checkpoint

package
v0.0.0-debug-20260702 Latest Latest
Warning

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

 Go to latest Published: Jul 2, 2026 License: Apache-2.0 Imports: 41 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	PrefetchData uint16 = iota
	PrefetchMetaIdx
	ReadMetaIdx
	ReadData
)
View Source 
const (
	CKPProtocolVersion_V1 uint8 = 1
	CKPProtocolVersion_V2 uint8 = 2

	CKPProtocolVersion_Curr = CKPProtocolVersion_V2
)
View Source 
const (
	CheckpointAttr_StartTS         = "start_ts"
	CheckpointAttr_EndTS           = "end_ts"
	CheckpointAttr_MetaLocation    = "meta_location"
	CheckpointAttr_EntryType       = "entry_type"
	CheckpointAttr_Version         = "version"
	CheckpointAttr_AllLocations    = "all_locations"
	CheckpointAttr_CheckpointLSN   = "checkpoint_lsn"
	CheckpointAttr_TruncateLSN     = "truncate_lsn"
	CheckpointAttr_Type            = "type"
	CheckpointAttr_TableIDLocation = "table_id_location"

	CheckpointAttr_StartTSIdx         = 0
	CheckpointAttr_EndTSIdx           = 1
	CheckpointAttr_MetaLocationIdx    = 2
	CheckpointAttr_EntryTypeIdx       = 3
	CheckpointAttr_VersionIdx         = 4
	CheckpointAttr_AllLocationsIdx    = 5
	CheckpointAttr_CheckpointLSNIdx   = 6
	CheckpointAttr_TruncateLSNIdx     = 7
	CheckpointAttr_TypeIdx            = 8
	CheckpointAttr_TableIDLocationIdx = 9

	CheckpointSchemaColumnCountV1 = 5 // start, end, loc, type, ver
	CheckpointSchemaColumnCountV2 = 9
	CheckpointSchemaColumnCountV3 = 10
)
View Source 
const (
	DefaultObjectReplayWorkerCount = 10
)

Variables

View Source 
var (
	CheckpointSchemaAttr = []string{
		CheckpointAttr_StartTS,
		CheckpointAttr_EndTS,
		CheckpointAttr_MetaLocation,
		CheckpointAttr_EntryType,
		CheckpointAttr_Version,
		CheckpointAttr_AllLocations,
		CheckpointAttr_CheckpointLSN,
		CheckpointAttr_TruncateLSN,
		CheckpointAttr_Type,
		CheckpointAttr_TableIDLocation,
	}
	CheckpointSchemaTypes = []types.Type{
		types.New(types.T_TS, 0, 0),
		types.New(types.T_TS, 0, 0),
		types.New(types.T_varchar, types.MaxVarcharLen, 0),
		types.New(types.T_bool, 0, 0),
		types.New(types.T_uint32, 0, 0),
		types.New(types.T_varchar, types.MaxVarcharLen, 0),
		types.New(types.T_uint64, 0, 0),
		types.New(types.T_uint64, 0, 0),
		types.New(types.T_int8, 0, 0),
		types.New(types.T_varchar, types.MaxVarcharLen, 0),
	}
)
View Source 
var (
	CheckpointSchema *catalog.Schema
)
View Source 
var ErrBadIntent = moerr.NewInternalErrorNoCtxf("bad intent")
View Source 
var ErrCheckpointDisabled = moerr.NewInternalErrorNoCtxf("checkpoint disabled")
View Source 
var ErrExecutorClosed = moerr.NewInternalErrorNoCtxf("executor closed")
View Source 
var ErrExecutorRestarted = moerr.NewInternalErrorNoCtxf("executor restarted")
View Source 
var ErrFlusherStopped = moerr.NewInternalErrorNoCtx("flusher stopped")
View Source 
var ErrPendingCheckpoint = moerr.NewPrevCheckpointNotFinished()
View Source 
var ErrStopRunner = moerr.NewInternalErrorNoCtxf("runner stopped")

Functions

func FilterSortedMetaFilesByTimestamp 

func FilterSortedMetaFilesByTimestamp(
	ts *types.TS,
	files []ioutil.TSRangeFile,
) []ioutil.TSRangeFile

func IsAllDirtyFlushed 

func IsAllDirtyFlushed(source logtail.Collector, cata *catalog.Catalog, start, end types.TS, doPrint bool) bool

func MakeMetadataBatch 

func MakeMetadataBatch() *batch.Batch

func MergeCkpMeta added in v1.1.0 

func MergeCkpMeta(
	ctx context.Context,
	sid string,
	fs fileservice.FileService,
	cnLocation, tnLocation objectio.Location,
	startTs, ts types.TS,
) (string, error)

func NewRunner added in v0.6.0 

func NewRunner(
	ctx context.Context,
	rt *dbutils.Runtime,
	catalog *catalog.Catalog,
	source logtail.Collector,
	wal wal.Store,
	cfg *CheckpointCfg,
	opts ...Option,
) *runner

Types

type CKPMetaReader 

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

func MakeCKPMetaDirReader 

func MakeCKPMetaDirReader(
	ctx context.Context,
	sid string,
	verbose int,
	fs fileservice.FileService,
) (*CKPMetaReader, error)

func NewCKPMetaReader 

func NewCKPMetaReader(
	sid string,
	dir string,
	files []string,
	verbose int,
	fs fileservice.FileService,
) *CKPMetaReader

func (*CKPMetaReader) Close 

func (r *CKPMetaReader) Close()

func (*CKPMetaReader) Next 

func (r *CKPMetaReader) Next(
	ctx context.Context,
	mp *mpool.MPool,
) (bats []*batch.Batch, release func(), err error)

type CheckpointCfg 

type CheckpointCfg struct {

	// min count of transaction to trigger incremental checkpoint
	// exception: force incremental checkpoint ignore min count
	MinCount int64

	// extra reserved wal entry count for incremental checkpoint
	IncrementalReservedWALCount uint64

	// min interval to trigger incremental checkpoint
	// exception: force incremental checkpoint ignore interval
	IncrementalInterval time.Duration

	// increment checkpoint min count to trigger global checkpoint
	// exception: force global checkpoint ignore min count
	GlobalMinCount int64

	// history duration to keep for a global checkpoint
	GlobalHistoryDuration time.Duration

	// history duration to keep for a table id
	TableIDHistoryDuration time.Duration
	// object size threshold to sink table id batch
	TableIDSinkerThreshold int
}

func (*CheckpointCfg) FillDefaults 

func (cfg *CheckpointCfg) FillDefaults()

func (CheckpointCfg) String 

func (cfg CheckpointCfg) String() string

type CheckpointEntry added in v0.6.0 

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

func InheritCheckpointEntry 

func InheritCheckpointEntry(
	from *CheckpointEntry,
	replaceOpts ...EntryOption,
) *CheckpointEntry

func ListSnapshotCheckpoint added in v1.2.0 

func ListSnapshotCheckpoint(
	ctx context.Context,
	sid string,
	fs fileservice.FileService,
	snapshot types.TS,
	allFiles map[string]struct{},
) ([]*CheckpointEntry, error)

func NewCheckpointEntry added in v0.6.0 

func NewCheckpointEntry(
	sid string, start, end types.TS, typ EntryType, opts ...EntryOption,
) *CheckpointEntry

func ReadEntriesFromMeta 

func ReadEntriesFromMeta(
	ctx context.Context,
	sid string,
	dir string,
	name string,
	verbose int,
	onEachEntry func(entry *CheckpointEntry),
	mp *mpool.MPool,
	fs fileservice.FileService,
) (entries []*CheckpointEntry, err error)

name is like `meta_<start_ts>_<end_ts>.<ext>` ReadEntriesFromMeta reads the checkpoint entries from the meta file. 

+--------------------------------+-------------------+
| Attrs                          | Types             |
+--------------------------------+-------------------+
| CheckpointAttr_Start           | types.TS          |
| CheckpointAttr_End             | types.TS          |
| CheckpointAttr_MetaLocation    | objectio.Location |
| CheckpointAttr_EntryType       | int8              |
| CheckpointAttr_Version         | uint32            |
| CheckpointAttr_AllLocations    | objectio.Location |
| CheckpointAttr_CheckpointLSN   | uint64            |
| CheckpointAttr_TruncateLSN     | uint64            |
| CheckpointAttr_Type            | int8              |
+--------------------------------+-------------------+

func ReplayCheckpointEntries 

func ReplayCheckpointEntries(bat *containers.Batch, checkpointVersion int) (entries []*CheckpointEntry, maxGlobalEnd types.TS)

func (*CheckpointEntry) Age 

func (e *CheckpointEntry) Age() time.Duration

func (*CheckpointEntry) AllChecked 

func (e *CheckpointEntry) AllChecked() bool

func (*CheckpointEntry) AllGE 

func (e *CheckpointEntry) AllGE(o *CheckpointEntry) bool

compare with other entry e.start >= o.end

func (*CheckpointEntry) DeferRetirement 

func (e *CheckpointEntry) DeferRetirement()

func (*CheckpointEntry) Done 

func (e *CheckpointEntry) Done()

it can only be called at most once it will unblock all Wait() calls

func (*CheckpointEntry) GetCheckpointMetaInfo 

func (e *CheckpointEntry) GetCheckpointMetaInfo(
	ctx context.Context,
	id uint64,
	mp *mpool.MPool,
	fs fileservice.FileService,
) (res *logtail.ObjectInfoJson, err error)

func (*CheckpointEntry) GetEnd added in v0.6.0 

func (e *CheckpointEntry) GetEnd() types.TS

func (*CheckpointEntry) GetLocation added in v0.6.0 

func (e *CheckpointEntry) GetLocation() objectio.Location

func (*CheckpointEntry) GetObjects 

func (e *CheckpointEntry) GetObjects(
	ctx context.Context,
	pinned map[string]bool,
	mp *mpool.MPool,
	fs fileservice.FileService,
) (err error)

func (*CheckpointEntry) GetStart added in v0.6.0 

func (e *CheckpointEntry) GetStart() types.TS

func (*CheckpointEntry) GetState added in v0.6.0 

func (e *CheckpointEntry) GetState() State

func (*CheckpointEntry) GetTNLocation added in v1.2.0 

func (e *CheckpointEntry) GetTNLocation() objectio.Location

func (*CheckpointEntry) GetTableByID 

func (e *CheckpointEntry) GetTableByID(
	ctx context.Context, fs fileservice.FileService, tid uint64, mp *mpool.MPool,
) (data *batch.Batch, err error)

Only for test Only fill in columns ObjectID, CreateTS and DeleteTS

func (*CheckpointEntry) GetTableIDLocation 

func (e *CheckpointEntry) GetTableIDLocation() objectio.LocationSlice

func (*CheckpointEntry) GetTableIDs 

func (e *CheckpointEntry) GetTableIDs(
	ctx context.Context,
	loc objectio.Location,
	mp *mpool.MPool,
	fs fileservice.FileService,
) (result []uint64, err error)

func (*CheckpointEntry) GetTableRanges 

func (e *CheckpointEntry) GetTableRanges(
	ctx context.Context,
	mp *mpool.MPool,
	fs fileservice.FileService,
) (ranges []ckputil.TableRange, err error)

func (*CheckpointEntry) GetTableRangesByID 

func (e *CheckpointEntry) GetTableRangesByID(
	ctx context.Context,
	tableID uint64,
	mp *mpool.MPool,
	fs fileservice.FileService,
) (ranges []ckputil.TableRange, err error)

func (*CheckpointEntry) GetTruncateLsn 

func (e *CheckpointEntry) GetTruncateLsn() uint64

func (*CheckpointEntry) GetType added in v1.1.0 

func (e *CheckpointEntry) GetType() EntryType

func (*CheckpointEntry) GetVersion added in v1.0.0 

func (e *CheckpointEntry) GetVersion() uint32

func (*CheckpointEntry) HasOverlap added in v0.6.0 

func (e *CheckpointEntry) HasOverlap(from, to types.TS) bool

false: e.start > to or e.end < from true: otherwise

func (*CheckpointEntry) IsBackup 

func (e *CheckpointEntry) IsBackup() bool

func (*CheckpointEntry) IsCommitted added in v0.6.0 

func (e *CheckpointEntry) IsCommitted() bool

func (*CheckpointEntry) IsCompact 

func (e *CheckpointEntry) IsCompact() bool

func (*CheckpointEntry) IsFinished added in v0.6.0 

func (e *CheckpointEntry) IsFinished() bool

func (*CheckpointEntry) IsFlushChecked 

func (e *CheckpointEntry) IsFlushChecked() bool

func (*CheckpointEntry) IsGlobal 

func (e *CheckpointEntry) IsGlobal() bool

func (*CheckpointEntry) IsIncremental added in v0.6.0 

func (e *CheckpointEntry) IsIncremental() bool

func (*CheckpointEntry) IsPendding added in v0.6.0 

func (e *CheckpointEntry) IsPendding() bool

func (*CheckpointEntry) IsPolicyChecked 

func (e *CheckpointEntry) IsPolicyChecked() bool

func (*CheckpointEntry) IsRunning added in v0.6.0 

func (e *CheckpointEntry) IsRunning() bool

func (*CheckpointEntry) IsYoungNeighbor 

func (e *CheckpointEntry) IsYoungNeighbor(o *CheckpointEntry) bool

true: o.start == e.end + 1

func (*CheckpointEntry) JsonString 

func (e *CheckpointEntry) JsonString() string

func (*CheckpointEntry) LEByTS 

func (e *CheckpointEntry) LEByTS(ts *types.TS) bool

e.end <= ts it means that e is before ts

func (*CheckpointEntry) LSN 

func (e *CheckpointEntry) LSN() uint64

func (*CheckpointEntry) LSNString added in v1.0.0 

func (e *CheckpointEntry) LSNString() string

func (*CheckpointEntry) MarshalJSON 

func (e *CheckpointEntry) MarshalJSON() ([]byte, error)

func (*CheckpointEntry) ResetAge 

func (e *CheckpointEntry) ResetAge()

func (*CheckpointEntry) SetFlushChecked 

func (e *CheckpointEntry) SetFlushChecked()

func (*CheckpointEntry) SetLSN added in v1.0.0 

func (e *CheckpointEntry) SetLSN(ckpLSN, truncateLSN uint64)

func (*CheckpointEntry) SetLocation added in v0.6.0 

func (e *CheckpointEntry) SetLocation(cn, tn objectio.Location)

func (*CheckpointEntry) SetPolicyChecked 

func (e *CheckpointEntry) SetPolicyChecked()

func (*CheckpointEntry) SetState added in v0.6.0 

func (e *CheckpointEntry) SetState(state State) (ok bool)

func (*CheckpointEntry) SetTableIDLocation 

func (e *CheckpointEntry) SetTableIDLocation(locations objectio.LocationSlice)

func (*CheckpointEntry) SetVersion added in v1.0.1 

func (e *CheckpointEntry) SetVersion(version uint32)

func (*CheckpointEntry) String added in v0.6.0 

func (e *CheckpointEntry) String() string

func (*CheckpointEntry) TooOld 

func (e *CheckpointEntry) TooOld() bool

func (*CheckpointEntry) Wait 

func (e *CheckpointEntry) Wait() <-chan struct{}

it can be called multiple times it will block until Done() is called

type CheckpointReader 

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

func NewCheckpointReader 

func NewCheckpointReader(
	ctx context.Context,
	checkpointEntries []*CheckpointEntry,
	maxGlobalEnd types.TS,
	mp *mpool.MPool,
	fs fileservice.FileService,
) *CheckpointReader

func (*CheckpointReader) Prepare 

func (reader *CheckpointReader) Prepare(ctx context.Context) (err error)

func (*CheckpointReader) Read 

func (reader *CheckpointReader) Read(
	ctx context.Context,
	bat *batch.Batch,
) (end bool, err error)

func (*CheckpointReader) Reset 

func (reader *CheckpointReader) Reset(ctx context.Context)

type CheckpointScheduler 

type CheckpointScheduler interface {
	TryScheduleCheckpoint(types.TS, bool) (Intent, error)
	RunnerReader
}

type CkpReplayer 

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

func (*CkpReplayer) Close 

func (c *CkpReplayer) Close()

func (*CkpReplayer) ReadCkpFiles 

func (c *CkpReplayer) ReadCkpFiles() (err error)

func (*CkpReplayer) ReplayCatalog 

func (c *CkpReplayer) ReplayCatalog(
	readTxn txnif.AsyncTxn,
	phase string,
) (err error)

func (*CkpReplayer) ReplayObjectlist 

func (c *CkpReplayer) ReplayObjectlist(ctx context.Context, phase string) (err error)

ReplayObjectlist replays the data part of the checkpoint.

func (*CkpReplayer) ReplayThreeTablesObjectlist 

func (c *CkpReplayer) ReplayThreeTablesObjectlist(phase string) (
	maxTs types.TS,
	maxLSN uint64,
	err error,
)

ReplayThreeTablesObjectlist replays the object list the three tables, and check the LSN and TS.

func (*CkpReplayer) Submit 

func (c *CkpReplayer) Submit(tid uint64, replayFn func())

type EntryOption 

type EntryOption func(*CheckpointEntry)

func WithCheckedEntryOption 

func WithCheckedEntryOption(policyChecked, flushedChecked bool) EntryOption

func WithEndEntryOption 

func WithEndEntryOption(end types.TS) EntryOption

func WithStateEntryOption 

func WithStateEntryOption(state State) EntryOption

type EntryType added in v0.7.0 

type EntryType int8
const (
	ET_Global EntryType = iota
	ET_Incremental
	ET_Backup
	ET_Compacted
)

func (EntryType) String 

func (e EntryType) String() string

type FlushCfg 

type FlushCfg struct {
	ForceFlushTimeout       time.Duration
	ForceFlushCheckInterval time.Duration
	FlushInterval           time.Duration
	CronPeriod              time.Duration
}

type FlushMutableCfg 

type FlushMutableCfg struct {
	ForceFlushTimeout       time.Duration
	ForceFlushCheckInterval time.Duration
}

type FlushRequest 

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

type Flusher 

type Flusher interface {
	IsAllChangesFlushed(start, end types.TS, doPrint bool) bool
	FlushTable(ctx context.Context, dbID, tableID uint64, ts types.TS) error
	ForceFlush(ctx context.Context, ts types.TS) error
	ForceFlushWithInterval(ctx context.Context, ts types.TS, flushInterval time.Duration) (err error)
	ChangeForceFlushTimeout(timeout time.Duration)
	ChangeForceCheckInterval(interval time.Duration)
	GetCfg() FlushCfg
	Restart(opts ...FlusherOption)
	IsNoop() bool
	Start()
	Stop()
}

func NewFlusher 

func NewFlusher(
	rt *dbutils.Runtime,
	checkpointSchduler CheckpointScheduler,
	catalogCache *catalog.Catalog,
	sourcer logtail.Collector,
	noop bool,
	opts ...FlusherOption,
) Flusher

type FlusherOption 

type FlusherOption func(*flushImpl)

func WithFlusherCfg 

func WithFlusherCfg(cfg FlushCfg) FlusherOption

func WithFlusherCronPeriod 

func WithFlusherCronPeriod(period time.Duration) FlusherOption

func WithFlusherForceCheckInterval 

func WithFlusherForceCheckInterval(interval time.Duration) FlusherOption

func WithFlusherForceTimeout 

func WithFlusherForceTimeout(timeout time.Duration) FlusherOption

func WithFlusherInterval 

func WithFlusherInterval(interval time.Duration) FlusherOption

func WithFlusherQueueSize 

func WithFlusherQueueSize(size int) FlusherOption

type Intent 

type Intent interface {
	String() string
	Wait() <-chan struct{}
}

type MetadataEntryGetter 

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

func (*MetadataEntryGetter) Close 

func (getter *MetadataEntryGetter) Close()

func (*MetadataEntryGetter) NextBatch 

func (getter *MetadataEntryGetter) NextBatch(
	ctx context.Context,
	onEachEntry func(entry *CheckpointEntry),
	mp *mpool.MPool,
) (
	entries []*CheckpointEntry, err error,
)

type Observer added in v0.6.0 

type Observer interface {
	OnNewCheckpoint(ts types.TS)
}

type Option added in v0.6.0 

type Option func(*runner)

func WithObserver added in v0.6.0 

func WithObserver(o Observer) Option

type ReplayClient 

type ReplayClient interface {
	AddCheckpointMetaFile(string)
	ReplayCKPEntry(*CheckpointEntry) error
}

type Runner added in v0.6.0 

type Runner interface {
	ReplayClient
	CheckpointScheduler
	TestRunner
	RunnerWriter
	RunnerReader

	Start()
	Stop()

	BuildReplayer(string) *CkpReplayer
	GCByTS(ctx context.Context, ts types.TS) error
}

type RunnerReader added in v0.7.0 

type RunnerReader interface {
	String() string
	GetAllIncrementalCheckpoints() []*CheckpointEntry
	GetAllGlobalCheckpoints() []*CheckpointEntry
	GetIncrementalCountAfterGlobal() int
	CollectCheckpointsInRange(ctx context.Context, start, end types.TS) (ckpLoc string, lastEnd types.TS, err error)
	ICKPSeekLT(ts types.TS, cnt int) []*CheckpointEntry
	GetLowWaterMark() types.TS
	MaxLSN() uint64
	GetCatalog() *catalog.Catalog
	GetCheckpointMetaFiles() map[string]struct{}
	ICKPRange(start, end *types.TS, cnt int) []*CheckpointEntry
	GetCompacted() *CheckpointEntry

	// for test, delete in next phase
	GetAllCheckpoints() []*CheckpointEntry
	GetAllCheckpointsForBackup(compact *CheckpointEntry) []*CheckpointEntry

	MaxGlobalCheckpoint() *CheckpointEntry
	MaxIncrementalCheckpoint() *CheckpointEntry
	PendingIncrementalCheckpoint() *CheckpointEntry
	MinIncrementalCheckpoint() *CheckpointEntry
	GetDirtyCollector() logtail.Collector
}

type RunnerWriter 

type RunnerWriter interface {
	RemoveCheckpointMetaFile(string)
	AddCheckpointMetaFile(string)
	UpdateCompacted(entry *CheckpointEntry) bool
}

type State added in v0.6.0 

type State int8
const (
	ST_Running State = iota
	ST_Pending
	ST_Finished
)

type TestRunner added in v0.7.0 

type TestRunner interface {
	EnableCheckpoint(*CheckpointCfg)
	DisableCheckpoint(ctx context.Context) (*CheckpointCfg, error)

	// special file for backup
	CreateSpecialCheckpointFile(ctx context.Context, start, end types.TS) (string, error)

	ForceGCKP(context.Context, types.TS, time.Duration) error
	ForceICKP(context.Context, *types.TS) error
	MaxLSNInRange(end types.TS) uint64
	GetICKPIntentOnlyForTest() *CheckpointEntry

	WaitRunningCKPDoneForTest(ctx context.Context, gckp bool) error

	GCNeeded() bool
}

Source Files

View all Source files

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.