loader

Documentation

Index

• func CollectDirFiles(path string) map[string]struct{}
• func RegisterMetrics(registry *prometheus.Registry)
• func SQLReplace(s, old, new string) string
• type CheckPoint
• type DBConn
• type DataFiles
• type FilePosSet
• type Loader
  • func NewLoader(cfg *config.SubTaskConfig) *Loader
  • func (l *Loader) Close()
  • func (l *Loader) Error() interface{}
  • func (l *Loader) Init(ctx context.Context) (err error)
  • func (l *Loader) IsFreshTask(ctx context.Context) (bool, error)
  • func (l *Loader) Pause()
  • func (l *Loader) PrintStatus(ctx context.Context)
  • func (l *Loader) Process(ctx context.Context, pr chan pb.ProcessResult)
  • func (l *Loader) Restore(ctx context.Context) error
  • func (l *Loader) Resume(ctx context.Context, pr chan pb.ProcessResult)
  • func (l *Loader) Status() interface{}
  • func (l *Loader) Type() pb.UnitType
  • func (l *Loader) Update(cfg *config.SubTaskConfig) error
• type RemoteCheckPoint
  • func (cp *RemoteCheckPoint) CalcProgress(allFiles map[string]Tables2DataFiles) error
  • func (cp *RemoteCheckPoint) Clear(tctx *tcontext.Context) error
  • func (cp *RemoteCheckPoint) Close()
  • func (cp *RemoteCheckPoint) Count(tctx *tcontext.Context) (int, error)
  • func (cp *RemoteCheckPoint) GenSQL(filename string, offset int64) string
  • func (cp *RemoteCheckPoint) GetAllRestoringFileInfo() map[string][]int64
  • func (cp *RemoteCheckPoint) GetRestoringFileInfo(db, table string) map[string][]int64
  • func (cp *RemoteCheckPoint) Init(tctx *tcontext.Context, filename string, endPos int64) error
  • func (cp *RemoteCheckPoint) IsTableFinished(db, table string) bool
  • func (cp *RemoteCheckPoint) Load(tctx *tcontext.Context) error
  • func (cp *RemoteCheckPoint) ResetConn(tctx *tcontext.Context) error
  • func (cp *RemoteCheckPoint) String() string
• type Tables2DataFiles
• type Worker
  • func NewWorker(loader *Loader, id int) (worker *Worker, err error)
  • func (w *Worker) Close()

Functions

func CollectDirFiles

func CollectDirFiles(path string) map[string]struct{}

CollectDirFiles gets files in path

func RegisterMetrics

func RegisterMetrics(registry *prometheus.Registry)

RegisterMetrics registers metrics

func SQLReplace

func SQLReplace(s, old, new string) string

SQLReplace works like strings.Replace but only supports one replacement. It uses backquote pairs to quote the old and new word.

Types

type CheckPoint

type CheckPoint interface {
	// Load loads all checkpoints recorded before.
	// because of no checkpoints updated in memory when error occurred
	// when resuming, Load will be called again to load checkpoints
	Load(tctx *tcontext.Context) error

	// GetRestoringFileInfo get restoring data files for table
	GetRestoringFileInfo(db, table string) map[string][]int64

	// GetAllRestoringFileInfo return all restoring files position
	GetAllRestoringFileInfo() map[string][]int64

	// IsTableFinished query if table has finished
	IsTableFinished(db, table string) bool

	// CalcProgress calculate which table has finished and which table partial restored
	CalcProgress(allFiles map[string]Tables2DataFiles) error

	// Init initialize checkpoint data in tidb
	Init(tctx *tcontext.Context, filename string, endpos int64) error

	// ResetConn resets database connections owned by the Checkpoint
	ResetConn(tctx *tcontext.Context) error

	// Close closes the CheckPoint
	Close()

	// Clear clears all recorded checkpoints
	Clear(tctx *tcontext.Context) error

	// Count returns recorded checkpoints' count
	Count(tctx *tcontext.Context) (int, error)

	// GenSQL generates sql to update checkpoint to DB
	GenSQL(filename string, offset int64) string
}

CheckPoint represents checkpoint status

type DBConn

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

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

type DataFiles

type DataFiles []string

DataFiles represent all data files for a single table

type FilePosSet

type FilePosSet map[string][]int64

FilePosSet represents a set in mathematics.

type Loader

type Loader struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Loader can load your mydumper data into TiDB database.

func NewLoader

func NewLoader(cfg *config.SubTaskConfig) *Loader

NewLoader creates a new Loader.

func (*Loader) Close

func (l *Loader) Close()

Close does graceful shutdown

func (*Loader) Error

func (l *Loader) Error() interface{}

Error implements SubTaskUnit.Error

func (*Loader) Init

func (l *Loader) Init(ctx context.Context) (err error)

Init initializes loader for a load task, but not start Process. if fail, it should not call l.Close.

func (*Loader) IsFreshTask

func (l *Loader) IsFreshTask(ctx context.Context) (bool, error)

IsFreshTask implements Unit.IsFreshTask

func (*Loader) Pause

func (l *Loader) Pause()

Pause pauses the process, and it can be resumed later should cancel context from external

func (*Loader) PrintStatus

func (l *Loader) PrintStatus(ctx context.Context)

PrintStatus prints status like progress percentage.

func (*Loader) Process

func (l *Loader) Process(ctx context.Context, pr chan pb.ProcessResult)

Process implements Unit.Process

func (*Loader) Restore

func (l *Loader) Restore(ctx context.Context) error

Restore begins the restore process.

func (*Loader) Resume

func (l *Loader) Resume(ctx context.Context, pr chan pb.ProcessResult)

Resume resumes the paused process

func (*Loader) Status

func (l *Loader) Status() interface{}

Status implements SubTaskUnit.Status

func (*Loader) Type

func (l *Loader) Type() pb.UnitType

Type implements Unit.Type

func (*Loader) Update

func (l *Loader) Update(cfg *config.SubTaskConfig) error

Update implements Unit.Update now, only support to update config for routes, filters, column-mappings, black-white-list now no config diff implemented, so simply re-init use new config no binlog filter for loader need to update

type RemoteCheckPoint

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

RemoteCheckPoint implements CheckPoint by saving status in remote database system, mostly in TiDB. it's not thread-safe

func (*RemoteCheckPoint) CalcProgress

func (cp *RemoteCheckPoint) CalcProgress(allFiles map[string]Tables2DataFiles) error

CalcProgress implements CheckPoint.CalcProgress

func (*RemoteCheckPoint) Clear

func (cp *RemoteCheckPoint) Clear(tctx *tcontext.Context) error

Clear implements CheckPoint.Clear

func (*RemoteCheckPoint) Close

func (cp *RemoteCheckPoint) Close()

Close implements CheckPoint.Close

func (*RemoteCheckPoint) Count

func (cp *RemoteCheckPoint) Count(tctx *tcontext.Context) (int, error)

Count implements CheckPoint.Count

func (*RemoteCheckPoint) GenSQL

func (cp *RemoteCheckPoint) GenSQL(filename string, offset int64) string

GenSQL implements CheckPoint.GenSQL

func (*RemoteCheckPoint) GetAllRestoringFileInfo

func (cp *RemoteCheckPoint) GetAllRestoringFileInfo() map[string][]int64

GetAllRestoringFileInfo implements CheckPoint.GetAllRestoringFileInfo

func (*RemoteCheckPoint) GetRestoringFileInfo

func (cp *RemoteCheckPoint) GetRestoringFileInfo(db, table string) map[string][]int64

GetRestoringFileInfo implements CheckPoint.GetRestoringFileInfo

func (*RemoteCheckPoint) Init

func (cp *RemoteCheckPoint) Init(tctx *tcontext.Context, filename string, endPos int64) error

Init implements CheckPoint.Init

func (*RemoteCheckPoint) IsTableFinished

func (cp *RemoteCheckPoint) IsTableFinished(db, table string) bool

IsTableFinished implements CheckPoint.IsTableFinished

func (*RemoteCheckPoint) Load

func (cp *RemoteCheckPoint) Load(tctx *tcontext.Context) error

Load implements CheckPoint.Load

func (*RemoteCheckPoint) ResetConn

func (cp *RemoteCheckPoint) ResetConn(tctx *tcontext.Context) error

ResetConn implements CheckPoint.ResetConn

func (*RemoteCheckPoint) String

func (cp *RemoteCheckPoint) String() string

type Tables2DataFiles

type Tables2DataFiles map[string]DataFiles

Tables2DataFiles represent all data files of a table collection as a map

type Worker

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

Worker represents a worker.

func NewWorker

func NewWorker(loader *Loader, id int) (worker *Worker, err error)

NewWorker returns a Worker.

func (*Worker) Close

func (w *Worker) Close()

Close closes worker