sql_execution_engine

Documentation

Index

• Constants
• Variables
• func DebugDmlMsg(msgBatch []*core.Msg) (interface{}, interface{})
• func GenerateInsertIgnoreSQL(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

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 MySQLReplaceEngine = "mysql-replace-engine"

Variables

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

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 GenerateInsertIgnoreSQL

func GenerateInsertIgnoreSQL(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
}