conn

Documentation

Index

• type BaseConn
  • func NewBaseConn(conn *sql.Conn, strategy retry.Strategy) *BaseConn
  • func (conn *BaseConn) ApplyRetryStrategy(tctx *tcontext.Context, params retry.Params, ...) (interface{}, int, error)
  • func (conn *BaseConn) ExecuteSQL(tctx *tcontext.Context, queries []string, args ...[]interface{}) (int, error)
  • func (conn *BaseConn) ExecuteSQLWithIgnoreError(tctx *tcontext.Context, ignoreErr func(error) bool, queries []string, ...) (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
  • 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

Types

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

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

NewBaseConn builds BaseConn to connect real DB

func (*BaseConn) ApplyRetryStrategy

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

ApplyRetryStrategy apply specify strategy for BaseConn

func (*BaseConn) ExecuteSQL

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

ExecuteSQL executes sql on real DB, return 1. failed: (the index of sqls executed error, error) 2. succeed: (len(sqls), nil)

func (*BaseConn) ExecuteSQLWithIgnoreError

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

ExecuteSQLWithIgnoreError executes sql on real DB, and will ignore some error and continue execute the next query. return 1. failed: (the index of sqls executed error, error) 2. succeed: (len(sqls), nil)

func (*BaseConn) QuerySQL

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

QuerySQL defines query statement, and connect to real DB

func (*BaseConn) SetRetryStrategy

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

SetRetryStrategy set retry strategy for baseConn

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

func NewBaseDB(db *sql.DB) *BaseDB

NewBaseDB returns *BaseDB object

func (*BaseDB) Close

func (d *BaseDB) Close() error

Close release *BaseDB resource

func (*BaseDB) CloseBaseConn

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

CloseBaseConn release BaseConn resource from BaseDB, and close BaseConn

func (*BaseDB) GetBaseConn

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

GetBaseConn retrieves *BaseConn which has own retryStrategy

type DBProvider

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

DBProvider providers BaseDB instance

var DefaultDBProvider DBProvider

DefaultDBProvider is global instance of DBProvider