checkpoint

package
v1.1.3 Latest Latest
Warning

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

 Go to latest Published: Apr 15, 2024 License: Apache-2.0 Imports: 35 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	PrefetchData uint16 = iota
	PrefetchMetaIdx
	ReadMetaIdx
	ReadData
)
View Source 
const (
	PrefixIncremental = "incremental"
	PrefixGlobal      = "global"
	PrefixMetadata    = "meta"
	CheckpointDir     = "ckp/"
)
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"

	CheckpointSchemaColumnCountV1 = 5 // start, end, loc, type, ver
	CheckpointSchemaColumnCountV2 = 9
)

Variables

View Source 
var (
	CheckpointSchemaAttr = []string{
		CheckpointAttr_StartTS,
		CheckpointAttr_EndTS,
		CheckpointAttr_MetaLocation,
		CheckpointAttr_EntryType,
		CheckpointAttr_Version,
		CheckpointAttr_AllLocations,
		CheckpointAttr_CheckpointLSN,
		CheckpointAttr_TruncateLSN,
		CheckpointAttr_Type,
	}
	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),
	}
)
View Source 
var (
	CheckpointSchema *catalog.Schema
)

Functions

func MergeCkpMeta added in v1.1.0 

func MergeCkpMeta(ctx context.Context, 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.Driver,
	opts ...Option) *runner

Types

type CheckpointEntry added in v0.6.0 

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

func NewCheckpointEntry added in v0.6.0 

func NewCheckpointEntry(start, end types.TS, typ EntryType) *CheckpointEntry

func (*CheckpointEntry) CheckPrintTime added in v0.8.0 

func (e *CheckpointEntry) CheckPrintTime() bool

func (*CheckpointEntry) GCEntry added in v0.7.0 

func (e *CheckpointEntry) GCEntry(fs *objectio.ObjectFS) error

func (*CheckpointEntry) GCMetadata added in v0.7.0 

func (e *CheckpointEntry) GCMetadata(fs *objectio.ObjectFS) error

func (*CheckpointEntry) GetByTableID added in v0.6.0 

func (e *CheckpointEntry) GetByTableID(ctx context.Context, fs *objectio.ObjectFS, tid uint64) (ins, del, cnIns, segDel *api.Batch, 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) 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) 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

func (*CheckpointEntry) IncrWaterLine added in v1.0.0 

func (e *CheckpointEntry) IncrWaterLine()

func (*CheckpointEntry) IsCommitted added in v0.6.0 

func (e *CheckpointEntry) IsCommitted() bool

func (*CheckpointEntry) IsFinished added in v0.6.0 

func (e *CheckpointEntry) IsFinished() 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) IsRunning added in v0.6.0 

func (e *CheckpointEntry) IsRunning() bool

func (*CheckpointEntry) LSNString added in v1.0.0 

func (e *CheckpointEntry) LSNString() string

func (*CheckpointEntry) LessEq added in v0.7.0 

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

func (*CheckpointEntry) Prefetch added in v0.8.0 

func (e *CheckpointEntry) Prefetch(
	ctx context.Context,
	fs *objectio.ObjectFS,
	data *logtail.CheckpointData,
) (err error)

func (*CheckpointEntry) PrefetchMetaIdx added in v1.0.0 

func (e *CheckpointEntry) PrefetchMetaIdx(
	ctx context.Context,
	fs *objectio.ObjectFS,
) (data *logtail.CheckpointData, err error)

func (*CheckpointEntry) Read added in v0.6.0 

func (e *CheckpointEntry) Read(
	ctx context.Context,
	fs *objectio.ObjectFS,
	data *logtail.CheckpointData,
) (err error)

func (*CheckpointEntry) ReadMetaIdx added in v1.0.0 

func (e *CheckpointEntry) ReadMetaIdx(
	ctx context.Context,
	fs *objectio.ObjectFS,
	data *logtail.CheckpointData,
) (err error)

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) SetState added in v0.6.0 

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

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

type DirtyCtx added in v0.6.0 

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

type EntryType added in v0.7.0 

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

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 WithCheckpointBlockRows added in v1.0.0 

func WithCheckpointBlockRows(rows int) Option

func WithCheckpointSize added in v1.0.0 

func WithCheckpointSize(size int) Option

func WithCollectInterval added in v0.6.0 

func WithCollectInterval(interval time.Duration) Option

func WithFlushInterval added in v0.6.0 

func WithFlushInterval(interval time.Duration) Option

func WithForceFlushCheckInterval added in v0.6.0 

func WithForceFlushCheckInterval(interval time.Duration) Option

func WithForceFlushTimeout added in v0.6.0 

func WithForceFlushTimeout(to time.Duration) Option

func WithGlobalMinCount added in v0.7.0 

func WithGlobalMinCount(count int) Option

func WithGlobalVersionInterval added in v0.7.0 

func WithGlobalVersionInterval(interval time.Duration) Option

func WithMinCount added in v0.6.0 

func WithMinCount(count int) Option

func WithMinIncrementalInterval added in v0.6.0 

func WithMinIncrementalInterval(interval time.Duration) Option

func WithObserver added in v0.6.0 

func WithObserver(o Observer) Option

func WithReserveWALEntryCount added in v1.0.0 

func WithReserveWALEntryCount(count uint64) Option

type Runner added in v0.6.0 

type Runner interface {
	TestRunner
	RunnerReader
	Start()
	Stop()
	String() string
	EnqueueWait(any) error
	Replay(catalog.DataFactory) (types.TS, uint64, bool, error)

	FlushTable(ctx context.Context, dbID, tableID uint64, ts types.TS) error
	GCByTS(ctx context.Context, ts types.TS) error

	// for test, delete in next phase
	DebugUpdateOptions(opts ...Option)
	GetAllCheckpoints() []*CheckpointEntry
}

type RunnerReader added in v0.7.0 

type RunnerReader interface {
	GetAllIncrementalCheckpoints() []*CheckpointEntry
	GetAllGlobalCheckpoints() []*CheckpointEntry
	GetPenddingIncrementalCount() int
	GetGlobalCheckpointCount() int
	CollectCheckpointsInRange(ctx context.Context, start, end types.TS) (ckpLoc string, lastEnd types.TS, err error)
	ICKPSeekLT(ts types.TS, cnt int) []*CheckpointEntry
	MaxLSN() uint64
}

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()
	DisableCheckpoint()

	CleanPenddingCheckpoint()
	ForceGlobalCheckpoint(end types.TS, versionInterval time.Duration) error
	ForceGlobalCheckpointSynchronously(ctx context.Context, end types.TS, versionInterval time.Duration) error
	ForceCheckpointForBackup(end types.TS) (string, error)
	ForceIncrementalCheckpoint(end types.TS, truncate bool) error
	IsAllChangesFlushed(start, end types.TS, printTree bool) bool
	MaxLSNInRange(end types.TS) uint64

	ExistPendingEntryToGC() bool
	MaxGlobalCheckpoint() *CheckpointEntry
	MaxCheckpoint() *CheckpointEntry
	ForceFlush(ts types.TS, ctx context.Context, duration time.Duration) (err error)
	ForceFlushWithInterval(ts types.TS, ctx context.Context, forceDuration, flushInterval time.Duration) (err error)
	GetDirtyCollector() logtail.Collector
}

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.