scan

Documentation

Index

• Constants
• func InitTableState(state *TableState, table *core.Table, conn *sql.DB) (err error)
• type MysqlScanInput
  • func NewMysqlScanInput() *MysqlScanInput
  • func (in *MysqlScanInput) Ack(msg *core.Message, err error)
  • func (in *MysqlScanInput) Configure(config core.StringMap) (err error)
  • func (in *MysqlScanInput) GetState() ([]byte, bool)
  • func (in *MysqlScanInput) SetState(state []byte) (err error)
  • func (in *MysqlScanInput) Start() (err error)
  • func (in *MysqlScanInput) Stop()
• type MysqlScanInputConfig
• type TableScanner
  • func NewTableScanner(id int, input *MysqlScanInput, tables chan *core.Table) *TableScanner
• type TableState

Constants

const (
	DefaultBatchSize   = 1000
	DefaultConcurrency = 8
)

Functions

func InitTableState

func InitTableState(state *TableState, table *core.Table, conn *sql.DB) (err error)

Types

type MysqlScanInput

type MysqlScanInput struct {
	*core.BaseInput
	Config *MysqlScanInputConfig
	// contains filtered or unexported fields
}

func NewMysqlScanInput

func NewMysqlScanInput() *MysqlScanInput

func (*MysqlScanInput) Ack

func (in *MysqlScanInput) Ack(msg *core.Message, err error)

func (*MysqlScanInput) Configure

func (in *MysqlScanInput) Configure(config core.StringMap) (err error)

func (*MysqlScanInput) GetState

func (in *MysqlScanInput) GetState() ([]byte, bool)

func (*MysqlScanInput) SetState

func (in *MysqlScanInput) SetState(state []byte) (err error)

func (*MysqlScanInput) Start

func (in *MysqlScanInput) Start() (err error)

func (*MysqlScanInput) Stop

func (in *MysqlScanInput) Stop()

type MysqlScanInputConfig

type MysqlScanInputConfig struct {
	Host           string
	Port           uint16
	User           string
	Password       string
	TableNameRegex string // select tables need to be replicated by regex
	BatchSize      int
	Concurrency    int
}

type TableScanner

type TableScanner struct {
	// contains filtered or unexported fields
}

func NewTableScanner

func NewTableScanner(id int, input *MysqlScanInput, tables chan *core.Table) *TableScanner

type TableState

type TableState struct {
	TableName         string
	DBName            string
	ColumnStatesValue atomic.Value  `json:"-"`
	ColumnStates      []interface{} // for json marshall and unmarshall
	EstimatedCount    int64
	FinishedCount     int64
	Done              bool
}