disttae

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

Documentation

Index

Constants

View Source 
const (
	TableStatsTableSize = iota
	TableStatsTableRows

	TableStatsTObjectCnt
	TableStatsDObjectCnt

	TableStatsTBlockCnt
	TableStatsDBlockCnt

	TableStatsCnt
)
View Source 
const (
	// MinExecutorConcurrency is the minimum concurrency for concurrentExecutor
	// which handles IO-intensive tasks (reading S3 objects).
	MinExecutorConcurrency = 32

	// MaxExecutorConcurrency is the maximum concurrency for concurrentExecutor
	// to avoid extreme cases in high-core systems.
	MaxExecutorConcurrency = 108

	// MinWorkerConcurrency is the minimum concurrency for updateWorker
	// which handles table-level update requests (coordinator role).
	MinWorkerConcurrency = 16

	// WorkerConcurrencyRatio is the ratio of updateWorker concurrency to executor concurrency.
	// updateWorker concurrency = executorConcurrency / WorkerConcurrencyRatio
	WorkerConcurrencyRatio = 4

	// SamplingThreshold is the minimum number of objects to enable sampling.
	// Below this threshold, full scan is used for accuracy.
	SamplingThreshold = 100

	// MinSampleObjects is the minimum number of objects to sample.
	// Set equal to SamplingThreshold to ensure sampling count never drops below full scan count.
	MinSampleObjects = 100

	// MaxSampleObjects is the maximum number of objects to sample.
	// Raise to allow large tables (>5w objects) to reach ~1-2% sampling.
	MaxSampleObjects = 5000
)
View Source 
const (
	// LargeTableThreshold is the object count threshold to classify a table as large.
	// Tables with fewer objects are considered small tables.
	LargeTableThreshold = 500

	// LargeTableChangeRateThreshold is the minimum change rate to trigger stats update for large tables.
	// Change rate = pendingChanges / baseObjectCount
	LargeTableChangeRateThreshold = 0.05 // 5%

	// LargeTableMaxUpdateInterval is the maximum interval between stats updates for large tables.
	// Even if change rate is below threshold, update will be triggered after this interval.
	LargeTableMaxUpdateInterval = 30 * time.Minute
)
View Source 
const (
	PREFETCH_THRESHOLD  = 256
	PREFETCH_ROUNDS     = 24
	SMALLSCAN_THRESHOLD = 150
	LARGESCAN_THRESHOLD = 1500
)
View Source 
const (
	INSERT = iota
	DELETE
	ALTER              // alter command for TN. Update batches for mo_tables and mo_columns will fall into the category of INSERT and DELETE.
	SOFT_DELETE_OBJECT // soft delete object command for TN
)
View Source 
const (
	MO_DATABASE_ID_NAME_IDX       = 1
	MO_DATABASE_ID_ACCOUNT_IDX    = 2
	MO_DATABASE_LIST_ACCOUNT_IDX  = 1
	MO_TABLE_ID_NAME_IDX          = 1
	MO_TABLE_ID_DATABASE_ID_IDX   = 2
	MO_TABLE_ID_ACCOUNT_IDX       = 3
	MO_TABLE_LIST_DATABASE_ID_IDX = 1
	MO_TABLE_LIST_ACCOUNT_IDX     = 2
	MO_PRIMARY_OFF                = 2
	INIT_ROWID_OFFSET             = math.MaxUint32
)
View Source 
const (
	CommitWorkspaceThreshold       uint64 = 1 * mpool.MB
	WriteWorkspaceThreshold        uint64 = 5 * mpool.MB
	ExtraWorkspaceThreshold        uint64 = 500 * mpool.MB
	InsertEntryThreshold                  = 5000
	GCBatchOfFileCount             int    = 1000
	GCPoolSize                     int    = 5
	CNTransferTxnLifespanThreshold        = time.Second * 5
)
View Source 
const (
	AllColumns = "*"
)
View Source 
const DefaultLoadParallism = 20
View Source 
const StreamReaderLease = time.Minute * 2

Variables

View Source 
var (
	GcCycle = 10 * time.Second
)
View Source 
var (
	// MinUpdateInterval is the minimal interval to update stats info as it
	// is necessary to update stats every time.
	MinUpdateInterval = time.Second * 15
)
View Source 
var NewPartitionStateChangesHandler = logtailreplay.NewChangesHandler

NewPartitionStateChangesHandler is the function used to create a ChangeHandler from the partition state. It is a variable so tests can stub it.

View Source 
var RequestSnapshotRead = func(ctx context.Context, tbl *txnTable, snapshot *types.TS) (resp any, err error) {
	whichTN := func(string) ([]uint64, error) { return nil, nil }
	payload := func(tnShardID uint64, parameter string, proc *process.Process) ([]byte, error) {
		req := cmd_util.SnapshotReadReq{}
		ts := snapshot.ToTimestamp()
		req.Snapshot = &ts

		return req.MarshalBinary()
	}

	responseUnmarshaler := func(payload []byte) (any, error) {
		checkpoints := &cmd_util.SnapshotReadResp{}
		if err = checkpoints.UnmarshalBinary(payload); err != nil {
			return nil, err
		}
		return checkpoints, nil
	}
	tableProc := tbl.proc.Load()
	proc := process.NewTopProcess(ctx, tableProc.GetMPool(),
		tableProc.Base.TxnClient, tableProc.GetTxnOperator(),
		tableProc.Base.FileService, tableProc.Base.LockService,
		tableProc.Base.QueryClient, tableProc.Base.Hakeeper,
		tableProc.Base.UdfService, tableProc.Base.Aicm,
		tableProc.Base.TaskService,
	)
	handler := ctl.GetTNHandlerFunc(api.OpCode_OpSnapshotRead, whichTN, payload, responseUnmarshaler)
	result, err := handler(proc, "DN", "", ctl.MoCtlTNCmdSender)
	sleep, sarg, exist := fault.TriggerFault("snapshot_read_timeout")
	if exist {
		time.Sleep(time.Duration(sleep) * time.Second)
		return nil, moerr.NewInternalError(ctx, sarg)
	}
	if moerr.IsMoErrCode(err, moerr.ErrNotSupported) {

		return nil, moerr.NewNotSupportedNoCtx("current tn version not supported `snapshot read`")
	}
	if err != nil {
		return nil, err
	}
	if len(result.Data.([]any)) == 0 {
		return cmd_util.SnapshotReadResp{Succeed: false}, nil
	}
	return result.Data.([]any)[0], nil
}
View Source 
var TableStatsName = [TableStatsCnt]string{
	"table_size",
	"table_rows",
	"tobject_cnt",
	"dobject_cnt",
	"tblock_cnt",
	"dblock_cnt",
}

Functions

func BuildLocalDataSource 

func BuildLocalDataSource(
	ctx context.Context,
	rel engine.Relation,
	ranges engine.RelData,
	txnOffset int,
) (source engine.DataSource, err error)

for ut

func CheckTxnIsValid added in v1.2.1 

func CheckTxnIsValid(txnOp client.TxnOperator) (err error)

func CollectAndCalculateStats 

func CollectAndCalculateStats(ctx context.Context, req *updateStatsRequest, executor ConcurrentExecutor) (float64, error)

CollectAndCalculateStats is the main function to calculate and update the stats for scan node. Returns the actual sampling ratio used (sampledObjects / totalObjects).

func ConstructObjStatsByLoadObjMeta added in v1.1.0 

func ConstructObjStatsByLoadObjMeta(
	ctx context.Context,
	metaLoc objectio.Location,
	fs fileservice.FileService,
) (stats objectio.ObjectStats, dataMeta objectio.ObjectDataMeta, err error)

func DefaultNewRpcStreamToTnLogTailService 

func DefaultNewRpcStreamToTnLogTailService(
	ctx context.Context,
	sid string,
	serviceAddr string,
	rpcClient morpc.RPCClient,
) (morpc.RPCClient, morpc.Stream, error)

XXX generate a rpc client and new a stream. we should hide these code into service's NewClient method next day.

func ForeachCommittedObjects added in v1.2.0 

func ForeachCommittedObjects(
	createObjs map[objectio.ObjectNameShort]struct{},
	delObjs map[objectio.ObjectNameShort]struct{},
	p *logtailreplay.PartitionState,
	onObj func(info objectio.ObjectEntry) error) (err error)

func ForeachSnapshotObjects added in v1.1.0 

func ForeachSnapshotObjects(
	ts timestamp.Timestamp,
	onObject func(obj objectio.ObjectEntry, isCommitted bool) error,
	tableSnapshot *logtailreplay.PartitionState,
	extraCommitted []objectio.ObjectStats,
	uncommitted ...objectio.ObjectStats,
) (err error)

func ForeachTombstoneObject 

func ForeachTombstoneObject(
	ts types.TS,
	onTombstone func(tombstone objectio.ObjectEntry) (next bool, err error),
	pState *logtailreplay.PartitionState,
) error

func ForeachVisibleObjects 

func ForeachVisibleObjects(
	state *logtailreplay.PartitionState,
	ts types.TS,
	fn func(obj objectio.ObjectEntry) error,
	executor ConcurrentExecutor,
	visitTombstone bool,
) (err error)

func GetChangedTableList 

func GetChangedTableList(
	ctx context.Context,
	service string,
	eng engine.Engine,
	accs []uint64,
	dbs []uint64,
	tbls []uint64,
	ts []timestamp.Timestamp,
	to *types.TS,
	minTS *types.TS,
	typ cmd_util.ChangedListType,
	forEachTable func(
		accountID int64,
		databaseID int64,
		tableID int64,
		tableName string,
		dbName string,
		relKind string,
		pkSequence int,
		snapshot types.TS,
	),
	handleFn func(
		ctx context.Context,
		meta txn.TxnMeta,
		req *cmd_util.GetChangedTableListReq,
		resp *cmd_util.GetChangedTableListResp,
	) (func(), error),
) (err error)

func GetMOTableStatsExecutor 

func GetMOTableStatsExecutor(
	service string,
	eng engine.Engine,
	sqlExecutor func() ie.InternalExecutor,
) func(ctx context.Context, task task.Task) error

func GetPartitionStateStart 

func GetPartitionStateStart(
	ctx context.Context,
	rel engine.Relation,
) (types.TS, error)

func GetSnapshotReadFnWithHandler 

func GetSnapshotReadFnWithHandler(
	handleFn func(
		ctx context.Context,
		meta txn.TxnMeta,
		req *cmd_util.SnapshotReadReq,
		resp *cmd_util.SnapshotReadResp,
	) (func(), error),
) func(ctx context.Context, tbl *txnTable, snapshot *types.TS) (resp any, err error)

func HandleShardingReadApproxObjectsNum 

func HandleShardingReadApproxObjectsNum(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

HandleShardingReadApproxObjectsNum handles sharding read ApproxObjectsNum

func HandleShardingReadBuildReader 

func HandleShardingReadBuildReader(
	ctx context.Context,
	shard shard.TableShard,
	e engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

HandleShardingReadReader handles sharding read Reader

func HandleShardingReadClose 

func HandleShardingReadClose(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

func HandleShardingReadCollectTombstones 

func HandleShardingReadCollectTombstones(
	ctx context.Context,
	shard shard.TableShard,
	eng engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

func HandleShardingReadGetColumMetadataScanInfo 

func HandleShardingReadGetColumMetadataScanInfo(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

HandleShardingReadGetColumMetadataScanInfo handles sharding read GetColumMetadataScanInfo

func HandleShardingReadMergeObjects 

func HandleShardingReadMergeObjects(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

HandleShardingReadMergeObjects handles sharding read MergeObjects

func HandleShardingReadNext 

func HandleShardingReadNext(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

func HandleShardingReadPrimaryKeysMayBeModified 

func HandleShardingReadPrimaryKeysMayBeModified(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

HandleShardingReadPrimaryKeysMayBeModified handles sharding read PrimaryKeysMayBeModified

func HandleShardingReadPrimaryKeysMayBeUpserted 

func HandleShardingReadPrimaryKeysMayBeUpserted(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

func HandleShardingReadRanges 

func HandleShardingReadRanges(
	ctx context.Context,
	shard shard.TableShard,
	eng engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

HandleShardingReadRanges handles sharding read Ranges

func HandleShardingReadRows 

func HandleShardingReadRows(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

HandleShardingReadRows handles sharding read rows

func HandleShardingReadSize 

func HandleShardingReadSize(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

HandleShardingReadSize handles sharding read size

func HandleShardingReadStatus 

func HandleShardingReadStatus(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

HandleShardingReadStatus handles sharding read status

func HandleShardingReadVisibleObjectStats 

func HandleShardingReadVisibleObjectStats(
	ctx context.Context,
	shard shard.TableShard,
	engine engine.Engine,
	param shard.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

func LinearSearchOffsetByValFactory added in v1.2.0 

func LinearSearchOffsetByValFactory(pk *vector.Vector) func(*vector.Vector) []int64

func ListTnService added in v1.1.0 

func ListTnService(
	service string,
	appendFn func(service *metadata.TNService),
)

ListTnService gets all tn service in the cluster

func MockShardingLocalReader 

func MockShardingLocalReader() engine.Reader

TODO::

func MockTableDelegate 

func MockTableDelegate(
	tableDelegate engine.Relation,
	service shardservice.ShardService,
) (engine.Relation, error)

func NewTableMetaReader 

func NewTableMetaReader(
	ctx context.Context,
	rel engine.Relation,
) (engine.Reader, error)

func ScanSnapshotWithCurrentRanges 

func ScanSnapshotWithCurrentRanges(
	ctx context.Context,
	caller string,
	rel engine.Relation,
	relData engine.RelData,
	snapshotTS types.TS,
	attrs []string,
	colTypes []types.Type,
	filterExpr *plan.Expr,
	scanParallelism int,
	mp *mpool.MPool,
	onBatch func(*batch.Batch) error,
) error

ScanSnapshotWithCurrentRanges reads rows at snapshotTS by reusing the current relation handle and its current-view ranges.

Snapshot scan reuses current-view disk ranges, but materializes the latest committed in-memory rows separately so reader fallback does not need to lazily consult historical pState state during scan execution.

Serial path uses a hybrid source that emits committed in-memory rows first, then reads persisted blocks through RemoteDataSource. Parallel path emits the same committed in-memory rows once, then scans persisted shards concurrently.

Types

type CCPRTxnCache 

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

CCPRTxnCache is a cache for tracking CCPR (Cross-Cluster Publication Replication) objects and their associated transactions. It maintains a mapping from object names to transaction IDs.

Thread-safety: All methods are thread-safe and use mutex for synchronization.

Lifecycle of an entry:

  1. WriteObject: creates entry with isWriting=true (if new file) or appends txnID
  2. OnFileWritten: clears isWriting; if already committed, deletes the entry
  3. OnTxnCommit: marks committed=true; if not isWriting, deletes the entry
  4. OnTxnRollback: removes txnID; if last txnID and not committed, GCs the file
  5. OnTxnUnknownResult: removes cache tracking without deleting the file

func NewCCPRTxnCache 

func NewCCPRTxnCache(gcPool *ants.Pool, fs fileservice.FileService) *CCPRTxnCache

NewCCPRTxnCache creates a new CCPRTxnCache instance

func (*CCPRTxnCache) OnFileWritten 

func (c *CCPRTxnCache) OnFileWritten(objectName string)

OnFileWritten is called after the file has been successfully written to fileservice. It clears the isWriting flag. If the entry is already committed, deletes the entry since the file is now safely persisted and no longer needs cache tracking.

func (*CCPRTxnCache) OnTxnCommit 

func (c *CCPRTxnCache) OnTxnCommit(txnID []byte)

OnTxnCommit is called when a transaction commits successfully. It marks the entry as committed and removes the entire entry if the file has already been written (!isWriting). If still writing, the entry stays until OnFileWritten cleans it up.

func (*CCPRTxnCache) OnTxnRollback 

func (c *CCPRTxnCache) OnTxnRollback(txnID []byte)

OnTxnRollback is called when a transaction rolls back. It removes the txnID from all associated object entries. If an object entry has no more txnIDs and was never committed, the file is GC'd. If the entry was committed (by another txn), the file is kept.

func (*CCPRTxnCache) OnTxnUnknownResult 

func (c *CCPRTxnCache) OnTxnUnknownResult(txnID []byte)

OnTxnUnknownResult is called when the transaction commit result is unknown. The object may already be committed, so this path must not GC object files. It removes cache tracking for affected objects so later rollback cleanup cannot delete a possibly-committed object through stale CCPR references.

func (*CCPRTxnCache) WriteObject 

func (c *CCPRTxnCache) WriteObject(ctx context.Context, objectName string, txnID []byte) (isNewFile bool, err error)

WriteObject checks if an object needs to be written and registers it with the given transaction ID. This method DOES NOT write the file - it only checks and registers in the cache. The caller is responsible for writing the file when isNewFile is true.

After the caller writes the file, it should call OnFileWritten to complete the registration.

Returns:

  • isNewFile: true if file needs to be written, false if file already exists or is being written
  • error: error if operation failed

type CheckpointChangesHandle 

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

func NewCheckpointChangesHandle 

func NewCheckpointChangesHandle(
	ctx context.Context,
	table *txnTable,
	end types.TS,
	mp *mpool.MPool,
) (*CheckpointChangesHandle, error)

func (*CheckpointChangesHandle) Close 

func (h *CheckpointChangesHandle) Close() error

func (*CheckpointChangesHandle) Next 

func (h *CheckpointChangesHandle) Next(
	ctx context.Context, mp *mpool.MPool,
) (
	data *batch.Batch,
	tombstone *batch.Batch,
	hint engine.ChangesHandle_Hint,
	err error,
)

type CloneTxnCache 

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

func (CloneTxnCache) AddSharedFile 

func (ctc CloneTxnCache) AddSharedFile(txnId []byte, name string)

func (CloneTxnCache) AddTxn 

func (ctc CloneTxnCache) AddTxn(txnId []byte, snapshot int64)

func (CloneTxnCache) AddTxnLocalSharedFile 

func (ctc CloneTxnCache) AddTxnLocalSharedFile(txnId []byte, name string)

func (CloneTxnCache) DeleteTxn 

func (ctc CloneTxnCache) DeleteTxn(txnId []byte)

func (CloneTxnCache) IsSharedFile 

func (ctc CloneTxnCache) IsSharedFile(txnId []byte, name string) bool

func (CloneTxnCache) IsTxnLocalSharedFile 

func (ctc CloneTxnCache) IsTxnLocalSharedFile(txnId []byte, name string) bool

func (CloneTxnCache) RemoveTxnLocalSharedFile 

func (ctc CloneTxnCache) RemoveTxnLocalSharedFile(txnId []byte, name string)

type CombinedRelData 

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

func (*CombinedRelData) AppendBlockInfo 

func (r *CombinedRelData) AppendBlockInfo(blk *objectio.BlockInfo)

func (*CombinedRelData) AppendBlockInfoSlice 

func (r *CombinedRelData) AppendBlockInfoSlice(objectio.BlockInfoSlice)

func (*CombinedRelData) AppendShardID 

func (r *CombinedRelData) AppendShardID(id uint64)

func (*CombinedRelData) AttachTombstones 

func (r *CombinedRelData) AttachTombstones(tombstones engine.Tombstoner) error

func (*CombinedRelData) BuildEmptyRelData 

func (r *CombinedRelData) BuildEmptyRelData(preAllocSize int) engine.RelData

func (*CombinedRelData) DataCnt 

func (r *CombinedRelData) DataCnt() int

func (*CombinedRelData) DataSlice 

func (r *CombinedRelData) DataSlice(begin, end int) engine.RelData

func (*CombinedRelData) GetBlockInfo 

func (r *CombinedRelData) GetBlockInfo(i int) objectio.BlockInfo

func (*CombinedRelData) GetBlockInfoSlice 

func (r *CombinedRelData) GetBlockInfoSlice() objectio.BlockInfoSlice

func (*CombinedRelData) GetShardID 

func (r *CombinedRelData) GetShardID(i int) uint64

func (*CombinedRelData) GetShardIDList 

func (r *CombinedRelData) GetShardIDList() []uint64

func (*CombinedRelData) GetTombstones 

func (r *CombinedRelData) GetTombstones() engine.Tombstoner

func (*CombinedRelData) GetType 

func (r *CombinedRelData) GetType() engine.RelDataType

func (*CombinedRelData) MarshalBinary 

func (r *CombinedRelData) MarshalBinary() ([]byte, error)

func (*CombinedRelData) SetBlockInfo 

func (r *CombinedRelData) SetBlockInfo(i int, blk *objectio.BlockInfo)

func (*CombinedRelData) SetShardID 

func (r *CombinedRelData) SetShardID(i int, id uint64)

func (*CombinedRelData) Split 

func (r *CombinedRelData) Split(i int) []engine.RelData

func (*CombinedRelData) String 

func (r *CombinedRelData) String() string

func (*CombinedRelData) UnmarshalBinary 

func (r *CombinedRelData) UnmarshalBinary(buf []byte) error

type ConcurrentExecutor added in v1.2.2 

type ConcurrentExecutor interface {
	// AppendTask append the concurrent task to the exuecutor.
	AppendTask(concurrentTask)
	// Run starts receive task to execute.
	Run(context.Context)
	// GetConcurrency returns the concurrency of this executor.
	GetConcurrency() int
}

ConcurrentExecutor is an interface that runs tasks concurrently.

type DNStore 

type DNStore = metadata.TNService

type Engine 

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

func New 

func New(
	ctx context.Context,
	service string,
	mp *mpool.MPool,
	fs fileservice.FileService,
	cli client.TxnClient,
	hakeeper logservice.CNHAKeeperClient,
	keyRouter client2.KeyRouter[pb.StatsInfoKey],
	updateWorkerFactor int,
	options ...EngineOptions,
) *Engine

func (*Engine) AcquireQuota 

func (e *Engine) AcquireQuota(v int64) (int64, bool)

func (*Engine) AllocateIDByKey added in v0.8.0 

func (e *Engine) AllocateIDByKey(ctx context.Context, key string) (uint64, error)

func (*Engine) BuildBlockReaders 

func (e *Engine) BuildBlockReaders(
	ctx context.Context,
	p any,
	ts timestamp.Timestamp,
	expr *plan.Expr,
	def *plan.TableDef,
	relData engine.RelData,
	num int) ([]engine.Reader, error)

func (*Engine) Close 

func (e *Engine) Close() error

func (*Engine) Create 

func (e *Engine) Create(ctx context.Context, name string, op client.TxnOperator) error

func (*Engine) Database 

func (e *Engine) Database(
	ctx context.Context,
	name string,
	op client.TxnOperator,
) (engine.Database, error)

func (*Engine) Databases 

func (e *Engine) Databases(ctx context.Context, op client.TxnOperator) ([]string, error)

func (*Engine) Delete 

func (e *Engine) Delete(ctx context.Context, name string, op client.TxnOperator) (err error)

func (*Engine) FS 

func (e *Engine) FS() fileservice.FileService

func (*Engine) ForceGC 

func (e *Engine) ForceGC(ctx context.Context, ts types.TS)

for UT

func (*Engine) GCCatalogCache 

func (e *Engine) GCCatalogCache(ctx context.Context, ago time.Duration) error

GCCatalogCache implements engine.CatalogCacheGCer.

func (*Engine) GetCCPRTxnCache 

func (e *Engine) GetCCPRTxnCache() *CCPRTxnCache

GetCCPRTxnCache returns the CCPR transaction cache

func (*Engine) GetGlobalStats 

func (e *Engine) GetGlobalStats() *GlobalStats

GetGlobalStats returns the GlobalStats instance

func (*Engine) GetLatestCatalogCache 

func (e *Engine) GetLatestCatalogCache() *cache.CatalogCache

func (*Engine) GetMessageCenter 

func (e *Engine) GetMessageCenter() any

func (*Engine) GetNameById added in v0.7.0 

func (e *Engine) GetNameById(ctx context.Context, op client.TxnOperator, tableId uint64) (dbName string, tblName string, err error)

func (*Engine) GetOrCreateLatestPart 

func (e *Engine) GetOrCreateLatestPart(
	ctx context.Context,
	accId uint64,
	databaseId,
	tableId uint64,
) *logtailreplay.Partition

func (*Engine) GetRelationById added in v0.7.0 

func (e *Engine) GetRelationById(ctx context.Context, op client.TxnOperator, tableId uint64) (dbName, tableName string, rel engine.Relation, err error)

func (*Engine) GetService 

func (e *Engine) GetService() string

func (*Engine) GetTNServices 

func (e *Engine) GetTNServices() []DNStore

func (*Engine) HandleMoTableStatsCtl 

func (d *Engine) HandleMoTableStatsCtl(cmd string) string

func (*Engine) Hints 

func (e *Engine) Hints() (h engine.Hints)

func (*Engine) InitLogTailPushModel added in v0.8.0 

func (e *Engine) InitLogTailPushModel(ctx context.Context, timestampWaiter client.TimestampWaiter) error

func (*Engine) LatestLogtailAppliedTime 

func (e *Engine) LatestLogtailAppliedTime() timestamp.Timestamp

func (*Engine) LaunchMTSTasksForUT 

func (d *Engine) LaunchMTSTasksForUT()

func (*Engine) LazyLoadLatestCkp 

func (e *Engine) LazyLoadLatestCkp(
	ctx context.Context,
	accId uint64,
	tableID uint64,
	tableName string,
	dbID uint64,
	dbName string) (*logtailreplay.Partition, error)

func (*Engine) LogDynamicCtx 

func (d *Engine) LogDynamicCtx() string

func (*Engine) MTSTableRows 

func (d *Engine) MTSTableRows(
	ctx context.Context,
	accs, dbs, tbls []uint64,
	eng engine.Engine,
	forceUpdate bool,
	resetUpdateTime bool,
) (sizes []uint64, err error)

func (*Engine) MTSTableSize 

func (d *Engine) MTSTableSize(
	ctx context.Context,
	accs, dbs, tbls []uint64,
	eng engine.Engine,
	forceUpdate bool,
	resetUpdateTime bool,
) (sizes []uint64, err error)

func (*Engine) New 

func (e *Engine) New(ctx context.Context, op client.TxnOperator) error

func (*Engine) Nodes 

func (e *Engine) Nodes(
	isInternal bool, tenant string, username string, cnLabel map[string]string,
) (engine.Nodes, error)

func (*Engine) NotifyCleanDeletes 

func (d *Engine) NotifyCleanDeletes()

func (*Engine) NotifyInsertNewTable 

func (d *Engine) NotifyInsertNewTable()

func (*Engine) NotifyUpdateForgotten 

func (d *Engine) NotifyUpdateForgotten()

func (*Engine) PackerPool 

func (e *Engine) PackerPool() *fileservice.Pool[*types.Packer]

func (*Engine) PrefetchTableMeta 

func (e *Engine) PrefetchTableMeta(ctx context.Context, key pb.StatsInfoKey) bool

return true if the prefetch is received return false if the prefetch is not rejected

func (*Engine) PushClient added in v1.2.0 

func (e *Engine) PushClient() *PushClient

func (*Engine) QueryTableStats 

func (d *Engine) QueryTableStats(
	ctx context.Context,
	wantedStatsIdxes []int,
	accs, dbs, tbls []uint64,
	forceUpdate bool,
	resetUpdateTime bool,
) (statsVals [][]any, err error, ok bool)

func (*Engine) QueryTableStatsByAccounts 

func (d *Engine) QueryTableStatsByAccounts(
	ctx context.Context,
	wantedStatsIdxes []int,
	accs []uint64,
	forceUpdate bool,
	resetUpdateTime bool,
) (statsVals [][]any, retAcc []uint64, ok bool, err error)

func (*Engine) QueryTableStatsByDatabase 

func (d *Engine) QueryTableStatsByDatabase(
	ctx context.Context,
	wantedStatsIdxes []int,
	dbIds []uint64,
	forceUpdate bool,
	resetUpdateTime bool,
) (statsVals [][]any, retDb []uint64, ok bool, err error)

func (*Engine) QueryTableStatsByTable 

func (d *Engine) QueryTableStatsByTable(
	ctx context.Context,
	wantedStatsIdxes []int,
	tblIds []uint64,
	forceUpdate bool,
	resetUpdateTime bool,
) (statsVals [][]any, retTbls []uint64, ok bool, err error)

func (*Engine) ReleaseQuota 

func (e *Engine) ReleaseQuota(quota int64) (left uint64)

func (*Engine) ResetGCWorkerPool 

func (e *Engine) ResetGCWorkerPool(pool *ants.Pool)

func (*Engine) RunGCScheduler 

func (e *Engine) RunGCScheduler(ctx context.Context)

RunGCScheduler runs all GC tasks in a single goroutine with different intervals

func (*Engine) SetService 

func (e *Engine) SetService(svr string)

func (*Engine) SetWorkspaceThreshold 

func (e *Engine) SetWorkspaceThreshold(commitThreshold, writeThreshold uint64) (commit, write uint64)

SetWorkspaceThreshold updates the commit and write workspace thresholds (in MB). Non-zero values override the current thresholds, while zero keeps them unchanged. Returns the previous thresholds (in MB).

func (*Engine) Stats added in v1.2.0 

func (e *Engine) Stats(ctx context.Context, key pb.StatsInfoKey, sync bool) *pb.StatsInfo

func (*Engine) TryToSubscribeTable added in v1.2.0 

func (e *Engine) TryToSubscribeTable(ctx context.Context, accId, dbID, tbID uint64, dbName, tblName string) error

TryToSubscribeTable implements the LogtailEngine interface.

func (*Engine) UnsubscribeTable added in v1.2.0 

func (e *Engine) UnsubscribeTable(ctx context.Context, accId, dbID, tbID uint64) error

UnsubscribeTable implements the LogtailEngine interface.

type EngineOptions 

type EngineOptions func(*Engine)

func WithCNTransferTxnLifespanThreshold 

func WithCNTransferTxnLifespanThreshold(th time.Duration) EngineOptions

func WithCommitWorkspaceThreshold 

func WithCommitWorkspaceThreshold(th uint64) EngineOptions

func WithExtraWorkspaceThreshold 

func WithExtraWorkspaceThreshold(th uint64) EngineOptions

func WithExtraWorkspaceThresholdQuota 

func WithExtraWorkspaceThresholdQuota(quota uint64) EngineOptions

func WithInsertEntryMaxCount 

func WithInsertEntryMaxCount(th int) EngineOptions

func WithMoServerStateChecker 

func WithMoServerStateChecker(checker func() bool) EngineOptions

func WithMoTableStatsConf 

func WithMoTableStatsConf(conf MoTableStatsConfig) EngineOptions

func WithPrefetchOnSubscribed 

func WithPrefetchOnSubscribed(th []string) EngineOptions

func WithSQLExecFunc 

func WithSQLExecFunc(f func() ie.InternalExecutor) EngineOptions

func WithWriteWorkspaceThreshold 

func WithWriteWorkspaceThreshold(th uint64) EngineOptions

type Entry 

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

Entry represents a delete/insert

func (*Entry) Bat 

func (e *Entry) Bat() *batch.Batch

func (*Entry) DatabaseId 

func (e *Entry) DatabaseId() uint64

func (*Entry) FileName 

func (e *Entry) FileName() string

func (*Entry) String 

func (e *Entry) String() string

func (*Entry) TableId 

func (e *Entry) TableId() uint64

func (*Entry) Type 

func (e *Entry) Type() int

type GlobalStats added in v1.2.0 

type GlobalStats struct {

	// KeyRouter is the router to decides which node should send to.
	KeyRouter client.KeyRouter[pb.StatsInfoKey]
	// contains filtered or unexported fields
}

func NewGlobalStats added in v1.2.0 

func NewGlobalStats(
	ctx context.Context, e *Engine, keyRouter client.KeyRouter[pb.StatsInfoKey], opts ...GlobalStatsOption,
) *GlobalStats

func (*GlobalStats) Get added in v1.2.0 

func (gs *GlobalStats) Get(ctx context.Context, key pb.StatsInfoKey, sync bool) *pb.StatsInfo

func (*GlobalStats) GetBaseObjectCnt 

func (gs *GlobalStats) GetBaseObjectCnt(key pb.StatsInfoKey) int64

func (*GlobalStats) GetSamplingRatio 

func (gs *GlobalStats) GetSamplingRatio(key pb.StatsInfoKey) float64

GetSamplingRatio returns the sampling ratio used in the last stats update for the given key. Returns 0 if no update record exists.

func (*GlobalStats) PatchStats 

func (gs *GlobalStats) PatchStats(key pb.StatsInfoKey, patch *PatchArgs) error

PatchStats partially updates the stats for the given key

func (*GlobalStats) PrefetchTableMeta 

func (gs *GlobalStats) PrefetchTableMeta(ctx context.Context, key pb.StatsInfoKey) bool

func (*GlobalStats) RefreshWithMode 

func (gs *GlobalStats) RefreshWithMode(ctx context.Context, key pb.StatsInfoKey, samplingMode string) error

RefreshWithMode triggers a stats refresh with the specified sampling mode

func (*GlobalStats) RemoveTid added in v1.2.2 

func (gs *GlobalStats) RemoveTid(tableID uint64)

RemoveTid removes all statsInfoMap entries for the given table ID. Called from cleanMemoryTableWithTable (1+ hour after unsubscribe/drop) to prevent unbounded map growth. Safe because no queries target a table that has been unsubscribed for over an hour.

type GlobalStatsConfig added in v1.2.0 

type GlobalStatsConfig struct {
	LogtailUpdateStatsThreshold int
}

type GlobalStatsOption added in v1.2.0 

type GlobalStatsOption func(s *GlobalStats)

func WithApproxObjectNumUpdater 

func WithApproxObjectNumUpdater(f func() int64) GlobalStatsOption

WithApproxObjectNumUpdater set the update function to update approx object num.

func WithUpdateWorkerFactor added in v1.2.2 

func WithUpdateWorkerFactor(f int) GlobalStatsOption

WithUpdateWorkerFactor set the update worker factor.

type IDGenerator 

type IDGenerator interface {
	AllocateID(ctx context.Context) (uint64, error)
	// AllocateIDByKey allocate a globally unique ID by key.
	AllocateIDByKey(ctx context.Context, key string) (uint64, error)
}

type ItemEntry 

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

ItemEntry represents an entry in the items BTree, sorted by objectName

func (ItemEntry) Less 

func (e ItemEntry) Less(other ItemEntry) bool

Less compares two ItemEntry by objectName for BTree ordering

type LocalDisttaeDataSource 

type LocalDisttaeDataSource struct {
	OrderBy []*plan.OrderBySpec
	Limit   uint64
	// contains filtered or unexported fields
}

func NewLocalDataSource 

func NewLocalDataSource(
	ctx context.Context,
	table *txnTable,
	txnOffset int,
	pState *logtailreplay.PartitionState,
	rangesSlice objectio.BlockInfoSlice,
	extraTombstones engine.Tombstoner,
	skipReadMem bool,
	policy engine.TombstoneApplyPolicy,
	category engine.DataSourceType,
) (source *LocalDisttaeDataSource, err error)

func (*LocalDisttaeDataSource) ApplyTombstones 

func (ls *LocalDisttaeDataSource) ApplyTombstones(
	ctx context.Context,
	bid *objectio.Blockid,
	rowsOffset []int64,
	dynamicPolicy engine.TombstoneApplyPolicy,
) ([]int64, error)

ApplyTombstones check if any deletes exist in

  1. unCommittedInmemDeletes: a. workspace writes b. flushed to s3 c. raw rowId offset deletes (not flush yet)
  2. committedInmemDeletes
  3. committedPersistedTombstone

func (*LocalDisttaeDataSource) Close 

func (ls *LocalDisttaeDataSource) Close()

func (*LocalDisttaeDataSource) GetOrderBy 

func (ls *LocalDisttaeDataSource) GetOrderBy() []*plan.OrderBySpec

func (*LocalDisttaeDataSource) GetTombstones 

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

func (*LocalDisttaeDataSource) Next 

func (ls *LocalDisttaeDataSource) Next(
	ctx context.Context,
	cols []string,
	types []types.Type,
	seqNums []uint16,
	pkSeqNum int32,
	filter any,
	mp *mpool.MPool,
	outBatch *batch.Batch,
) (info *objectio.BlockInfo, state engine.DataState, err error)

func (*LocalDisttaeDataSource) SetFilterZM 

func (ls *LocalDisttaeDataSource) SetFilterZM(zm objectio.ZoneMap)

func (*LocalDisttaeDataSource) SetOrderBy 

func (ls *LocalDisttaeDataSource) SetOrderBy(orderby []*plan.OrderBySpec)

func (*LocalDisttaeDataSource) String 

func (ls *LocalDisttaeDataSource) String() string

type MoTableStatsConfig 

type MoTableStatsConfig struct {
	UpdateDuration     time.Duration `toml:"update-duration"`
	ForceUpdate        bool          `toml:"force-update"`
	GetTableListLimit  int           `toml:"get-table-list-limit"`
	StatsUsingOldImpl  bool          `toml:"stats-using-old-impl"`
	DisableStatsTask   bool          `toml:"disable-stats-task"`
	CorrectionDuration time.Duration `toml:"correction-duration"`
}

type NoteLevel 

type NoteLevel string
const (
	DATABASE NoteLevel = "database"
	TABLE    NoteLevel = "table"
	COLUMN   NoteLevel = "column"
)

type PartitionChangesHandle 

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

func NewPartitionChangesHandle 

func NewPartitionChangesHandle(
	ctx context.Context,
	tbl *txnTable,
	from, to types.TS,
	skipDeletes bool,
	snapshotReadPolicy engine.SnapshotReadPolicy,
	mp *mpool.MPool,
) (*PartitionChangesHandle, error)

func (*PartitionChangesHandle) Close 

func (h *PartitionChangesHandle) Close() error

func (*PartitionChangesHandle) Next 

func (h *PartitionChangesHandle) Next(ctx context.Context, mp *mpool.MPool) (data, tombstone *batch.Batch, hint engine.ChangesHandle_Hint, err error)

type PatchArgs 

type PatchArgs struct {
	// Table-level stats
	TableCnt             *float64 `json:"table_cnt,omitempty"`
	BlockNumber          *int64   `json:"block_number,omitempty"`
	AccurateObjectNumber *int64   `json:"accurate_object_number,omitempty"`

	// Column-level stats (merge mode)
	NdvMap     map[string]float64 `json:"ndv_map,omitempty"`
	MinValMap  map[string]float64 `json:"min_val_map,omitempty"`
	MaxValMap  map[string]float64 `json:"max_val_map,omitempty"`
	NullCntMap map[string]uint64  `json:"null_cnt_map,omitempty"`
	SizeMap    map[string]uint64  `json:"size_map,omitempty"`

	// ShuffleRange partial updates (fine-grained control per column)
	// Each column can have its Overlap/Uniform/Result fields updated independently
	ShuffleRangeMap map[string]*ShuffleRangePartialUpdate `json:"shuffle_range_map,omitempty"`
}

PatchArgs defines arguments for patch command

type PushClient added in v1.2.0 

type PushClient struct {
	LogtailRPCClientFactory func(context.Context, string, string, morpc.RPCClient) (morpc.RPCClient, morpc.Stream, error)
	// contains filtered or unexported fields
}

PushClient is a structure responsible for all operations related to the log tail push model. It provides the following methods: 

	-----------------------------------------------------------------------------------------------------
	 1. checkTxnTimeIsLegal : block the process until we have received enough log tail (T_log >= T_txn)
	 2. TryToSubscribeTable : block the process until we subscribed a table succeed.
	 3. subscribeTable	   : send a table subscribe request to service.
	 4. subSysTables : subscribe mo_databases, mo_tables, mo_columns
	 5. receiveTableLogTailContinuously   : start (1 + consumerNumber) routine to receive log tail from service.

 Watch out for the following points:
	 1. if we want to lock both subscriber and subscribed, we should lock subscriber first.
	-----------------------------------------------------------------------------------------------------

func (*PushClient) Disconnect 

func (c *PushClient) Disconnect() error

func (*PushClient) GetState added in v1.2.0 

func (c *PushClient) GetState() State

func (*PushClient) IsSubscribed 

func (c *PushClient) IsSubscribed(tblId uint64) bool

func (*PushClient) IsSubscriberReady 

func (c *PushClient) IsSubscriberReady() bool

func (*PushClient) LatestLogtailAppliedTime 

func (c *PushClient) LatestLogtailAppliedTime() timestamp.Timestamp

func (*PushClient) SetReconnectHandler 

func (c *PushClient) SetReconnectHandler(handler func())

func (*PushClient) SetSubscribeState 

func (c *PushClient) SetSubscribeState(dbId, tblId uint64, state SubscribeState)

Only used for ut

func (*PushClient) TryGC 

func (c *PushClient) TryGC(ctx context.Context)

TryGC checks if ready and runs GC for PushClient resources (unused tables)

func (*PushClient) TryToSubscribeTable added in v1.2.0 

func (c *PushClient) TryToSubscribeTable(
	ctx context.Context,
	accId, dbId, tblId uint64,
	dbName, tblName string,
) (err error)

TryToSubscribeTable subscribe a table and block until subscribe succeed.

func (*PushClient) UnsubscribeTable added in v1.2.0 

func (c *PushClient) UnsubscribeTable(ctx context.Context, accId, dbID, tbID uint64) error

UnsubscribeTable implements the LogtailEngine interface.

type ReaderPhase 

type ReaderPhase uint8
const (
	InLocal ReaderPhase = iota
	InRemote
	InEnd
)

type ShuffleRangePartialUpdate 

type ShuffleRangePartialUpdate struct {
	Overlap *float64  `json:"overlap,omitempty"`
	Uniform *float64  `json:"uniform,omitempty"`
	Result  []float64 `json:"result,omitempty"`
}

ShuffleRangePartialUpdate contains fields that can be independently updated in ShuffleRange

type SnapshotGCConfig 

type SnapshotGCConfig struct {
	Enabled              bool          // Whether snapshot GC is enabled
	GCInterval           time.Duration // Interval between GC runs
	MaxAge               time.Duration // Max age for inactive snapshots
	MaxSnapshotsPerTable int           // Max snapshots per table
	MaxTotalSnapshots    int           // Max total snapshots across all tables
}

SnapshotGCConfig holds configuration for snapshot GC

func DefaultSnapshotGCConfig 

func DefaultSnapshotGCConfig() SnapshotGCConfig

DefaultSnapshotGCConfig returns the default configuration

type SnapshotGCState 

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

SnapshotGCState holds the runtime state for snapshot GC

type SnapshotManager 

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

SnapshotManager manages all tracked partitions across all tables

func NewSnapshotManager 

func NewSnapshotManager() *SnapshotManager

NewSnapshotManager creates a new SnapshotManager

func (*SnapshotManager) Add 

func (sm *SnapshotManager) Add(
	dbID, tableID uint64,
	partition *logtailreplay.Partition,
	tableName string,
	ts types.TS,
) *logtailreplay.PartitionState

Add adds a new partition to the manager for a table

func (*SnapshotManager) Find 

func (sm *SnapshotManager) Find(dbID, tableID uint64, ts types.TS) *logtailreplay.PartitionState

Find searches for a snapshot that can serve the given timestamp for a table

func (*SnapshotManager) GetConfig 

func (sm *SnapshotManager) GetConfig() SnapshotGCConfig

GetConfig returns a copy of the configuration

func (*SnapshotManager) GetMetrics 

func (sm *SnapshotManager) GetMetrics() *SnapshotMetrics

GetMetrics returns a pointer to current metrics

func (*SnapshotManager) GetOrCreate 

func (sm *SnapshotManager) GetOrCreate(dbID, tableID uint64) *TrackedPartitions

GetOrCreate gets or creates a TrackedPartitions for the given table

func (*SnapshotManager) Init 

func (sm *SnapshotManager) Init()

Init initializes the snapshot manager with configuration

func (*SnapshotManager) MaybeStartGC 

func (sm *SnapshotManager) MaybeStartGC()

MaybeStartGC checks if GC should run and starts it if needed

func (*SnapshotManager) RunGC 

func (sm *SnapshotManager) RunGC()

RunGC performs garbage collection on all tracked partitions

type SnapshotMetrics 

type SnapshotMetrics struct {
	TotalSnapshots  atomic.Int64 // Current total snapshots
	TotalTables     atomic.Int64 // Current tables with snapshots
	GCRuns          atomic.Int64 // Total GC runs
	SnapshotsGCed   atomic.Int64 // Total snapshots GCed
	LastGCDuration  atomic.Int64 // Last GC duration in nanoseconds
	SnapshotHits    atomic.Int64 // Snapshot cache hits (reuse)
	SnapshotMisses  atomic.Int64 // Snapshot cache misses (create)
	SnapshotCreates atomic.Int64 // Total snapshots created
	LRUEvictions    atomic.Int64 // Evictions due to count limit
	AgeEvictions    atomic.Int64 // Evictions due to age limit
}

SnapshotMetrics holds metrics for snapshot management

type State added in v1.2.0 

type State struct {
	LatestTS  timestamp.Timestamp
	SubTables map[uint64]SubTableStatus
}

type SubTableStatus added in v1.2.0 

type SubTableStatus struct {
	DBID       uint64
	SubState   SubscribeState
	LatestTime time.Time
}

SubTableStatus is used for external API compatibility (e.g., GetState).

type SubscribeState added in v1.2.1 

type SubscribeState int32
const (
	InvalidSubState SubscribeState = iota
	Subscribing
	SubRspReceived
	Subscribed
	Unsubscribing
	Unsubscribed
	SubRspTableNotExist

	FakeLogtailServerAddress = "fake address for ut"
)

type Summary 

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

type TableMetaReader 

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

func (*TableMetaReader) Close 

func (r *TableMetaReader) Close() error

func (*TableMetaReader) GetOrderBy 

func (r *TableMetaReader) GetOrderBy() []*plan.OrderBySpec

func (*TableMetaReader) GetTableDef 

func (r *TableMetaReader) GetTableDef() *plan.TableDef

func (*TableMetaReader) GetTxnInfo 

func (r *TableMetaReader) GetTxnInfo() string

func (*TableMetaReader) Read 

func (r *TableMetaReader) Read(
	ctx context.Context,
	_ []string,
	_ *plan.Expr,
	mp *mpool.MPool,
	outBatch *batch.Batch,
) (end bool, err error)

func (*TableMetaReader) SetFilterZM 

func (r *TableMetaReader) SetFilterZM(zoneMap objectio.ZoneMap)

func (*TableMetaReader) SetIndexParam 

func (r *TableMetaReader) SetIndexParam(param *plan.IndexReaderParam)

func (*TableMetaReader) SetOrderBy 

func (r *TableMetaReader) SetOrderBy(specs []*plan.OrderBySpec)

type TrackedPartition 

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

TrackedPartition wraps a partition with access tracking information

func NewTrackedPartition 

func NewTrackedPartition(partition *logtailreplay.Partition) *TrackedPartition

NewTrackedPartition creates a new TrackedPartition

func (*TrackedPartition) GetAge 

func (tp *TrackedPartition) GetAge() time.Duration

GetAge returns the age since last access

func (*TrackedPartition) GetCreateTime 

func (tp *TrackedPartition) GetCreateTime() time.Time

GetCreateTime returns the creation time

func (*TrackedPartition) GetLastAccessTime 

func (tp *TrackedPartition) GetLastAccessTime() time.Time

GetLastAccessTime returns the last access time

func (*TrackedPartition) GetPartition 

func (tp *TrackedPartition) GetPartition() *logtailreplay.Partition

GetPartition returns the underlying partition and updates access time

type TrackedPartitions 

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

TrackedPartitions manages a collection of tracked partitions for a table

func NewTrackedPartitions 

func NewTrackedPartitions(dbID, tableID uint64, metrics *SnapshotMetrics) *TrackedPartitions

NewTrackedPartitions creates a new TrackedPartitions manager

func (*TrackedPartitions) Add 

func (tp *TrackedPartitions) Add(
	partition *logtailreplay.Partition,
	tableName string,
	ts types.TS,
	maxCount int,
) *logtailreplay.PartitionState

Add adds a new tracked partition, enforcing the maximum count limit with LRU eviction Returns the partition state of the added partition

func (*TrackedPartitions) Count 

func (tp *TrackedPartitions) Count() int

Count returns the current number of tracked partitions

func (*TrackedPartitions) Find 

func (tp *TrackedPartitions) Find(ts types.TS) *logtailreplay.PartitionState

Find searches for a partition that can serve the given timestamp Returns the partition state if found, or nil if not found Automatically updates access time and metrics

func (*TrackedPartitions) GC 

func (tp *TrackedPartitions) GC(maxAge time.Duration) int

GC removes partitions that haven't been accessed within maxAge Returns the number of partitions removed

type Transaction 

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

Transaction represents a transaction

func NewTxnWorkSpace 

func NewTxnWorkSpace(eng *Engine, proc *process.Process) *Transaction

func (*Transaction) Adjust added in v1.0.0 

func (txn *Transaction) Adjust(writeOffset uint64) error

Adjust adjust writes order after the current statement finished.

func (*Transaction) ApproximateInMemInsertSize 

func (txn *Transaction) ApproximateInMemInsertSize() uint64

ApproximateInMemInsertSize returns the approximate total size of in-memory insert entries in the workspace. Intended for testing and diagnostics.

func (*Transaction) BindTxnOp added in v1.2.0 

func (txn *Transaction) BindTxnOp(op client.TxnOperator)

func (*Transaction) CloneSnapshotWS added in v1.2.0 

func (txn *Transaction) CloneSnapshotWS() client.Workspace

func (*Transaction) Commit added in v0.8.0 

func (txn *Transaction) Commit(ctx context.Context) (reqs []txn.TxnRequest, err error)

func (*Transaction) EndStatement added in v1.0.0 

func (txn *Transaction) EndStatement()

func (*Transaction) FinalizeCommit 

func (txn *Transaction) FinalizeCommit(context.Context)

func (*Transaction) FinalizeCommitWithUnknownResult 

func (txn *Transaction) FinalizeCommitWithUnknownResult(context.Context)

func (*Transaction) ForEachTableWrites 

func (txn *Transaction) ForEachTableWrites(databaseId uint64, tableId uint64, offset int, f func(Entry))

func (*Transaction) GCObjsByIdxRange 

func (txn *Transaction) GCObjsByIdxRange(start, end int) (err error)

[start, end]

func (*Transaction) GCObjsByStats 

func (txn *Transaction) GCObjsByStats(sl ...objectio.ObjectStats) (err error)

func (*Transaction) GetCCPRTaskID 

func (txn *Transaction) GetCCPRTaskID() string

GetCCPRTaskID returns the CCPR task ID for this transaction. Returns empty string if no task ID is set.

func (*Transaction) GetHaveDDL added in v1.2.2 

func (txn *Transaction) GetHaveDDL() bool

func (*Transaction) GetProc 

func (txn *Transaction) GetProc() *process.Process

func (*Transaction) GetSQLCount added in v1.0.0 

func (txn *Transaction) GetSQLCount() uint64

func (*Transaction) GetSnapshotWriteOffset added in v1.1.2 

func (txn *Transaction) GetSnapshotWriteOffset() int

func (*Transaction) GetSyncProtectionJobID 

func (txn *Transaction) GetSyncProtectionJobID() string

GetSyncProtectionJobID returns the sync protection job ID for this transaction. Returns empty string if no job ID is set.

func (*Transaction) IncrSQLCount added in v1.0.0 

func (txn *Transaction) IncrSQLCount()

func (*Transaction) IncrStatementID added in v0.8.0 

func (txn *Transaction) IncrStatementID(ctx context.Context, commit bool) error

func (*Transaction) IsCCPRTxn 

func (txn *Transaction) IsCCPRTxn() bool

IsCCPRTxn returns true if this transaction is a CCPR transaction.

func (*Transaction) PPString 

func (txn *Transaction) PPString() string

func (*Transaction) ReadOnly 

func (txn *Transaction) ReadOnly() bool

detecting whether a transaction is a read-only transaction

func (*Transaction) Readonly 

func (txn *Transaction) Readonly() bool

func (*Transaction) Rollback added in v0.8.0 

func (txn *Transaction) Rollback(ctx context.Context) error

func (*Transaction) RollbackLastStatement added in v0.8.0 

func (txn *Transaction) RollbackLastStatement(ctx context.Context) error

func (*Transaction) SetCCPRTaskID 

func (txn *Transaction) SetCCPRTaskID(taskID string)

SetCCPRTaskID sets the CCPR task ID for this transaction. When a CCPR task ID is set, the transaction can bypass shared object read-only checks.

func (*Transaction) SetCCPRTxn 

func (txn *Transaction) SetCCPRTxn()

SetCCPRTxn marks this transaction as a CCPR transaction. CCPR transactions will call CCPRTxnCache.OnTxnCommit/OnTxnRollback when committing/rolling back.

func (*Transaction) SetCloneTxn 

func (txn *Transaction) SetCloneTxn(snapshot int64)

func (*Transaction) SetHaveDDL added in v1.2.2 

func (txn *Transaction) SetHaveDDL(haveDDL bool)

func (*Transaction) SetSyncProtectionJobID 

func (txn *Transaction) SetSyncProtectionJobID(jobID string)

SetSyncProtectionJobID sets the sync protection job ID for this transaction. This is used to pass the job ID to TN for commit-time validation.

func (*Transaction) StartStatement added in v1.0.0 

func (txn *Transaction) StartStatement()

func (*Transaction) StashFlushedTombstones 

func (txn *Transaction) StashFlushedTombstones(stats objectio.ObjectStats)

func (*Transaction) String 

func (txn *Transaction) String() string

func (*Transaction) UpdateSnapshotWriteOffset added in v1.1.2 

func (txn *Transaction) UpdateSnapshotWriteOffset()

func (*Transaction) WriteBatch 

func (txn *Transaction) WriteBatch(
	typ int, note string,
	accountId uint32,
	databaseId uint64,
	tableId uint64,
	databaseName string,
	tableName string,
	bat *batch.Batch,
	tnStore DNStore) (genRowidVec *vector.Vector, err error)

WriteBatch used to write data to the transaction buffer insert/delete/update all use this api truncate : it denotes the batch with typ DELETE on mo_tables is generated when Truncating a table.

NOTE: For Insert type, rowid is generated in this function. Be carefule use this function multiple times for the same batch

func (*Transaction) WriteFile 

func (txn *Transaction) WriteFile(
	typ int,
	accountId uint32,
	databaseId,
	tableId uint64,
	databaseName,
	tableName string,
	fileName string,
	bat *batch.Batch,
	tnStore DNStore,
) error

WriteFile used to add a s3 file information to the transaction buffer insert/delete/update all use this api

func (*Transaction) WriteFileLocked added in v1.0.0 

func (txn *Transaction) WriteFileLocked(
	typ int,
	accountId uint32,
	databaseId,
	tableId uint64,
	databaseName,
	tableName string,
	fileName string,
	inputBat *batch.Batch,
	tnStore DNStore,
) (err error)

func (*Transaction) WriteFileLockedSkipTransfer 

func (txn *Transaction) WriteFileLockedSkipTransfer(
	typ int,
	accountId uint32,
	databaseId,
	tableId uint64,
	databaseName,
	tableName string,
	fileName string,
	inputBat *batch.Batch,
	tnStore DNStore,
) (err error)

WriteFileLockedSkipTransfer is similar to WriteFileLocked but marks the entry to skip transfer processing. Used by CCPR for cross-cluster tombstones.

func (*Transaction) WriteOffset added in v1.1.1 

func (txn *Transaction) WriteOffset() uint64

writeOffset returns the offset of the first write in the workspace

type TransferFlow 

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

func ConstructCNTombstoneObjectsTransferFlow 

func ConstructCNTombstoneObjectsTransferFlow(
	ctx context.Context,
	start, end types.TS,
	table *txnTable,
	txn *Transaction,
	mp *mpool.MPool,
	fs fileservice.FileService,
) (*TransferFlow, []zap.Field, error)

func ConstructTransferFlow 

func ConstructTransferFlow(
	table *txnTable,
	hiddenSelection objectio.HiddenColumnSelection,
	sourcer engine.Reader,
	isObjectDeletedFn func(*objectio.ObjectId) bool,
	newDataObjects []objectio.ObjectStats,
	mp *mpool.MPool,
	fs fileservice.FileService,
	opts ...TransferOption,
) *TransferFlow

func (*TransferFlow) Close 

func (flow *TransferFlow) Close() error

func (*TransferFlow) GetResult 

func (flow *TransferFlow) GetResult() ([]objectio.ObjectStats, []*batch.Batch)

func (*TransferFlow) Process 

func (flow *TransferFlow) Process(ctx context.Context) error

type TransferOption 

type TransferOption func(*TransferFlow)

type TxnIndexEntry 

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

TxnIndexEntry represents an entry in the txnIndex BTree, sorted by txnID It stores all objectNames associated with a transaction for fast lookup

func (TxnIndexEntry) Less 

func (e TxnIndexEntry) Less(other TxnIndexEntry) bool

Less compares two TxnIndexEntry by txnID for BTree ordering

type UT_ForceTransCheck 

type UT_ForceTransCheck struct{}

Source Files

View all Source files

Directories

Path Synopsis

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.