ioutil

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

Documentation

Index

Constants

View Source 
const (
	// checkpint related
	CheckpointExt = "ckp"
	CompactExt    = "cpt"

	// gc related
	GCFullExt   = "fgc"
	SnapshotExt = "snap"
	AcctExt     = "acct"
	TmpExt      = "tmp"
	GCMetaExt   = CheckpointExt

	// invalid ext
	InvalidExt = "%$%#"

	PrefixMetadata = "meta"
	SuffixMetadata = ".ckp"

	PrefixGCMeta   = "gc"
	PrefixSnapMeta = "snap"
	PrefixAcctMeta = "acct"
)
View Source 
const (
	TableObjectsAttr_Accout_Idx     = 0
	TableObjectsAttr_DB_Idx         = 1
	TableObjectsAttr_Table_Idx      = 2
	TableObjectsAttr_ObjectType_Idx = 3
	TableObjectsAttr_ID_Idx         = 4
	TableObjectsAttr_CreateTS_Idx   = 5
	TableObjectsAttr_DeleteTS_Idx   = 6
	TableObjectsAttr_Cluster_Idx    = 7
)
View Source 
const (
	ObjectType_Invalid int8 = iota
	ObjectType_Data
	ObjectType_Tombstone
)
View Source 
const (
	JTLoad tasks.JobType = 200 + iota
	JTFlush
)
View Source 
const DefaultInMemoryStagedSize = mpool.MB * 16

Variables

View Source 
var PipelineFlushKey = pipelineFlushKeyType{}

PipelineFlushKey is the context key used to enable pipeline flush in Sinker. Set the context value to true to enable pipeline flush for INSERT operations (e.g., ALTER TABLE ADD PRIMARY KEY).

Functions

func CheckTombstoneFile 

func CheckTombstoneFile(
	ctx context.Context,
	prefixPattern []byte,
	getTombstoneFileFn func() (*objectio.ObjectStats, error),
	onBlockSelectedFn func(*objectio.ObjectStats, int) (bool, error),
	fs fileservice.FileService,
) (
	tombstoneObjectCnt int,
	skipObjectCnt int,
	totalBlkCnt int,
	err error,
)

func CloseDefaultSinkPool 

func CloseDefaultSinkPool()

CloseDefaultSinkPool drains and closes the global sink pool, allowing a new one to be created on the next GetDefaultSinkPool call. Safe to call during process shutdown or between tests for isolation.

func CoarseFilterTombstoneObject 

func CoarseFilterTombstoneObject(
	ctx context.Context,
	nextDeletedDataObject func() *objectio.ObjectId,
	tombstoneObjects []objectio.ObjectStats,
	fs fileservice.FileService,
) (filtered []objectio.ObjectStats, err error)

CoarseFilterTombstoneObject It is used to filter out tombstone objects that do not contain any deleted data objects. This is a coarse filter using ZM, so false positives may occur

func EncodeAcctMetadataName 

func EncodeAcctMetadataName(start, end types.TS) string

func EncodeCKPMetadataFullName 

func EncodeCKPMetadataFullName(
	start, end types.TS,
) string

with dirname input: start=0, end=100 output: /ckp/meta_0_100.ckp

func EncodeCKPMetadataName 

func EncodeCKPMetadataName(
	start, end types.TS,
) string

without dirname

func EncodeCompactCKPMetadataFullName 

func EncodeCompactCKPMetadataFullName(
	start, end types.TS,
) string

func EncodeGCMetadataName 

func EncodeGCMetadataName(start, end types.TS) string

func EncodeSnapshotMetadataName 

func EncodeSnapshotMetadataName(start, end types.TS) string

func EncodeTmpFileName 

func EncodeTmpFileName(dir, prefix string, ts int64) string

func EvalDeleteMaskFromCNCreatedTombstones 

func EvalDeleteMaskFromCNCreatedTombstones(
	bid *types.Blockid,
	deletedRows *vector.Vector,
) (rows objectio.Bitmap)

func EvalDeleteMaskFromDNCreatedTombstones 

func EvalDeleteMaskFromDNCreatedTombstones(
	deletedRows *vector.Vector,
	commitTSVec *vector.Vector,
	meta objectio.BlockObject,
	ts *types.TS,
	blockid *types.Blockid,
) (rows objectio.Bitmap)

func FillBlockDeleteMask 

func FillBlockDeleteMask(
	ctx context.Context,
	snapshotTS *types.TS,
	blockId *types.Blockid,
	location objectio.Location,
	fs fileservice.FileService,
	createdByCN bool,
) (deleteMask objectio.Bitmap, err error)

func FindStartEndOfBlockFromSortedRowids 

func FindStartEndOfBlockFromSortedRowids(rowids []types.Rowid, id *types.Blockid) (start int, end int)

func FindTombstonesOfObject 

func FindTombstonesOfObject(
	ctx context.Context,
	objectId *objectio.ObjectId,
	tombstoneObjects []objectio.ObjectStats,
	fs fileservice.FileService,
) (sels bitmap.Bitmap, err error)

func GetCheckpointDir 

func GetCheckpointDir() string

func GetGCDir 

func GetGCDir() string

func GetTombstonesByBlockId 

func GetTombstonesByBlockId(
	ctx context.Context,
	ts *types.TS,
	blockId *objectio.Blockid,
	getTombstoneFileFn func() (*objectio.ObjectStats, error),
	deletedMask *objectio.Bitmap,
	fs fileservice.FileService,
) (err error)

func InheritGCMetadataName 

func InheritGCMetadataName(name string, start, end *types.TS) string

func IsCKPExt 

func IsCKPExt(ext string) bool

func IsFullName 

func IsFullName(name string) bool

func IsGCMetaExt 

func IsGCMetaExt(ext string) bool

func IsMetadataName 

func IsMetadataName(name string) bool

func IsRowDeleted 

func IsRowDeleted(
	ctx context.Context,
	ts *types.TS,
	row *types.Rowid,
	getTombstoneFileFn func() (*objectio.ObjectStats, error),
	fs fileservice.FileService,
) (bool, error)

func IsRowDeletedByLocation 

func IsRowDeletedByLocation(
	ctx context.Context,
	snapshotTS *types.TS,
	row *objectio.Rowid,
	location objectio.Location,
	fs fileservice.FileService,
	createdByCN bool,
) (deleted bool, err error)

func LoadColumns 

func LoadColumns(
	ctx context.Context,
	cols []uint16,
	typs []types.Type,
	fs fileservice.FileService,
	location objectio.Location,
	cacheVectors containers.Vectors,
	m *mpool.MPool,
	policy fileservice.Policy,
) (release func(), fromCache bool, err error)

func LoadColumns2 

func LoadColumns2(
	ctx context.Context,
	cols []uint16,
	typs []types.Type,
	fs fileservice.FileService,
	location objectio.Location,
	policy fileservice.Policy,
	needCopy bool,
	vPool *containers.VectorPool,
) (vectors []containers.Vector, release func(), err error)

LoadColumns2 load columns data from file service for TN need to copy data from vPool to avoid releasing cache

func LoadColumnsData 

func LoadColumnsData(
	ctx context.Context,
	columns []uint16,
	typs []types.Type,
	fs fileservice.FileService,
	location objectio.Location,
	cacheVectors containers.Vectors,
	m *mpool.MPool,
	policy fileservice.Policy,
) (dataMeta objectio.ObjectDataMeta, release func(), fromCache bool, err error)

func LoadColumnsData2 

func LoadColumnsData2(
	ctx context.Context,
	cols []uint16,
	typs []types.Type,
	fs fileservice.FileService,
	location objectio.Location,
	policy fileservice.Policy,
	needCopy bool,
	vPool *containers.VectorPool,
) (vectors []containers.Vector, release func(), err error)

func LoadOneBlock 

func LoadOneBlock(
	ctx context.Context,
	fs fileservice.FileService,
	key objectio.Location,
	metaType objectio.DataMetaType,
) (*batch.Batch, uint16, error)

func LoadOneBlockWithIndex 

func LoadOneBlockWithIndex(
	ctx context.Context,
	fs fileservice.FileService,
	idxes []uint16,
	key objectio.Location,
	metaType objectio.DataMetaType,
) (*batch.Batch, uint16, error)

func LoadTombstoneColumns 

func LoadTombstoneColumns(
	ctx context.Context,
	cols []uint16,
	typs []types.Type,
	fs fileservice.FileService,
	location objectio.Location,
	cacheVectors containers.Vectors,
	m *mpool.MPool,
	policy fileservice.Policy,
) (meta objectio.ObjectDataMeta, release func(), err error)

func MakeFullName 

func MakeFullName(dir, name string) string

it is very special for history reason <@_@>! here the dir should be like : ckp/ or gc/ or debug/

func MakeGCFullName 

func MakeGCFullName(name string) string

func Prefetch 

func Prefetch(
	sid string,
	service fileservice.FileService,
	key objectio.Location,
) error

func PrefetchMeta 

func PrefetchMeta(
	sid string,
	service fileservice.FileService,
	key objectio.Location,
) error

func ReadDeletes 

func ReadDeletes(
	ctx context.Context,
	deltaLoc objectio.Location,
	fs fileservice.FileService,
	isPersistedByCN bool,
	cacheVectors containers.Vectors,
	pkType *types.Type,
) (objectio.ObjectDataMeta, func(), error)

ReadDeletes will read the pk column if pk type not nil

func RunPipelineTest 

func RunPipelineTest(
	fn func(),
)

func Start 

func Start(sid string)

func Stop 

func Stop(sid string)

Types

type BlockReader 

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

func NewFileReader 

func NewFileReader(
	service fileservice.FileService,
	name string,
) (*BlockReader, error)

func NewFileReaderNoCache 

func NewFileReaderNoCache(service fileservice.FileService, name string) (*BlockReader, error)

func NewObjectReader 

func NewObjectReader(
	service fileservice.FileService,
	key objectio.Location,
	opts ...objectio.ReaderOptionFunc,
) (*BlockReader, error)

func (*BlockReader) GetName 

func (r *BlockReader) GetName() string

func (*BlockReader) LoadAllBlocks 

func (r *BlockReader) LoadAllBlocks(ctx context.Context, m *mpool.MPool) ([]objectio.BlockObject, error)

func (*BlockReader) LoadAllColumns 

func (r *BlockReader) LoadAllColumns(
	ctx context.Context,
	idxs []uint16,
	m *mpool.MPool,
) ([]*batch.Batch, func(), error)

func (*BlockReader) LoadColumns 

func (r *BlockReader) LoadColumns(
	ctx context.Context,
	cols []uint16,
	typs []types.Type,
	blk uint16,
	m *mpool.MPool,
) (bat *batch.Batch, release func(), err error)

LoadColumns needs typs to generate columns, if the target table has no schema change, nil can be passed.

func (*BlockReader) LoadObjectMeta 

func (r *BlockReader) LoadObjectMeta(ctx context.Context, m *mpool.MPool) (objectio.ObjectDataMeta, error)

func (*BlockReader) LoadOneBF 

func (r *BlockReader) LoadOneBF(
	ctx context.Context,
	blk uint16,
) (objectio.StaticFilter, uint32, error)

func (*BlockReader) LoadOneSubColumns 

func (r *BlockReader) LoadOneSubColumns(
	ctx context.Context,
	cols []uint16,
	typs []types.Type,
	dataType uint16,
	blk uint16,
	m *mpool.MPool,
) (bat *batch.Batch, release func(), err error)

LoadColumns needs typs to generate columns, if the target table has no schema change, nil can be passed.

func (*BlockReader) LoadSubColumns 

func (r *BlockReader) LoadSubColumns(
	ctx context.Context,
	cols []uint16,
	typs []types.Type,
	blk uint16,
	m *mpool.MPool,
) (bats []*batch.Batch, releases func(), err error)

LoadColumns needs typs to generate columns, if the target table has no schema change, nil can be passed.

func (*BlockReader) LoadZoneMap 

func (r *BlockReader) LoadZoneMap(
	ctx context.Context,
	seqnums []uint16,
	block objectio.BlockObject,
	m *mpool.MPool) ([]objectio.ZoneMap, error)

func (*BlockReader) LoadZoneMaps 

func (r *BlockReader) LoadZoneMaps(
	ctx context.Context,
	seqnums []uint16,
	id uint16,
	m *mpool.MPool,
) ([]objectio.ZoneMap, error)

type BlockWriter 

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

func ConstructTombstoneWriter 

func ConstructTombstoneWriter(
	hiddenSelection objectio.HiddenColumnSelection,
	fs fileservice.FileService,
) *BlockWriter

ConstructTombstoneWriter hiddenSelection: true for add hidden columns `commitTs` and `abort` object file created by `CN` with `hiddenSelection` is false

func ConstructTombstoneWriterWithArena 

func ConstructTombstoneWriterWithArena(
	hiddenSelection objectio.HiddenColumnSelection,
	fs fileservice.FileService,
	arena *objectio.WriteArena,
) *BlockWriter

func ConstructWriter 

func ConstructWriter(
	ver uint32,
	seqnums []uint16,
	sortkeyPos int,
	sortkeyIsPK bool,
	isTombstone bool,
	fs fileservice.FileService,
) *BlockWriter

func ConstructWriterWithArena 

func ConstructWriterWithArena(
	ver uint32,
	seqnums []uint16,
	sortkeyPos int,
	sortkeyIsPK bool,
	isTombstone bool,
	fs fileservice.FileService,
	arena *objectio.WriteArena,
) *BlockWriter

func ConstructWriterWithSegmentID 

func ConstructWriterWithSegmentID(
	segmentID *objectio.Segmentid,
	num uint16,
	ver uint32,
	seqnums []uint16,
	sortkeyPos int,
	sortkeyIsPK bool,
	isTombstone bool,
	fs fileservice.FileService,
	arena *objectio.WriteArena,
) *BlockWriter

func NewBlockWriter 

func NewBlockWriter(fs fileservice.FileService, name string) (*BlockWriter, error)

func NewBlockWriterNew 

func NewBlockWriterNew(
	fs fileservice.FileService,
	name objectio.ObjectName,
	schemaVer uint32,
	seqnums []uint16,
	isTombstone bool,
) (*BlockWriter, error)

seqnums is the column's seqnums of the batch written by `WriteBatch`. `WriteBatchWithoutIndex` will ignore the seqnums

func NewBlockWriterWithArena 

func NewBlockWriterWithArena(
	fs fileservice.FileService,
	name objectio.ObjectName,
	schemaVer uint32,
	seqnums []uint16,
	isTombstone bool,
	arena *objectio.WriteArena,
) (*BlockWriter, error)

func (*BlockWriter) GetName 

func (w *BlockWriter) GetName() objectio.ObjectName

func (*BlockWriter) GetObjectStats 

func (w *BlockWriter) GetObjectStats(opts ...objectio.ObjectStatsOptions) objectio.ObjectStats

func (*BlockWriter) GetWrittenOriginalSize 

func (w *BlockWriter) GetWrittenOriginalSize() uint32

func (*BlockWriter) SetAppendable 

func (w *BlockWriter) SetAppendable()

func (*BlockWriter) SetFakePK 

func (w *BlockWriter) SetFakePK(idx uint16)

SetFakePK marks a fake PK column so its NDV is set to totalRows in Sync.

func (*BlockWriter) SetPrimaryKey 

func (w *BlockWriter) SetPrimaryKey(idx uint16)

func (*BlockWriter) SetPrimaryKeyWithType 

func (w *BlockWriter) SetPrimaryKeyWithType(idx uint16, pkType uint8, prefix ...index.PrefixFn)

func (*BlockWriter) SetSortKey 

func (w *BlockWriter) SetSortKey(idx uint16)

func (*BlockWriter) SetTombstone 

func (w *BlockWriter) SetTombstone()

func (*BlockWriter) Stats 

func (w *BlockWriter) Stats() objectio.ObjectStats

func (*BlockWriter) String 

func (w *BlockWriter) String(
	blocks []objectio.BlockObject,
) string

func (*BlockWriter) Sync 

func (w *BlockWriter) Sync(ctx context.Context) ([]objectio.BlockObject, objectio.Extent, error)

func (*BlockWriter) WriteBatch 

func (w *BlockWriter) WriteBatch(batch *batch.Batch) (objectio.BlockObject, error)

WriteBatch write a batch whose schema is decribed by seqnum in NewBlockWriterNew, write batch to memroy cache, not S3

func (*BlockWriter) WriteSubBatch 

func (w *BlockWriter) WriteSubBatch(batch *batch.Batch, dataType objectio.DataMetaType) (objectio.BlockObject, int, error)

type FSinkerImpl 

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

func NewFSinkerImpl 

func NewFSinkerImpl(
	seqnums []uint16,
	sortKeyPos int,
	isPrimaryKey bool,
	isTombstone bool,
	schemaVersion uint32,
	mp *mpool.MPool,
	fs fileservice.FileService,
) *FSinkerImpl

func NewTombstoneFSinkerImpl 

func NewTombstoneFSinkerImpl(
	hidden objectio.HiddenColumnSelection,
	mp *mpool.MPool,
	fs fileservice.FileService,
) *FSinkerImpl

func (*FSinkerImpl) Close 

func (s *FSinkerImpl) Close() error

func (*FSinkerImpl) Reset 

func (s *FSinkerImpl) Reset()

func (*FSinkerImpl) Sink 

func (s *FSinkerImpl) Sink(ctx context.Context, b *batch.Batch) error

func (*FSinkerImpl) Sync 

func (s *FSinkerImpl) Sync(ctx context.Context) (*objectio.ObjectStats, error)

type FetchFunc 

type FetchFunc = func(ctx context.Context, params fetchParams) (any, error)

type FileSinker 

type FileSinker interface {
	Sink(context.Context, *batch.Batch) error
	Sync(context.Context) (*objectio.ObjectStats, error)
	Reset()
	Close() error
}

type FileSinkerFactory 

type FileSinkerFactory func(*mpool.MPool, fileservice.FileService) FileSinker

FileSinkerFactory is a factory function to create a FileSinker

func NewFSinkerImplFactory 

func NewFSinkerImplFactory(
	seqnums []uint16,
	sortKeyPos int,
	isPrimaryKey bool,
	isTombstone bool,
	schemaVersion uint32,
) FileSinkerFactory

This factory is used to create a FileSinker for all kinds of objects the user should provide the seqnums, sortKeyPos, isPrimaryKey, isTombstone, schemaVersion

func NewTombstoneFSinkerImplFactory 

func NewTombstoneFSinkerImplFactory(hidden objectio.HiddenColumnSelection) FileSinkerFactory

This factory is used to create a FileSinker for tombstone hiddenSelection: whether to write hidden tombstone 

true: TN created tombstone objects
false: CN created tombstone objects

type IOJobFactory 

type IOJobFactory func(context.Context, fetchParams) *tasks.Job

type IoPipeline 

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

func MustGetPipeline 

func MustGetPipeline(sid string) *IoPipeline

func NewIOPipeline 

func NewIOPipeline(
	opts ...Option,
) *IoPipeline

func (*IoPipeline) Fetch 

func (p *IoPipeline) Fetch(
	ctx context.Context,
	params fetchParams,
) (res any, err error)

func (*IoPipeline) Prefetch 

func (p *IoPipeline) Prefetch(params PrefetchParams) (err error)

func (*IoPipeline) Start 

func (p *IoPipeline) Start()

func (*IoPipeline) Stop 

func (p *IoPipeline) Stop()

type ObjectColumnMetasBuilder 

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

func NewObjectColumnMetasBuilder 

func NewObjectColumnMetasBuilder(colIdx int) *ObjectColumnMetasBuilder

func (*ObjectColumnMetasBuilder) AddPKData 

func (b *ObjectColumnMetasBuilder) AddPKData(data containers.Vector)

func (*ObjectColumnMetasBuilder) AddRowCnt 

func (b *ObjectColumnMetasBuilder) AddRowCnt(rows int)

func (*ObjectColumnMetasBuilder) Build 

func (b *ObjectColumnMetasBuilder) Build() (uint32, []objectio.ColumnMeta)

func (*ObjectColumnMetasBuilder) GetPKData 

func (b *ObjectColumnMetasBuilder) GetPKData() []containers.Vector

func (*ObjectColumnMetasBuilder) GetTotalRow 

func (b *ObjectColumnMetasBuilder) GetTotalRow() uint32

func (*ObjectColumnMetasBuilder) InspectVector 

func (b *ObjectColumnMetasBuilder) InspectVector(idx int, vec containers.Vector, isPK bool)

func (*ObjectColumnMetasBuilder) SetPKNdv 

func (b *ObjectColumnMetasBuilder) SetPKNdv(idx uint16, ndv uint32)

func (*ObjectColumnMetasBuilder) UpdateZm 

func (b *ObjectColumnMetasBuilder) UpdateZm(idx int, zm index.ZM)

type Option 

type Option func(*IoPipeline)

func WithFetchParallism 

func WithFetchParallism(num int) Option

func WithJobFactory 

func WithJobFactory(factory IOJobFactory) Option

func WithPrefetchParallism 

func WithPrefetchParallism(num int) Option

type PrefetchFunc 

type PrefetchFunc = func(params PrefetchParams) error

type PrefetchParams 

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

PrefetchParams is the parameter of the executed IoPipeline.PrefetchParams, which provides the merge function, which can merge the PrefetchParams requests of multiple blocks in an object/file

func BuildPrefetchParams 

func BuildPrefetchParams(service fileservice.FileService, key objectio.Location) (PrefetchParams, error)

type PrefetchType 

type PrefetchType uint8
const (
	PrefetchFileType PrefetchType = iota
	PrefetchMetaType
)

type SinkPool 

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

SinkPool is a global worker pool for parallelizing Sinker serialization (Sink) and IO (Sync) operations. All Sinkers with pipeline mode enabled share the same pool, so concurrent queries do not multiply memory usage.

Architecture: 

Main goroutine → [sinkChan] → N sink workers → [syncChan] → M sync workers

Memory is bounded globally: 

(sinkChan cap + N) × 128MB raw + (syncChan cap + M) × compressed size

func GetDefaultSinkPool 

func GetDefaultSinkPool() *SinkPool

GetDefaultSinkPool returns the process-global sink pool, creating it lazily. Sink workers cap serialization fanout, while sync workers cap concurrent IO.

func NewSinkPool 

func NewSinkPool(sinkWorkers, syncWorkers int) *SinkPool

NewSinkPool creates a new worker pool with the given number of serialization (sink) workers and IO (sync) workers.

func (*SinkPool) Close 

func (p *SinkPool) Close()

Close shuts down the pool, waiting for all workers to finish.

func (*SinkPool) Submit 

func (p *SinkPool) Submit(job *poolSinkJob) error

Submit enqueues a sink job for async processing. The job's result.pending WaitGroup is incremented; it will be decremented when the job fully completes (after both Sink and Sync).

type Sinker 

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

func NewSinker 

func NewSinker(
	sortKeyIdx int,
	attrs []string,
	attrTypes []types.Type,
	factory FileSinkerFactory,
	mp *mpool.MPool,
	fs fileservice.FileService,
	opts ...SinkerOption,
) *Sinker

func NewTombstoneSinker 

func NewTombstoneSinker(
	hidden objectio.HiddenColumnSelection,
	pkType types.Type,
	mp *mpool.MPool,
	fs fileservice.FileService,
	opts ...SinkerOption,
) *Sinker

func (*Sinker) Close 

func (sinker *Sinker) Close() error

func (*Sinker) GetInMemoryData 

func (sinker *Sinker) GetInMemoryData() []*batch.Batch

func (*Sinker) GetInMemoryThreshold 

func (sinker *Sinker) GetInMemoryThreshold() int

func (*Sinker) GetMPool 

func (sinker *Sinker) GetMPool() *mpool.MPool

func (*Sinker) GetResult 

func (sinker *Sinker) GetResult() ([]objectio.ObjectStats, []*batch.Batch)

func (*Sinker) Reset 

func (sinker *Sinker) Reset()

Reset discards all staged and result data without tearing down the sinker. It is safe to call after an aborted pipeline execution so that the next Write/Sync cycle starts from a clean state. The buf pool, file-sinker executor (and its arena), mpool, and fileservice references are all kept alive so they can be reused without reallocation.

func (*Sinker) StagedSize 

func (sinker *Sinker) StagedSize() int

StagedSize returns the current size of in-memory staged data.

func (*Sinker) String 

func (sinker *Sinker) String() string

func (*Sinker) Sync 

func (sinker *Sinker) Sync(ctx context.Context) error

func (*Sinker) SyncAndTakeResults 

func (sinker *Sinker) SyncAndTakeResults(ctx context.Context) ([]objectio.ObjectStats, []*batch.Batch, error)

SyncAndTakeResults calls Sync, then extracts and returns the accumulated results, resetting the sinker so it can accept more Write calls.

func (*Sinker) TakeStagedBatches 

func (sinker *Sinker) TakeStagedBatches() []*batch.Batch

TakeStagedBatches extracts the un-synced in-memory staged batches, transferring ownership to the caller. The sinker's staged area is cleared.

func (*Sinker) Write 

func (sinker *Sinker) Write(
	ctx context.Context,
	data *batch.Batch,
) (err error)

Write always copy the data

func (*Sinker) WriteOwned 

func (sinker *Sinker) WriteOwned(
	ctx context.Context,
	data *batch.Batch,
) (owned bool, err error)

WriteOwned stages a full-block batch directly without copying. The caller transfers ownership: the sinker will clean the batch later. If the batch is not exactly BlockMaxRows or there is a partially-filled staged buffer, falls back to Write (which copies).

type SinkerOption 

type SinkerOption func(*Sinker)

func WithAllMergeSorted 

func WithAllMergeSorted() SinkerOption

func WithBuffer 

func WithBuffer(buffer containers.IBatchBuffer, isOwner bool) SinkerOption

func WithDedupAll 

func WithDedupAll() SinkerOption

func WithMemorySizeThreshold 

func WithMemorySizeThreshold(size int) SinkerOption

func WithOffHeap 

func WithOffHeap() SinkerOption

func WithPipelineFlush 

func WithPipelineFlush() SinkerOption

func WithTailSizeCap 

func WithTailSizeCap(size int) SinkerOption

type TSRangeFile 

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

func DecodeCKPMetaName 

func DecodeCKPMetaName(name string) (meta TSRangeFile)

func DecodeGCMetadataName 

func DecodeGCMetadataName(name string) (ret TSRangeFile)

func DecodeTSRangeFile 

func DecodeTSRangeFile(name string) (meta TSRangeFile)

name is expected to be like `meta_<start_ts>_<end_ts>.<ext>`

func ListTSRangeFiles 

func ListTSRangeFiles(
	ctx context.Context,
	dir string,
	fs fileservice.FileService,
) (files []TSRangeFile, err error)

func ListTSRangeFilesInGCDir 

func ListTSRangeFilesInGCDir(
	ctx context.Context,
	fs fileservice.FileService,
) (files []TSRangeFile, err error)

func NewTSRangeFile 

func NewTSRangeFile(name, ext string, start, end types.TS) TSRangeFile

func TryDecodeTSRangeFile 

func TryDecodeTSRangeFile(name string) (dir string, ret TSRangeFile)

name may contain dir name name is expected to be a encoded from TSRangeFile

func (*TSRangeFile) AsString 

func (m *TSRangeFile) AsString(prefix string) string

func (*TSRangeFile) GetCKPFullName 

func (m *TSRangeFile) GetCKPFullName() string

func (*TSRangeFile) GetEnd 

func (m *TSRangeFile) GetEnd() *types.TS

func (*TSRangeFile) GetExt 

func (m *TSRangeFile) GetExt() string

func (*TSRangeFile) GetGCFullName 

func (m *TSRangeFile) GetGCFullName() string

func (*TSRangeFile) GetIdx 

func (m *TSRangeFile) GetIdx() int

func (*TSRangeFile) GetName 

func (m *TSRangeFile) GetName() string

func (*TSRangeFile) GetStart 

func (m *TSRangeFile) GetStart() *types.TS

func (TSRangeFile) IsAcctExt 

func (m TSRangeFile) IsAcctExt() bool

func (TSRangeFile) IsCKPFile 

func (m TSRangeFile) IsCKPFile() bool

func (TSRangeFile) IsCompactExt 

func (m TSRangeFile) IsCompactExt() bool

func (TSRangeFile) IsFullGCExt 

func (m TSRangeFile) IsFullGCExt() bool

func (TSRangeFile) IsMetadataFile 

func (m TSRangeFile) IsMetadataFile() bool

func (TSRangeFile) IsSnapshotExt 

func (m TSRangeFile) IsSnapshotExt() bool

func (TSRangeFile) IsValid 

func (m TSRangeFile) IsValid() bool

func (*TSRangeFile) RangeEqual 

func (m *TSRangeFile) RangeEqual(start, end *types.TS) bool

func (*TSRangeFile) SetExt 

func (m *TSRangeFile) SetExt(ext string)

func (*TSRangeFile) SetIdx 

func (m *TSRangeFile) SetIdx(idx int)

func (*TSRangeFile) SetInvalid 

func (m *TSRangeFile) SetInvalid()

Source Files

View all Source files

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.