conn

type BaseConn ¶

type BaseConn struct {
	DBConn *sql.Conn

	RetryStrategy retry.Strategy
}

BaseConn is the basic connection we use in dm BaseDB -> BaseConn correspond to sql.DB -> sql.Conn In our scenario, there are two main reasons why we need BaseConn

1. we often need one fixed DB connection to execute sql
2. we need own retry policy during execute failed

So we split a fixed sql.Conn out of sql.DB, and wraps it to BaseConn And Similar with sql.Conn, all BaseConn generated from one BaseDB shares this BaseDB to reset

Basic usage: For Syncer and Loader Unit, they both have different amount of connections due to config Currently we have some types of connections exist

Syncer:
	Worker Connection:
		DML connection:
			execute some DML on Downstream DB, one unit has `syncer.WorkerCount` worker connections
		DDL Connection:
			execute some DDL on Downstream DB, one unit has one connection
	CheckPoint Connection:
		interact with CheckPoint DB, one unit has one connection
	OnlineDDL connection:
		interact with Online DDL DB, one unit has one connection
	ShardGroupKeeper connection:
		interact with ShardGroupKeeper DB, one unit has one connection

Loader:
	Worker Connection:
		execute some DML to Downstream DB, one unit has `loader.PoolSize` worker connections
	CheckPoint Connection:
		interact with CheckPoint DB, one unit has one connection
	Restore Connection:
		only use to create schema and table in restoreData,
		it ignore already exists error and it should be removed after use, one unit has one connection

each connection should have ability to retry on some common errors (e.g. tmysql.ErrTiKVServerTimeout) or maybe some specify errors in the future and each connection also should have ability to reset itself during some specify connection error (e.g. driver.ErrBadConn)

func NewBaseConn(conn *sql.Conn, strategy retry.Strategy) *BaseConn

func (conn *BaseConn) ApplyRetryStrategy(tctx *tcontext.Context, params retry.Params,
	operateFn func(*tcontext.Context) (interface{}, error)) (interface{}, int, error)

func (conn *BaseConn) ExecuteSQL(tctx *tcontext.Context, hVec *metricsproxy.HistogramVecProxy, task string, queries []string, args ...[]interface{}) (int, error)

func (conn *BaseConn) ExecuteSQLWithIgnoreError(tctx *tcontext.Context, hVec *metricsproxy.HistogramVecProxy, task string, ignoreErr func(error) bool, queries []string, args ...[]interface{}) (int, error)

func (conn *BaseConn) QuerySQL(tctx *tcontext.Context, query string, args ...interface{}) (*sql.Rows, error)

func (conn *BaseConn) SetRetryStrategy(strategy retry.Strategy) error

type BaseDB ¶

type BaseDB struct {
	DB *sql.DB

	Retry retry.Strategy
	// contains filtered or unexported fields
}

BaseDB wraps *sql.DB, control the BaseConn

func NewBaseDB(db *sql.DB) *BaseDB

func (d *BaseDB) Close() error

func (d *BaseDB) CloseBaseConn(conn *BaseConn) error

func (d *BaseDB) GetBaseConn(ctx context.Context) (*BaseConn, error)

type DBProvider ¶

type DBProvider interface {
	Apply(config config.DBConfig) (*BaseDB, error)
}

DBProvider providers BaseDB instance

var DefaultDBProvider DBProvider