checkpoint

Documentation

Index

• Constants
• Variables
• func NewRunner(fs *objectio.ObjectFS, catalog *catalog.Catalog, scheduler tasks.TaskScheduler, ...) *runner
• type CheckpointEntry
  • func NewCheckpointEntry(start, end types.TS) *CheckpointEntry
  • func (e *CheckpointEntry) GetByTableID(fs *objectio.ObjectFS, tid uint64) (ins, del, cnIns *api.Batch, err error)
  • func (e *CheckpointEntry) GetEnd() types.TS
  • func (e *CheckpointEntry) GetLocation() string
  • func (e *CheckpointEntry) GetStart() types.TS
  • func (e *CheckpointEntry) GetState() State
  • func (e *CheckpointEntry) HasOverlap(from, to types.TS) bool
  • func (e *CheckpointEntry) IsCommitted() bool
  • func (e *CheckpointEntry) IsFinished() bool
  • func (e *CheckpointEntry) IsIncremental() bool
  • func (e *CheckpointEntry) IsPendding() bool
  • func (e *CheckpointEntry) IsRunning() bool
  • func (e *CheckpointEntry) Read(fs *objectio.ObjectFS) (data *logtail.CheckpointData, err error)
  • func (e *CheckpointEntry) Replay(c *catalog.Catalog, fs *objectio.ObjectFS, dataFactory catalog.DataFactory) (readDuration, applyDuration time.Duration, err error)
  • func (e *CheckpointEntry) SetLocation(location string)
  • func (e *CheckpointEntry) SetState(state State) (ok bool)
  • func (e *CheckpointEntry) String() string
• type DirtyCtx
• type Observer
• type Option
  • func WithCollectInterval(interval time.Duration) Option
  • func WithFlushInterval(interval time.Duration) Option
  • func WithForceFlushCheckInterval(interval time.Duration) Option
  • func WithForceFlushTimeout(to time.Duration) Option
  • func WithMinCount(count int) Option
  • func WithMinGlobalInterval(interval time.Duration) Option
  • func WithMinIncrementalInterval(interval time.Duration) Option
  • func WithObserver(o Observer) Option
• type Runner
• type State

Constants

const (
	PrefixIncremental = "incremental"
	PrefixGlobal      = "global"
	PrefixMetadata    = "meta"
	CheckpointDir     = "ckp/"
)

const (
	CheckpointAttr_StartTS      = "start_ts"
	CheckpointAttr_EndTS        = "end_ts"
	CheckpointAttr_MetaLocation = "meta_location"
)

Variables

var (
	CheckpointSchemaAttr = []string{
		CheckpointAttr_StartTS,
		CheckpointAttr_EndTS,
		CheckpointAttr_MetaLocation,
	}
	CheckpointSchemaTypes = []types.Type{
		types.New(types.T_TS, 0, 0, 0),
		types.New(types.T_TS, 0, 0, 0),
		types.New(types.T_varchar, 0, 0, 0),
	}
)

var (
	CheckpointSchema *catalog.Schema
)

Functions

func NewRunner

func NewRunner(
	fs *objectio.ObjectFS,
	catalog *catalog.Catalog,
	scheduler tasks.TaskScheduler,
	source logtail.Collector,
	wal wal.Driver,
	opts ...Option) *runner

Types

type CheckpointEntry

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

func NewCheckpointEntry

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

func (*CheckpointEntry) GetByTableID

func (e *CheckpointEntry) GetByTableID(fs *objectio.ObjectFS, tid uint64) (ins, del, cnIns *api.Batch, err error)

func (*CheckpointEntry) GetEnd

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

func (*CheckpointEntry) GetLocation

func (e *CheckpointEntry) GetLocation() string

func (*CheckpointEntry) GetStart

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

func (*CheckpointEntry) GetState

func (e *CheckpointEntry) GetState() State

func (*CheckpointEntry) HasOverlap

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

func (*CheckpointEntry) IsCommitted

func (e *CheckpointEntry) IsCommitted() bool

func (*CheckpointEntry) IsFinished

func (e *CheckpointEntry) IsFinished() bool

func (*CheckpointEntry) IsIncremental

func (e *CheckpointEntry) IsIncremental() bool

func (*CheckpointEntry) IsPendding

func (e *CheckpointEntry) IsPendding() bool

func (*CheckpointEntry) IsRunning

func (e *CheckpointEntry) IsRunning() bool

func (*CheckpointEntry) Read

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

func (*CheckpointEntry) Replay

func (e *CheckpointEntry) Replay(
	c *catalog.Catalog,
	fs *objectio.ObjectFS,
	dataFactory catalog.DataFactory) (readDuration, applyDuration time.Duration, err error)

func (*CheckpointEntry) SetLocation

func (e *CheckpointEntry) SetLocation(location string)

func (*CheckpointEntry) SetState

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

func (*CheckpointEntry) String

func (e *CheckpointEntry) String() string

type DirtyCtx

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

type Observer

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

type Option

type Option func(*runner)

func WithCollectInterval

func WithCollectInterval(interval time.Duration) Option

func WithFlushInterval

func WithFlushInterval(interval time.Duration) Option

func WithForceFlushCheckInterval

func WithForceFlushCheckInterval(interval time.Duration) Option

func WithForceFlushTimeout

func WithForceFlushTimeout(to time.Duration) Option

func WithMinCount

func WithMinCount(count int) Option

func WithMinGlobalInterval

func WithMinGlobalInterval(interval time.Duration) Option

func WithMinIncrementalInterval

func WithMinIncrementalInterval(interval time.Duration) Option

func WithObserver

func WithObserver(o Observer) Option

type Runner

type Runner interface {
	Start()
	Stop()
	EnqueueWait(any) error
	Replay(catalog.DataFactory) (types.TS, error)
	MaxLSN() uint64

	MockCheckpoint(end types.TS)
	FlushTable(dbID, tableID uint64, ts types.TS) error

	// for test, delete in next phase
	TestCheckpoint(entry *CheckpointEntry)
	DebugUpdateOptions(opts ...Option)
	GetAllCheckpoints() []*CheckpointEntry
	CollectCheckpointsInRange(start, end types.TS) (ckpLoc string, lastEnd types.TS)
}

type State

type State int8

const (
	ST_Running State = iota
	ST_Pending
	ST_Finished
)