checkpoints

package
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 23, 2023 License: Apache-2.0, Apache-2.0 Imports: 28 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// the table names to store each kind of checkpoint in the checkpoint database
	// remember to increase the version number in case of incompatible change.
	CheckpointTableNameTask   = "task_v2"
	CheckpointTableNameTable  = "table_v7"
	CheckpointTableNameEngine = "engine_v5"
	CheckpointTableNameChunk  = "chunk_v5"
)
View Source 
const (
	// shared by MySQLCheckpointsDB and GlueCheckpointsDB
	CreateDBTemplate        = "CREATE DATABASE IF NOT EXISTS %s;"
	CreateTaskTableTemplate = `` /* 376-byte string literal not displayed */

	CreateTableTableTemplate = `` /* 580-byte string literal not displayed */

	CreateEngineTableTemplate = `` /* 338-byte string literal not displayed */

	CreateChunkTableTemplate = `` /* 827-byte string literal not displayed */

	InitTaskTemplate = `` /* 174-byte string literal not displayed */

	InitTableTemplate = `` /* 182-byte string literal not displayed */

	ReadTaskTemplate = `` /* 139-byte string literal not displayed */

	ReadEngineTemplate = `
		SELECT engine_id, status FROM %s.%s WHERE table_name = ? ORDER BY engine_id DESC;`
	ReadChunkTemplate = `` /* 268-byte string literal not displayed */

	ReadTableRemainTemplate = `
		SELECT status, alloc_base, table_id, kv_bytes, kv_kvs, kv_checksum FROM %s.%s WHERE table_name = ?;`
	ReplaceEngineTemplate = `
		REPLACE INTO %s.%s (table_name, engine_id, status) VALUES (?, ?, ?);`
	ReplaceChunkTemplate = `` /* 346-byte string literal not displayed */

	UpdateChunkTemplate = `` /* 168-byte string literal not displayed */

	UpdateTableRebaseTemplate = `
		UPDATE %s.%s SET alloc_base = GREATEST(?, alloc_base) WHERE table_name = ?;`
	UpdateTableStatusTemplate = `
		UPDATE %s.%s SET status = ? WHERE table_name = ?;`
	UpdateTableChecksumTemplate = `UPDATE %s.%s SET kv_bytes = ?, kv_kvs = ?, kv_checksum = ? WHERE table_name = ?;`
	UpdateEngineTemplate        = `
		UPDATE %s.%s SET status = ? WHERE (table_name, engine_id) = (?, ?);`
	DeleteCheckpointRecordTemplate = "DELETE FROM %s.%s WHERE table_name = ?;"
)
View Source 
const WholeTableEngineID = math.MaxInt32

Variables

This section is empty.

Functions

func IsCheckpointTable 

func IsCheckpointTable(name string) bool

func IsCheckpointsDBExists 

func IsCheckpointsDBExists(ctx context.Context, cfg *config.Config) (bool, error)

func Transact 

func Transact(ctx context.Context, purpose string, s Session, logger log.Logger, action func(context.Context, Session) error) error

Types

type CheckpointStatus 

type CheckpointStatus uint8
const (
	CheckpointStatusMissing         CheckpointStatus = 0
	CheckpointStatusMaxInvalid      CheckpointStatus = 25
	CheckpointStatusLoaded          CheckpointStatus = 30
	CheckpointStatusAllWritten      CheckpointStatus = 60
	CheckpointStatusClosed          CheckpointStatus = 90
	CheckpointStatusImported        CheckpointStatus = 120
	CheckpointStatusIndexImported   CheckpointStatus = 140
	CheckpointStatusAlteredAutoInc  CheckpointStatus = 150
	CheckpointStatusChecksumSkipped CheckpointStatus = 170
	CheckpointStatusChecksummed     CheckpointStatus = 180
	CheckpointStatusAnalyzeSkipped  CheckpointStatus = 200
	CheckpointStatusAnalyzed        CheckpointStatus = 210
)

func (CheckpointStatus) MetricName 

func (status CheckpointStatus) MetricName() string

type ChunkCheckpoint 

type ChunkCheckpoint struct {
	Key               ChunkCheckpointKey
	FileMeta          mydump.SourceFileMeta
	ColumnPermutation []int
	Chunk             mydump.Chunk
	Checksum          verify.KVChecksum
	Timestamp         int64
}

func (*ChunkCheckpoint) DeepCopy 

func (ccp *ChunkCheckpoint) DeepCopy() *ChunkCheckpoint

func (*ChunkCheckpoint) FinishedSize 

func (ccp *ChunkCheckpoint) FinishedSize() int64

func (*ChunkCheckpoint) TotalSize 

func (ccp *ChunkCheckpoint) TotalSize() int64

func (*ChunkCheckpoint) UnfinishedSize 

func (ccp *ChunkCheckpoint) UnfinishedSize() int64

type ChunkCheckpointKey 

type ChunkCheckpointKey struct {
	Path   string
	Offset int64
}

func (*ChunkCheckpointKey) String 

func (key *ChunkCheckpointKey) String() string

type ChunkCheckpointMerger 

type ChunkCheckpointMerger struct {
	EngineID          int32
	Key               ChunkCheckpointKey
	Checksum          verify.KVChecksum
	Pos               int64
	RowID             int64
	ColumnPermutation []int
	EndOffset         int64 // For test only.
}

func (*ChunkCheckpointMerger) MergeInto 

func (merger *ChunkCheckpointMerger) MergeInto(cpd *TableCheckpointDiff)

type DB 

type DB interface {
	Initialize(ctx context.Context, cfg *config.Config, dbInfo map[string]*TidbDBInfo) error
	TaskCheckpoint(ctx context.Context) (*TaskCheckpoint, error)
	Get(ctx context.Context, tableName string) (*TableCheckpoint, error)
	Close() error
	// InsertEngineCheckpoints initializes the checkpoints related to a table.
	// It assumes the entire table has not been imported before and will fill in
	// default values for the column permutations and checksums.
	InsertEngineCheckpoints(ctx context.Context, tableName string, checkpoints map[int32]*EngineCheckpoint) error
	Update(taskCtx context.Context, checkpointDiffs map[string]*TableCheckpointDiff) error

	RemoveCheckpoint(ctx context.Context, tableName string) error
	// MoveCheckpoints renames the checkpoint schema to include a suffix
	// including the taskID (e.g. `tidb_lightning_checkpoints.1234567890.bak`).
	MoveCheckpoints(ctx context.Context, taskID int64) error
	// GetLocalStoringTables returns a map containing tables have engine files stored on local disk.
	// currently only meaningful for local backend
	GetLocalStoringTables(ctx context.Context) (map[string][]int32, error)
	IgnoreErrorCheckpoint(ctx context.Context, tableName string) error
	DestroyErrorCheckpoint(ctx context.Context, tableName string) ([]DestroyedTableCheckpoint, error)
	DumpTables(ctx context.Context, csv io.Writer) error
	DumpEngines(ctx context.Context, csv io.Writer) error
	DumpChunks(ctx context.Context, csv io.Writer) error
}

func OpenCheckpointsDB 

func OpenCheckpointsDB(ctx context.Context, cfg *config.Config) (DB, error)

type DestroyedTableCheckpoint 

type DestroyedTableCheckpoint struct {
	TableName   string
	MinEngineID int32
	MaxEngineID int32
}

type EngineCheckpoint 

type EngineCheckpoint struct {
	Status CheckpointStatus
	Chunks []*ChunkCheckpoint // a sorted array
}

func (*EngineCheckpoint) DeepCopy 

func (engine *EngineCheckpoint) DeepCopy() *EngineCheckpoint

type FileCheckpointsDB 

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

func NewFileCheckpointsDB 

func NewFileCheckpointsDB(ctx context.Context, path string) (*FileCheckpointsDB, error)

func NewFileCheckpointsDBWithExstorageFileName 

func NewFileCheckpointsDBWithExstorageFileName(
	ctx context.Context,
	path string,
	s storage.ExternalStorage,
	fileName string,
) (*FileCheckpointsDB, error)

func (*FileCheckpointsDB) Close 

func (cpdb *FileCheckpointsDB) Close() error

func (*FileCheckpointsDB) DestroyErrorCheckpoint 

func (cpdb *FileCheckpointsDB) DestroyErrorCheckpoint(_ context.Context, targetTableName string) ([]DestroyedTableCheckpoint, error)

func (*FileCheckpointsDB) DumpChunks 

func (cpdb *FileCheckpointsDB) DumpChunks(context.Context, io.Writer) error

func (*FileCheckpointsDB) DumpEngines 

func (cpdb *FileCheckpointsDB) DumpEngines(context.Context, io.Writer) error

func (*FileCheckpointsDB) DumpTables 

func (cpdb *FileCheckpointsDB) DumpTables(context.Context, io.Writer) error

func (*FileCheckpointsDB) Get 

func (cpdb *FileCheckpointsDB) Get(_ context.Context, tableName string) (*TableCheckpoint, error)

func (*FileCheckpointsDB) GetLocalStoringTables 

func (cpdb *FileCheckpointsDB) GetLocalStoringTables(_ context.Context) (map[string][]int32, error)

func (*FileCheckpointsDB) IgnoreErrorCheckpoint 

func (cpdb *FileCheckpointsDB) IgnoreErrorCheckpoint(_ context.Context, targetTableName string) error

func (*FileCheckpointsDB) Initialize 

func (cpdb *FileCheckpointsDB) Initialize(ctx context.Context, cfg *config.Config, dbInfo map[string]*TidbDBInfo) error

func (*FileCheckpointsDB) InsertEngineCheckpoints 

func (cpdb *FileCheckpointsDB) InsertEngineCheckpoints(_ context.Context, tableName string, checkpoints map[int32]*EngineCheckpoint) error

func (*FileCheckpointsDB) MoveCheckpoints 

func (cpdb *FileCheckpointsDB) MoveCheckpoints(ctx context.Context, taskID int64) error

func (*FileCheckpointsDB) RemoveCheckpoint 

func (cpdb *FileCheckpointsDB) RemoveCheckpoint(_ context.Context, tableName string) error

func (*FileCheckpointsDB) TaskCheckpoint 

func (cpdb *FileCheckpointsDB) TaskCheckpoint(_ context.Context) (*TaskCheckpoint, error)

func (*FileCheckpointsDB) Update 

func (cpdb *FileCheckpointsDB) Update(_ context.Context, checkpointDiffs map[string]*TableCheckpointDiff) error

type GlueCheckpointsDB 

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

GlueCheckpointsDB is almost same with MySQLCheckpointsDB, but it uses TiDB's internal data structure which requires a lot to keep same with database/sql. TODO: Encapsulate Begin/Commit/Rollback txn, form SQL with args and query/iter/scan TiDB's RecordSet into a interface to reuse MySQLCheckpointsDB.

func NewGlueCheckpointsDB 

func NewGlueCheckpointsDB(ctx context.Context, se Session, f func() (Session, error), schemaName string) (*GlueCheckpointsDB, error)

func (GlueCheckpointsDB) Close 

func (g GlueCheckpointsDB) Close() error

func (GlueCheckpointsDB) DestroyErrorCheckpoint 

func (g GlueCheckpointsDB) DestroyErrorCheckpoint(ctx context.Context, tableName string) ([]DestroyedTableCheckpoint, error)

func (GlueCheckpointsDB) DumpChunks 

func (g GlueCheckpointsDB) DumpChunks(ctx context.Context, csv io.Writer) error

func (GlueCheckpointsDB) DumpEngines 

func (g GlueCheckpointsDB) DumpEngines(ctx context.Context, csv io.Writer) error

func (GlueCheckpointsDB) DumpTables 

func (g GlueCheckpointsDB) DumpTables(ctx context.Context, csv io.Writer) error

func (GlueCheckpointsDB) Get 

func (g GlueCheckpointsDB) Get(ctx context.Context, tableName string) (*TableCheckpoint, error)

func (GlueCheckpointsDB) GetLocalStoringTables 

func (g GlueCheckpointsDB) GetLocalStoringTables(ctx context.Context) (map[string][]int32, error)

func (GlueCheckpointsDB) IgnoreErrorCheckpoint 

func (g GlueCheckpointsDB) IgnoreErrorCheckpoint(ctx context.Context, tableName string) error

func (GlueCheckpointsDB) Initialize 

func (g GlueCheckpointsDB) Initialize(ctx context.Context, cfg *config.Config, dbInfo map[string]*TidbDBInfo) error

func (GlueCheckpointsDB) InsertEngineCheckpoints 

func (g GlueCheckpointsDB) InsertEngineCheckpoints(ctx context.Context, tableName string, checkpointMap map[int32]*EngineCheckpoint) error

func (GlueCheckpointsDB) MoveCheckpoints 

func (g GlueCheckpointsDB) MoveCheckpoints(ctx context.Context, taskID int64) error

func (GlueCheckpointsDB) RemoveCheckpoint 

func (g GlueCheckpointsDB) RemoveCheckpoint(ctx context.Context, tableName string) error

func (GlueCheckpointsDB) TaskCheckpoint 

func (g GlueCheckpointsDB) TaskCheckpoint(ctx context.Context) (*TaskCheckpoint, error)

func (GlueCheckpointsDB) Update 

func (g GlueCheckpointsDB) Update(ctx context.Context, checkpointDiffs map[string]*TableCheckpointDiff) error

type MySQLCheckpointsDB 

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

func NewMySQLCheckpointsDB 

func NewMySQLCheckpointsDB(ctx context.Context, db *sql.DB, schemaName string) (*MySQLCheckpointsDB, error)

func (*MySQLCheckpointsDB) Close 

func (cpdb *MySQLCheckpointsDB) Close() error

func (*MySQLCheckpointsDB) DestroyErrorCheckpoint 

func (cpdb *MySQLCheckpointsDB) DestroyErrorCheckpoint(ctx context.Context, tableName string) ([]DestroyedTableCheckpoint, error)

func (*MySQLCheckpointsDB) DumpChunks 

func (cpdb *MySQLCheckpointsDB) DumpChunks(ctx context.Context, writer io.Writer) error

func (*MySQLCheckpointsDB) DumpEngines 

func (cpdb *MySQLCheckpointsDB) DumpEngines(ctx context.Context, writer io.Writer) error

func (*MySQLCheckpointsDB) DumpTables 

func (cpdb *MySQLCheckpointsDB) DumpTables(ctx context.Context, writer io.Writer) error

func (*MySQLCheckpointsDB) Get 

func (cpdb *MySQLCheckpointsDB) Get(ctx context.Context, tableName string) (*TableCheckpoint, error)

func (*MySQLCheckpointsDB) GetLocalStoringTables 

func (cpdb *MySQLCheckpointsDB) GetLocalStoringTables(ctx context.Context) (map[string][]int32, error)

func (*MySQLCheckpointsDB) IgnoreErrorCheckpoint 

func (cpdb *MySQLCheckpointsDB) IgnoreErrorCheckpoint(ctx context.Context, tableName string) error

func (*MySQLCheckpointsDB) Initialize 

func (cpdb *MySQLCheckpointsDB) Initialize(ctx context.Context, cfg *config.Config, dbInfo map[string]*TidbDBInfo) error

func (*MySQLCheckpointsDB) InsertEngineCheckpoints 

func (cpdb *MySQLCheckpointsDB) InsertEngineCheckpoints(ctx context.Context, tableName string, checkpoints map[int32]*EngineCheckpoint) error

func (*MySQLCheckpointsDB) MoveCheckpoints 

func (cpdb *MySQLCheckpointsDB) MoveCheckpoints(ctx context.Context, taskID int64) error

func (*MySQLCheckpointsDB) RemoveCheckpoint 

func (cpdb *MySQLCheckpointsDB) RemoveCheckpoint(ctx context.Context, tableName string) error

func (*MySQLCheckpointsDB) TaskCheckpoint 

func (cpdb *MySQLCheckpointsDB) TaskCheckpoint(ctx context.Context) (*TaskCheckpoint, error)

func (*MySQLCheckpointsDB) Update 

func (cpdb *MySQLCheckpointsDB) Update(taskCtx context.Context, checkpointDiffs map[string]*TableCheckpointDiff) error

type NullCheckpointsDB 

type NullCheckpointsDB struct{}

NullCheckpointsDB is a checkpoints database with no checkpoints.

func NewNullCheckpointsDB 

func NewNullCheckpointsDB() *NullCheckpointsDB

func (*NullCheckpointsDB) Close 

func (*NullCheckpointsDB) Close() error

func (*NullCheckpointsDB) DestroyErrorCheckpoint 

func (*NullCheckpointsDB) DestroyErrorCheckpoint(context.Context, string) ([]DestroyedTableCheckpoint, error)

func (*NullCheckpointsDB) DumpChunks 

func (*NullCheckpointsDB) DumpChunks(context.Context, io.Writer) error

func (*NullCheckpointsDB) DumpEngines 

func (*NullCheckpointsDB) DumpEngines(context.Context, io.Writer) error

func (*NullCheckpointsDB) DumpTables 

func (*NullCheckpointsDB) DumpTables(context.Context, io.Writer) error

func (*NullCheckpointsDB) Get 

func (*NullCheckpointsDB) Get(_ context.Context, _ string) (*TableCheckpoint, error)

func (*NullCheckpointsDB) GetLocalStoringTables 

func (*NullCheckpointsDB) GetLocalStoringTables(context.Context) (map[string][]int32, error)

func (*NullCheckpointsDB) IgnoreErrorCheckpoint 

func (*NullCheckpointsDB) IgnoreErrorCheckpoint(context.Context, string) error

func (*NullCheckpointsDB) Initialize 

func (*NullCheckpointsDB) Initialize(context.Context, *config.Config, map[string]*TidbDBInfo) error

func (*NullCheckpointsDB) InsertEngineCheckpoints 

func (*NullCheckpointsDB) InsertEngineCheckpoints(_ context.Context, _ string, _ map[int32]*EngineCheckpoint) error

func (*NullCheckpointsDB) MoveCheckpoints 

func (*NullCheckpointsDB) MoveCheckpoints(context.Context, int64) error

func (*NullCheckpointsDB) RemoveCheckpoint 

func (*NullCheckpointsDB) RemoveCheckpoint(context.Context, string) error

func (*NullCheckpointsDB) TaskCheckpoint 

func (*NullCheckpointsDB) TaskCheckpoint(ctx context.Context) (*TaskCheckpoint, error)

func (*NullCheckpointsDB) Update 

func (*NullCheckpointsDB) Update(context.Context, map[string]*TableCheckpointDiff) error

type RebaseCheckpointMerger 

type RebaseCheckpointMerger struct {
	AllocBase int64
}

func (*RebaseCheckpointMerger) MergeInto 

func (merger *RebaseCheckpointMerger) MergeInto(cpd *TableCheckpointDiff)

type Session 

type Session interface {
	Close()
	Execute(context.Context, string) ([]sqlexec.RecordSet, error)
	CommitTxn(context.Context) error
	RollbackTxn(context.Context)
	PrepareStmt(sql string) (stmtID uint32, paramCount int, fields []*ast.ResultField, err error)
	ExecutePreparedStmt(ctx context.Context, stmtID uint32, param []types.Datum) (sqlexec.RecordSet, error)
	DropPreparedStmt(stmtID uint32) error
}

type StatusCheckpointMerger 

type StatusCheckpointMerger struct {
	EngineID int32 // WholeTableEngineID == apply to whole table.
	Status   CheckpointStatus
}

func (*StatusCheckpointMerger) MergeInto 

func (merger *StatusCheckpointMerger) MergeInto(cpd *TableCheckpointDiff)

func (*StatusCheckpointMerger) SetInvalid 

func (merger *StatusCheckpointMerger) SetInvalid()

type TableCheckpoint 

type TableCheckpoint struct {
	Status    CheckpointStatus
	AllocBase int64
	Engines   map[int32]*EngineCheckpoint
	TableID   int64
	// remote checksum before restore
	Checksum verify.KVChecksum
}

func (*TableCheckpoint) Apply 

func (cp *TableCheckpoint) Apply(cpd *TableCheckpointDiff)

Apply the diff to the existing chunk and engine checkpoints in `cp`.

func (*TableCheckpoint) CountChunks 

func (cp *TableCheckpoint) CountChunks() int

func (*TableCheckpoint) DeepCopy 

func (cp *TableCheckpoint) DeepCopy() *TableCheckpoint

type TableCheckpointDiff 

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

func NewTableCheckpointDiff 

func NewTableCheckpointDiff() *TableCheckpointDiff

func (*TableCheckpointDiff) String 

func (cpd *TableCheckpointDiff) String() string

type TableCheckpointMerger 

type TableCheckpointMerger interface {
	// MergeInto the table checkpoint diff from a status update or chunk update.
	// If there are multiple updates to the same table, only the last one will
	// take effect. Therefore, the caller must ensure events for the same table
	// are properly ordered by the global time (an old event must be merged
	// before the new one).
	MergeInto(cpd *TableCheckpointDiff)
}

type TableChecksumMerger 

type TableChecksumMerger struct {
	Checksum verify.KVChecksum
}

func (*TableChecksumMerger) MergeInto 

func (m *TableChecksumMerger) MergeInto(cpd *TableCheckpointDiff)

type TaskCheckpoint 

type TaskCheckpoint struct {
	TaskID       int64
	SourceDir    string
	Backend      string
	ImporterAddr string
	TiDBHost     string
	TiDBPort     int
	PdAddr       string
	SortedKVDir  string
	LightningVer string
}

type TidbDBInfo 

type TidbDBInfo struct {
	ID     int64
	Name   string
	Tables map[string]*TidbTableInfo
}

type TidbTableInfo 

type TidbTableInfo struct {
	ID   int64
	DB   string
	Name string
	Core *model.TableInfo
}

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.