checkpoint

Documentation

Overview

Package checkpoint contains a utility for persisting checkpoint (fka resolved) timestamps.

Index

• Variables
• type Checkpoints
  • func ProvideCheckpoints(ctx context.Context, pool *types.StagingPool, meta ident.StagingSchema) (*Checkpoints, error)
  • func (r *Checkpoints) ScanForTargetSchemas(ctx context.Context) ([]ident.Schema, error)
  • func (r *Checkpoints) Start(ctx *stopper.Context, group *types.TableGroup, bounds *notify.Var[hlc.Range], ...) (*Group, error)
• type Group
  • func (r *Group) Advance(ctx context.Context, partition ident.Ident, ts hlc.Time) error
  • func (r *Group) Commit(ctx context.Context, rng hlc.Range) error
  • func (r *Group) Ensure(ctx context.Context, partitions []ident.Ident) error
  • func (r *Group) Refresh()
  • func (r *Group) TableGroup() *types.TableGroup
• type Option
  • func LimitLookahead(limit int) Option

Variables

var Set = wire.NewSet(ProvideCheckpoints)

Set is used by Wire.

Types

type Checkpoints

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

Checkpoints is a factory for Group instances, which manage checkpoint timestamps associated with a group of tables.

func ProvideCheckpoints

func ProvideCheckpoints(
	ctx context.Context, pool *types.StagingPool, meta ident.StagingSchema,
) (*Checkpoints, error)

ProvideCheckpoints is called by Wire.

func (*Checkpoints) ScanForTargetSchemas

func (r *Checkpoints) ScanForTargetSchemas(ctx context.Context) ([]ident.Schema, error)

ScanForTargetSchemas reports any group names that have unresolved timestamps.

func (*Checkpoints) Start

func (r *Checkpoints) Start(
	ctx *stopper.Context, group *types.TableGroup, bounds *notify.Var[hlc.Range], options ...Option,
) (*Group, error)

Start a background goroutine to update the provided bounds variable. The returned Group facade allows the bounds to be modified in conjunction with updating the checkpoint timestamp staging table. The returned Group is not memoized.

type Group

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

Group provides durable storage of the checkpoint (FKA resolved) timestamps associated with a types.TableGroup.

Timestamps to be processed are recorded with Group.Advance and their completion is recorded via Group.Commit. A checkpoint may consist of an arbitrary number of partitions, which may be pre-created with Group.Ensure.

This Group type will collaborate with other components by driving a
[notify.Var] containing an [hlc.Range]. This range represents a
window of eligible checkpoints that require processing, where the
minimum is called "committed" and the maximum is called "proposed".
The proposed time is the least common maximum value across all
partitions of a checkpoint, whereas the commit time is always common
to all partitions.

func (*Group) Advance

func (r *Group) Advance(ctx context.Context, partition ident.Ident, ts hlc.Time) error

Advance extends the proposed checkpoint timestamp associated with the partition of the Group. It is an error if the timestamp does not advance beyond its current point, as this will indicate a violation of changefeed invariants. If successful, this method will asynchronously refresh the Group.

func (*Group) Commit

func (r *Group) Commit(ctx context.Context, rng hlc.Range) error

Commit updates the applied-at timestamp associated with the checkpoints in the open range [min,max). This will asynchronously refresh the Group.

func (*Group) Ensure

func (r *Group) Ensure(ctx context.Context, partitions []ident.Ident) error

Ensure that a checkpoint exists for all named partitions. If no checkpoint exists for a given partition, an applied, minimum-valued checkpoint will be created. This method can be used to expand the number of partitions associated with a group at any point in time.

func (*Group) Refresh

func (r *Group) Refresh()

Refresh the Group asynchronously. This is intended for testing.

func (*Group) TableGroup

func (r *Group) TableGroup() *types.TableGroup

TableGroup returns the types.TableGroup whose checkpoints are being persisted.

type Option

type Option interface {
	// contains filtered or unexported methods
}

An Option to Checkpoints.Start.

func LimitLookahead

func LimitLookahead(limit int) Option

LimitLookahead limits the number of resolved timestamps that are used to calculate the extent of the resolving range.