sql_execution_engine

package

v0.9.71 Latest Latest Go to latest Published: Jun 29, 2020 License: Apache-2.0 Imports: 17 Imported by: 1

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/moiot/gravity

Links

Open Source Insights

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 ¶

View Source

const (
	ConflictDetectEngine = "conflict-detect-engine"

	ConflictFileName = "conflict.log"
)

View Source

const (
	QueryExist queryResult = iota
	QueryNotExist
	QueryUnknown
)

View Source

const ManualEngine = "manual-engine"

View Source

const MySQLInsertIgnore = "mysql-insert-ignore"

View Source

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

View Source

const MySQLReplaceEngine = "mysql-replace-engine"

Variables ¶

View Source

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

View Source

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

View Source

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

View Source

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

Functions ¶

func DebugDmlMsg ¶

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

func ExecWithInternalTxnTag ¶ added in v0.9.40

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 ¶ added in v0.9.39

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 ¶ added in v0.9.40

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

type MysqlReplaceEngineConfig ¶ added in v0.9.16

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

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL