dbconn

Documentation

Index

• func CloseBaseDB(logCtx *tcontext.Context, baseDB *conn.BaseDB)
• func CloseUpstreamConn(tctx *tcontext.Context, conn *UpStreamConn)
• func GetCharsetAndCollationInfo(tctx *tcontext.Context, conn *DBConn) (map[string]string, map[int]string, error)
• func GetParserForConn(tctx *tcontext.Context, dbConn *DBConn) (*parser.Parser, error)
• func GetSchemaCreateSQL(tctx *tcontext.Context, conn *DBConn, schemaID string) (sql string, err error)
• func GetTableCreateSQL(tctx *tcontext.Context, conn *DBConn, tableID string) (sql string, err error)
• func NewTCPConnWithIOCounter(conn *net.TCPConn, c *atomic.Uint64) net.Conn
• type DBConn
  • func CreateConns(tctx *tcontext.Context, cfg *config.SubTaskConfig, dbCfg conn.ScopedDBConfig, ...) (*conn.BaseDB, []*DBConn, error)
  • func NewDBConn(cfg *config.SubTaskConfig, baseConn *conn.BaseConn) *DBConn
  • func (conn *DBConn) ExecuteSQL(tctx *tcontext.Context, metricProxies *metrics.Proxies, queries []string, ...) (int, error)
  • func (conn *DBConn) ExecuteSQLAutoSplit(tctx *tcontext.Context, metricProxies *metrics.Proxies, queries []string, ...) error
  • func (conn *DBConn) ExecuteSQLWithIgnore(tctx *tcontext.Context, metricProxies *metrics.Proxies, ...) (int, error)
  • func (conn *DBConn) QuerySQL(tctx *tcontext.Context, metricProxies *metrics.Proxies, query string, ...) (*sql.Rows, error)
  • func (conn *DBConn) ResetConn(tctx *tcontext.Context) error
  • func (conn *DBConn) Scope() terror.ErrScope
• type TCPConnWithIOCounter
  • func (t *TCPConnWithIOCounter) Read(b []byte) (n int, err error)
  • func (t *TCPConnWithIOCounter) Write(b []byte) (n int, err error)
• type UpStreamConn
  • func (c *UpStreamConn) FetchAllDoTables(ctx context.Context, bw *filter.Filter) (map[string][]string, error)
  • func (c *UpStreamConn) GetMasterStatus(ctx *tcontext.Context, flavor string) (mysql.Position, mysql.GTIDSet, error)
  • func (c *UpStreamConn) GetParser(ctx context.Context) (*parser.Parser, error)
  • func (c *UpStreamConn) GetServerUUID(ctx context.Context, flavor string) (string, error)
  • func (c *UpStreamConn) GetServerUnixTS(ctx context.Context) (int64, error)
  • func (c *UpStreamConn) KillConn(ctx context.Context, connID uint32) error

Functions

func CloseBaseDB

func CloseBaseDB(logCtx *tcontext.Context, baseDB *conn.BaseDB)

CloseBaseDB closes baseDB to release all connection generated by this baseDB and this baseDB.

func CloseUpstreamConn

func CloseUpstreamConn(tctx *tcontext.Context, conn *UpStreamConn)

CloseUpstreamConn closes the UpStreamConn.

func GetCharsetAndCollationInfo

func GetCharsetAndCollationInfo(tctx *tcontext.Context, conn *DBConn) (map[string]string, map[int]string, error)

GetCharsetAndCollationInfo returns charset and collation info.

func GetParserForConn

func GetParserForConn(tctx *tcontext.Context, dbConn *DBConn) (*parser.Parser, error)

func GetSchemaCreateSQL

func GetSchemaCreateSQL(tctx *tcontext.Context, conn *DBConn, schemaID string) (sql string, err error)

GetSchemaCreateSQL gets table create sql by 'show create table schema.table'.

func GetTableCreateSQL

func GetTableCreateSQL(tctx *tcontext.Context, conn *DBConn, tableID string) (sql string, err error)

GetTableCreateSQL gets table create sql by 'show create table schema.table'.

func NewTCPConnWithIOCounter

func NewTCPConnWithIOCounter(conn *net.TCPConn, c *atomic.Uint64) net.Conn

NewTCPConnWithIOCounter creates a new TCPConnWithIOCounter.

Types

type DBConn

type DBConn struct {

	// generate new BaseConn and close old one
	ResetBaseConnFn func(*tcontext.Context, *conn.BaseConn) (*conn.BaseConn, error)
	// contains filtered or unexported fields
}

DBConn represents a live DB connection it's not thread-safe.

func CreateConns

func CreateConns(tctx *tcontext.Context, cfg *config.SubTaskConfig, dbCfg conn.ScopedDBConfig, count int, ioCounter *atomic.Uint64, uuid string) (*conn.BaseDB, []*DBConn, error)

CreateConns returns a opened DB from dbCfg and number of `count` connections of that DB.

func NewDBConn

func NewDBConn(cfg *config.SubTaskConfig, baseConn *conn.BaseConn) *DBConn

func (*DBConn) ExecuteSQL

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

ExecuteSQL does some SQL executions.

func (*DBConn) ExecuteSQLAutoSplit

func (conn *DBConn) ExecuteSQLAutoSplit(
	tctx *tcontext.Context,
	metricProxies *metrics.Proxies,
	queries []string,
	args ...[]interface{},
) error

ExecuteSQLAutoSplit wraps BaseConn.ExecuteSQLAutoSplit. TODO: refine DBConn and BaseConn.

func (*DBConn) ExecuteSQLWithIgnore

func (conn *DBConn) ExecuteSQLWithIgnore(
	tctx *tcontext.Context,
	metricProxies *metrics.Proxies,
	ignoreError func(error) bool,
	queries []string,
	args ...[]interface{},
) (int, error)

ExecuteSQLWithIgnore do some SQL executions and can ignore some error by `ignoreError`.

func (*DBConn) QuerySQL

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

QuerySQL does one query.

func (*DBConn) ResetConn

func (conn *DBConn) ResetConn(tctx *tcontext.Context) error

ResetConn reset one worker connection from specify *BaseDB.

func (*DBConn) Scope

func (conn *DBConn) Scope() terror.ErrScope

Scope return connection scope.

type TCPConnWithIOCounter

type TCPConnWithIOCounter struct {
	*net.TCPConn
	// contains filtered or unexported fields
}

TCPConnWithIOCounter is a wrapper of net.TCPConn with counter that accumulates the bytes this connection reads/writes.

func (*TCPConnWithIOCounter) Read

func (t *TCPConnWithIOCounter) Read(b []byte) (n int, err error)

func (*TCPConnWithIOCounter) Write

func (t *TCPConnWithIOCounter) Write(b []byte) (n int, err error)

type UpStreamConn

type UpStreamConn struct {
	BaseDB *conn.BaseDB
}

UpStreamConn connect to upstream DB Normally, we need to get some upstream information through some helper functions these helper functions are all easy query functions, so we use a pool of connections here maybe change to one connection some day.

func (*UpStreamConn) FetchAllDoTables

func (c *UpStreamConn) FetchAllDoTables(ctx context.Context, bw *filter.Filter) (map[string][]string, error)

FetchAllDoTables returns tables matches allow-list.

func (*UpStreamConn) GetMasterStatus

func (c *UpStreamConn) GetMasterStatus(ctx *tcontext.Context, flavor string) (mysql.Position, mysql.GTIDSet, error)

GetMasterStatus returns binlog location that extracted from SHOW MASTER STATUS.

func (*UpStreamConn) GetParser

func (c *UpStreamConn) GetParser(ctx context.Context) (*parser.Parser, error)

GetParser returns the parser with correct flag for upstream.

func (*UpStreamConn) GetServerUUID

func (c *UpStreamConn) GetServerUUID(ctx context.Context, flavor string) (string, error)

GetServerUUID returns upstream server UUID.

func (*UpStreamConn) GetServerUnixTS

func (c *UpStreamConn) GetServerUnixTS(ctx context.Context) (int64, error)

GetServerUnixTS returns the result of current timestamp in upstream.

func (*UpStreamConn) KillConn

func (c *UpStreamConn) KillConn(ctx context.Context, connID uint32) error

KillConn kills a connection in upstream.