gc

Documentation

Index

• Constants
• Variables
• type BatchType
• type Cleaner
  • func NewCheckpointCleaner(ctx context.Context, fs *objectio.ObjectFS, ckpClient checkpoint.RunnerReader, ...) Cleaner
• type CleanerState
• type DiskCleaner
  • func NewDiskCleaner(diskCleaner Cleaner) *DiskCleaner
  • func (cleaner *DiskCleaner) GC(ctx context.Context) (err error)
  • func (cleaner *DiskCleaner) GetCleaner() Cleaner
  • func (cleaner *DiskCleaner) Start()
  • func (cleaner *DiskCleaner) Stop()
• type GCTable
  • func NewGCTable() *GCTable
  • func (t *GCTable) Compare(table *GCTable) bool
  • func (t *GCTable) Merge(GCTable *GCTable)
  • func (t *GCTable) ReadTable(ctx context.Context, name string, size int64, fs *objectio.ObjectFS, ...) error
  • func (t *GCTable) SaveFullTable(start, end types.TS, fs *objectio.ObjectFS, files []string) ([]objectio.BlockObject, error)
  • func (t *GCTable) SaveTable(start, end types.TS, fs *objectio.ObjectFS, files []string) ([]objectio.BlockObject, error)
  • func (t *GCTable) SoftGC(table *GCTable, ts types.TS, snapShotList map[uint32]containers.Vector, ...) ([]string, map[uint32][]types.TS)
  • func (t *GCTable) String() string
  • func (t *GCTable) UpdateTable(data *logtail.CheckpointData)
• type GCWorker
  • func NewGCWorker(fs *objectio.ObjectFS, cleaner *checkpointCleaner) *GCWorker
  • func (g *GCWorker) ExecDelete(ctx context.Context, names []string, disableGC bool) error
  • func (g *GCWorker) Idle()
  • func (g *GCWorker) Start() bool
• type ObjectEntry
• type TombstoneEntry

Constants

const (
	MessgeReplay = iota
	MessgeNormal
)

const (
	PrefixGCMeta   = "gc"
	PrefixSnapMeta = "snap"
	PrefixAcctMeta = "acct"
	GCMetaDir      = "gc/"
	CKPMetaDir     = "ckp/"
)

const (
	GCAttrObjectName = "name"
	GCAttrBlockId    = "block_id"
	GCAttrTableId    = "table_id"
	GCAttrDBId       = "db_id"
	GCAttrCommitTS   = "commit_ts"
	GCCreateTS       = "create_time"
	GCDeleteTS       = "delete_time"
	GCAttrTombstone  = "tombstone"
	GCAttrVersion    = "version"
)

const (
	AddChecker    = "add_checker"
	RemoveChecker = "remove_checker"
)

const (
	CheckerKeyTTL   = "ttl"
	CheckerKeyMinTS = "min_ts"
)

const CurrentVersion = uint16(3)

const MinMergeCount = 20

const NotFoundLimit = 10

Variables

var (
	BlockSchemaAttr = []string{
		GCAttrObjectName,
		GCCreateTS,
		GCDeleteTS,
		GCAttrCommitTS,
		GCAttrTableId,
	}
	BlockSchemaTypes = []types.Type{
		types.New(types.T_varchar, 5000, 0),
		types.New(types.T_TS, types.MaxVarcharLen, 0),
		types.New(types.T_TS, types.MaxVarcharLen, 0),
		types.New(types.T_TS, types.MaxVarcharLen, 0),
		types.New(types.T_uint64, 0, 0),
	}

	BlockSchemaAttrV1 = []string{
		GCAttrBlockId,
		GCAttrTableId,
		GCAttrDBId,
		GCAttrObjectName,
	}
	BlockSchemaTypesV1 = []types.Type{
		types.New(types.T_Blockid, 0, 0),
		types.New(types.T_uint64, 0, 0),
		types.New(types.T_uint64, 0, 0),
		types.New(types.T_varchar, 5000, 0),
	}

	TombstoneSchemaAttr = []string{
		GCAttrTombstone,
		GCAttrObjectName,
		GCAttrCommitTS,
	}

	TombstoneSchemaTypes = []types.Type{
		types.New(types.T_varchar, 5000, 0),
		types.New(types.T_varchar, 5000, 0),
		types.New(types.T_TS, types.MaxVarcharLen, 0),
	}

	VersionsSchemaAttr = []string{
		GCAttrVersion,
	}

	VersionsSchemaTypes = []types.Type{
		types.New(types.T_uint16, 0, 0),
	}

	DropTableSchemaAttr = []string{
		GCAttrTableId,
		GCAttrDBId,
	}
	DropTableSchemaTypes = []types.Type{
		types.New(types.T_uint64, 0, 0),
		types.New(types.T_uint64, 0, 0),
	}

	DropDBSchemaAtt = []string{
		GCAttrDBId,
	}
	DropDBSchemaTypes = []types.Type{
		types.New(types.T_uint64, 0, 0),
	}

	DeleteFileSchemaAtt = []string{
		GCAttrObjectName,
	}
	DeleteFileSchemaTypes = []types.Type{
		types.New(types.T_varchar, 5000, 0),
	}
)

Types

type BatchType

type BatchType int8

const (
	Versions BatchType = iota
	ObjectList
	TombstoneList
)

const (
	CreateBlock BatchType = iota
	DeleteBlock
	DropTable
	DropDB
	DeleteFile
	Tombstone
)

type Cleaner

type Cleaner interface {
	Replay() error
	Process()
	TryGC() error
	AddChecker(checker func(item any) bool, key string) int
	RemoveChecker(key string) error
	GetMaxConsumed() *checkpoint.CheckpointEntry
	Stop()
	// for test
	SetMinMergeCountForTest(count int)
	GetMinMerged() *checkpoint.CheckpointEntry
	CheckGC() error
	GetInputs() *GCTable
	SetTid(tid uint64)
	EnableGCForTest()
	DisableGCForTest()
	SetCheckGC(enable bool)
	GetMPool() *mpool.MPool
	GetSnapshots() (map[uint32]containers.Vector, error)
}

func NewCheckpointCleaner

func NewCheckpointCleaner(
	ctx context.Context,
	fs *objectio.ObjectFS,
	ckpClient checkpoint.RunnerReader,
	disableGC bool,
) Cleaner

type CleanerState

type CleanerState int8

const (
	Idle CleanerState = iota
	Running
)

type DiskCleaner

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

DiskCleaner is the main structure of gc operation, and provides "JobFactory" to let tae notify itself to perform a gc

func NewDiskCleaner

func NewDiskCleaner(
	diskCleaner Cleaner,
) *DiskCleaner

func (*DiskCleaner) GC

func (cleaner *DiskCleaner) GC(ctx context.Context) (err error)

func (*DiskCleaner) GetCleaner

func (cleaner *DiskCleaner) GetCleaner() Cleaner

func (*DiskCleaner) Start

func (cleaner *DiskCleaner) Start()

func (*DiskCleaner) Stop

func (cleaner *DiskCleaner) Stop()

type GCTable

type GCTable struct {
	sync.Mutex
	// contains filtered or unexported fields
}

GCTable is a data structure in memory after consuming checkpoint

func NewGCTable

func NewGCTable() *GCTable

func (*GCTable) Compare

func (t *GCTable) Compare(table *GCTable) bool

For test

func (*GCTable) Merge

func (t *GCTable) Merge(GCTable *GCTable)

Merge can merge two GCTables

func (*GCTable) ReadTable

func (t *GCTable) ReadTable(ctx context.Context, name string, size int64, fs *objectio.ObjectFS, ts types.TS) error

ReadTable reads an s3 file and replays a GCTable in memory

func (*GCTable) SaveFullTable

func (t *GCTable) SaveFullTable(start, end types.TS, fs *objectio.ObjectFS, files []string) ([]objectio.BlockObject, error)

SaveFullTable is to write data to s3

func (*GCTable) SaveTable

func (t *GCTable) SaveTable(start, end types.TS, fs *objectio.ObjectFS, files []string) ([]objectio.BlockObject, error)

SaveTable is to write data to s3

func (*GCTable) SoftGC

func (t *GCTable) SoftGC(
	table *GCTable,
	ts types.TS,
	snapShotList map[uint32]containers.Vector,
	meta *logtail.SnapshotMeta,
) ([]string, map[uint32][]types.TS)

SoftGC is to remove objectentry that can be deleted from GCTable

func (*GCTable) String

func (t *GCTable) String() string

func (*GCTable) UpdateTable

func (t *GCTable) UpdateTable(data *logtail.CheckpointData)

type GCWorker

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

func NewGCWorker

func NewGCWorker(fs *objectio.ObjectFS, cleaner *checkpointCleaner) *GCWorker

func (*GCWorker) ExecDelete

func (g *GCWorker) ExecDelete(ctx context.Context, names []string, disableGC bool) error

func (*GCWorker) Idle

func (g *GCWorker) Idle()

func (*GCWorker) Start

func (g *GCWorker) Start() bool

type ObjectEntry

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

type TombstoneEntry

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