logtail

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: 53 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	TableIDAttr_Account     = "account"
	TableIDAttr_DBID        = "db_id"
	TableIDAttr_TableID     = "table_id"
	TableIDAttr_ObjectStart = "object_start"
	TableIDAttr_ObjectEnd   = "object_end"
)
View Source 
const (
	CKPTableIDBatch_SpecialTableID = 0

	BatchRowCountThreshold = 8192
)
View Source 
const (
	SnapshotTypeIdx types.Enum = iota
	SnapshotTypeCluster
	SnapshotTypeAccount
	SnapshotTypeDatabase
	SnapshotTypeTable
)
View Source 
const (
	ColSnapshotId uint16 = iota
	ColSName
	ColTS
	ColLevel
	ColAccountName
	ColDatabaseName
	ColTableName
	ColObjId
)

mo_snapshot's schema

View Source 
const (
	TableInfoTypeIdx types.Enum = iota
	SnapshotTidIdx
	AObjectDelIdx
	PitrTidIdx
	IscpTidIdx
)
View Source 
const (
	PitrUnitYear   = "y"
	PitrUnitMonth  = "mo"
	PitrUnitDay    = "d"
	PitrUnitHour   = "h"
	PitrUnitMinute = "m"
)
View Source 
const (
	PitrLevelCluster  = "cluster"
	PitrLevelAccount  = "account"
	PitrLevelDatabase = "database"
	PitrLevelTable    = "table"
)
View Source 
const (
	ColPitrId uint16 = iota
	ColPitrName
	ColPitrCreateAccount
	ColPitrCreateTime
	ColPitrModifiedTime
	ColPitrLevel
	ColPitrAccountId
	ColPitrAccountName
	ColPitrDatabaseName
	ColPitrTableName
	ColPitrObjId
	ColPitrLength
	ColPitrUnit
)

pitr's schema

View Source 
const (
	ColIscpAccountId uint16 = iota
	ColIscpTableId
	ColIscpJobName
	ColIscpJobId
	ColIscpJobSpec
	ColIscpJobState
	ColIscpWatermark
	ColIscpJobStatus
	ColIscpCreateAt
	ColIscpDropAt
)

iscp schema

View Source 
const (
	UsageAccID uint8 = iota
	UsageDBID
	UsageTblID
	UsageObjID
	UsageSize

	UsageMAX
)
View Source 
const (
	SnapshotAttr_TID                              = catalog.SnapshotAttr_TID
	SnapshotAttr_DBID                             = catalog.SnapshotAttr_DBID
	ObjectAttr_ObjectStats                        = catalog.ObjectAttr_ObjectStats
	EntryNode_CreateAt                            = catalog.EntryNode_CreateAt
	EntryNode_DeleteAt                            = catalog.EntryNode_DeleteAt
	SnapshotMetaAttr_BlockInsertBatchLocation     = "block_insert_batch_location"
	SnapshotMetaAttr_BlockDeleteBatchLocation     = "block_delete_batch_location"
	SnapshotMetaAttr_DataObjectBatchLocation      = "data_object_batch_location"
	SnapshotMetaAttr_TombstoneObjectBatchLocation = "tombstone_object_batch_location"
	CheckpointMetaAttr_BlockLocation              = "checkpoint_meta_block_location"
	CheckpointMetaAttr_SchemaType                 = "checkpoint_meta_schema_type"

	CheckpointMetaAttr_StorageUsageInsLocation = "checkpoint_meta_storage_usage_ins_location"
	CheckpointMetaAttr_StorageUsageDelLocation = "checkpoint_meta_storage_usage_del_location"

	// supporting `show accounts` in checkpoint
	CheckpointMetaAttr_ObjectSize = "checkpoint_meta_object_size"
	CheckpointMetaAttr_ObjectID   = "checkpoint_meta_object_id"
)
View Source 
const (
	CheckpointVersion12 uint32 = 12
	CheckpointVersion13 uint32 = 13

	CheckpointCurrentVersion = CheckpointVersion13
)
View Source 
const (
	MetaIDX uint16 = iota

	TNMetaIDX

	StorageUsageInsIDX
	ObjectInfoIDX
	StorageUsageDelIDX
	TombstoneObjectInfoIDX
)
View Source 
const (
	Checkpoint_Meta_TID_IDX                  = 2
	Checkpoint_Meta_Insert_Block_LOC_IDX     = 3
	Checkpoint_Meta_Delete_Block_LOC_IDX     = 4
	Checkpoint_Meta_Data_Object_LOC_IDX      = 5
	Checkpoint_Meta_Tombstone_Object_LOC_IDX = 6
	Checkpoint_Meta_Usage_Ins_LOC_IDX        = 7
	Checkpoint_Meta_Usage_Del_LOC_IDX        = 8
)
View Source 
const (
	LocationOffset      = 0
	LocationLength      = objectio.LocationLen
	StartOffsetOffset   = LocationOffset + LocationLength
	StartOffsetLength   = 8
	EndOffsetOffset     = StartOffsetOffset + StartOffsetLength
	EndOffsetLength     = 8
	BlockLocationLength = EndOffsetOffset + EndOffsetLength
)

func (l BlockLocations) append iterator

Location is a fixed-length unmodifiable byte array. Layout: Location(objectio.Location) | StartOffset(uint64) | EndOffset(uint64)

View Source 
const (
	BlockInsert = iota
	BlockDelete
	DataObject
	TombstoneObject
	StorageUsageIns
	StorageUsageDel
)
View Source 
const DefaultCheckpointBlockRows = 10000
View Source 
const DefaultCheckpointSize = 512 * 1024 * 1024
View Source 
const (
	LogtailHeartbeatDuration = time.Millisecond * 2
)
View Source 
const MaxIDX = TombstoneObjectInfoIDX + 1
View Source 
const MetaMaxIdx = StorageUsageDel + 1
View Source 
const MoTablesPK = "mo_tables_pk"
View Source 
const Size90M = 90 * 1024 * 1024
View Source 
const StorageUsageMagic uint64 = 0x1A2B3C4D5E6F
View Source 
const UsageBatMetaTableId uint64 = StorageUsageMagic

Variables

View Source 
var (
	BaseAttr = []string{
		catalog.PhyAddrColumnName,
		objectio.TombstoneAttr_CommitTs_Attr,
	}

	BaseTypes = []types.Type{
		types.T_Rowid.ToType(),
		types.T_TS.ToType(),
	}

	MetaSchema         *catalog.Schema
	TNMetaSchema       *catalog.Schema
	ObjectInfoSchema   *catalog.Schema
	StorageUsageSchema *catalog.Schema

	MetaSchemaAttr = []string{
		SnapshotAttr_TID,
		SnapshotMetaAttr_BlockInsertBatchLocation,
		SnapshotMetaAttr_BlockDeleteBatchLocation,
		SnapshotMetaAttr_DataObjectBatchLocation,
		SnapshotMetaAttr_TombstoneObjectBatchLocation,
		CheckpointMetaAttr_StorageUsageInsLocation,
		CheckpointMetaAttr_StorageUsageDelLocation,
	}

	MetaShcemaTypes = []types.Type{
		types.New(types.T_uint64, 0, 0),
		types.New(types.T_varchar, types.MaxVarcharLen, 0),
		types.New(types.T_varchar, types.MaxVarcharLen, 0),
		types.New(types.T_varchar, types.MaxVarcharLen, 0),
		types.New(types.T_varchar, types.MaxVarcharLen, 0),
		types.New(types.T_varchar, types.MaxVarcharLen, 0),
		types.New(types.T_varchar, types.MaxVarcharLen, 0),
	}

	MetaSchemaSeqnums = []uint16{0, 1, 2, 3, 4, 5, 6}

	MetaSchema_Tid_Idx             = 0
	MetaSchema_DataObject_Idx      = 3
	MetaSchema_TombstoneObject_Idx = 4

	TNMetaSchemaAttr = []string{
		CheckpointMetaAttr_BlockLocation,
		CheckpointMetaAttr_SchemaType,
	}
	TNMetaShcemaTypes = []types.Type{
		types.New(types.T_varchar, types.MaxVarcharLen, 0),
		types.New(types.T_uint16, 0, 0),
	}

	ObjectInfoAttr = []string{
		ObjectAttr_ObjectStats,
		SnapshotAttr_DBID,
		SnapshotAttr_TID,
		EntryNode_CreateAt,
		EntryNode_DeleteAt,
		txnbase.SnapshotAttr_StartTS,
		txnbase.SnapshotAttr_PrepareTS,
		txnbase.SnapshotAttr_CommitTS,
	}
	ObjectInfoTypes = []types.Type{
		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_TS, 0, 0),
		types.New(types.T_TS, 0, 0),
		types.New(types.T_TS, 0, 0),
		types.New(types.T_TS, 0, 0),
		types.New(types.T_TS, 0, 0),
	}
	ObjectInfoSeqnums          = []uint16{0, 1, 2, 3, 4, 5, 6, 7}
	ObjectInfo_ObjectStats_Idx = 0
	ObjectInfo_DBID_Idx        = 1
	ObjectInfo_TID_Idx         = 2
	ObjectInfo_CreateAt_Idx    = 3
	ObjectInfo_DeleteAt_Idx    = 4

	StorageUsageSchemaAttrs = []string{
		pkgcatalog.SystemColAttr_AccID,
		SnapshotAttr_DBID,
		SnapshotAttr_TID,
		CheckpointMetaAttr_ObjectID,
		CheckpointMetaAttr_ObjectSize,
	}

	StorageUsageSchemaTypes = []types.Type{
		types.New(types.T_uint64, 0, 0),
		types.New(types.T_uint64, 0, 0),
		types.New(types.T_uint64, 0, 0),
		types.New(types.T_uuid, 0, 0),
		types.New(types.T_uint64, 0, 0),
	}
)
View Source 
var TableIDAttrs = []string{
	TableIDAttr_Account,
	TableIDAttr_DBID,
	TableIDAttr_TableID,
	TableIDAttr_ObjectStart,
	TableIDAttr_ObjectEnd,
}
View Source 
var TableIDSeqnums = []uint16{0, 1, 2, 3, 4}
View Source 
var TableIDTypes = []types.Type{
	types.T_uint32.ToType(),
	types.T_uint64.ToType(),
	types.T_uint64.ToType(),
	types.T_TS.ToType(),
	types.T_TS.ToType(),
}

Functions

func AddDate 

func AddDate(t time.Time, year, month, day int) time.Time

func BackupCheckpointDataFactory added in v1.1.0 

func BackupCheckpointDataFactory(
	start, end types.TS,
	fs fileservice.FileService,
) func(c *catalog.Catalog) (*CheckpointData_V2, error)

func BatchToString 

func BatchToString(name string, bat *containers.Batch, isSpecialRowID bool) string

func ConsumeCheckpointEntries 

func ConsumeCheckpointEntries(
	ctx context.Context,
	sid string,
	metaLoc string,
	tableID uint64,
	tableName string,
	dbID uint64,
	dbName string,
	forEachObject func(ctx context.Context, fs fileservice.FileService, obj objectio.ObjectEntry, isTombstone bool) (err error),
	mp *mpool.MPool,
	fs fileservice.FileService) (err error)

metaLoc records locaion of checkpoints. In v1, for each checkpoint, it records location and version separated by ';'. e.g. ckp1-location;ckp1-version;ckp2-location;ckp2-version; In v2, it first records the version of the whole metaloc, then follows the same format as v1 for each ckp. e.g. 2;ckp1-location;ckp1-version;ckp2-location;ckp2-version;

func CorrectUsageWrongPlacement added in v1.1.0 

func CorrectUsageWrongPlacement(c *catalog.Catalog) (int, float64, error)

func DataChangeToLogtailBatch added in v0.8.0 

func DataChangeToLogtailBatch(src *containers.BatchWithVersion) *containers.Batch

GetDataWindowForLogtail returns the batch according to the writeSchema. columns are sorted by seqnum and vacancy is filled with zero value

func DebugBatchToString 

func DebugBatchToString(name string, bat *containers.Batch, isSpecialRowID bool, lvl zapcore.Level) string

func FillUsageBatOfCompacted 

func FillUsageBatOfCompacted(
	ctx context.Context,
	usage *TNUsageMemo,
	data *batch.Batch,
	meta *SnapshotMeta,
	snapshots *SnapshotInfo,
	pitrs *PitrInfo,
	_ int,
)

func ForeachTombstoneObject 

func ForeachTombstoneObject(
	onTombstone func(tombstone *objData) (next bool, err error),
	ds map[string]*objData,
) error

func GetObjectsFromCKPMeta 

func GetObjectsFromCKPMeta(
	ctx context.Context,
	reader *CKPReader,
	pinned map[string]bool,
) (err error)

only need to read meta

func GetTableIDsFromCheckpoint 

func GetTableIDsFromCheckpoint(
	ctx context.Context,
	reader *CKPReader,
) (result []uint64, err error)

func GetTombstonesByBlockId 

func GetTombstonesByBlockId(
	bid *objectio.Blockid,
	deleteMask *objectio.Bitmap,
	scanOp func(func(tombstone *objData) (bool, error)) error,
	needShrink bool,
) (err error)

func GlobalCheckpointDataFactory added in v0.7.0 

func GlobalCheckpointDataFactory(
	end types.TS,
	versionInterval time.Duration,
	fs fileservice.FileService,
) func(c *catalog.Catalog) (*CheckpointData_V2, error)

func HandleSyncLogTailReq 

func HandleSyncLogTailReq(
	ctx context.Context,
	ckpClient CheckpointClient,
	mgr *Manager,
	c *catalog.Catalog,
	req api.SyncLogTailReq,
	canRetry bool) (resp api.SyncLogTailResp, closeCB func(), err error)

func IDXString added in v1.1.0 

func IDXString(idx uint16) string

func IncrementalCheckpointDataFactory 

func IncrementalCheckpointDataFactory(
	start, end types.TS,
	size int,
	fs fileservice.FileService,
) func(c *catalog.Catalog) (*CheckpointData_V2, error)

func IsMoTable 

func IsMoTable(tid uint64) bool

func LoadBlkColumnsByMeta added in v0.8.0 

func LoadBlkColumnsByMeta(
	version uint32,
	cxt context.Context,
	colTypes []types.Type,
	colNames []string,
	id uint16,
	reader *ioutil.BlockReader,
	mp *mpool.MPool,
) ([]*containers.Batch, error)

func LoadCheckpointLocations added in v1.2.0 

func LoadCheckpointLocations(
	ctx context.Context,
	sid string,
	reader *CKPReader,
) (map[string]objectio.Location, error)

func MockCallback added in v0.8.0 

func MockCallback(from, to timestamp.Timestamp, closeCB func(), tails ...logtail.TableLogtail) error

func MockCheckpointV12 

func MockCheckpointV12(
	ctx context.Context,
	c *catalog.Catalog,
	start, end types.TS,
	mp *mpool.MPool,
	fs fileservice.FileService,
) (location objectio.Location, err error)

func MockTableIDBatch 

func MockTableIDBatch(
	ctx context.Context,
	start, end types.TS,
	sinkerThreshold int,
	rowCount int,
	mp *mpool.MPool,
	fs fileservice.FileService,
) (locations objectio.LocationSlice, err error)

func NewDirtyCollector 

func NewDirtyCollector(
	sourcer *Manager,
	clock clock.Clock,
	catalog *catalog.Catalog,
	interceptor DirtyEntryInterceptor) *dirtyCollector

func ObjectIsSnapshotRefers 

func ObjectIsSnapshotRefers(
	obj *objectio.ObjectStats,
	pitr, createTS, dropTS *types.TS,
	snapshots []types.TS,
) bool

func PairAccountVsDB added in v1.1.0 

func PairAccountVsDB(c *catalog.Catalog) map[uint64]uint64

func ReWriteCheckpointAndBlockFromKey added in v1.1.0 

func ReWriteCheckpointAndBlockFromKey(
	ctx context.Context,
	sid string,
	fs, dstFs fileservice.FileService,
	loc objectio.Location,
	lastCkpData *CKPReader,
	version uint32, ts types.TS,
) (objectio.Location, objectio.Location, []string, error)

func ReplayCheckpoint 

func ReplayCheckpoint(
	ctx context.Context,
	c *catalog.Catalog,
	forSys bool,
	reader *CKPReader,
) (err error)

func SyncTableIDBatch 

func SyncTableIDBatch(
	ctx context.Context,
	start, end types.TS,
	ttl time.Duration,
	sinkerThreshold int,
	ckpLocation objectio.Location,
	ckpVersion uint32,
	prevTableIDLocation objectio.LocationSlice,
	mp *mpool.MPool,
	fs fileservice.FileService,
) (locations objectio.LocationSlice, err error)

func ToStringTemplate 

func ToStringTemplate(vec containers.Vector, printN int, opts ...common.TypePrintOpt) string

func TombstoneChangeToLogtailBatch 

func TombstoneChangeToLogtailBatch(src *containers.BatchWithVersion) *containers.Batch

Types

type BackupDeltaLocDataSource 

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

func NewBackupDeltaLocDataSource 

func NewBackupDeltaLocDataSource(
	ctx context.Context,
	fs fileservice.FileService,
	ts types.TS,
	ds map[string]*objData,
) *BackupDeltaLocDataSource

func NewSnapshotDataSource 

func NewSnapshotDataSource(
	ctx context.Context,
	fs fileservice.FileService,
	ts types.TS,
	stats []objectio.ObjectStats,
) *BackupDeltaLocDataSource

func (*BackupDeltaLocDataSource) ApplyTombstones 

func (d *BackupDeltaLocDataSource) ApplyTombstones(
	_ context.Context,
	_ *objectio.Blockid,
	_ []int64,
	_ engine.TombstoneApplyPolicy,
) ([]int64, error)

func (*BackupDeltaLocDataSource) Close 

func (d *BackupDeltaLocDataSource) Close()

func (*BackupDeltaLocDataSource) GetOrderBy 

func (d *BackupDeltaLocDataSource) GetOrderBy() []*plan.OrderBySpec

func (*BackupDeltaLocDataSource) GetTombstones 

func (d *BackupDeltaLocDataSource) GetTombstones(
	ctx context.Context, bid *objectio.Blockid,
) (deletedRows objectio.Bitmap, err error)

func (*BackupDeltaLocDataSource) Next 

func (d *BackupDeltaLocDataSource) Next(
	_ context.Context,
	_ []string,
	_ []types.Type,
	_ []uint16,
	_ int32,
	_ any,
	_ *mpool.MPool,
	_ *batch.Batch,
) (*objectio.BlockInfo, engine.DataState, error)

func (*BackupDeltaLocDataSource) SetFilterZM 

func (d *BackupDeltaLocDataSource) SetFilterZM(zm objectio.ZoneMap)

func (*BackupDeltaLocDataSource) SetOrderBy 

func (d *BackupDeltaLocDataSource) SetOrderBy(orderby []*plan.OrderBySpec)

func (*BackupDeltaLocDataSource) SetTS 

func (d *BackupDeltaLocDataSource) SetTS(
	ts types.TS,
)

func (*BackupDeltaLocDataSource) String 

func (d *BackupDeltaLocDataSource) String() string

type BaseCollector_V2 

type BaseCollector_V2 struct {
	*catalog.LoopProcessor
	// contains filtered or unexported fields
}

func NewBackupCollector_V2 

func NewBackupCollector_V2(start, end types.TS, fs fileservice.FileService) *BaseCollector_V2

func NewBaseCollector_V2 

func NewBaseCollector_V2(start, end types.TS, size int, fs fileservice.FileService) *BaseCollector_V2

func (*BaseCollector_V2) Close 

func (collector *BaseCollector_V2) Close()

func (*BaseCollector_V2) Collect 

func (collector *BaseCollector_V2) Collect(c *catalog.Catalog) (err error)

func (*BaseCollector_V2) OrphanData 

func (collector *BaseCollector_V2) OrphanData() *CheckpointData_V2

type BlockLocation added in v1.0.0 

type BlockLocation []byte

func BuildBlockLoaction added in v1.0.0 

func BuildBlockLoaction(id uint16, start, end uint64) BlockLocation

func BuildBlockLoactionWithLocation added in v1.0.0 

func BuildBlockLoactionWithLocation(name objectio.ObjectName, extent objectio.Extent, rows uint32, id uint16, start, end uint64) BlockLocation

func (BlockLocation) Contains added in v1.0.0 

func (l BlockLocation) Contains(i common.ClosedInterval) bool

func (BlockLocation) GetEndOffset added in v1.0.0 

func (l BlockLocation) GetEndOffset() uint64

func (BlockLocation) GetID added in v1.0.0 

func (l BlockLocation) GetID() uint16

func (BlockLocation) GetLocation added in v1.0.0 

func (l BlockLocation) GetLocation() objectio.Location

func (BlockLocation) GetStartOffset added in v1.0.0 

func (l BlockLocation) GetStartOffset() uint64

func (BlockLocation) SetEndOffset added in v1.0.0 

func (l BlockLocation) SetEndOffset(end uint64)

func (BlockLocation) SetID added in v1.0.0 

func (l BlockLocation) SetID(id uint16)

func (BlockLocation) SetLocation added in v1.0.0 

func (l BlockLocation) SetLocation(location objectio.Location)

func (BlockLocation) SetStartOffset added in v1.0.0 

func (l BlockLocation) SetStartOffset(start uint64)

func (BlockLocation) String added in v1.0.0 

func (l BlockLocation) String() string

type BlockLocations added in v1.0.0 

type BlockLocations []byte

func (*BlockLocations) Append added in v1.0.0 

func (l *BlockLocations) Append(loc BlockLocation)

func (BlockLocations) MakeIterator added in v1.0.0 

func (l BlockLocations) MakeIterator() *BlockLocationsIterator

func (BlockLocations) String added in v1.0.0 

func (l BlockLocations) String() string

type BlockLocationsIterator added in v1.0.0 

type BlockLocationsIterator struct {
	*BlockLocations
	// contains filtered or unexported fields
}

func (*BlockLocationsIterator) HasNext added in v1.0.0 

func (i *BlockLocationsIterator) HasNext() bool

func (*BlockLocationsIterator) Next added in v1.0.0 

func (i *BlockLocationsIterator) Next() BlockLocation

type BlockT added in v0.7.0 

type BlockT = *txnBlock

type BoundTableOperator 

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

func (*BoundTableOperator) Report 

func (c *BoundTableOperator) Report() string

func (*BoundTableOperator) Run 

func (c *BoundTableOperator) Run() error

type CKPReader 

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

func GetCheckpointReader 

func GetCheckpointReader(
	ctx context.Context,
	sid string,
	fs fileservice.FileService,
	location objectio.Location,
	version uint32,
) (*CKPReader, error)

func LoadCheckpointEntriesFromKey added in v1.0.0 

func LoadCheckpointEntriesFromKey(
	ctx context.Context,
	sid string,
	fs fileservice.FileService,
	location objectio.Location,
	version uint32,
	softDeletes *map[string]bool,
	baseTS *types.TS,
) ([]*objectio.BackupObject, *CKPReader, error)

func NewCKPReader 

func NewCKPReader(
	version uint32,
	location objectio.Location,
	mp *mpool.MPool,
	fs fileservice.FileService,
) *CKPReader

func NewCKPReaderWithTableID_V2 

func NewCKPReaderWithTableID_V2(
	version uint32,
	location objectio.Location,
	tableID uint64,
	mp *mpool.MPool,
	fs fileservice.FileService,
) *CKPReader

func (*CKPReader) ConsumeCheckpointWithTableID 

func (reader *CKPReader) ConsumeCheckpointWithTableID(
	ctx context.Context,
	forEachObject func(ctx context.Context, fs fileservice.FileService, obj objectio.ObjectEntry, isTombstone bool) (err error),
) (err error)

func (*CKPReader) ForEachRow 

func (reader *CKPReader) ForEachRow(
	ctx context.Context,
	forEachRow func(
		account uint32,
		dbid, tid uint64,
		objectType int8,
		objectStats objectio.ObjectStats,
		create, delete types.TS,
		rowID types.Rowid,
	) error,
) (err error)

func (*CKPReader) GetCheckpointData 

func (reader *CKPReader) GetCheckpointData(ctx context.Context) (ckpData *batch.Batch, err error)

func (*CKPReader) GetLocations 

func (reader *CKPReader) GetLocations() []objectio.Location

func (*CKPReader) GetTableRanges 

func (reader *CKPReader) GetTableRanges(
	ctx context.Context,
) (ranges []ckputil.TableRange, err error)

must called after ReadMeta if withTableID, return from dataRanges and tombstoneRanges else load from reader

func (*CKPReader) LoadBatchData 

func (reader *CKPReader) LoadBatchData(
	ctx context.Context,
	_ []string,
	_ *plan.Expr,
	_ *mpool.MPool,
	data *batch.Batch,
) (end bool, err error)

func (*CKPReader) PrefetchData 

func (reader *CKPReader) PrefetchData(sid string)

func (*CKPReader) ReadMeta 

func (reader *CKPReader) ReadMeta(
	ctx context.Context,
) (err error)

type CheckpointClient 

type CheckpointClient interface {
	CollectCheckpointsInRange(ctx context.Context, start, end types.TS) (ckpLoc string, lastEnd types.TS, err error)
	FlushTable(ctx context.Context, accoutID uint32, dbID, tableID uint64, ts types.TS) error
}

type CheckpointData 

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

type CheckpointData_V2 

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

func NewCheckpointDataWithSinker 

func NewCheckpointDataWithSinker(sinker *ioutil.Sinker, allocator *mpool.MPool) *CheckpointData_V2

func NewCheckpointData_V2 

func NewCheckpointData_V2(
	allocator *mpool.MPool,
	objSize int,
	fs fileservice.FileService,
) *CheckpointData_V2

func (*CheckpointData_V2) Close 

func (data *CheckpointData_V2) Close()

func (*CheckpointData_V2) ExportStats 

func (data *CheckpointData_V2) ExportStats(prefix string) []zap.Field

func (*CheckpointData_V2) Sync 

func (data *CheckpointData_V2) Sync(
	ctx context.Context,
	fs fileservice.FileService,
) (location objectio.Location, ckpfiles []string, err error)

type CheckpointMeta 

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

func NewCheckpointMeta 

func NewCheckpointMeta() *CheckpointMeta

func (*CheckpointMeta) DecodeFromString added in v1.0.0 

func (m *CheckpointMeta) DecodeFromString(keys [][]byte) (err error)

func (*CheckpointMeta) String added in v1.0.0 

func (m *CheckpointMeta) String() string

type Collector 

type Collector interface {
	String() string
	Run(lag time.Duration)
	ScanInRange(from, to types.TS) (*DirtyTreeEntry, int)
	ScanInRangePruned(from, to types.TS) *DirtyTreeEntry
	GetAndRefreshMerged() *DirtyTreeEntry
	Merge() *DirtyTreeEntry
	GetMaxLSN(from, to types.TS) uint64
	Init(maxts types.TS)
}

type DirtyEntryInterceptor 

type DirtyEntryInterceptor = catalog.Processor

type DirtyTreeEntry 

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

func NewDirtyTreeEntry 

func NewDirtyTreeEntry(start, end types.TS, tree *model.Tree) *DirtyTreeEntry

func NewEmptyDirtyTreeEntry 

func NewEmptyDirtyTreeEntry() *DirtyTreeEntry

func (*DirtyTreeEntry) GetTimeRange 

func (entry *DirtyTreeEntry) GetTimeRange() (from, to types.TS)

func (*DirtyTreeEntry) GetTree 

func (entry *DirtyTreeEntry) GetTree() (tree *model.Tree)

func (*DirtyTreeEntry) IsEmpty 

func (entry *DirtyTreeEntry) IsEmpty() bool

func (*DirtyTreeEntry) Merge 

func (entry *DirtyTreeEntry) Merge(o *DirtyTreeEntry)

func (*DirtyTreeEntry) String 

func (entry *DirtyTreeEntry) String() string

type GlobalCollector_V2 

type GlobalCollector_V2 struct {
	BaseCollector_V2
	// contains filtered or unexported fields
}

func NewGlobalCollector_V2 

func NewGlobalCollector_V2(
	fs fileservice.FileService,
	end types.TS,
	versionInterval time.Duration,
) *GlobalCollector_V2

type Logtailer added in v0.7.0 

type Logtailer interface {
	// RangeLogtail returns logtail for all tables within the range (from, to].
	// NOTE: caller should keep time range monotonous, or there would be a checkpoint.
	RangeLogtail(
		ctx context.Context, from, to timestamp.Timestamp,
	) ([]logtail.TableLogtail, []func(), error)

	RegisterCallback(cb func(from, to timestamp.Timestamp, closeCB func(), tails ...logtail.TableLogtail) error)

	// TableLogtail returns logtail for the specified table.
	//
	// NOTE: If table not exist, logtail.TableLogtail shouldn't be a simple zero value.
	TableLogtail(
		ctx context.Context, table api.TableID, from, to timestamp.Timestamp,
	) (logtail.TableLogtail, func(), error)

	// Now is a time getter from TxnManager. Users of Logtailer should get a timestamp
	// from Now and use the timestamp to collect logtail, in that case, all txn prepared
	// before it are visible.
	Now() (timestamp.Timestamp, timestamp.Timestamp)
}

Logtailer provides logtail for the specified table.

type LogtailerImpl added in v0.7.0 

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

func NewLogtailer added in v0.7.0 

func NewLogtailer(
	ctx context.Context,
	ckpClient CheckpointClient,
	mgr *Manager,
	c *catalog.Catalog) *LogtailerImpl

func (*LogtailerImpl) Now added in v0.8.0 

func (l *LogtailerImpl) Now() (timestamp.Timestamp, timestamp.Timestamp)

Now is a time getter from TxnManager. Users of Logtailer should get a timestamp from Now and use the timestamp to collect logtail, in that case, all txn prepared before it are visible.

func (*LogtailerImpl) RangeLogtail added in v0.7.0 

func (l *LogtailerImpl) RangeLogtail(
	ctx context.Context, from, to timestamp.Timestamp,
) ([]logtail.TableLogtail, []func(), error)

RangeLogtail returns logtail for all tables that are modified within the range (from, to]. Check out all dirty tables in the time window and collect logtails for every table

func (*LogtailerImpl) RegisterCallback added in v0.8.0 

func (l *LogtailerImpl) RegisterCallback(cb func(from, to timestamp.Timestamp, closeCB func(), tails ...logtail.TableLogtail) error)

func (*LogtailerImpl) TableLogtail added in v0.7.0 

func (l *LogtailerImpl) TableLogtail(
	ctx context.Context, table api.TableID, from, to timestamp.Timestamp,
) (logtail.TableLogtail, func(), error)

TableLogtail returns logtail for the specified table. It boils down to calling `HandleSyncLogTailReq`

type Manager added in v0.7.0 

type Manager struct {
	txnbase.NoopCommitListener
	// contains filtered or unexported fields
}

Logtail manager holds sorted txn handles. Its main jobs:

- Insert new txn handle - Efficiently iterate over arbitrary range of txn handles on a snapshot - Truncate unneceessary txn handles according to GC timestamp

func NewManager added in v0.7.0 

func NewManager(
	rt *dbutils.Runtime,
	blockSize int,
	nowClock func() types.TS,
) *Manager

func (*Manager) GCByTS added in v0.7.0 

func (mgr *Manager) GCByTS(ctx context.Context, ts types.TS) (updated bool)

func (*Manager) GetMaxCommittedLSN 

func (mgr *Manager) GetMaxCommittedLSN() uint64

func (*Manager) GetReader added in v0.7.0 

func (mgr *Manager) GetReader(from, to types.TS) *Reader

GetReader get a snapshot of all txn prepared between from and to.

func (*Manager) GetTableOperator added in v0.7.0 

func (mgr *Manager) GetTableOperator(
	from, to types.TS,
	tableEntry *catalog.TableEntry,
	visitor *TableLogtailRespBuilder,
) *BoundTableOperator

func (*Manager) GetTruncateTS 

func (mgr *Manager) GetTruncateTS() types.TS

func (*Manager) OnEndPrePrepare added in v0.7.0 

func (mgr *Manager) OnEndPrePrepare(txn txnif.AsyncTxn)

OnEndPrePrepare is a listener for TxnManager. When a txn completes PrePrepare, add it to the logtail manager

func (*Manager) OnEndPrepareWAL added in v0.8.0 

func (mgr *Manager) OnEndPrepareWAL(txn txnif.AsyncTxn)

func (*Manager) RegisterCallback added in v0.8.0 

func (mgr *Manager) RegisterCallback(cb func(from, to timestamp.Timestamp, closeCB func(), tails ...logtail.TableLogtail) error) error

func (*Manager) Start added in v0.8.0 

func (mgr *Manager) Start()

func (*Manager) Stop added in v0.8.0 

func (mgr *Manager) Stop()

func (*Manager) TryCompactTable added in v1.0.0 

func (mgr *Manager) TryCompactTable()

func (*Manager) UpdateMaxCommittedLSN 

func (mgr *Manager) UpdateMaxCommittedLSN(lsn uint64)

type ObjectAbstract added in v1.2.3 

type ObjectAbstract struct {
	TotalObjCnt  int
	TotalObjSize int
	TotalBlkCnt  int
	TotalRowCnt  int
}

type ObjectInfoJson 

type ObjectInfoJson struct {
	TableCnt     int    `json:"table_count,omitempty"`
	ObjectCnt    uint64 `json:"object_count"`
	ObjectAddCnt uint64 `json:"object_add_count"`
	ObjectDelCnt uint64 `json:"object_del_count"`
	TombstoneCnt int    `json:"tombstone_count"`

	Tables []TableInfoJson `json:"tables,omitempty"`
}

func GetCheckpointMetaInfo 

func GetCheckpointMetaInfo(
	ctx context.Context,
	id uint64,
	reader *CKPReader,
) (res *ObjectInfoJson, err error)

type PitrInfo 

type PitrInfo = SnapshotInfo

PitrInfo is an alias for backward compatibility

func NewPitrInfo 

func NewPitrInfo() *PitrInfo

type Reader added in v0.7.0 

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

Reader is a snapshot of all txn prepared between from and to. Dirty tables/objects/blocks can be queried based on those txn

func (*Reader) GetDirty added in v0.7.0 

func (r *Reader) GetDirty() (tree *model.Tree, count int)

Merge all dirty table/object/block into one dirty tree

func (*Reader) GetMaxLSN added in v0.7.0 

func (r *Reader) GetMaxLSN() (maxLsn uint64)

TODO: optimize

func (*Reader) IsDirtyOnTable 

func (r *Reader) IsDirtyOnTable(DbID, id uint64) bool

type RowT added in v0.7.0 

type RowT = *txnRow

type SnapshotInfo 

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

SnapshotInfo represents snapshot information at different levels Shared structure for both PITR and Snapshot functionality

func NewSnapshotInfo 

func NewSnapshotInfo() *SnapshotInfo

func (*SnapshotInfo) GetSnapshotsByLevel 

func (p *SnapshotInfo) GetSnapshotsByLevel(level string, objID uint64) []types.TS

GetSnapshotsByLevel returns all snapshots for a specific level and object ID

func (*SnapshotInfo) GetTS 

func (p *SnapshotInfo) GetTS(
	accountID uint32,
	dbID uint64,
	tableID uint64,
) (ts types.TS)

GetTS returns the earliest applicable timestamp for PITR usage For PITR, we only need the first (earliest) timestamp from each level

func (*SnapshotInfo) IsEmpty 

func (p *SnapshotInfo) IsEmpty() bool

func (*SnapshotInfo) MinTS 

func (p *SnapshotInfo) MinTS() (ts types.TS)

func (*SnapshotInfo) ToTsList 

func (p *SnapshotInfo) ToTsList() []types.TS

type SnapshotMeta added in v1.2.0 

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

func NewSnapshotMeta added in v1.2.0 

func NewSnapshotMeta() *SnapshotMeta

func (*SnapshotMeta) AccountToTableSnapshots 

func (sm *SnapshotMeta) AccountToTableSnapshots(
	snapshots *SnapshotInfo,
	pitr *PitrInfo,
) (
	tableSnapshots map[uint64][]types.TS,
	tablePitrs map[uint64]*types.TS,
)

AccountToTableSnapshots returns a map from table id to its snapshots. The snapshots parameter contains all levels of snapshots. The pitr is the pitr info.

func (*SnapshotMeta) GetAccountId 

func (sm *SnapshotMeta) GetAccountId(tid uint64) (uint32, bool)

func (*SnapshotMeta) GetAllTableIDs 

func (sm *SnapshotMeta) GetAllTableIDs() map[uint64]bool

GetAllTableIDs returns a copy of all table IDs in the snapshot meta

func (*SnapshotMeta) GetISCP 

func (sm *SnapshotMeta) GetISCP(
	ctx context.Context,
	sid string,
	fs fileservice.FileService,
	mp *mpool.MPool,
) (map[uint64]types.TS, error)

func (*SnapshotMeta) GetPITR 

func (sm *SnapshotMeta) GetPITR(
	ctx context.Context,
	sid string,
	gcTime time.Time,
	fs fileservice.FileService,
	mp *mpool.MPool,
) (*PitrInfo, error)

func (*SnapshotMeta) GetPitrByTable 

func (sm *SnapshotMeta) GetPitrByTable(
	pitr *PitrInfo, dbID, tableID uint64,
) *types.TS

func (*SnapshotMeta) GetSnapshot added in v1.2.0 

func (sm *SnapshotMeta) GetSnapshot(
	ctx context.Context,
	sid string,
	fs fileservice.FileService,
	mp *mpool.MPool,
	extraClusterTS ...types.TS,
) (*SnapshotInfo, error)

func (*SnapshotMeta) GetSnapshotListLocked 

func (sm *SnapshotMeta) GetSnapshotListLocked(snapshots *SnapshotInfo, tid uint64) []types.TS

func (*SnapshotMeta) GetTableDropAt 

func (sm *SnapshotMeta) GetTableDropAt(tid uint64) (types.TS, bool)

func (*SnapshotMeta) GetTablePK 

func (sm *SnapshotMeta) GetTablePK(tid uint64) string

for test

func (*SnapshotMeta) InitTableInfo added in v1.2.0 

func (sm *SnapshotMeta) InitTableInfo(
	ctx context.Context,
	fs fileservice.FileService,
	data *CKPReader,
	startts, endts types.TS,
)

func (*SnapshotMeta) MergeTableInfo added in v1.2.0 

func (sm *SnapshotMeta) MergeTableInfo(
	snapshots *SnapshotInfo,
	pitr *PitrInfo,
) error

func (*SnapshotMeta) ReadMeta added in v1.2.0 

func (sm *SnapshotMeta) ReadMeta(ctx context.Context, name string, fs fileservice.FileService) error

func (*SnapshotMeta) ReadTableInfo added in v1.2.0 

func (sm *SnapshotMeta) ReadTableInfo(ctx context.Context, name string, fs fileservice.FileService) error

func (*SnapshotMeta) Rebuild added in v1.2.0 

func (sm *SnapshotMeta) Rebuild(
	ins *containers.Batch,
	objects *map[uint64]map[objectio.Segmentid]*objectInfo,
	objects2 *map[objectio.Segmentid]*objectInfo,
	objects3 *map[objectio.Segmentid]*objectInfo,
)

func (*SnapshotMeta) RebuildAObjectDel 

func (sm *SnapshotMeta) RebuildAObjectDel(ins *containers.Batch)

func (*SnapshotMeta) RebuildIscp 

func (sm *SnapshotMeta) RebuildIscp(ins *containers.Batch)

func (*SnapshotMeta) RebuildPitr 

func (sm *SnapshotMeta) RebuildPitr(ins *containers.Batch)

func (*SnapshotMeta) RebuildTableInfo added in v1.2.0 

func (sm *SnapshotMeta) RebuildTableInfo(ins *containers.Batch)

func (*SnapshotMeta) RebuildTid added in v1.2.1 

func (sm *SnapshotMeta) RebuildTid(ins *containers.Batch)

func (*SnapshotMeta) SaveMeta added in v1.2.0 

func (sm *SnapshotMeta) SaveMeta(name string, fs fileservice.FileService) (uint32, error)

func (*SnapshotMeta) SaveTableInfo added in v1.2.0 

func (sm *SnapshotMeta) SaveTableInfo(name string, fs fileservice.FileService) (uint32, error)

func (*SnapshotMeta) SetTid added in v1.2.0 

func (sm *SnapshotMeta) SetTid(tid uint64)

func (*SnapshotMeta) String added in v1.2.0 

func (sm *SnapshotMeta) String() string

func (*SnapshotMeta) TableInfoString added in v1.2.0 

func (sm *SnapshotMeta) TableInfoString() string

func (*SnapshotMeta) Update added in v1.2.0 

func (sm *SnapshotMeta) Update(
	ctx context.Context,
	fs fileservice.FileService,
	data *CKPReader,
	startts, endts types.TS,
	taskName string,
) (err error)

type StorageUsageCache added in v1.1.0 

type StorageUsageCache struct {
	// when two requests happens within [lastUpdate, lastUpdate + lazyThreshold],
	// it will reuse the cached result, no new query to TN.
	sync.Mutex
	// contains filtered or unexported fields
}

func NewStorageUsageCache added in v1.1.0 

func NewStorageUsageCache(opts ...StorageUsageCacheOption) *StorageUsageCache

func (*StorageUsageCache) CacheLen added in v1.1.0 

func (c *StorageUsageCache) CacheLen() int

func (*StorageUsageCache) ClearForUpdate added in v1.1.0 

func (c *StorageUsageCache) ClearForUpdate()

func (*StorageUsageCache) Delete added in v1.1.0 

func (c *StorageUsageCache) Delete(usage UsageData)

func (*StorageUsageCache) GatherAccountSize added in v1.1.0 

func (c *StorageUsageCache) GatherAccountSize(id uint64) (size, snapshotSize uint64, exist bool)

func (*StorageUsageCache) GatherAllAccSize added in v1.1.0 

func (c *StorageUsageCache) GatherAllAccSize() (usages map[uint64][]uint64)

func (*StorageUsageCache) GatherObjectAbstractForAccounts added in v1.2.3 

func (c *StorageUsageCache) GatherObjectAbstractForAccounts() (abstract map[uint64]ObjectAbstract)

func (*StorageUsageCache) Get added in v1.1.0 

func (c *StorageUsageCache) Get(usage UsageData) (ret UsageData, exist bool)

func (*StorageUsageCache) IsExpired added in v1.1.0 

func (c *StorageUsageCache) IsExpired() bool

func (*StorageUsageCache) Iter added in v1.1.0 

func (c *StorageUsageCache) Iter() btree.IterG[UsageData]

func (*StorageUsageCache) LessFunc added in v1.1.0 

func (c *StorageUsageCache) LessFunc() func(a UsageData, b UsageData) bool

func (*StorageUsageCache) MemUsed added in v1.1.0 

func (c *StorageUsageCache) MemUsed() float64

MemUsed returns the memory used in megabytes

func (*StorageUsageCache) SetOrReplace added in v1.1.1 

func (c *StorageUsageCache) SetOrReplace(usage UsageData)

func (*StorageUsageCache) String added in v1.1.0 

func (c *StorageUsageCache) String() string

type StorageUsageCacheOption added in v1.1.0 

type StorageUsageCacheOption = func(c *StorageUsageCache)

func WithLazyThreshold added in v1.1.0 

func WithLazyThreshold(lazy int) StorageUsageCacheOption

WithLazyThreshold sets lazyThreshold to lazy seconds

func WithLessFunc added in v1.1.0 

func WithLessFunc(less func(a UsageData, b UsageData) bool) StorageUsageCacheOption

type SyncTableIDReader 

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

func NewSyncTableIDReader 

func NewSyncTableIDReader(
	locations objectio.LocationSlice,
	mp *mpool.MPool,
	fs fileservice.FileService,
) (reader *SyncTableIDReader, err error)

func (*SyncTableIDReader) Read 

func (reader *SyncTableIDReader) Read(ctx context.Context) (release func(), bat *batch.Batch, isEnd bool, err error)

type TNUsageMemo added in v1.1.0 

type TNUsageMemo struct {
	sync.Mutex

	C *catalog.Catalog
	// contains filtered or unexported fields
}

func NewTNUsageMemo added in v1.1.0 

func NewTNUsageMemo(c *catalog.Catalog) *TNUsageMemo

func (*TNUsageMemo) AddReqTrace added in v1.1.0 

func (m *TNUsageMemo) AddReqTrace(accountId uint64, tSize uint64, t time.Time, hint string)

func (*TNUsageMemo) CacheLen added in v1.1.0 

func (m *TNUsageMemo) CacheLen() int

func (*TNUsageMemo) Clear added in v1.1.0 

func (m *TNUsageMemo) Clear()

func (*TNUsageMemo) ClearDroppedAccounts added in v1.1.0 

func (m *TNUsageMemo) ClearDroppedAccounts(reserved map[uint64]struct{}) string

func (*TNUsageMemo) ClearNewAccCache added in v1.1.0 

func (m *TNUsageMemo) ClearNewAccCache()

func (*TNUsageMemo) Delete added in v1.1.0 

func (m *TNUsageMemo) Delete(usage UsageData)

func (*TNUsageMemo) DeltaUpdate added in v1.1.1 

func (m *TNUsageMemo) DeltaUpdate(delta UsageData, del bool)

DeltaUpdate does setting or updating with delta size (delta.Size)

func (*TNUsageMemo) EnterProcessing added in v1.1.0 

func (m *TNUsageMemo) EnterProcessing()

func (*TNUsageMemo) EstablishFromCKPs added in v1.1.0 

func (m *TNUsageMemo) EstablishFromCKPs()

EstablishFromCKPs replays usage info which stored in ckps into the tn cache

func (*TNUsageMemo) GatherAccountSize added in v1.1.0 

func (m *TNUsageMemo) GatherAccountSize(id uint64) (size, snapshotSize uint64, exist bool)

func (*TNUsageMemo) GatherAllAccSize added in v1.1.0 

func (m *TNUsageMemo) GatherAllAccSize() (usages map[uint64][]uint64)

func (*TNUsageMemo) GatherNewAccountSize added in v1.1.0 

func (m *TNUsageMemo) GatherNewAccountSize(id uint64) (size, snapshotSize uint64, exist bool)

func (*TNUsageMemo) GatherObjectAbstractForAllAccount added in v1.2.3 

func (m *TNUsageMemo) GatherObjectAbstractForAllAccount() map[uint64]ObjectAbstract

func (*TNUsageMemo) GatherSpecialTableSize added in v1.2.1 

func (m *TNUsageMemo) GatherSpecialTableSize() (size uint64)

func (*TNUsageMemo) Get added in v1.1.0 

func (m *TNUsageMemo) Get(usage UsageData) (old UsageData, exist bool)

func (*TNUsageMemo) GetAllReqTrace added in v1.1.0 

func (m *TNUsageMemo) GetAllReqTrace() (accountIds []uint64, timestamps []time.Time, sizes []uint64, hints []string)

func (*TNUsageMemo) GetCache added in v1.1.0 

func (m *TNUsageMemo) GetCache() *StorageUsageCache

func (*TNUsageMemo) GetDelayed added in v1.1.0 

func (m *TNUsageMemo) GetDelayed() map[uint64]UsageData

func (*TNUsageMemo) GetNewAccCacheLatestUpdate added in v1.1.0 

func (m *TNUsageMemo) GetNewAccCacheLatestUpdate() types.TS

func (*TNUsageMemo) HasUpdate added in v1.1.0 

func (m *TNUsageMemo) HasUpdate() bool

func (*TNUsageMemo) LeaveProcessing added in v1.1.0 

func (m *TNUsageMemo) LeaveProcessing()

func (*TNUsageMemo) MemoryUsed added in v1.1.0 

func (m *TNUsageMemo) MemoryUsed() float64

func (*TNUsageMemo) PrepareReplay added in v1.1.0 

func (m *TNUsageMemo) PrepareReplay(datas []*CheckpointData, vers []uint32)

func (*TNUsageMemo) Replace added in v1.1.1 

func (m *TNUsageMemo) Replace(new UsageData)

Replace replaces the old usage with newUsage

func (*TNUsageMemo) UpdateNewAccCache added in v1.1.0 

func (m *TNUsageMemo) UpdateNewAccCache(usage UsageData, del bool)

type TableInfoJson 

type TableInfoJson struct {
	ID             uint64 `json:"id"`
	Add            uint64 `json:"add,omitempty"`
	Delete         uint64 `json:"delete,omitempty"`
	TombstoneRows  uint64 `json:"tombstone_rows,omitempty"`
	TombstoneCount uint64 `json:"tombstone_count,omitempty"`
}

type TableLogtailRespBuilder 

type TableLogtailRespBuilder struct {
	*catalog.LoopProcessor
	// contains filtered or unexported fields
}

func NewTableLogtailRespBuilder 

func NewTableLogtailRespBuilder(ctx context.Context, ckp string, start, end types.TS, tbl *catalog.TableEntry) *TableLogtailRespBuilder

func (*TableLogtailRespBuilder) BuildResp 

func (b *TableLogtailRespBuilder) BuildResp() (api.SyncLogTailResp, error)

func (*TableLogtailRespBuilder) Close 

func (b *TableLogtailRespBuilder) Close()

func (*TableLogtailRespBuilder) VisitObj added in v1.1.0 

func (b *TableLogtailRespBuilder) VisitObj(e *catalog.ObjectEntry) error

type TableMeta added in v1.0.0 

type TableMeta struct {
	common.ClosedInterval
	// contains filtered or unexported fields
}

func NewTableMeta added in v1.0.0 

func NewTableMeta() *TableMeta

func (*TableMeta) String added in v1.0.0 

func (m *TableMeta) String() string

type TableRespKind added in v0.8.0 

type TableRespKind int
const (
	TableRespKind_Data TableRespKind = iota
	TableRespKind_DataMeta
	TableRespKind_TombstoneMeta
)

type TempFKey added in v1.0.0 

type TempFKey struct{}

type TempFilter added in v1.0.0 

type TempFilter struct {
	sync.RWMutex
	// contains filtered or unexported fields
}
var TempF *TempFilter

func (*TempFilter) Add added in v1.0.0 

func (f *TempFilter) Add(id uint64)

func (*TempFilter) Check added in v1.0.0 

func (f *TempFilter) Check(id uint64) (skip bool)

type TxnLogtailRespBuilder added in v0.8.0 

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

func NewTxnLogtailRespBuilder added in v0.8.0 

func NewTxnLogtailRespBuilder(rt *dbutils.Runtime) *TxnLogtailRespBuilder

func (*TxnLogtailRespBuilder) BuildResp added in v0.8.0 

func (b *TxnLogtailRespBuilder) BuildResp()

func (*TxnLogtailRespBuilder) Close added in v0.8.0 

func (b *TxnLogtailRespBuilder) Close()

func (*TxnLogtailRespBuilder) CollectLogtail added in v0.8.0 

func (b *TxnLogtailRespBuilder) CollectLogtail(txn txnif.AsyncTxn) (*[]logtail.TableLogtail, func())

type TxnTable added in v0.7.0 

type TxnTable struct {
	*model.AOT[BlockT, RowT]
}

func NewTxnTable added in v0.7.0 

func NewTxnTable(blockSize int, nowClock func() types.TS) *TxnTable

func (*TxnTable) AddTxn added in v0.7.0 

func (table *TxnTable) AddTxn(txn txnif.AsyncTxn) (err error)

func (*TxnTable) ForeachRowInBetween added in v0.7.0 

func (table *TxnTable) ForeachRowInBetween(
	from, to types.TS,
	skipBlkOp func(blk BlockT) bool,
	postBlkOp func(blk BlockT),
	rowOp func(row RowT) (goNext bool),
) (readRows int)

func (*TxnTable) TruncateByTimeStamp added in v0.7.0 

func (table *TxnTable) TruncateByTimeStamp(ts types.TS) (cnt int)

func (*TxnTable) TryCompact added in v1.0.0 

func (table *TxnTable) TryCompact(from types.TS, rt *dbutils.Runtime) (to types.TS)

type UsageData added in v1.0.1 

type UsageData struct {
	AccId        uint64
	DbId         uint64
	TblId        uint64
	Size         uint64
	SnapshotSize uint64

	// this will not persist
	// only global ckp will update
	ObjectAbstract
	// contains filtered or unexported fields
}

func MockUsageData added in v1.1.0 

func MockUsageData(accCnt, dbCnt, tblCnt int, allocator *atomic.Uint64) (result []UsageData)

MockUsageData generates accCnt * dbCnt * tblCnt UsageDatas. the accIds, dbIds and tblIds are random produced. this func ensure that all ids are different.

func Objects2Usages added in v1.2.3 

func Objects2Usages(objs []*catalog.ObjectEntry, isGlobal bool) (usages []UsageData)

func (UsageData) IsZero added in v1.1.0 

func (u UsageData) IsZero() bool

func (*UsageData) Merge added in v1.2.3 

func (u *UsageData) Merge(other UsageData, delete bool)

func (UsageData) String added in v1.1.0 

func (u UsageData) String() string

Source Files

View all Source files

Directories

Path Synopsis
This package implements client and server for logtail push model.
 This package implements client and server for logtail push model.

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.