sql_execution_engine

Documentation

Index

• Constants
• Variables
• func DebugDmlMsg(msgBatch []*core.Msg) (interface{}, interface{})
• func ExecWithInternalTxnTag(pipelineName string, internalTxnTaggerCfg *InternalTxnTaggerCfg, db *sql.DB, ...) error
• func GenerateInsertIgnoreSQL(msgBatch []*core.Msg, tableDef *schema_store.Table) (string, []interface{}, error)
• func GenerateInsertOnDuplicateKeyUpdate(msgBatch []*core.Msg, tableDef *schema_store.Table) (string, []interface{}, error)
• func GenerateReplaceSQLWithMultipleValues(msgBatch []*core.Msg, tableDef *schema_store.Table) (string, []interface{}, error)
• func GenerateSingleDeleteSQL(msg *core.Msg, tableDef *schema_store.Table) (string, []interface{}, error)
• func GetSingleSqlPlaceHolderAndArgWithEncodedData(msg *core.Msg, tableDef *schema_store.Table) (string, []interface{}, error)
• func PlaceHoldersAndArgsFromEncodedData(msgBatch []*core.Msg, tableDef *schema_store.Table) ([]string, []interface{}, error)
• func SQLWithAnnotation(sql string, annotationContent string) string
• func ValidateSchema(msg *core.Msg, tableDef *schema_store.Table) error
• type EngineExecutor
  • func NewConflictPreviewEngine(db *sql.DB, maxRetryCount int, sleepDuration time.Duration, enableDelete bool) EngineExecutor
  • func NewEngineExecutor(pipelineName string, engineName string, db *sql.DB, ...) EngineExecutor
• type EngineInitializer
• type InternalTxnTaggerCfg
• type MysqlReplaceEngineConfig

Constants

const (
	ConflictDetectEngine = "conflict-detect-engine"

	ConflictFileName = "conflict.log"
)

const (
	QueryExist queryResult = iota
	QueryNotExist
	QueryUnknown
)

const ManualEngine = "manual-engine"

const MySQLInsertIgnore = "mysql-insert-ignore"

const MySQLInsertOnDuplicateKeyUpdate = "mysql-insert-on-duplicate-key-update"

const MySQLReplaceEngine = "mysql-replace-engine"

Variables

var DefaultInternalTxnTaggerCfg = map[string]interface{}{
	"tag-internal-txn": false,
	"sql-annotation":   "",
}

var DefaultMySQLReplaceEngineConfig = map[string]interface{}{
	MySQLReplaceEngine: DefaultInternalTxnTaggerCfg,
}

var ErrDeleteRowSkip = errors.New("sql_execution_engine: skip this delete event")

var ErrRowConflict = errors.New("sql_execution_engine: this row conflicts")

Functions

func DebugDmlMsg

func DebugDmlMsg(msgBatch []*core.Msg) (interface{}, interface{})

func ExecWithInternalTxnTag

func ExecWithInternalTxnTag(
	pipelineName string,
	internalTxnTaggerCfg *InternalTxnTaggerCfg,
	db *sql.DB,
	query string,
	args []interface{}) error

func GenerateInsertIgnoreSQL

func GenerateInsertIgnoreSQL(msgBatch []*core.Msg, tableDef *schema_store.Table) (string, []interface{}, error)

func GenerateInsertOnDuplicateKeyUpdate

func GenerateInsertOnDuplicateKeyUpdate(msgBatch []*core.Msg, tableDef *schema_store.Table) (string, []interface{}, error)

func GenerateReplaceSQLWithMultipleValues

func GenerateReplaceSQLWithMultipleValues(msgBatch []*core.Msg, tableDef *schema_store.Table) (string, []interface{}, error)

func GenerateSingleDeleteSQL

func GenerateSingleDeleteSQL(msg *core.Msg, tableDef *schema_store.Table) (string, []interface{}, error)

func GetSingleSqlPlaceHolderAndArgWithEncodedData

func GetSingleSqlPlaceHolderAndArgWithEncodedData(msg *core.Msg, tableDef *schema_store.Table) (string, []interface{}, error)

func PlaceHoldersAndArgsFromEncodedData

func PlaceHoldersAndArgsFromEncodedData(msgBatch []*core.Msg, tableDef *schema_store.Table) ([]string, []interface{}, error)

func SQLWithAnnotation

func SQLWithAnnotation(sql string, annotationContent string) string

func ValidateSchema

func ValidateSchema(msg *core.Msg, tableDef *schema_store.Table) error

Types

type EngineExecutor

type EngineExecutor interface {
	Execute(msgBatch []*core.Msg, tableDef *schema_store.Table) error
}

func NewConflictPreviewEngine

func NewConflictPreviewEngine(db *sql.DB, maxRetryCount int, sleepDuration time.Duration, enableDelete bool) EngineExecutor

func NewEngineExecutor

func NewEngineExecutor(pipelineName string, engineName string, db *sql.DB, data map[string]interface{}) EngineExecutor

type EngineInitializer

type EngineInitializer interface {
	Init(db *sql.DB) error
}

type InternalTxnTaggerCfg

type InternalTxnTaggerCfg struct {
	TagInternalTxn bool   `mapstructure:"tag-internal-txn" json:"tag-internal-txn"`
	SQLAnnotation  string `mapstructure:"sql-annotation" json:"sql-annotation"`
}

type MysqlReplaceEngineConfig

type MysqlReplaceEngineConfig struct {
	InternalTxnTaggerCfg `mapstructure:",squash"`
}