immutable

package

Go to main page

Versions in this module

v1

v1.5.3-rc4

Jan 15, 2026

v1.5.2

Jan 9, 2026

v1.5.1

Dec 24, 2025

v1.5.0

Nov 27, 2025

Changes in this version

+ const SkipParquetTask

+ const TSSPFileNameSize

+ const TsspFileSuffix

+ func AppendFragmentsToPKRecord(dst *record.Record, fragments []int64)

+ func BuildPKFilePathFromTSSP(path string) string

+ func DirDepth(path string) int

+ func GetUnorderDir(mstPath string) string

+ func RemoveTmpSuffix(s string) string

+ func RenameIndexFilePathFromTSSP(old string, tsspFile string) string

+ func RenameMergeFiles(srcName TSSPFileName, addSeq uint64) string

+ func SetColStoreCompactorFactory(fn func(tb *MmsTables, mst string) ColStoreCompactor)

+ func ShardInParquetProcess(path string) bool

type BaseFilterOptions

+ func (opts *BaseFilterOptions) Init(fieldSize int)

+ type CSConsumeIterator struct

+ func NewCSConsumeIterator(its []*CSFileIterator) *CSConsumeIterator

+ func (csi *CSConsumeIterator) Next() (*record.ConsumeRecord, error)

+ func (csi *CSConsumeIterator) Release()

+ type CSFileIterator struct

+ func NewCSFileIterator(file TSSPFile, skSchema record.Schemas, tcDuration int64) (*CSFileIterator, error)

+ func (ci *CSFileIterator) AppendRecordRowTo(dst *record.Record, rowIdx int)

+ func (ci *CSFileIterator) NextFragment() bool

+ func (ci *CSFileIterator) NextSegment() error

+ func (ci *CSFileIterator) ReadAll() *record.Record

+ func (ci *CSFileIterator) ReadFragmentPK() map[string]any

+ func (ci *CSFileIterator) RemainRecordRow() int

+ func (ci *CSFileIterator) SegmentRecord() *record.Record

+ type CSIteratorCursor struct

+ func (c *CSIteratorCursor) Init(fragments []int64)

+ func (c *CSIteratorCursor) NextFragment() (int, bool)

+ func (c *CSIteratorCursor) NextSegment() (int, bool)

+ type CSTSSPFile interface

+ GetPkInfo func() *colstore.PKInfo

+ GetSkipIndexInfo func() []*colstore.SkipIndexInfo

+ SetPkInfo func(pkInfo *colstore.PKInfo)

+ SetSkipIndexInfo func(info []*colstore.SkipIndexInfo)

type ChunkIterator

+ func (c *ChunkIterator) WithSchema(schema record.Schemas)

type ChunkIterators

+ func NewChunkIterators(files []TSSPFile, dropping int64, signal chan struct{}, lg *Log.Logger, ...) *ChunkIterators

type ChunkMeta

+ func (m *ChunkMeta) FilterColMeta(filterMap map[string]int)

+ func (m *ChunkMeta) SetColMeta(newColMeta []ColumnMeta)

type ChunkMetaCodecCtx

+ func NewChunkMetaCodecCtx() *ChunkMetaCodecCtx

type ChunkMetaContext

+ func (ctx *ChunkMetaContext) InitColumns(schema record.Schemas)

+ type ChunkMetaWriter struct

+ func NewChunkMetaWriter() *ChunkMetaWriter

+ func (cmw *ChunkMetaWriter) FlushChunkMeta(w fileops.FileWriter) (int, error)

+ func (cmw *ChunkMetaWriter) FlushMetaIndex(w fileops.FileWriter) (int, error)

+ func (cmw *ChunkMetaWriter) GetChunkMetaHeader() *ChunkMetaHeader

+ func (cmw *ChunkMetaWriter) MemoryCount() int

+ func (cmw *ChunkMetaWriter) MemorySize() int

+ func (cmw *ChunkMetaWriter) MetaIndexCount() int

+ func (cmw *ChunkMetaWriter) Reset()

+ func (cmw *ChunkMetaWriter) Size() int64

+ func (cmw *ChunkMetaWriter) UpdateMetaIndex(minT, maxT int64, sid uint64)

+ func (cmw *ChunkMetaWriter) Write(cm *ChunkMeta) (int, error)

+ type ColStoreCompactor interface

+ Compact func(ident colstore.MeasurementIdent, files []TSSPFile) ([]TSSPFile, error)

+ func NewColStoreCompactor(tb *MmsTables, mst string) ColStoreCompactor

+ type ColStoreIndexWriter struct

+ func NewColStoreIndexWriter(dir, mst string, fileName TSSPFileName, lock *string, ...) *ColStoreIndexWriter

+ func (iw *ColStoreIndexWriter) BuildSkipIndexInfo() []*colstore.SkipIndexInfo

+ func (iw *ColStoreIndexWriter) Flush() error

+ func (iw *ColStoreIndexWriter) MustClose()

+ func (iw *ColStoreIndexWriter) Open()

+ func (iw *ColStoreIndexWriter) Write(ref record.Field, col *record.ColVal) error

+ type ColumnStoreTSSPWriter struct

+ func NewColumnStoreTSSPWriter(sw *StreamWriteFile, mst *colstore.Measurement) *ColumnStoreTSSPWriter

+ func (b *ColumnStoreTSSPWriter) WriteRecord(rec *record.Record) (*ColumnStoreWriteResult, error)

+ type ColumnStoreWriteResult struct

+ File TSSPFile

+ Fragments []int64

+ IndexFilePath string

+ PKRecord *record.Record

+ type DetachedFragmentIterator interface

type FragmentIterator

+ func NewFragmentIterator(skSchema record.Schemas) *FragmentIterator

+ func (fi *FragmentIterator) AddCSFileIterator(it *CSFileIterator)

+ func (fi *FragmentIterator) AppendTo(dst *record.Record, batchSize int) error

+ func (fi *FragmentIterator) BuildCompactedSchema(its []*CSFileIterator)

+ func (fi *FragmentIterator) Reset()

+ type FragmentIteratorBuilder struct

+ func NewFragmentIteratorBuilder(its []*CSFileIterator, skSchema record.Schemas) *FragmentIteratorBuilder

+ func (b *FragmentIteratorBuilder) BuildNext() (*FragmentIterator, []byte, error)

+ func (b *FragmentIteratorBuilder) Swap(i, j int)

+ type HotFile interface

+ FreeMemory func()

+ InMemSize func() int64

+ MinMaxTime func() (int64, int64, error)

+ type HotFiles struct

+ func (fs *HotFiles) Append(f HotFile)

type IndexMergeSet

+ HasDeletedTSID func(tsid uint64) bool

type MeasurementInProcess

+ func (m *MeasurementInProcess) MatchPrefix(prefix string) bool

type MmsIdTime

+ func (m *MmsIdTime) CreateIdInfo(id uint64) (info *idInfo)

+ func (m *MmsIdTime) Size() int

type MmsTables

+ func (m *MmsTables) ClearOldTsspFiles(oldOrders map[string][]TSSPFile, oldUnOrders map[string][]TSSPFile) error

+ func (m *MmsTables) GetCSFilesRef(name string) ([]TSSPFile, bool)

+ func (m *MmsTables) GetLockPath() *string

+ func (m *MmsTables) Path() string

+ func (m *MmsTables) ReplaceByNoClearShardId(supplier clearevent.NoClearShardAndIndexSupplier) (map[string][]TSSPFile, map[string][]TSSPFile, error)

+ func (m *MmsTables) SetDbRp(db, rp string)

type MsBuilder

+ func (b *MsBuilder) WritePrimaryIndex(writeRec *record.Record, pkSchema record.Schemas, filepath, lockpath string, ...) error

type Segment

+ func (s *Segment) OffsetSize() (int64, uint32)

type StreamWriteFile

+ func NewStreamWriteFile(mst string, Conf *Config, dir string, signal chan struct{}, lock *string) *StreamWriteFile

+ func (c *StreamWriteFile) Clean()

+ func (c *StreamWriteFile) GetDir() string

+ func (c *StreamWriteFile) GetFileName() TSSPFileName

+ func (c *StreamWriteFile) IdTimePairs() *IdTimePairs

+ func (c *StreamWriteFile) InitCompactFile(f TSSPFile) error

+ func (c *StreamWriteFile) InitFlushFile(seq uint64, order bool) error

+ func (c *StreamWriteFile) WriteRecord(sid uint64, rec *record.Record) error

type TSSPFileName

+ func (n *TSSPFileName) AddSeq(seq uint64)

+ func (n *TSSPFileName) GetExtend() uint16

+ func (n *TSSPFileName) GetLevel() uint16

+ func (n *TSSPFileName) GetMerge() uint16

+ func (n *TSSPFileName) GetSeq() uint64

+ func (n *TSSPFileName) SetLock(l *string)

type TSSPFiles

+ func (f *TSSPFiles) Copy() []TSSPFile

+ func (f *TSSPFiles) GetFilesAndUnref() []TSSPFile

type TablesStore

+ ClearOldTsspFiles func(map[string][]TSSPFile, map[string][]TSSPFile) error

+ GetCSFilesRef func(measurement string) ([]TSSPFile, bool)

+ NewStreamWriteFile func(mst string) *StreamWriteFile

+ ReplaceByNoClearShardId func(supplier clearevent.NoClearShardAndIndexSupplier) (map[string][]TSSPFile, map[string][]TSSPFile, error)

type UnorderedReader

+ func (r *UnorderedReader) MatchMaxTime(maxTime int64) int64

v1.4.3

Sep 1, 2025

v1.4.2

Sep 1, 2025

v1.4.1

Jun 5, 2025

Changes in this version

type ChunkMetaCodecCtx

+ func (ctx *ChunkMetaCodecCtx) Instance() *ChunkMetaCodecCtx

type FirstLastReader

+ func (r *FirstLastReader) ReadTime(ctx *ReadContext, copied bool, ioPriority int) error

type HotFileWriter

+ func (w *HotFileWriter) Instance() *HotFileWriter

v1.4.0

Apr 12, 2025

Changes in this version

+ const ChunkMetaCompressSelf

+ var DDLRespDataFactory = make(map[hybridqp.DDLType]GenDDLRespDataFunc)

+ var DDLSequenceHandlerFactory = make(map[hybridqp.DDLType]GenSequenceHandlerFunc)

+ var ErrParquetStopped = errors.New("parquet task stopped")

+ func AddTSSP2ParquetProcess(files ...string)

+ func DecodeAggTimes(buf []byte) ([]byte, int64, int64, error)

+ func DelTSSP2ParquetProcess(files ...string)

+ func FilesMergedTire(files []TSSPFile) uint64

+ func IsChunkMetaCompressSelf() bool

+ func MarshalChunkMeta(ctx *ChunkMetaCodecCtx, cm *ChunkMeta, dst []byte) ([]byte, error)

+ func MarshalColumnMeta(ctx *ChunkMetaCodecCtx, col *ColumnMeta, dst []byte) []byte

+ func MarshalTimeRange(ctx *ChunkMetaCodecCtx, sr []SegmentRange, dst []byte) []byte

+ func NewFileLoadContext() *fileLoadContext

+ func RefFilesReader(files ...TSSPFile)

+ func RegistryDDLRespData(ddl hybridqp.DDLType, f GenDDLRespDataFunc)

+ func RegistryDDLSequenceHandler(ddl hybridqp.DDLType, f GenSequenceHandlerFunc)

+ func ReloadSpecifiedFiles(m *MmsTables, mst string, tsspFiles *TSSPFiles)

+ func ResetQueryFileCache()

+ func SearchChunkMetaBlock(data []byte, itemCount uint32, sid uint64) []byte

+ func UnmarshalChunkMeta(ctx *ChunkMetaCodecCtx, cm *ChunkMeta, buf []byte) ([]byte, error)

+ func UnmarshalChunkMetaAdaptive(ctx *ChunkMetaCodecCtx, cm *ChunkMeta, columns []string, buf []byte) ([]byte, error)

+ func UnmarshalChunkMetaBaseAttr(ctx *ChunkMetaCodecCtx, cm *ChunkMeta, buf []byte) ([]byte, error)

+ func UnmarshalChunkMetaWithColumns(ctx *ChunkMetaCodecCtx, cm *ChunkMeta, columns []string, buf []byte) ([]byte, error)

+ func UnmarshalColumnMeta(ctx *ChunkMetaCodecCtx, segCount int, col *ColumnMeta, buf []byte) ([]byte, error)

+ func UnmarshalColumnMetaWithoutName(ctx *ChunkMetaCodecCtx, segCount int, col *ColumnMeta, buf []byte) ([]byte, error)

+ func UnmarshalColumnName(ctx *ChunkMetaCodecCtx, buf []byte) ([]byte, string)

+ func UnmarshalTimeRange(ctx *ChunkMetaCodecCtx, cm *ChunkMeta, buf []byte) ([]byte, bool)

+ func UseIndexCompressWriter() bool

+ type CSParquetManager struct

+ func NewCSParquetManager() *CSParquetManager

+ func (m *CSParquetManager) Convert(files []TSSPFile, db string, rp string, mst string)

+ func (m *CSParquetManager) Recover()

+ func (m *CSParquetManager) Run()

+ func (m *CSParquetManager) Stop()

+ func (m *CSParquetManager) Wait()

+ type CSParquetPlan struct

+ DstFile string

+ Id uint64

+ Mst string

+ TSSPFile string

+ func (p *CSParquetPlan) BeforeRun() error

+ func (p *CSParquetPlan) IterRecord(handler func(*record.Record) error) error

+ func (p *CSParquetPlan) SetLogFile(file string)

+ type CSParquetTask struct

+ func (t *CSParquetTask) Execute()

+ func (t *CSParquetTask) LockFiles()

+ func (t *CSParquetTask) Stop()

+ func (t *CSParquetTask) UnLockFiles()

type ChunkIterator

+ func (c *ChunkIterator) IncrChunkUsed()

type ChunkMeta

+ func (m *ChunkMeta) Validation()

+ type ChunkMetaCodecCtx struct

+ func GetChunkMetaCodecCtx() *ChunkMetaCodecCtx

+ func (ctx *ChunkMetaCodecCtx) GetHeader() *ChunkMetaHeader

+ func (ctx *ChunkMetaCodecCtx) GetIndex(val string) uint64

+ func (ctx *ChunkMetaCodecCtx) GetValue(idx int) string

+ func (ctx *ChunkMetaCodecCtx) MemSize() int

+ func (ctx *ChunkMetaCodecCtx) Release()

+ func (ctx *ChunkMetaCodecCtx) SetTrailer(trailer *Trailer)

type ChunkMetaContext

+ func (ctx *ChunkMetaContext) CodecCtx() *ChunkMetaCodecCtx

+ type ChunkMetaHeader struct

+ func (h *ChunkMetaHeader) AppendValue(val string)

+ func (h *ChunkMetaHeader) CopyTo(dst *ChunkMetaHeader)

+ func (h *ChunkMetaHeader) GetValue(idx int) string

+ func (h *ChunkMetaHeader) Len() int

+ func (h *ChunkMetaHeader) Marshal(dst []byte) []byte

+ func (h *ChunkMetaHeader) Reset()

+ func (h *ChunkMetaHeader) Unmarshal(buf []byte)

+ type DDLBasePlan interface

+ Execute func(dst map[string]DDLRespData, mstKeys map[string][][]byte, ...) error

+ Stop func()

+ func NewDDLBasePlan(table TablesStore, idx IndexMergeSet, logger *Log.Logger, sh EngineShard, ...) DDLBasePlan

+ type DDLRespData interface

+ Add func(key, value string)

+ Count func() int

+ ForEach func(process func(key, value string))

+ func NewSeriesKeys() DDLRespData

+ type ExtraData struct

+ ChunkMetaCompressFlag uint8

+ ChunkMetaHeader *ChunkMetaHeader

+ TimeStoreFlag uint8

+ func (e *ExtraData) CopyTo(dst *ExtraData)

+ func (e *ExtraData) EnableTimeStore()

+ func (e *ExtraData) MarshalExtraData(dst []byte) []byte

+ func (e *ExtraData) Reset()

+ func (e *ExtraData) SetChunkMetaCompressFlag(v uint8)

+ func (e *ExtraData) SetChunkMetaHeader(header *ChunkMetaHeader)

+ func (e *ExtraData) UnmarshalExtraData(src []byte) ([]byte, error)

type FirstLastReader

+ func (r *FirstLastReader) ReadMaxFromPreAgg(colMeta *ColumnMeta) (interface{}, int64, bool)

+ func (r *FirstLastReader) ReadMinFromPreAgg(colMeta *ColumnMeta) (interface{}, int64, bool)

type FloatPreAgg

+ func (m *FloatPreAgg) VLCDecode(src []byte) ([]byte, error)

+ func (m *FloatPreAgg) VLCEncode(dst []byte) []byte

+ type GenDDLRespDataFunc func() DDLRespData

+ func GetDDLRespData(ddl hybridqp.DDLType) GenDDLRespDataFunc

+ type GenSequenceHandlerFunc func(idx IndexMergeSet, condition influxql.Expr, tr *util.TimeRange, limit int) SequenceIteratorHandler

+ func GetDDLDDLSequenceHandler(ddl hybridqp.DDLType) GenSequenceHandlerFunc

+ type HotFileManager struct

+ func NewHotFileManager() *HotFileManager

+ func (m *HotFileManager) Add(f TSSPFile)

+ func (m *HotFileManager) AddAll(files []TSSPFile)

+ func (m *HotFileManager) AllocLoadMemory(size int64) bool

+ func (m *HotFileManager) BackgroundFree()

+ func (m *HotFileManager) Free()

+ func (m *HotFileManager) InHotDuration(f TSSPFile) bool

+ func (m *HotFileManager) IncrMemorySize(size int64)

+ func (m *HotFileManager) Run()

+ func (m *HotFileManager) SetMaxMemorySize(size int64)

+ func (m *HotFileManager) Stop()

+ type HotFileReader struct

+ func NewHotFileReader(r fileops.BasicFileReader, buf []byte) *HotFileReader

+ func (r *HotFileReader) IsMmapRead() bool

+ func (r *HotFileReader) ReadAt(off int64, size uint32, dstPtr *[]byte, ioPriority int) ([]byte, error)

+ func (r *HotFileReader) Release()

+ func (r *HotFileReader) Size() (int64, error)

+ type HotFileWriter struct

+ func NewHotFileWriter(w fileops.FileWriter) *HotFileWriter

+ func (w *HotFileWriter) AppendChunkMetaToData() error

+ func (w *HotFileWriter) BuildHotFileReader(r fileops.BasicFileReader) *HotFileReader

+ func (w *HotFileWriter) MemSize() int

+ func (w *HotFileWriter) Release()

+ func (w *HotFileWriter) WriteChunkMeta(b []byte) (int, error)

+ func (w *HotFileWriter) WriteData(b []byte) (int, error)

type IdTimePairs

+ func (p *IdTimePairs) UnmarshalBlocks(decTimes bool, src []byte, startIdx int, decoder *encoding.CoderContext) (int, int, int, error)

+ func (p *IdTimePairs) UnmarshalHeader(src []byte) (uint32, error)

type IndexMergeSet

+ GetSeriesBytes func(sid uint64, buf []byte, condition influxql.Expr, ...) error

+ SearchSeriesKeys func(series [][]byte, name []byte, condition influxql.Expr) ([][]byte, error)

type IntegerPreAgg

+ func (m *IntegerPreAgg) VLCDecode(src []byte) ([]byte, error)

+ func (m *IntegerPreAgg) VLCEncode(dst []byte) []byte

type MetaIndex

+ func (m *MetaIndex) GetCount() uint32

type MsBuilder

+ func (b *MsBuilder) SetToHot()

type ParquetTask

+ func (t *ParquetTask) GetTagKeys() (map[string]struct{}, error)

+ func (t *ParquetTask) LockFiles()

+ func (t *ParquetTask) RemoveLog()

+ func (t *ParquetTask) Stop()

+ func (t *ParquetTask) UnLockFiles()

type SequenceIteratorHandler

+ func NewSeriesKeysIteratorHandler(idx IndexMergeSet, condition influxql.Expr, tr *util.TimeRange, limit int) SequenceIteratorHandler

+ type SeriesKeys struct

+ func (s *SeriesKeys) Add(key, _ string)

+ func (s *SeriesKeys) Count() int

+ func (s *SeriesKeys) ForEach(process func(key, _ string))

+ type SeriesKeysIteratorHandler struct

+ func (h *SeriesKeysIteratorHandler) Begin()

+ func (h *SeriesKeysIteratorHandler) Finish()

+ func (h *SeriesKeysIteratorHandler) Init(param map[string]interface{}) error

+ func (h *SeriesKeysIteratorHandler) Limited() bool

+ func (h *SeriesKeysIteratorHandler) NextChunkMeta(cm *ChunkMeta) error

+ func (h *SeriesKeysIteratorHandler) NextFile(TSSPFile)

type TSSPFiles

+ func (f *TSSPFiles) AppendReloadFiles(file ...TSSPInfo)

+ func (f *TSSPFiles) Lock()

+ func (f *TSSPFiles) MergedLevelCount(level uint16) int

+ func (f *TSSPFiles) Unlock()

+ type TSSPInfo interface

+ FileName func() *TSSPFileName

+ FileNameExtend func() uint16

+ FilePath func() string

+ LevelAndSequence func() (uint16, uint64)

+ Order func() bool

type TagSets

+ func (s *TagSets) Count() int

type Trailer

+ func (t *Trailer) Marshal(dst []byte) []byte

+ func (t *Trailer) Unmarshal(src []byte) ([]byte, error)

v1.3.1

Jan 2, 2025

Changes in this version

type MmsTables

+ func (m *MmsTables) GetAllMstList() []string

type TablesStore

+ GetAllMstList func() []string

v1.3.0

Nov 12, 2024

Changes in this version

+ const CapacityBinFile

+ const ChunkMetaCompressEnd

+ const ChunkMetaCompressLZ4

+ const ChunkMetaReadNum

+ const DefaultLevelMergeFileNum

+ const InitParamKeyDst

+ const InitParamKeyKeys

+ const InitParamKeyMeasurement

+ const META_DATA_N_BYTES

+ const META_DATA_SIZE

+ const META_STORE_N_BYTES

+ const MergeSelfFastModeMaxLevel

+ const MetaIndexConsumeNum

+ const ReaderContentDuration

+ const ReaderContentNumSpan

+ const ReaderContentSizeSpan

+ const ReaderFilterDuration

+ const ShardMoveLogDir

+ const SortLimitCursorDuration

+ var LevelMergeFileNum = []int

+ func CleanTempFile(f fileops.File)

+ func CompareT[T int | int64 | float64 | string](s1, s2 T, isAscending bool) (bool, bool)

+ func FlushRemoteEnabled(tier uint64) bool

+ func GenParquetLogName() string

+ func GetCursorsBy(path *sparseindex.OBSFilterPath, tr util.TimeRange, isAscending bool) (int, uint64, error)

+ func GetDetachedFlushEnabled() bool

+ func GetPKItems(path string, obsOpts *obs.ObsOptions, miChunkIds []int64) (*colstore.DetachedPKMetaInfo, []*colstore.DetachedPKInfo, error)

+ func InParquetProcess(files ...string) bool

+ func ProcParquetLog(logDir string, lockPath *string, ctx EventContext) error

+ func ReadPKDataAll(path string, opts *obs.ObsOptions, offset, length []int64, ...) ([]*colstore.DetachedPKData, error)

+ func ReadPKMetaAll(path string, opts *obs.ObsOptions, offset, length []int64) ([]*colstore.DetachedPKMeta, error)

+ func ReadPKMetaInfoAll(path string, opts *obs.ObsOptions) (*colstore.DetachedPKMetaInfo, error)

+ func ReadReliabilityLog(file string, dst interface{}) error

+ func SaveReliabilityLog(data interface{}, dir string, lockFile string, nameGenerator func() string) (string, error)

type ChunkMeta

+ func NewChunkMeta(sid uint64, minT, maxT int64, count int) *ChunkMeta

+ func (m *ChunkMeta) GetTimeRangeBy(index int) SegmentRange

type ColumnIterator

+ func (itr *ColumnIterator) IterCurrentChunk(p ColumnIteratorPerformer) error

type CompactGroup

+ func (g *CompactGroup) Add(item string)

+ func (g *CompactGroup) Len() int

+ func (g *CompactGroup) UpdateLevel(lv uint16)

+ type CompactGroupBuilder struct

+ func (b *CompactGroupBuilder) AddFile(f TSSPFile) bool

+ func (b *CompactGroupBuilder) Init(name string, closing *int64, size int)

+ func (b *CompactGroupBuilder) Limited() bool

+ func (b *CompactGroupBuilder) Release()

+ func (b *CompactGroupBuilder) SwitchGroup()

+ type CompactTask struct

+ func NewCompactTask(table *MmsTables, plan *CompactGroup, full bool) *CompactTask

+ func (t *CompactTask) BeforeExecute() bool

+ func (t *CompactTask) Execute()

+ func (t *CompactTask) Finish()

+ func (t *CompactTask) IncrFull(n int64)

type DetachedMetaDataReader

+ func (reader *DetachedMetaDataReader) Close()

type DetachedMetaIndexReader

+ func (reader *DetachedMetaIndexReader) Close()

type DetachedPKDataReader

+ func (reader *DetachedPKDataReader) Close()

type DetachedPKMetaInfoReader

+ func (reader *DetachedPKMetaInfoReader) Close()

type DetachedPKMetaReader

+ func (reader *DetachedPKMetaReader) Close()

+ type EngineShard interface

+ GetDataPath func() string

+ GetIdent func() *meta.ShardIdentifier

+ IsOpened func() bool

+ OpenAndEnable func(client metaclient.MetaClient) error

+ type Event interface

+ Enable func() bool

+ Init func(mst string, level uint16)

+ Instance func() Event

+ OnFinish func(ctx EventContext)

+ OnInterrupt func()

+ OnNewFile func(f TSSPFile)

+ OnReplaceFile func(shardDir string, lockFile string) error

+ OnWriteChunkMeta func(cm *ChunkMeta)

+ OnWriteRecord func(rec *record.Record)

+ type EventBus struct

+ func DefaultEventBus() *EventBus

+ func (b *EventBus) NewEvents(typ EventType, mst string, level uint16) *Events

+ func (b *EventBus) Register(typ EventType, e Event)

+ type EventContext struct

+ func NewEventContext(idx IndexMergeSet, scheduler *scheduler.TaskScheduler) *EventContext

+ type EventType int

+ const EventTypeEnd

+ const EventTypeMergeSelf

+ const EventTypeStreamCompact

+ type Events struct

+ func (es *Events) Finish(success bool, ctx EventContext)

+ func (es *Events) Instance() *Events

+ func (es *Events) Register(e Event)

+ func (es *Events) TriggerNewFile(f TSSPFile)

+ func (es *Events) TriggerReplaceFile(shardDir, lock string) error

+ func (es *Events) TriggerWriteChunkMeta(cm *ChunkMeta)

+ func (es *Events) TriggerWriteRecord(rec *record.Record)

+ type FileInfoExtend struct

+ FileInfo meta.FileInfo

+ Name string

type FileReader

+ ChunkMetaCompressMode func() uint8

type FileReaderContext

+ func (f *FileReaderContext) GetSchemas() record.Schemas

type ImmTable

+ FullyCompacted func(m *MmsTables) bool

+ type IndexMergeSet interface

+ GetSeries func(sid uint64, buf []byte, condition influxql.Expr, ...) error

type Location

+ func (d *Location) SetClosedSignal(s *bool)

type LocationCursor

+ func (l *LocationCursor) Reset()

+ type MatchAllOperator struct

+ func (r *MatchAllOperator) Compute(rec *record.Record)

type MeasurementInProcess

+ func (m *MeasurementInProcess) Has(name string) bool

+ type MergeContext struct

+ func BuildMergeContext(mst string, files *TSSPFiles, full bool, lmt *lastMergeTime) []*MergeContext

+ func (ctx *MergeContext) AddUnordered(f TSSPFile)

+ func (ctx *MergeContext) Limited() bool

+ func (ctx *MergeContext) MergeSelf() bool

+ func (ctx *MergeContext) MergeSelfFast() bool

+ func (ctx *MergeContext) Release()

+ func (ctx *MergeContext) Sort()

+ func (ctx *MergeContext) ToLevel() uint16

+ func (ctx *MergeContext) UnorderedLen() int

+ func (ctx *MergeContext) UpdateLevel(l uint16)

+ type MergePerformers struct

+ func NewMergePerformers(ur *UnorderedReader) *MergePerformers

+ func (c *MergePerformers) Close()

+ func (c *MergePerformers) Closed() bool

+ func (c *MergePerformers) Len() int

+ func (c *MergePerformers) Less(i, j int) bool

+ func (c *MergePerformers) Next() error

+ func (c *MergePerformers) Pop() interface{}

+ func (c *MergePerformers) Push(v interface{})

+ func (c *MergePerformers) Release()

+ func (c *MergePerformers) Swap(i, j int)

+ type MergeSelf struct

+ func NewMergeSelf(mts *MmsTables, lg *logger.Logger) *MergeSelf

+ func (m *MergeSelf) InitEvents(ctx *MergeContext) *Events

+ func (m *MergeSelf) Merge(mst string, toLevel uint16, files []TSSPFile) (TSSPFile, error)

+ func (m *MergeSelf) Stop()

+ type MergeSelfParquetEvent struct

+ func (e *MergeSelfParquetEvent) Instance() Event

+ func (e *MergeSelfParquetEvent) OnWriteRecord(rec *record.Record)

+ type MetaControl interface

+ IsEmpty func() bool

+ Pop func() (MetaDataInfo, bool)

+ Push func(MetaDataInfo)

+ func NewMetaControl(isQueue bool, count int) MetaControl

+ type MetaData struct

+ func NewMetaData(blockIndex int64, data []byte, offset int32) (*MetaData, error)

+ func (m *MetaData) GetBlockIndex() int64

+ func (m *MetaData) GetContentBlockLength() int32

+ func (m *MetaData) GetContentBlockOffset() int64

+ func (m *MetaData) GetMaxTime() int64

+ func (m *MetaData) GetMinTime() int64

+ type MetaDataInfo interface

+ GetMaxTime func() int64

+ GetMinTime func() int64

+ type MetaDatas []*MetaData

+ func (a MetaDatas) Len() int

+ func (a MetaDatas) Less(i, j int) bool

+ func (a MetaDatas) Swap(i, j int)

type MetaIndex

+ func GetMetaIndexAndBlockId(path string, obsOpts *obs.ObsOptions, chunkCount int64, tr util.TimeRange) ([]int64, []*MetaIndex, error)

+ func (m *MetaIndex) GetID() uint64

+ func (m *MetaIndex) GetOffset() int64

+ func (m *MetaIndex) GetSize() uint32

+ type MetaQueue []MetaDataInfo

+ func (q *MetaQueue) IsEmpty() bool

+ func (q *MetaQueue) Pop() (MetaDataInfo, bool)

+ func (q *MetaQueue) Push(v MetaDataInfo)

+ type MetaStack []MetaDataInfo

+ func (s *MetaStack) IsEmpty() bool

+ func (s *MetaStack) Pop() (MetaDataInfo, bool)

+ func (s *MetaStack) Push(value MetaDataInfo)

type MmsIdTime

+ func (m *MmsIdTime) Get(id uint64) (int64, int64)

type MmsTables

+ func (m *MmsTables) CopyCSFiles(name string) []TSSPFile

+ func (m *MmsTables) GetMstList(isOrder bool) []string

+ func (m *MmsTables) GetObsOption() *obs.ObsOptions

+ func (m *MmsTables) GetShardID() uint64

+ func (m *MmsTables) GetTableFileNum(name string, order bool) int

+ func (m *MmsTables) LoadSequencer()

+ func (m *MmsTables) RenameFileToLevel(plan *CompactGroup) error

+ func (m *MmsTables) SetIndexMergeSet(idx IndexMergeSet)

+ func (m *MmsTables) SetObsOption(option *obs.ObsOptions)

type MsBuilder

+ FilesInfo []FileInfoExtend

+ RowCount int64

+ ShardID uint64

+ func (b *MsBuilder) CloseIndexWriters() error

+ func (b *MsBuilder) GetIndexBuilder() *index.IndexWriterBuilder

+ func (b *MsBuilder) NewIndexWriterBuilder(schema record.Schemas, indexRelation influxql.IndexRelation)

+ func (b *MsBuilder) WriteDetachedIndex(writeRec *record.Record, rowsPerSegment []int) error

+ type ParquetTask struct

+ func (t *ParquetTask) Execute(ctx EventContext)

+ func (t *ParquetTask) GetSeries(sId uint64, ctx EventContext) (string, error)

type ReadContext

+ func (d *ReadContext) IsAborted() bool

+ type SegmentSequenceReader struct

+ func NewSegmentSequenceReader(path *sparseindex.OBSFilterPath, taskID int, count uint64, ...) (*SegmentSequenceReader, error)

+ func (reader *SegmentSequenceReader) Close()

+ func (reader *SegmentSequenceReader) ConsumeDateByShard() ([]map[string]interface{}, bool, int64, uint64, int64, error)

+ type SequenceIterator interface

+ AddFiles func(files []TSSPFile)

+ Buffer func() *pool.Buffer

+ Release func()

+ Run func() error

+ SetChunkMetasReader func(reader SequenceIteratorChunkMetaReader)

+ Stop func()

+ func NewSequenceIterator(handler SequenceIteratorHandler, logger *Log.Logger) SequenceIterator

+ type SequenceIteratorChunkMetaReader interface

+ ReadChunkMetas func(f TSSPFile, idx int) ([]byte, []uint32, error)

+ type SequenceIteratorHandler interface

+ Begin func()

+ Finish func()

+ Init func(map[string]interface{}) error

+ Limited func() bool

+ NextChunkMeta func([]byte) error

+ NextFile func(TSSPFile)

type Sequencer

+ func (s *Sequencer) GetMmsIdTime(name string) *MmsIdTime

+ type ShowTagValuesPlan interface

+ Execute func(dst map[string]*TagSets, tagKeys map[string][][]byte, condition influxql.Expr, ...) error

+ Stop func()

+ func NewShowTagValuesPlan(table TablesStore, idx IndexMergeSet, logger *Log.Logger, sh EngineShard, ...) ShowTagValuesPlan

+ type SortLimitCursor struct

+ func NewSortLimitCursor(options hybridqp.Options, schemas record.Schemas, input comm.TimeCutKeyCursor, ...) *SortLimitCursor

+ func (t *SortLimitCursor) Close() error

+ func (t *SortLimitCursor) EndSpan()

+ func (t *SortLimitCursor) GetInput() comm.TimeCutKeyCursor

+ func (t *SortLimitCursor) GetSchema() record.Schemas

+ func (t *SortLimitCursor) Name() string

+ func (t *SortLimitCursor) Next() (*record.Record, comm.SeriesInfoIntf, error)

+ func (t *SortLimitCursor) NextAggData() (*record.Record, *comm.FileInfo, error)

+ func (t *SortLimitCursor) SetOps(ops []*comm.CallOption)

+ func (t *SortLimitCursor) SinkPlan(plan hybridqp.QueryNode)

+ func (t *SortLimitCursor) StartSpan(span *tracing.Span)

+ type SortLimitRows struct

+ func NewSortLimitRows(sortIndex []int, schema record.Schemas, shardId int64) *SortLimitRows

+ func (rs *SortLimitRows) Len() int

+ func (rs *SortLimitRows) Less(i, j int) bool

+ func (rs *SortLimitRows) Pop() interface{}

+ func (rs *SortLimitRows) PopToRec() *record.Record

+ func (rs *SortLimitRows) Push(x interface{})

+ func (rs *SortLimitRows) Swap(i, j int)

+ type StreamCompactParquetEvent struct

+ func (e *StreamCompactParquetEvent) Instance() Event

+ func (e *StreamCompactParquetEvent) OnWriteChunkMeta(cm *ChunkMeta)

type StreamIterators

+ func (c *StreamIterators) InitEvents(level uint16) *Events

+ func (c *StreamIterators) RemoveTmpFiles()

+ type TSSP2ParquetEvent struct

+ func (e *TSSP2ParquetEvent) Enable() bool

+ func (e *TSSP2ParquetEvent) Init(mst string, level uint16)

+ func (e *TSSP2ParquetEvent) OnFinish(ctx EventContext)

+ func (e *TSSP2ParquetEvent) OnInterrupt()

+ func (e *TSSP2ParquetEvent) OnNewFile(f TSSPFile)

+ func (e *TSSP2ParquetEvent) OnReplaceFile(shardDir string, lockFile string) error

+ type TSSP2ParquetPlan struct

+ Files []string

+ Mst string

+ Schema map[string]uint8

+ func (p *TSSP2ParquetPlan) Init(mst string, level uint16)

type TSSPFile

+ ChunkMetaCompressMode func() uint8

+ UpdateLevel func(level uint16)

type TSSPFileDetachedReader

+ func (t *TSSPFileDetachedReader) UpdateTime(time int64)

type TSSPFiles

+ func (f *TSSPFiles) MaxMerged() uint16

type TablesStore

+ CopyCSFiles func(name string) []TSSPFile

+ GetMstList func(isOrder bool) []string

+ GetObsOption func() *obs.ObsOptions

+ GetShardID func() uint64

+ GetTableFileNum func(string, bool) int

+ LoadSequencer func()

+ SetIndexMergeSet func(idx IndexMergeSet)

+ SetObsOption func(option *obs.ObsOptions)

+ type TagSets struct

+ func NewTagSets() *TagSets

+ func (s *TagSets) Add(key, val string)

+ func (s *TagSets) ForEach(process func(tagKey, tagValue string))

+ func (s *TagSets) TagKVCount() int

+ type TagValuesIteratorHandler struct

+ func NewTagValuesIteratorHandler(idx IndexMergeSet, condition influxql.Expr, tr *util.TimeRange, limit int) *TagValuesIteratorHandler

+ func (h *TagValuesIteratorHandler) Begin()

+ func (h *TagValuesIteratorHandler) Finish()

+ func (h *TagValuesIteratorHandler) Init(param map[string]interface{}) error

+ func (h *TagValuesIteratorHandler) Limited() bool

+ func (h *TagValuesIteratorHandler) NextChunkMeta(buf []byte) error

+ func (h *TagValuesIteratorHandler) NextFile(TSSPFile)

+ type UnnestOperator interface

+ Compute func(rec *record.Record)

+ func GetUnnestFuncOperator(unnest *influxql.Unnest, schema record.Schemas) (UnnestOperator, error)

+ func NewMatchAllOperator(unnest *influxql.Unnest, schemas record.Schemas) UnnestOperator

+ type UnorderedColumn struct

+ func (c *UnorderedColumn) Init(meta *ColumnMeta, idx int)

type UnorderedColumnReader

+ func (r *UnorderedColumnReader) ChangeColumn(sid uint64, ref *record.Field)

+ func (r *UnorderedColumnReader) ChangeSeries(sid uint64) error

+ func (r *UnorderedColumnReader) Close()

+ func (r *UnorderedColumnReader) HasColumn() bool

+ func (r *UnorderedColumnReader) MatchSeries(sid uint64) bool

+ func (r *UnorderedColumnReader) ReadSchemas(sid uint64, maxTime int64, dst map[string]record.Field)

+ func (r *UnorderedColumnReader) ReadTime(sid uint64, maxTime int64) []int64

+ type UnorderedColumns struct

+ func NewUnorderedColumns() *UnorderedColumns

+ func (c *UnorderedColumns) ChangeColumn(name string) *UnorderedColumn

+ func (c *UnorderedColumns) GetLineOffset(name string) int

+ func (c *UnorderedColumns) GetSegOffset(name string) int

+ func (c *UnorderedColumns) IncrLineOffset(name string, n int)

+ func (c *UnorderedColumns) IncrSegOffset(name string, n int)

+ func (c *UnorderedColumns) Init(cm *ChunkMeta)

+ func (c *UnorderedColumns) ReadCompleted() bool

+ func (c *UnorderedColumns) SetRemainLine(n int)

+ func (c *UnorderedColumns) TimeMeta() *ColumnMeta

+ func (c *UnorderedColumns) Walk(callback func(meta *ColumnMeta))

type UnorderedReader

+ func (r *UnorderedReader) ChangeColumn(sid uint64, ref *record.Field)

+ func (r *UnorderedReader) ChangeSeries(sid uint64) error

+ func (r *UnorderedReader) CloseFile()

v1.3.0-rc.1

Sep 2, 2024

v1.2.0

Feb 29, 2024

Changes in this version

+ const BatchReaderRecordNum

+ const ChunkMetaCompressNone

+ const ChunkMetaCompressSnappy

+ const ChunkMetaFile

+ const ChunkMetaSize

+ const ChunkMetaTTL

+ const CompressModMaxChunkMetaItemCount

+ const DataFile

+ const DefaultExpectedSegmentSize

+ const DetachedMetaIndexLen

+ const FD_OUTSIDE

+ const IndexOfChunkMetaCompressFlag

+ const MetaIndexFile

+ const MetaIndexHeaderSize

+ const MetaIndexItemSize

+ const MetaIndexLimitNum

+ const MetaIndexSegmentNum

+ const PKDataLimitNum

+ const PKMetaInfoLength

+ const PKMetaLimitNum

+ const PKMetaPrefixSize

+ const PkMetaHeaderSize

+ const PrimaryKeyFile

+ const PrimaryMetaFile

+ const QueryMetaCacheTTL

+ const QueryMetaDataCacheSize

+ const SwapperCompressNone

+ const SwapperCompressSnappy

+ const SwapperCompressZSTD

+ var ChunkMetaCache = cache.NewCache(ChunkMetaSize, ChunkMetaTTL)

+ var DetachedMetaDataCache = cache.NewCache(QueryMetaDataCacheSize, QueryMetaCacheTTL)

+ var PrefixDataPath string

+ func CanEncodeOneRowMode(col *record.ColVal) bool

+ func DecodeColumnHeader(col *record.ColVal, data []byte, colType uint8) ([]byte, []byte, error)

+ func DecodeColumnOfOneValue(data []byte, col *record.ColVal, typ uint8)

+ func EncodeColumnHeader(col *record.ColVal, dst []byte, typ uint8) []byte

+ func GenFixRowsPerSegment(data *record.Record, rowNumPerSegment int) []int

+ func GenRecByReserveIds(rec, filterRec *record.Record, rowNumber []int, redIdxMap map[int]struct{}) *record.Record

+ func GetBloomFilterBuf() *bloomFilter

+ func GetChunkMetaCompressMode() uint8

+ func GetMetaIndexChunkCount(obsOptions *obs.ObsOptions, dataPath string) (int64, error)

+ func GetMetaIndexOffsetAndLengthByChunkId(chunkId int64) (offset, length int64)

+ func GetPKMetaOffsetLengthByChunkId(pkMetaInfo *colstore.DetachedPKMetaInfo, chunkId int) (offset, length int64)

+ func GetPrefixDataPath() string

+ func IsFlushToFinalFile(totalSegmentCnt, flushToFinalFileLimit uint64) bool

+ func LeftBound(nums []uint32, target uint32, left int) int

+ func NewCsImmTableImpl() *csImmTableImpl

+ func NewObsWriter(path, fileName string, obsOpts *obs.ObsOptions) (*obsWriter, error)

+ func NewObsWriterByFd(fd fileops.File, obsOpts *obs.ObsOptions) (*obsWriter, error)

+ func NewTsImmTable() *tsImmTableImpl

+ func OpenObsFile(path, fileName string, obsOpts *obs.ObsOptions) (fileops.File, error)

+ func PreAggOnlyOneRow(buf []byte) bool

+ func PutBloomFilterBuf(key *bloomFilter)

+ func PutChunkMeta(filePath string, chunkMeta *ChunkMeta)

+ func PutDetachedSegmentTask(queryID string, meta IndexFrags)

+ func RenameIndexFiles(fname string, indexList []string) error

+ func RenameTmpFullTextIdxFile(msb *MsBuilder) error

+ func SetChunkMetaCompressMode(mode int)

+ func SetCompactionEnabled(compactionEnabled bool)

+ func SetDetachedFlushEnabled(detachFlushEnabled bool)

+ func SetIndexCompressMode(mode int)

+ func SetPrefixDataPath(dataPath string)

+ func TimeSorted(sortKeys []string) bool

+ func UpdateChunkMetaFunc(_, _ cache.Entry) bool

+ func UpdateDetachedMetaDataCache(old, new cache.Entry) bool

+ type AccumulateMetaIndex struct

+ func (a *AccumulateMetaIndex) GetBlockId() uint64

+ func (a *AccumulateMetaIndex) SetAccumulateMetaIndex(pkDataOffset uint32, blockId uint64, dataOffset, offset int64)

+ type BloomFilterIterator struct

+ func NewBloomFilterIterator(f *FragmentIterators, oldFiles []TSSPFile, bfCols []string) (*BloomFilterIterator, error)

+ func (bfi *BloomFilterIterator) AppendFileIdx(fileIdx int)

+ func (bfi *BloomFilterIterator) Write(toLocal bool) error

type ChunkMeta

+ func GetChunkMeta(filePath string) (*ChunkMeta, bool)

+ func (m *ChunkMeta) UnmarshalWithColumns(src []byte, columns []string) ([]byte, error)

+ type ChunkMetaContext struct

+ func NewChunkMetaContext(schema record.Schemas) *ChunkMetaContext

+ func (ctx *ChunkMetaContext) Instance() pool.Object

+ func (ctx *ChunkMetaContext) MemSize() int

+ func (ctx *ChunkMetaContext) Release()

+ type ChunkMetaEntry struct

+ func NewChunkMetaEntry(filePath string) *ChunkMetaEntry

+ func (e *ChunkMetaEntry) GetKey() string

+ func (e *ChunkMetaEntry) GetTime() time.Time

+ func (e *ChunkMetaEntry) GetValue() interface{}

+ func (e *ChunkMetaEntry) SetTime(time time.Time)

+ func (e *ChunkMetaEntry) SetValue(value interface{})

+ func (e *ChunkMetaEntry) Size() int64

+ type ColAux struct

+ func NewColAux(rec *record.Record, filterOption *BaseFilterOptions) *ColAux

type ColumnBuilder

+ func (b *ColumnBuilder) EncodeColumnBySize(ref record.Field, col *record.ColVal, timeCols []record.ColVal, ...) ([]byte, error)

+ func (b *ColumnBuilder) SetEncodeMode(detached bool)

type ColumnMeta

+ func (m *ColumnMeta) GetPreAgg() []byte

+ func (m *ColumnMeta) GetSegment(i int) (int64, uint32)

+ func (m *ColumnMeta) Type() uint8

type ColumnReader

+ UnrefCachePage func(cachePage *readcache.CachePage)

type Config

+ func (c *Config) GetCompactionEnabled() bool

+ func (c *Config) GetDetachedFlushEnabled() bool

+ func (c *Config) SetExpectedSegmentSize(n uint32)

+ type CsChunkDataImp struct

+ func (c *CsChunkDataImp) EncodeChunk(b *ChunkDataBuilder, id uint64, offset int64, rec *record.Record, dst []byte, ...) ([]byte, error)

+ func (c *CsChunkDataImp) EncodeChunkForCompaction(b *ChunkDataBuilder, offset int64, rec *record.Record, dst []byte, ...) ([]byte, error)

+ func (c *CsChunkDataImp) SetAccumulateRowsIndex(accumulateRowsIndex []int)

+ func (c *CsChunkDataImp) SetDetachedInfo(writeDetached bool)

+ type DetachedChunkMetaReader struct

+ func NewDetachedChunkMetaReader(path string, obsOpts *obs.ObsOptions) (*DetachedChunkMetaReader, error)

+ func (reader *DetachedChunkMetaReader) ReadChunkMeta(offset, length []int64) ([]*ChunkMeta, error)

+ type DetachedMetaDataReader struct

+ func NewDetachedMetaDataReader(path string, obsOpts *obs.ObsOptions, isSort bool) (*DetachedMetaDataReader, error)

+ func (reader *DetachedMetaDataReader) InitReadBatch(s []*SegmentMeta, schema record.Schemas)

+ func (reader *DetachedMetaDataReader) ReadBatch(dst *record.Record, decs *ReadContext) (*record.Record, error)

+ type DetachedMetaIndexReader struct

+ func NewDetachedMetaIndexReader(path string, obsOpts *obs.ObsOptions) (*DetachedMetaIndexReader, error)

+ func (reader *DetachedMetaIndexReader) ReadMetaIndex(offset, length []int64) ([]*MetaIndex, error)

+ type DetachedPKDataReader struct

+ func NewDetachedPKDataReader(path string, opts *obs.ObsOptions) (*DetachedPKDataReader, error)

+ func (reader *DetachedPKDataReader) Read(offset, length []int64, metas []*colstore.DetachedPKMeta) ([]*colstore.DetachedPKData, error)

+ func (reader *DetachedPKDataReader) SetPkMetaInfo(metaInfo *colstore.DetachedPKMetaInfo)

+ type DetachedPKMetaInfoReader struct

+ func NewDetachedPKMetaInfoReader(path string, opts *obs.ObsOptions) (*DetachedPKMetaInfoReader, error)

+ func (reader *DetachedPKMetaInfoReader) Read() (*colstore.DetachedPKMetaInfo, error)

+ type DetachedPKMetaReader struct

+ func NewDetachedPKMetaReader(path string, opts *obs.ObsOptions) (*DetachedPKMetaReader, error)

+ func (reader *DetachedPKMetaReader) Read(offset, length []int64) ([]*colstore.DetachedPKMeta, error)

+ type DetachedSegmentEntry struct

+ func NewSegmentMetaDataEntry(segmentID string) *DetachedSegmentEntry

+ func (e *DetachedSegmentEntry) GetKey() string

+ func (e *DetachedSegmentEntry) GetTime() time.Time

+ func (e *DetachedSegmentEntry) GetValue() interface{}

+ func (e *DetachedSegmentEntry) SetTime(time time.Time)

+ func (e *DetachedSegmentEntry) SetValue(value interface{})

+ func (e *DetachedSegmentEntry) Size() int64

+ type EncodeChunkData interface

+ EncodeChunk func(b *ChunkDataBuilder, id uint64, offset int64, rec *record.Record, dst []byte, ...) ([]byte, error)

+ EncodeChunkForCompaction func(b *ChunkDataBuilder, offset int64, rec *record.Record, dst []byte, ...) ([]byte, error)

+ SetAccumulateRowsIndex func(rowsPerSegment []int)

+ SetDetachedInfo func(writeDetached bool)

+ type EncodeColumnMode interface

type FileIterator

+ func (itr *FileIterator) ReadData(offset int64, size uint32) ([]byte, error)

type FileReader

+ RenameOnObs func(obsName string) error

+ type FileReaderContext struct

+ func NewFileReaderContext(tr util.TimeRange, schemas record.Schemas, decs *ReadContext, ...) *FileReaderContext

+ type FileSwapper struct

+ func NewFileSwapper(file string, lock string, limitCompact bool, compressMode int) (*FileSwapper, error)

+ func (s *FileSwapper) CopyTo(to io.Writer, buf []byte) (int64, error)

+ func (s *FileSwapper) MustClose()

+ func (s *FileSwapper) SetWriter(w io.WriteCloser)

+ func (s *FileSwapper) Write(b []byte) (int, error)

+ type FragmentIterator interface

type FragmentIterators

+ TimeClusterResult *record.Record

+ func (f *FragmentIterators) IsEmpty() bool

type ImmTable

+ AddBothTSSPFiles func(flushed *bool, m *MmsTables, name string, orderFiles []TSSPFile, ...)

+ GetCompactionType func(name string) config.CompactionType

+ GetEngineType func() config.EngineType

+ GetMstInfo func(name string) (*meta.MeasurementInfo, bool)

+ UpdateAccumulateMetaIndexInfo func(name string, index *AccumulateMetaIndex)

+ type IndexCompressWriter struct

+ func (w *IndexCompressWriter) BlockSize() int

+ func (w *IndexCompressWriter) Close() error

+ func (w *IndexCompressWriter) CopyTo(to io.Writer) (int, error)

+ func (w *IndexCompressWriter) GetWriter() *bufio.Writer

+ func (w *IndexCompressWriter) Init(name string, lock *string, cacheMeta bool, limitCompact bool)

+ func (w *IndexCompressWriter) MetaDataBlocks(dst [][]byte) [][]byte

+ func (w *IndexCompressWriter) Size() int

+ func (w *IndexCompressWriter) SwitchMetaBuffer() (int, error)

+ func (w *IndexCompressWriter) Write(p []byte) (int, error)

+ type IndexFrags interface

+ AddFragCount func(int64)

+ AppendFragRanges func(...fragment.FragmentRanges)

+ AppendIndexes func(...interface{})

+ BasePath func() string

+ FragCount func() int64

+ FragRanges func() []fragment.FragmentRanges

+ GetErr func() error

+ IndexCount func() int

+ Indexes func() interface{}

+ SetErr func(error)

+ Size func() int

+ func GetDetachedSegmentTask(queryID string) (IndexFrags, bool)

+ type IndexWriter interface

+ BlockSize func() int

+ Close func() error

+ CopyTo func(to io.Writer) (int, error)

+ Init func(name string, lock *string, cacheMeta bool, limitCompact bool)

+ MetaDataBlocks func(dst [][]byte) [][]byte

+ Size func() int

+ SwitchMetaBuffer func() (int, error)

+ Write func(p []byte) (int, error)

+ func NewPKIndexWriter(indexName string, cacheMeta bool, limitCompact bool, lockPath *string) IndexWriter

+ type IteratorByBlock struct

+ func NewIteratorByBlock(f *FragmentIterators, conf *Config, group FilesInfo, ...) *IteratorByBlock

+ func (ib *IteratorByBlock) Flush(pkSchema record.Schemas, readFinal bool, tbStore *MmsTables) error

+ func (ib *IteratorByBlock) WriteDetachedMeta(pkSchema record.Schemas) error

+ type IteratorByRow struct

+ func NewIteratorByRow(f *FragmentIterators, conf *Config) *IteratorByRow

+ func (ir *IteratorByRow) Flush(tbStore *MmsTables, pkSchema record.Schemas, final bool) error

+ func (ir *IteratorByRow) GetBreakPoint()

+ func (ir *IteratorByRow) NextWithBreakPoint()

type Location

+ func (l *Location) SetChunkMeta(chunkMeta *ChunkMeta)

+ type MeasurementInProcess struct

+ func NewMeasurementInProcess() *MeasurementInProcess

+ func (m *MeasurementInProcess) Add(name string) bool

+ func (m *MeasurementInProcess) Del(name string)

type MetaIndex

+ func (m *MetaIndex) IsExist(tr util.TimeRange) bool

type MmsTables

+ func (m *MmsTables) AddBothTSSPFiles(flushed *bool, name string, orderFiles []TSSPFile, unorderFiles []TSSPFile)

+ func (m *MmsTables) FullyCompacted() bool

+ func (m *MmsTables) GetMstInfo(name string) (*meta.MeasurementInfo, bool)

+ func (m *MmsTables) SetAccumulateMetaIndex(name string, aMetaIndex *AccumulateMetaIndex)

+ func (m *MmsTables) SetLockPath(lock *string)

+ func (m *MmsTables) SetTier(tier uint64)

type MsBuilder

+ EncodeChunkDataImp EncodeChunkData

+ func NewDetachedMsBuilder(dir, name string, lockPath *string, conf *Config, idCount int, ...) (*MsBuilder, error)

+ func (b *MsBuilder) BloomFilterNeedDetached(filterDetachedWriteTimes int) bool

+ func (b *MsBuilder) GetChunkBuilder() *ChunkDataBuilder

+ func (b *MsBuilder) GetFullTextIdx() bool

+ func (b *MsBuilder) GetLocalBfCount() int64

+ func (b *MsBuilder) GetSkipIndex() *sparseindex.SkipIndex

+ func (b *MsBuilder) NewSkipIndex(schema record.Schemas, indexRelation influxql.IndexRelation)

+ func (b *MsBuilder) SetEncodeChunkDataImp(engineType config.EngineType)

+ func (b *MsBuilder) SetFullTextIdx(fullTextIdx bool)

+ func (b *MsBuilder) SetLocalBfCount(count int64)

+ func (b *MsBuilder) SetTimeSorted(timeSorted bool)

+ func (b *MsBuilder) SwitchChunkMeta() error

+ func (b *MsBuilder) WriteChunkMeta(cm *ChunkMeta) (int, error)

+ func (b *MsBuilder) WriteDetached(id uint64, data *record.Record, pkSchema record.Schemas, firstFlush bool, ...) error

+ func (b *MsBuilder) WriteDetachedMetaAndIndex(writeRec *record.Record, pkSchema record.Schemas, firstFlush bool, ...) error

+ func (b *MsBuilder) WriteDetachedSkipIndex(writeRec *record.Record, rowsPerSegment []int) error

+ func (b *MsBuilder) WriteRecordByCol(id uint64, data *record.Record, schema record.Schemas, ...) (*MsBuilder, error)

+ type PageCacheReader struct

+ func NewPageCacheReader(t *Trailer, r *tsspFileReader) *PageCacheReader

+ func (pcr *PageCacheReader) GetCachePageIdsAndOffsets(start int64, size uint32) ([]int64, []int64, error)

+ func (pcr *PageCacheReader) GetMaxPageIdAndOffset() (int64, int64)

+ func (pcr *PageCacheReader) Init()

+ func (pcr *PageCacheReader) Read(offset int64, size uint32, buf *[]byte, ioPriority int) ([]byte, *readcache.CachePage, error)

+ func (pcr *PageCacheReader) ReadFixPageSize(offset int64, size uint32, buf *[]byte, ioPriority int) ([]byte, *readcache.CachePage, error)

+ func (pcr *PageCacheReader) ReadSinglePage(cacheKey string, pageOffset int64, pageSize int64, buf *[]byte, ioPriority int) (*readcache.CachePage, []byte, error)

+ func (pcr *PageCacheReader) ReadVariablePageSize(offset int64, size uint32, buf *[]byte, ioPriority int) ([]byte, *readcache.CachePage, error)

type PreAggBuilders

+ func (b *PreAggBuilders) FloatBuilder() *FloatPreAgg

+ func (b *PreAggBuilders) IntegerBuilder() *IntegerPreAgg

type ReadContext

+ func (d *ReadContext) SetClosedSignal(s *int32)

+ type SegmentMeta struct

+ func NewSegmentMeta(id int, c *ChunkMeta) *SegmentMeta

+ func (s *SegmentMeta) GetMaxTime() int64

+ func (s *SegmentMeta) GetMinTime() int64

+ type SegmentTask struct

type StreamIterators

+ func (c *StreamIterators) ListenCloseSignal(finish chan struct{})

+ func (c *StreamIterators) SetWriter(w fileops.FileWriter)

+ func (c *StreamIterators) SwitchChunkMeta() error

+ func (c *StreamIterators) WriteChunkMeta(cm *ChunkMeta) (int, error)

type StreamWriteFile

+ func (c *StreamWriteFile) SortColumns()

+ func (c *StreamWriteFile) SwitchChunkMeta() error

+ func (c *StreamWriteFile) WriteChunkMeta(cm *ChunkMeta) (int, error)

type TSSPFile

+ RenameOnObs func(obsName string) error

+ type TSSPFileAttachedReader struct

+ func NewTSSPFileAttachedReader(files []TSSPFile, fragRanges []fragment.FragmentRanges, ctx *FileReaderContext, ...) (*TSSPFileAttachedReader, error)

+ func (t *TSSPFileAttachedReader) Close() error

+ func (t *TSSPFileAttachedReader) EndSpan()

+ func (t *TSSPFileAttachedReader) GetSchema() record.Schemas

+ func (t *TSSPFileAttachedReader) Name() string

+ func (t *TSSPFileAttachedReader) Next() (*record.Record, comm.SeriesInfoIntf, error)

+ func (t *TSSPFileAttachedReader) NextAggData() (*record.Record, *comm.FileInfo, error)

+ func (t *TSSPFileAttachedReader) ResetBy(files []TSSPFile, fragRanges []fragment.FragmentRanges) error

+ func (t *TSSPFileAttachedReader) SetOps(ops []*comm.CallOption)

+ func (t *TSSPFileAttachedReader) SinkPlan(plan hybridqp.QueryNode)

+ func (t *TSSPFileAttachedReader) StartSpan(span *tracing.Span)

+ type TSSPFileDetachedReader struct

+ func NewTSSPFileDetachedReader(metaIndex []*MetaIndex, blocks [][]int, ctx *FileReaderContext, ...) (*TSSPFileDetachedReader, error)

+ func (t *TSSPFileDetachedReader) Close() error

+ func (t *TSSPFileDetachedReader) EndSpan()

+ func (t *TSSPFileDetachedReader) GetSchema() record.Schemas

+ func (t *TSSPFileDetachedReader) Name() string

+ func (t *TSSPFileDetachedReader) Next() (*record.Record, comm.SeriesInfoIntf, error)

+ func (t *TSSPFileDetachedReader) NextAggData() (*record.Record, *comm.FileInfo, error)

+ func (t *TSSPFileDetachedReader) ResetBy(metaIndex []*MetaIndex, blocks [][]int, ctx *FileReaderContext)

+ func (t *TSSPFileDetachedReader) SetOps(ops []*comm.CallOption)

+ func (t *TSSPFileDetachedReader) SinkPlan(plan hybridqp.QueryNode)

+ func (t *TSSPFileDetachedReader) StartSpan(span *tracing.Span)

type TSSPFiles

+ func (f *TSSPFiles) RLock()

+ func (f *TSSPFiles) RUnlock()

type TablesStore

+ AddBothTSSPFiles func(flushed *bool, name string, orderFiles []TSSPFile, unorderFiles []TSSPFile)

+ FullyCompacted func() bool

+ GetMstInfo func(name string) (*meta.MeasurementInfo, bool)

+ SetAccumulateMetaIndex func(name string, aMetaIndex *AccumulateMetaIndex)

+ SetLockPath func(lock *string)

+ SetTier func(tier uint64)

type Trailer

+ func (t *Trailer) DataSize() int64

+ func (t *Trailer) GetData(idx int, def uint8) uint8

+ func (t *Trailer) IndexSize() int64

+ func (t *Trailer) MetaIndexSize() int64

+ func (t *Trailer) SetChunkMetaCompressFlag()

+ type TsChunkDataImp struct

+ func (t *TsChunkDataImp) EncodeChunk(b *ChunkDataBuilder, id uint64, offset int64, rec *record.Record, dst []byte, ...) ([]byte, error)

+ func (t *TsChunkDataImp) EncodeChunkForCompaction(b *ChunkDataBuilder, offset int64, rec *record.Record, dst []byte, ...) ([]byte, error)

+ func (t *TsChunkDataImp) SetAccumulateRowsIndex(accumulateRowsIndex []int)

+ func (t *TsChunkDataImp) SetDetachedInfo(writeDetached bool)

+ type UnnestMatch interface

+ Get func([][]byte) [][]byte

v1.1.2

Oct 30, 2023

v1.1.1

Nov 7, 2023

Changes in this version

+ var LevelCompactRuleForCs = []uint16

+ func GetSortKeyColVal(fi *FileIterator, sortKey []record.Field, ctx *ReadContext, segPosition int) ([]*record.ColVal, []record.SortItem, error)

+ func SetFragmentsNumPerFlush(fragmentsNumPerFlush int)

+ func WriteIntoFile(msb *MsBuilder, tmp bool, withPKIndex bool) error

type BaseFilterOptions

+ RedIdxMap map[int]struct{}

type ChunkDataBuilder

+ func (b *ChunkDataBuilder) EncodeChunkForColumnStore(offset int64, rec *record.Record, dst []byte) ([]byte, error)

type Config

+ FragmentsNumPerFlush int

+ func NewColumnStoreConfig() *Config

+ type FragmentIterators struct

+ Conf *Config

+ PkRec []*record.Record

+ RecordResult *record.Record

+ SortKeyFileds []record.Field

+ func (f *FragmentIterators) Close()

+ func (f *FragmentIterators) CompareWithBreakPoint(curPos, breakPos int) bool

+ func (f *FragmentIterators) Flush(tbStore *MmsTables, pkSchema record.Schemas, final bool) error

+ func (f *FragmentIterators) GetBreakPoint()

+ func (f *FragmentIterators) Len() int

+ func (f *FragmentIterators) Less(i, j int) bool

+ func (f *FragmentIterators) NextWithBreakPoint()

+ func (f *FragmentIterators) Pop() interface{}

+ func (f *FragmentIterators) Push(v interface{})

+ func (f *FragmentIterators) Swap(i, j int)

+ func (f *FragmentIterators) WithLog(log *Log.Logger)

+ type FragmentIteratorsPool struct

+ func NewFragmentIteratorsPool(n int) *FragmentIteratorsPool

type ImmTable

+ LevelPlan func(m *MmsTables, level uint16) []*CompactGroup

+ NewFileIterators func(m *MmsTables, group *CompactGroup) (FilesInfo, error)

+ SetMstInfo func(name string, mstInfo *MeasurementInfo)

+ type MeasurementInfo struct

+ Name string

+ PrimaryKey []string

+ Schema map[string]int32

+ SortKey []string

type MmsTables

+ func (m *MmsTables) ReplacePKFile(mstName string, file string, rec *record.Record, mark fragment.IndexFragment, ...) error

+ func (m *MmsTables) SetMstInfo(name string, mstInfo *MeasurementInfo)

type MsBuilder

+ func (b *MsBuilder) SetTCLocation(tcLocation int8)

+ type SortKeyIterator struct

+ func NewSortKeyIterator(fi *FileIterator, sortKeyFields []record.Field, ctx *ReadContext, ...) (*SortKeyIterator, error)

+ func (s *SortKeyIterator) GetNewRecord() error

+ func (s *SortKeyIterator) NextSingleFragment(tbStore *MmsTables, f *FragmentIterators, pkSchema record.Schemas) (*record.Record, error)

type TablesStore

+ SetMstInfo func(name string, mstInfo *MeasurementInfo)

v1.1.0

Sep 28, 2023

Changes in this version

+ const ColumnStoreDirName

+ const CountBinFile

+ const DefaultMaxRowsPerSegment4ColStore

+ const DefaultMaxRowsPerSegment4TsStore

+ const DefaultMaxSegmentLimit4ColStore

+ const DefaultSnapshotTblNum

+ const IndexOfTimeStoreFlag

+ const TimeStoreFlag

+ func FillNilCol(col *record.ColVal, size int, ref *record.Field)

+ func FilterByFieldFuncs(rec, filterRec *record.Record, filterOption *BaseFilterOptions, ...) *record.Record

+ func GetDir(engineType config.EngineType, path string) string

+ func GetMaxRowsPerSegment4TsStore() int

+ func GetMergeFlag4TsStore() int32

+ func GetTmpFileSuffix() string

+ func InitQueryFileCache(cap uint32, enable bool)

+ func InitWriterPool(size int)

+ func MergeTimes(a []int64, b []int64, dst []int64) []int64

+ func RemoveTsspSuffix(dataPath string) string

+ func RenameTmpFilesWithPKIndex(newFiles []TSSPFile) error

+ func SetMaxRowsPerSegment4TsStore(maxRowsPerSegmentLimit int)

+ func SetMaxSegmentLimit4TsStore(limit int)

+ func SetMergeFlag4TsStore(v int32)

+ func SetSnapshotTblNum(snapshotTblNum int)

+ type BaseFilterOptions struct

+ CondFunctions binaryfilterfunc.CondFunctions

+ FieldsIdx []int

+ FilterTags []string

+ FiltersMap map[string]interface{}

+ TimeCondFunction binaryfilterfunc.IdxFunctions

type ChunkMeta

+ func (m *ChunkMeta) AllocColMeta(ref *record.Field) *ColumnMeta

+ func (m *ChunkMeta) DelEmptyColMeta()

type ColumnIteratorPerformer

+ HasSeries func(uint64) bool

+ WriteOriginal func(fi *FileIterator) error

type ColumnMeta

+ func (m *ColumnMeta) Equal(name string, ty int) bool

+ func (m *ColumnMeta) IsTime() bool

+ func (m *ColumnMeta) Name() string

type Config

+ SnapshotTblNum int

+ func GetColStoreConfig() *Config

+ func GetTsStoreConfig() *Config

+ func NewTsStoreConfig() *Config

+ func (c *Config) GetMaxRowsPerSegment() int

+ func (c *Config) GetMaxSegmentLimit() int

+ type FileReader interface

+ AverageChunkRows func() int

+ ChunkMeta func(id uint64, offset int64, size, itemCount uint32, metaIdx int, dst *ChunkMeta, ...) (*ChunkMeta, error)

+ Close func() error

+ Contains func(id uint64, tm util.TimeRange) bool

+ ContainsId func(id uint64) bool

+ ContainsTime func(tm util.TimeRange) bool

+ FileName func() string

+ FileSize func() int64

+ FreeFileHandle func() error

+ FreeMemory func() int64

+ GetFileReaderRef func() int64

+ InMemSize func() int64

+ LoadComponents func() error

+ LoadIdTimes func(isOrder bool, p *IdTimePairs) error

+ LoadIntoMemory func() error

+ MaxChunkRows func() int

+ MetaIndex func(id uint64, tr util.TimeRange) (int, *MetaIndex, error)

+ MetaIndexAt func(idx int) (*MetaIndex, error)

+ MinMaxSeriesID func() (min, max uint64, err error)

+ MinMaxTime func() (min, max int64, err error)

+ Name func() string

+ Open func() error

+ Read func(offset int64, size uint32, dst *[]byte, ioPriority int) ([]byte, error)

+ ReadChunkMetaData func(metaIdx int, m *MetaIndex, dst []ChunkMeta, ioPriority int) ([]ChunkMeta, error)

+ ReadData func(cm *ChunkMeta, segment int, dst *record.Record, ctx *ReadContext, ...) (*record.Record, error)

+ ReadDataBlock func(offset int64, size uint32, dst *[]byte, ioPriority int) ([]byte, error)

+ ReadMetaBlock func(metaIdx int, id uint64, offset int64, size uint32, count uint32, dst *[]byte, ...) ([]byte, error)

+ Ref func()

+ Rename func(newName string) error

+ Stat func() *Trailer

+ Unref func() int64

+ Version func() uint64

type FilterOptions

+ func (fo *FilterOptions) GetCond() influxql.Expr

+ func (fo *FilterOptions) SetCondFuncs(filterOption *BaseFilterOptions)

+ type FirstLastReader struct

+ func (r *FirstLastReader) Init(cm *ChunkMeta, cr ColumnReader, ref *record.Field, dst *record.Record, ...) *FirstLastReader

+ func (r *FirstLastReader) Read(ctx *ReadContext, copied bool, ioPriority int) error

+ func (r *FirstLastReader) Release()

+ type ImmTable interface

+ AddTSSPFiles func(m *MmsTables, name string, isOrder bool, files ...TSSPFile)

type Location

+ func (l *Location) AscendingDone()

+ func (l *Location) DescendingDone()

+ func (l *Location) SetFragmentRanges(frs []*fragment.FragmentRange)

type LocationCursor

+ func (l *LocationCursor) AddFilterRecPool(pool *record.CircularRecordPool)

+ func (l *LocationCursor) Close()

+ func (l *LocationCursor) FragmentCount() int

+ func (l *LocationCursor) RowCount() int

+ type MergeColPool struct

+ func (p *MergeColPool) Get() *record.ColVal

+ func (p *MergeColPool) Put(col *record.ColVal)

type MmsTables

+ CSFiles map[string]*TSSPFiles

+ ImmTable ImmTable

+ PKFiles map[string]*colstore.PKFiles

+ func (m *MmsTables) AddPKFile(mstName string, file string, rec *record.Record, mark fragment.IndexFragment)

+ func (m *MmsTables) GetCSFiles(name string) (files *TSSPFiles, ok bool)

+ func (m *MmsTables) GetPKFile(mstName string, file string) (pkInfo *colstore.PKInfo, ok bool)

+ func (m *MmsTables) ReloadSequencer(seq *Sequencer, async bool)

+ func (m *MmsTables) SeriesTotal() uint64

+ func (m *MmsTables) SetImmTableType(engineType config.EngineType)

type MsBuilder

+ func NewMsBuilder(dir, name string, lockPath *string, conf *Config, idCount int, ...) *MsBuilder

+ func (b *MsBuilder) GetPKInfoNum() int

+ func (b *MsBuilder) GetPKMark(i int) fragment.IndexFragment

+ func (b *MsBuilder) GetPKRecord(i int) *record.Record

+ func (b *MsBuilder) NewPKIndexWriter()

+ func (b *MsBuilder) StoreTimes()

+ type QueryfileCache struct

+ func GetQueryfileCache() *QueryfileCache

+ func NewQueryfileCache(cap uint32) *QueryfileCache

+ func (qfc *QueryfileCache) Get()

+ func (qfc *QueryfileCache) GetCap() uint32

+ func (qfc *QueryfileCache) Put(f TSSPFile)

type ReadContext

+ func (d *ReadContext) GetCoder() *encoding.CoderContext

+ func (d *ReadContext) GetReadBuff() []byte

+ func (d *ReadContext) SetSpan(readSpan, filterSpan *tracing.Span)

type Sequencer

+ func (s *Sequencer) DelMmsIdTime(name string)

+ func (s *Sequencer) ResetMmsIdTime()

+ func (s *Sequencer) SeriesTotal() uint64

+ func (s *Sequencer) SetStat(free, loading bool)

+ func (s *Sequencer) SetToInLoading() bool

+ func (s *Sequencer) UnRef()

+ type SeriesCounter struct

+ func (sc *SeriesCounter) DecrN(n uint64)

+ func (sc *SeriesCounter) Get() uint64

+ func (sc *SeriesCounter) Incr()

+ func (sc *SeriesCounter) Reset()

type StreamWriteFile

+ func (c *StreamWriteFile) SetValidate(en bool)

type TSSPFile

+ GetFileReaderRef func() int64

type TablesStore

+ AddPKFile func(name, file string, rec *record.Record, mark fragment.IndexFragment)

+ GetCSFiles func(mm string) (*TSSPFiles, bool)

+ GetPKFile func(mstName string, file string) (pkInfo *colstore.PKInfo, ok bool)

+ SeriesTotal func() uint64

+ SetImmTableType func(engineType config.EngineType)

type Trailer

+ func (t *Trailer) EqualData(idx int, v byte) bool

+ func (t *Trailer) SetData(idx int, v byte)

type UnorderedReader

+ func (r *UnorderedReader) AllocNilCol(size int, ref *record.Field) *record.ColVal

+ func (r *UnorderedReader) HasSeries(sid uint64) bool

+ type UnorderedReaderContext struct

v1.1.0-rc1

Aug 31, 2023

v1.0.1

Mar 29, 2023

Changes in this version

+ func FilterByOpts(rec *record.Record, opt *FilterOptions) *record.Record

type UnorderedReader

+ func (r *UnorderedReader) InitTimes(sid uint64, maxTime int64) error

+ func (r *UnorderedReader) ReadAllTimes() []int64

v1.0.0

Feb 23, 2023

Changes in this version

+ const BLOOMFILTER_SIZE

+ const COMPRESSION_RATIO

+ const DownSampleLogDir

+ const LOAD

+ const PRELOAD

+ const SERIESKEY_STATISTIC_SIZE

+ var ErrDownSampleStopped = errors.New("downSample stopped")

+ func FileOperation(f TSSPFile, op func())

+ func GenLogFileName(logSeq *uint64) string

+ func GetTmpTsspFileSuffix() string

+ func NewLastMergeTime() *lastMergeTime

+ func NewMergeContext(mst string) *mergeContext

+ func NewMergePerformer(ur *UnorderedReader, stat *statistics.MergeStatItem) *mergePerformer

+ func ReleaseMsBuilder(msb *MsBuilder)

+ func SumFilesSize(files []TSSPFile) int64

+ func UnrefAll(files ...*TSSPFiles)

+ func UnrefFilesReader(files ...TSSPFile)

+ func UnrefTSSPFiles(files *TSSPFiles)

+ type BufferReader struct

+ func NewBufferReader(maxSize uint32) *BufferReader

+ func (br *BufferReader) Read(offset int64, size uint32) ([]byte, error)

+ func (br *BufferReader) Reset(r TSSPFile)

type ChunkIterator

+ func (c *ChunkIterator) GetRecord() *record.Record

+ func (c *ChunkIterator) GetSeriesID() uint64

type ChunkMeta

+ func (m *ChunkMeta) Clone() *ChunkMeta

+ func (m *ChunkMeta) GetColMeta() []ColumnMeta

+ func (m *ChunkMeta) Len() int

+ func (m *ChunkMeta) Less(i, j int) bool

+ func (m *ChunkMeta) SegmentCount() int

+ func (m *ChunkMeta) Swap(i, j int)

type ColumnBuilder

+ func (b *ColumnBuilder) BuildPreAgg()

+ type ColumnIterator struct

+ func NewColumnIterator(fi *FileIterator) *ColumnIterator

+ func (itr *ColumnIterator) Close()

+ func (itr *ColumnIterator) Error() error

+ func (itr *ColumnIterator) IncrChunkUsed()

+ func (itr *ColumnIterator) NextChunkMeta() bool

+ func (itr *ColumnIterator) NextColumn(colIdx int) (*record.Field, bool)

+ func (itr *ColumnIterator) PutCol(col *record.ColVal)

+ func (itr *ColumnIterator) Run(p ColumnIteratorPerformer) error

+ type ColumnIteratorPerformer interface

+ ColumnChanged func(record.Field) error

+ Finish func() error

+ Handle func(col *record.ColVal, times []int64, lastSeg bool) error

+ SeriesChanged func(uint64, []int64) error

type ColumnMeta

+ func (m *ColumnMeta) Clone() ColumnMeta

type DiskFileReader

+ FreeFileHandle func() error

+ IsOpen func() bool

+ ReOpen func() error

type FileIterator

+ func (itr *FileIterator) GetCurtChunkMeta() *ChunkMeta

type Float

+ func NewFloat() *Float

type MmsTables

+ func (m *MmsTables) DisableCompAndMerge()

+ func (m *MmsTables) EnableCompAndMerge()

+ func (m *MmsTables) FreeSequencer() bool

+ func (m *MmsTables) GetBothFilesRef(measurement string, hasTimeFilter bool, tr record.TimeRange) ([]TSSPFile, []TSSPFile)

+ func (m *MmsTables) GetFileSeq() uint64

+ func (m *MmsTables) GetTSSPFiles(name string, isOrder bool) (files *TSSPFiles, ok bool)

+ func (m *MmsTables) IsOutOfOrderFilesExist() bool

+ func (m *MmsTables) Listen(signal chan struct{}, onClose func())

+ func (m *MmsTables) NewStreamWriteFile(mst string) *StreamWriteFile

+ func (m *MmsTables) ReplaceDownSampleFiles(mstNames []string, originFiles [][]TSSPFile, newFiles [][]TSSPFile, ...) (err error)

+ func (m *MmsTables) SetAddFunc(addFunc func(int64))

+ func (m *MmsTables) SetOpId(shardId uint64, opId uint64)

+ func (m *MmsTables) UnRefSequencer()

+ type Offset struct

+ type SegmentReader struct

+ func NewSegmentReader(fi *FileIterator) *SegmentReader

+ func (sr *SegmentReader) Read(seg Segment, ref *record.Field, col *record.ColVal) error

+ func (sr *SegmentReader) ResetContext()

type Sequencer

+ func (s *Sequencer) IsLoading() bool

+ type StreamWriteFile struct

+ Conf *Config

+ func NewWriteScanFile(mst string, m *MmsTables, file TSSPFile, schema record.Schemas) (*StreamWriteFile, error)

+ func (c *StreamWriteFile) AppendColumn(ref record.Field) error

+ func (c *StreamWriteFile) ChangeColumn(ref record.Field) error

+ func (c *StreamWriteFile) ChangeSid(sid uint64)

+ func (c *StreamWriteFile) Close(isError bool)

+ func (c *StreamWriteFile) Flush() error

+ func (c *StreamWriteFile) GetTSSPFile() TSSPFile

+ func (c *StreamWriteFile) Init(id uint64, chunkDataOffset int64, schema record.Schemas)

+ func (c *StreamWriteFile) InitFile(seq uint64) error

+ func (c *StreamWriteFile) InitMergedFile(f TSSPFile) error

+ func (c *StreamWriteFile) NewFile(addFileExt bool) error

+ func (c *StreamWriteFile) NewTSSPFile(tmp bool) (TSSPFile, error)

+ func (c *StreamWriteFile) Size() int64

+ func (c *StreamWriteFile) WriteCurrentMeta() error

+ func (c *StreamWriteFile) WriteData(id uint64, ref record.Field, col record.ColVal, timeCol *record.ColVal) error

+ func (c *StreamWriteFile) WriteFile() error

+ func (c *StreamWriteFile) WriteMeta(cm *ChunkMeta) error

type TSSPFile

+ FileName func() TSSPFileName

+ FreeFileHandle func() error

+ LoadComponents func() error

+ MetaIndexItemNum func() int64

+ RefFileReader func()

+ UnrefFileReader func()

type TSSPFileReader

+ func (r *TSSPFileReader) FreeFileHandle() error

+ func (r *TSSPFileReader) LoadComponents() error

+ func (r *TSSPFileReader) Ref()

+ func (r *TSSPFileReader) Unref()

type TableReader

+ FreeFileHandle func() error

+ LoadComponents func() error

+ Ref func()

+ Unref func()

type TablesStore

+ DisableCompAndMerge func()

+ EnableCompAndMerge func()

+ FreeSequencer func() bool

+ GetBothFilesRef func(measurement string, hasTimeFilter bool, tr record.TimeRange) ([]TSSPFile, []TSSPFile)

+ GetFileSeq func() uint64

+ GetTSSPFiles func(mm string, isOrder bool) (*TSSPFiles, bool)

+ IsOutOfOrderFilesExist func() bool

+ ReplaceDownSampleFiles func(mstNames []string, originFiles [][]TSSPFile, newFiles [][]TSSPFile, ...) error

+ SetAddFunc func(addFunc func(int64))

+ SetOpId func(shardId uint64, opId uint64)

+ UnRefSequencer func()

type Trailer

+ func (t *Trailer) MetaIndexItemNum() int64

+ type UnorderedColumnReader struct

+ func (r *UnorderedColumnReader) Read(ref *record.Field, maxTime int64) (*record.ColVal, []int64, error)

+ func (r *UnorderedColumnReader) ReadSchema(res map[string]record.Field, maxTime int64)

+ type UnorderedReader struct

+ func NewUnorderedReader(log *logger.Logger) *UnorderedReader

+ func (r *UnorderedReader) AddFiles(files []TSSPFile)

+ func (r *UnorderedReader) Close()

+ func (r *UnorderedReader) InitSeriesTimes(sid uint64) error

+ func (r *UnorderedReader) Read(sid uint64, ref *record.Field, maxTime int64) (*record.ColVal, []int64, error)

+ func (r *UnorderedReader) ReadRemain(sid uint64, cb remainCallback) error

+ func (r *UnorderedReader) ReadSeriesSchemas(sid uint64, maxTime int64) record.Schemas

+ func (r *UnorderedReader) ReadTimes(min, max int64) []int64

v0

v0.2.0

Dec 7, 2022

v0.1.0

Sep 27, 2022

Changes in this version

+ const AutoCompact

+ const BlockBoolean

+ const BlockFloat64

+ const BlockInteger

+ const BlockString

+ const CompactLevels

+ const DefaultMaxChunkMetaItemCount

+ const DefaultMaxChunkMetaItemSize

+ const DefaultMaxRowsPerSegment

+ const GB

+ const MB

+ const MergeFirstAvgSize

+ const MergeFirstDstSize

+ const MergeFirstMergeTime

+ const MergeFirstRatio

+ const MergeFirstSeqLen

+ const MergeFirstSeqSize

+ const NonStreamingCompact

+ const StreamingCompact

+ const StringCompressedZstd

+ const TsspDirName

+ var CLog = Log.NewLogger(errno.ModuleCompact)

+ var ChunkMetaLen = int(unsafe.Sizeof(ChunkMeta{})-24*2) + MinMaxTimeLen

+ var ChunkMetaMinLen = ChunkMetaLen + ColumnMetaLenMin*2

+ var ColumnMetaLenMin = (ColumnMeta{}).bytes(1)

+ var EnableMergeOutOfOrder = true

+ var ErrCompStopped = errors.New("compact stopped")

+ var ErrDirtyLog = errors.New("incomplete log file")

+ var ErrDroppingMst = errors.New("measurement is dropped")

+ var LeveLMinGroupFiles = [CompactLevels]int

+ var LevelCompactRule = []uint16

+ var MaxNumOfFileToMerge = 256

+ var MaxSizeOfFileToMerge int64 = 512 * 1024 * 1024

+ var MetaIndexLen = int(unsafe.Sizeof(MetaIndex{}))

+ var MinMaxTimeLen = int(unsafe.Sizeof(SegmentRange{}))

+ var SegmentLen = (Segment{}).bytes()

+ func AggregateData(newRec, baseRec *record.Record, ops []*comm.CallOption)

+ func CacheDataInMemory() bool

+ func CacheMetaInMemory() bool

+ func CheckCrc32En() bool

+ func CompactRecovery(path string, group *CompactGroup)

+ func DecodeBooleanBlock(in []byte, out *[]byte, ctx *CoderContext) ([]bool, error)

+ func DecodeFloatBlock(in []byte, out *[]byte, ctx *CoderContext) ([]float64, error)

+ func DecodeIntegerBlock(in []byte, out *[]byte, ctx *CoderContext) ([]int64, error)

+ func DecodeStringBlock(in []byte, out *[]byte, dstOffset *[]uint32, ctx *CoderContext) ([]byte, []uint32, error)

+ func DecodeTimestampBlock(in []byte, out *[]byte, ctx *CoderContext) ([]int64, error)

+ func DecodeUnsignedBlock(in []byte, out *[]byte, ctx *CoderContext) ([]uint64, error)

+ func EnableMmapRead(en bool)

+ func EnableReadCache(readCacheLimit int)

+ func EncodeBooleanBlock(in, out []byte, ctx *CoderContext) ([]byte, error)

+ func EncodeFloatBlock(in []byte, out []byte, ctx *CoderContext) ([]byte, error)

+ func EncodeIntegerBlock(in, out []byte, ctx *CoderContext) ([]byte, error)

+ func EncodeStringBlock(in []byte, offset []uint32, out []byte, ctx *CoderContext) ([]byte, error)

+ func EncodeTimestampBlock(in, out []byte, ctx *CoderContext) ([]byte, error)

+ func EncodeUnsignedBlock(in, out []byte, ctx *CoderContext) ([]byte, error)

+ func EstimateBufferSize(recSize int, rows int) int

+ func FilterByField(rec *record.Record, filterMap map[string]interface{}, con influxql.Expr, ...) *record.Record

+ func FilterByTime(rec *record.Record, tr record.TimeRange) *record.Record

+ func FilterByTimeDescend(rec *record.Record, tr record.TimeRange) *record.Record

+ func Init()

+ func InitDecFunctions()

+ func IsInterfaceNil(value interface{}) bool

+ func IsTempleFile(name string) bool

+ func MaxRowsPerSegment() int

+ func MergeFlag() int32

+ func MergeRecovery(path string, name string, inputs *OutOfOrderMergeContext)

+ func NewDiskFileReader(f fileops.File) *diskFileReader

+ func NewDiskWriter(lw NameReadWriterCloser, bufferSize int) *diskWriter

+ func NewIndexWriter(indexName string, cacheMeta bool, limitCompact bool) *indexWriter

+ func NewMemReaderEvictCtx() *memReaderEvictCtx

+ func NewMergeHelper(logger *logger.Logger, tier uint64, name string, path string, ...) *mergeHelper

+ func NonStreamingCompaction(fi FilesInfo) bool

+ func PutDataCoder(coder DataCoder)

+ func PutIDTimePairs(pair *IdTimePairs)

+ func PutMsBuilder(msBuilder *MsBuilder)

+ func ReleaseColumnBuilder(b PreAggBuilder)

+ func RenameTmpFiles(newFiles []TSSPFile) error

+ func ResetAggregateData(newRec *record.Record, ops []*comm.CallOption)

+ func SegMergeFlag(v int32)

+ func SetCacheDataBlock(en bool)

+ func SetCacheMetaData(en bool)

+ func SetCompactLimit(bytesPerSec int64, burstLimit int64)

+ func SetImmTableMaxMemoryPercentage(sysTotalMem, percentage int)

+ func SetMaxCompactor(n int)

+ func SetMaxFullCompactor(n int)

+ func SetMaxRowsPerSegment(maxRowsPerSegmentLimit int)

+ func SetMaxSegmentLimit(limit int)

+ func SetSnapshotLimit(bytesPerSec int64, burstLimit int64)

+ func SnapshotLimit() bool

+ func UnrefFiles(files ...TSSPFile)

+ func ZSTDCompressBound(srcSize int) int

+ func ZigZagDecode(v uint64) int64

+ func ZigZagEncode(v int64) uint64

+ type Boolean struct

+ func GetBoolCoder() *Boolean

+ func (enc *Boolean) Decoding(in []byte, out []byte) ([]byte, error)

+ func (enc *Boolean) Encoding(in []byte, out []byte) ([]byte, error)

+ func (enc *Boolean) SetEncodingType(ty int)

+ type BooleanPreAgg struct

+ func NewBooleanPreAgg() *BooleanPreAgg

+ type BytesBuffer struct

+ func NewBytesBuffer(buf []byte) *BytesBuffer

+ func (w *BytesBuffer) Bytes() []byte

+ func (w *BytesBuffer) Len() int

+ func (w *BytesBuffer) Read(p []byte) (n int, err error)

+ func (w *BytesBuffer) ReadByte() (b byte, err error)

+ func (w *BytesBuffer) Reset(b []byte)

+ func (w *BytesBuffer) Write(p []byte) (n int, err error)

+ func (w *BytesBuffer) WriteByte(b byte) error

+ type ChunkDataBuilder struct

+ func NewChunkDataBuilder(maxRowsPerSegment, maxSegmentLimit int) *ChunkDataBuilder

+ func (b *ChunkDataBuilder) EncodeChunk(id uint64, offset int64, rec *record.Record, dst []byte) ([]byte, error)

+ func (b *ChunkDataBuilder) EncodeTime(offset int64) error

+ type ChunkIterator struct

+ func NewChunkIterator(r *FileIterator) *ChunkIterator

+ func (c *ChunkIterator) Close()

+ func (c *ChunkIterator) Next() bool

+ func (c *ChunkIterator) WithLog(log *Log.Logger)

+ type ChunkIterators struct

+ func (c *ChunkIterators) Close()

+ func (c *ChunkIterators) Len() int

+ func (c *ChunkIterators) Less(i, j int) bool

+ func (c *ChunkIterators) Next() (uint64, *record.Record, error)

+ func (c *ChunkIterators) Pop() interface{}

+ func (c *ChunkIterators) Push(v interface{})

+ func (c *ChunkIterators) Swap(i, j int)

+ func (c *ChunkIterators) WithLog(log *Log.Logger)

+ type ChunkMeta struct

+ func (m *ChunkMeta) GetSid() (sid uint64)

+ func (m *ChunkMeta) MinMaxTime() (min int64, max int64)

+ func (m *ChunkMeta) Reset()

+ func (m *ChunkMeta) Rows(ab PreAggBuilder) int

+ func (m *ChunkMeta) Size() int

+ func (m *ChunkMeta) TimeMeta() *ColumnMeta

+ type CoderContext struct

+ func NewCoderContext() *CoderContext

+ func (ctx *CoderContext) Release()

+ type ColumnBuilder struct

+ func NewColumnBuilder() *ColumnBuilder

+ func (b *ColumnBuilder) EncodeColumn(ref record.Field, col *record.ColVal, timeCols []record.ColVal, ...) ([]byte, error)

+ type ColumnMeta struct

+ func (m *ColumnMeta) RowCount(ref *record.Field, decs *ReadContext) (int64, error)

+ type ColumnReader interface

+ ReadDataBlock func(offset int64, size uint32, dst *[]byte) ([]byte, error)

+ ReadMetaBlock func(metaIdx int, id uint64, offset int64, size uint32, count uint32, dst *[]byte) ([]byte, error)

+ type CompactGroup struct

+ func NewCompactGroup(name string, toLevle uint16, count int) *CompactGroup

+ type CompactedFileInfo struct

+ IsOrder bool

+ Name string

+ NewFile []string

+ OldFile []string

+ type Config struct

+ func NewConfig() *Config

+ func (c *Config) MaxRowsPerSegment() int

+ func (c *Config) SetFilesLimit(n int64)

+ func (c *Config) SetMaxRowsPerSegment(maxRowsPerSegmentLimit int)

+ func (c *Config) SetMaxSegmentLimit(n int)

+ type DataCoder interface

+ Decoding func(in []byte, out []byte) ([]byte, error)

+ Encoding func(in []byte, out []byte) ([]byte, error)

+ SetEncodingType func(ty int)

+ type DiskFileReader interface

+ Close func() error

+ IsMmapRead func() bool

+ Name func() string

+ ReadAt func(off int64, size uint32, dst *[]byte) ([]byte, error)

+ Rename func(newName string) error

+ type FileIterator struct

+ func NewFileIterator(r TSSPFile, log *Log.Logger) *FileIterator

+ func (itr *FileIterator) Close()

+ func (itr *FileIterator) NextChunkMeta() bool

+ func (itr *FileIterator) WithLog(log *Log.Logger)

+ type FileIterators []*FileIterator

+ func (i FileIterators) AverageRows() int

+ func (i FileIterators) Close()

+ func (i FileIterators) MaxChunkRows() int

+ func (i FileIterators) MaxColumns() int

+ type FileWriter interface

+ AppendChunkMetaToData func() error

+ ChunkMetaSize func() int64

+ Close func() error

+ DataSize func() int64

+ DataWriter func() io.Writer

+ MetaDataBlocks func(dst [][]byte) [][]byte

+ Name func() string

+ SwitchMetaBuffer func()

+ WriteChunkMeta func(b []byte) (int, error)

+ WriteData func(b []byte) (int, error)

+ type FilesInfo struct

+ type FilterOptions struct

+ func NewFilterOpts(cond influxql.Expr, filterMap map[string]interface{}, fieldsIdx []int, ...) *FilterOptions

+ type Float struct

+ func GetFloatCoder() *Float

+ func (enc *Float) Decoding(in []byte, out []byte) ([]byte, error)

+ func (enc *Float) Encoding(in []byte, out []byte) ([]byte, error)

+ func (enc *Float) Reset(dst []byte)

+ func (enc *Float) SetEncodingType(ty int)

+ type FloatPreAgg struct

+ func NewFloatPreAgg() *FloatPreAgg

+ type IdTimePairs struct

+ Ids []uint64

+ Name string

+ Rows []int64

+ Tms []int64

+ func GetIDTimePairs(name string) *IdTimePairs

+ func (p *IdTimePairs) Add(id uint64, tm int64)

+ func (p *IdTimePairs) AddRowCounts(rowCounts int64)

+ func (p *IdTimePairs) Len() int

+ func (p *IdTimePairs) Marshal(isOrder bool, dst []byte, ctx *CoderContext) []byte

+ func (p *IdTimePairs) Reset(name string)

+ func (p *IdTimePairs) Unmarshal(isOrder bool, src []byte) ([]byte, error)

+ type InMerge struct

+ func NewInMerge() *InMerge

+ func (m *InMerge) Add(name string) bool

+ func (m *InMerge) Del(name string)

+ type Integer struct

+ func GetInterCoder() *Integer

+ func (enc *Integer) Decoding(in []byte, out []byte) ([]byte, error)

+ func (enc *Integer) Encoding(in []byte, out []byte) ([]byte, error)

+ func (enc *Integer) SetEncodingType(ty int)

+ type IntegerPreAgg struct

+ func NewIntegerPreAgg() *IntegerPreAgg

+ type LimitWriter struct

+ func (w *LimitWriter) Close() error

+ func (w *LimitWriter) Name() string

+ func (w *LimitWriter) Read(b []byte) (int, error)

+ func (w *LimitWriter) Write(p []byte) (int, error)

+ type Limiter interface

+ Burst func() int

+ Limit func() rate.Limit

+ SetBurst func(newBurst int)

+ SetLimit func(newLimit rate.Limit)

+ WaitN func(ctx context.Context, n int) (err error)

+ func NewLimiter(bytesPerSec, burstLimit int) Limiter

+ type Location struct

+ func NewLocation(r TSSPFile, decs *ReadContext) *Location

+ func (l *Location) Contains(sid uint64, tr record.TimeRange) (bool, error)

+ func (l *Location) GetChunkMeta() *ChunkMeta

+ func (l *Location) Next() bool

+ func (l *Location) ReadData(filterOpts *FilterOptions, dst *record.Record) (*record.Record, error)

+ func (l *Location) ReadDone()

+ func (l *Location) ResetMeta()

+ type LocationCursor struct

+ func NewLocationCursor(n int) *LocationCursor

+ func (l *LocationCursor) AddLocation(loc *Location)

+ func (l *LocationCursor) AddRef()

+ func (l *LocationCursor) ReadData(filterOpts *FilterOptions, dst *record.Record) (*record.Record, error)

+ func (l *LocationCursor) ReadMeta(filterOpts *FilterOptions, dst *record.Record) (*record.Record, error)

+ func (l *LocationCursor) ReadOutOfOrderMeta(filterOpts *FilterOptions, dst *record.Record) (*record.Record, error)

+ func (l *LocationCursor) Unref()

+ func (l LocationCursor) Len() int

+ func (l LocationCursor) Less(i, j int) bool

+ func (l LocationCursor) Reverse()

+ func (l LocationCursor) Swap(i, j int)

+ type MemBlock struct

+ func (mb *MemBlock) AppendDataBlock(srcData []byte)

+ func (mb *MemBlock) CopyBlocks(src MemoryReader)

+ func (mb *MemBlock) DataBlocks() [][]byte

+ func (mb *MemBlock) DataInMemory() bool

+ func (mb *MemBlock) FreeMemory() int64

+ func (mb *MemBlock) LoadIntoMemory(dr DiskFileReader, tr *Trailer, metaIndexItems []MetaIndex) error

+ func (mb *MemBlock) MetaBlocks() [][]byte

+ func (mb *MemBlock) MetaInMemory() bool

+ func (mb *MemBlock) ReadChunkMetaBlock(metaIdx int, sid uint64, count uint32) []byte

+ func (mb *MemBlock) ReadDataBlock(offset int64, size uint32, dstPtr *[]byte) ([]byte, error)

+ func (mb *MemBlock) ReserveDataBlock(n int)

+ func (mb *MemBlock) ReserveMetaBlock(n int)

+ func (mb *MemBlock) Reset()

+ func (mb *MemBlock) SetMetaBlocks(blocks [][]byte)

+ func (mb *MemBlock) Size() int64

+ type MemoryReader interface

+ AppendDataBlock func(srcData []byte)

+ CopyBlocks func(src MemoryReader)

+ DataBlocks func() [][]byte

+ DataInMemory func() bool

+ FreeMemory func() int64

+ LoadIntoMemory func(dr DiskFileReader, tr *Trailer, metaIndexItems []MetaIndex) error

+ MetaBlocks func() [][]byte

+ MetaInMemory func() bool

+ ReadChunkMetaBlock func(metaIdx int, sid uint64, count uint32) []byte

+ ReadDataBlock func(offset int64, size uint32, dstPtr *[]byte) ([]byte, error)

+ ReserveDataBlock func(n int)

+ ReserveMetaBlock func(n int)

+ Reset func()

+ SetMetaBlocks func(blocks [][]byte)

+ Size func() int64

+ func NewMemReader() MemoryReader

+ func NewMemoryReader(blkSize int) MemoryReader

+ type MetaIndex struct

+ type MmsIdTime struct

+ func NewMmsIdTime() *MmsIdTime

+ type MmsReaders struct

+ Orders TableReaders

+ OutOfOrders TableReaders

+ type MmsTables struct

+ Conf *Config

+ Order map[string]*TSSPFiles

+ OutOfOrder map[string]*TSSPFiles

+ func NewTableStore(dir string, tier *uint64, compactRecovery bool, config *Config) *MmsTables

+ func (m *MmsTables) AddRowCountsBySid(measurement string, sid uint64, rowCounts int64)

+ func (m *MmsTables) AddTSSPFiles(name string, isOrder bool, files ...TSSPFile)

+ func (m *MmsTables) AddTable(mb *MsBuilder, isOrder bool, tmp bool)

+ func (m *MmsTables) Close() error

+ func (m *MmsTables) CompactDone(files []string)

+ func (m *MmsTables) CompactionDisable()

+ func (m *MmsTables) CompactionEnable()

+ func (m *MmsTables) CompactionEnabled() bool

+ func (m *MmsTables) DropMeasurement(_ context.Context, name string) error

+ func (m *MmsTables) File(mstName string, fileName string, isOrder bool) TSSPFile

+ func (m *MmsTables) FreeAllMemReader()

+ func (m *MmsTables) FullCompact(shid uint64) error

+ func (m *MmsTables) GetFilesRef(measurement string, isOrder bool) []TSSPFile

+ func (m *MmsTables) GetFilesRefByAscending(measurement string, isOrder bool, ascending bool, tr record.TimeRange) []TSSPFile

+ func (m *MmsTables) GetLastFlushTimeBySid(measurement string, sid uint64) int64

+ func (m *MmsTables) GetMstFileStat() *statistics.FileStat

+ func (m *MmsTables) GetOutOfOrderFileNum() int

+ func (m *MmsTables) GetRowCountsBySid(measurement string, sid uint64) int64

+ func (m *MmsTables) LevelCompact(level uint16, shid uint64) error

+ func (m *MmsTables) LevelPlan(level uint16) []*CompactGroup

+ func (m *MmsTables) MergeDisable()

+ func (m *MmsTables) MergeEnable()

+ func (m *MmsTables) MergeEnabled() bool

+ func (m *MmsTables) MergeOutOfOrder(shId uint64) error

+ func (m *MmsTables) NewChunkIterators(group FilesInfo) (*ChunkIterators, error)

+ func (m *MmsTables) NewFileIterators(group *CompactGroup) (FilesInfo, error)

+ func (m *MmsTables) NewStreamIterators(group FilesInfo) (*StreamIterators, error)

+ func (m *MmsTables) NextSequence() uint64

+ func (m *MmsTables) Open() (int64, int64, error)

+ func (m *MmsTables) ReplaceFiles(name string, oldFiles, newFiles []TSSPFile, isOrder bool, log *Log.Logger) (err error)

+ func (m *MmsTables) Sequencer() *Sequencer

+ func (m *MmsTables) Tier() uint64

+ func (m *MmsTables) Wait()

+ type MsBuilder struct

+ Conf *Config

+ FileName TSSPFileName

+ Files []TSSPFile

+ MaxIds int

+ Path string

+ func AllocMsBuilder(dir string, name string, conf *Config, idCount int, fileName TSSPFileName, ...) *MsBuilder

+ func GetMsBuilder(dir, name string, fileName TSSPFileName, conf *Config, sequencer *Sequencer, ...) *MsBuilder

+ func (b *MsBuilder) FileVersion() uint64

+ func (b *MsBuilder) Flush() error

+ func (b *MsBuilder) MaxRowsPerSegment() int

+ func (b *MsBuilder) Name() string

+ func (b *MsBuilder) NewTSSPFile(tmp bool) (TSSPFile, error)

+ func (b *MsBuilder) Reset()

+ func (b *MsBuilder) Size() int64

+ func (b *MsBuilder) WithLog(log *Log.Logger)

+ func (b *MsBuilder) WriteData(id uint64, data *record.Record) error

+ func (b *MsBuilder) WriteRecord(id uint64, data *record.Record, ...) (*MsBuilder, error)

+ type MsBuilderPool struct

+ func NewMsBuilderPool() *MsBuilderPool

+ type NameReadWriterCloser interface

+ Name func() string

+ func NewLimitWriter(w NameReadWriterCloser, l Limiter) NameReadWriterCloser

+ type OrderInfo struct

+ func (inf *OrderInfo) Len() int

+ func (inf *OrderInfo) Less(i, j int) bool

+ func (inf *OrderInfo) Swap(i, j int)

+ type OutOfOrderMergeContext struct

+ Names []string

+ Seqs []uint64

+ func NewOutOfOrderMergeContext() *OutOfOrderMergeContext

+ func (ctx *OutOfOrderMergeContext) Add(f TSSPFile) bool

+ func (ctx *OutOfOrderMergeContext) Release()

+ type PreAggBuilder interface

+ type PreAggBuilders struct

+ func (b *PreAggBuilders) Release()

+ type ReadContext struct

+ Ascending bool

+ func NewReadContext(ascending bool) *ReadContext

+ func (d *ReadContext) GetOps() []*comm.CallOption

+ func (d *ReadContext) InitPreAggBuilder()

+ func (d *ReadContext) MatchPreAgg() bool

+ func (d *ReadContext) Release()

+ func (d *ReadContext) Reset()

+ func (d *ReadContext) Set(ascending bool, tr record.TimeRange, onlyFirstOrLast bool, ...)

+ func (d *ReadContext) SetOps(c []*comm.CallOption)

+ func (d *ReadContext) SetTr(tr record.TimeRange)

+ type Segment struct

+ type SegmentRange [2]int64

+ type Sequencer struct

+ func NewSequencer() *Sequencer

+ func (s *Sequencer) AddRowCounts(mn string, id uint64, rowCounts int64)

+ func (s *Sequencer) BatchUpdate(p *IdTimePairs)

+ func (s *Sequencer) BatchUpdateCheckTime(p *IdTimePairs)

+ func (s *Sequencer) Get(mn string, id uint64) (lastFlushTime, rowCnt int64)

+ type StreamIterator struct

+ func NewStreamStreamIterator(fi *FileIterator) *StreamIterator

+ type StreamIterators struct

+ Conf *Config

+ func (c *StreamIterators) Close()

+ func (c *StreamIterators) FileVersion() uint64

+ func (c *StreamIterators) Flush() error

+ func (c *StreamIterators) Init(id uint64, chunkDataOffset int64, schema record.Schemas)

+ func (c *StreamIterators) Len() int

+ func (c *StreamIterators) Less(i, j int) bool

+ func (c *StreamIterators) NewFile(addFileExt bool) error

+ func (c *StreamIterators) NewTSSPFile(tmp bool) (TSSPFile, error)

+ func (c *StreamIterators) Pop() interface{}

+ func (c *StreamIterators) Push(v interface{})

+ func (c *StreamIterators) Size() int64

+ func (c *StreamIterators) Swap(i, j int)

+ func (c *StreamIterators) WithLog(log *Log.Logger)

+ type StreamIteratorsPool struct

+ func NewStreamIteratorsPool(n int) *StreamIteratorsPool

+ type String struct

+ func GetStringCoder() *String

+ func (enc *String) Decoding(in []byte, out []byte) ([]byte, error)

+ func (enc *String) Encoding(in []byte, out []byte) ([]byte, error)

+ func (enc *String) MaxEncodedLen(size int) int

+ func (enc *String) SetEncodingType(ty int)

+ type StringPreAgg struct

+ func NewStringPreAgg() *StringPreAgg

+ type TSSPFile interface

+ AddToEvictList func(level uint16)

+ AverageChunkRows func() int

+ ChunkAt func(index int) (*ChunkMeta, error)

+ ChunkMeta func(id uint64, offset int64, size, itemCount uint32, metaIdx int, dst *ChunkMeta) (*ChunkMeta, error)

+ Close func() error

+ Contains func(id uint64) (bool, error)

+ ContainsByTime func(tr record.TimeRange) (bool, error)

+ ContainsValue func(id uint64, tr record.TimeRange) (bool, error)

+ CreateTime func() int64

+ Delete func(ids []int64) error

+ DeleteRange func(ids []int64, min, max int64) error

+ FileNameExtend func() uint16

+ FileNameMerge func() uint16

+ FileSize func() int64

+ FileStat func() *Trailer

+ FreeMemory func(evictLock bool) int64

+ HasTombstones func() bool

+ InMemSize func() int64

+ Inuse func() bool

+ IsOrder func() bool

+ LevelAndSequence func() (uint16, uint64)

+ LoadIdTimes func(p *IdTimePairs) error

+ LoadIndex func() error

+ LoadIntoMemory func() error

+ MaxChunkRows func() int

+ MetaIndex func(id uint64, tr record.TimeRange) (int, *MetaIndex, error)

+ MetaIndexAt func(idx int) (*MetaIndex, error)

+ MinMaxTime func() (int64, int64, error)

+ Name func() string

+ Open func() error

+ Path func() string

+ Read func(id uint64, tr record.TimeRange, dst *record.Record) (*record.Record, error)

+ ReadAt func(cm *ChunkMeta, segment int, dst *record.Record, decs *ReadContext) (*record.Record, error)

+ ReadChunkMetaData func(metaIdx int, m *MetaIndex, dst []ChunkMeta) ([]ChunkMeta, error)

+ ReadData func(offset int64, size uint32, dst *[]byte) ([]byte, error)

+ Ref func()

+ Remove func() error

+ RemoveFromEvictList func(level uint16)

+ Rename func(newName string) error

+ Stop func()

+ TombstoneFiles func() []TombstoneFile

+ Unref func()

+ Version func() uint64

+ func OpenTSSPFile(name string, isOrder bool, cacheData bool) (TSSPFile, error)

+ type TSSPFileName struct

+ func NewTSSPFileName(seq uint64, level, merge, extent uint16, order bool) TSSPFileName

+ func (n *TSSPFileName) Equal(other *TSSPFileName) bool

+ func (n *TSSPFileName) ParseFileName(name string) error

+ func (n *TSSPFileName) Path(dir string, tmp bool) string

+ func (n *TSSPFileName) SetExtend(extend uint16)

+ func (n *TSSPFileName) SetLevel(l uint16)

+ func (n *TSSPFileName) SetMerge(merge uint16)

+ func (n *TSSPFileName) SetOrder(v bool)

+ func (n *TSSPFileName) SetSeq(seq uint64)

+ func (n *TSSPFileName) String() string

+ func (n *TSSPFileName) TmpPath(dir string) string

+ type TSSPFileReader struct

+ func CreateTSSPFileReader(size int64, fd fileops.File, trailer *Trailer, tb *TableData, ver uint64, ...) (*TSSPFileReader, error)

+ func NewTSSPFileReader(name string) (*TSSPFileReader, error)

+ func (r *TSSPFileReader) AverageChunkRows() int

+ func (r *TSSPFileReader) BlockHeader(meta *ChunkMeta, dst []record.Field) ([]record.Field, error)

+ func (r *TSSPFileReader) ChunkMeta(id uint64, offset int64, size, itemCount uint32, metaIdx int, dst *ChunkMeta) (*ChunkMeta, error)

+ func (r *TSSPFileReader) ChunkMetaAt(index int) (*ChunkMeta, error)

+ func (r *TSSPFileReader) Close() error

+ func (r *TSSPFileReader) Contains(id uint64, tm record.TimeRange) bool

+ func (r *TSSPFileReader) ContainsId(id uint64) bool

+ func (r *TSSPFileReader) ContainsTime(tm record.TimeRange) bool

+ func (r *TSSPFileReader) CreateTime() int64

+ func (r *TSSPFileReader) FileName() string

+ func (r *TSSPFileReader) FileSize() int64

+ func (r *TSSPFileReader) FreeMemory() int64

+ func (r *TSSPFileReader) GetTSSPFileBytes(offset int64, size uint32, buf *[]byte) ([]byte, error)

+ func (r *TSSPFileReader) InMemSize() int64

+ func (r *TSSPFileReader) LoadIndex() error

+ func (r *TSSPFileReader) LoadIntoMemory() error

+ func (r *TSSPFileReader) MaxChunkRows() int

+ func (r *TSSPFileReader) MetaIndex(id uint64, tr record.TimeRange) (int, *MetaIndex, error)

+ func (r *TSSPFileReader) MetaIndexAt(idx int) (*MetaIndex, error)

+ func (r *TSSPFileReader) MinMaxSeriesID() (min, max uint64, err error)

+ func (r *TSSPFileReader) MinMaxTime() (min, max int64, err error)

+ func (r *TSSPFileReader) Name() string

+ func (r *TSSPFileReader) Open() error

+ func (r *TSSPFileReader) Read(offset int64, size uint32, dst *[]byte) ([]byte, error)

+ func (r *TSSPFileReader) ReadChunkMetaData(metaIdx int, m *MetaIndex, dst []ChunkMeta) ([]ChunkMeta, error)

+ func (r *TSSPFileReader) ReadData(cm *ChunkMeta, segment int, dst *record.Record, decs *ReadContext) (*record.Record, error)

+ func (r *TSSPFileReader) ReadDataBlock(offset int64, size uint32, dst *[]byte) (rb []byte, err error)

+ func (r *TSSPFileReader) ReadMetaBlock(metaIdx int, id uint64, offset int64, size uint32, count uint32, dst *[]byte) (rb []byte, err error)

+ func (r *TSSPFileReader) Rename(newName string) error

+ func (r *TSSPFileReader) Stat() *Trailer

+ func (r *TSSPFileReader) Version() uint64

+ type TSSPFiles struct

+ func NewTSSPFiles() *TSSPFiles

+ func (f *TSSPFiles) Append(file TSSPFile)

+ func (f *TSSPFiles) Files() []TSSPFile

+ func (f *TSSPFiles) Len() int

+ func (f *TSSPFiles) Less(i, j int) bool

+ func (f *TSSPFiles) StopFiles()

+ func (f *TSSPFiles) Swap(i, j int)

+ type TableData struct

+ type TableReader interface

+ AverageChunkRows func() int

+ BlockHeader func(meta *ChunkMeta, dst []record.Field) ([]record.Field, error)

+ ChunkMeta func(id uint64, offset int64, size, itemCount uint32, metaIdx int, dst *ChunkMeta) (*ChunkMeta, error)

+ ChunkMetaAt func(index int) (*ChunkMeta, error)

+ Close func() error

+ Contains func(id uint64, tm record.TimeRange) bool

+ ContainsId func(id uint64) bool

+ ContainsTime func(tm record.TimeRange) bool

+ CreateTime func() int64

+ FileName func() string

+ FileSize func() int64

+ FreeMemory func() int64

+ InMemSize func() int64

+ LoadIndex func() error

+ LoadIntoMemory func() error

+ MaxChunkRows func() int

+ MetaIndex func(id uint64, tr record.TimeRange) (int, *MetaIndex, error)

+ MetaIndexAt func(idx int) (*MetaIndex, error)

+ MinMaxSeriesID func() (min, max uint64, err error)

+ MinMaxTime func() (min, max int64, err error)

+ Name func() string

+ Open func() error

+ Read func(offset int64, size uint32, dst *[]byte) ([]byte, error)

+ ReadChunkMetaData func(metaIdx int, m *MetaIndex, dst []ChunkMeta) ([]ChunkMeta, error)

+ ReadData func(cm *ChunkMeta, segment int, dst *record.Record, decs *ReadContext) (*record.Record, error)

+ ReadDataBlock func(offset int64, size uint32, dst *[]byte) ([]byte, error)

+ ReadMetaBlock func(metaIdx int, id uint64, offset int64, size uint32, count uint32, dst *[]byte) ([]byte, error)

+ Rename func(newName string) error

+ Stat func() *Trailer

+ Version func() uint64

+ type TableReaders []TSSPFile

+ func (tables TableReaders) Len() int

+ func (tables TableReaders) Less(i, j int) bool

+ func (tables TableReaders) Swap(i, j int)

+ type TableStat struct

+ type TableStoreGC struct

+ func (sgc *TableStoreGC) Add(free bool, files ...TSSPFile)

+ func (sgc *TableStoreGC) GC()

+ type TablesGC interface

+ Add func(free bool, files ...TSSPFile)

+ GC func()

+ func NewTableStoreGC() TablesGC

+ type TablesStore interface

+ AddRowCountsBySid func(measurement string, sid uint64, rowCounts int64)

+ AddTSSPFiles func(name string, isOrder bool, f ...TSSPFile)

+ AddTable func(ms *MsBuilder, isOrder bool, tmp bool)

+ Close func() error

+ CompactDone func(seq []string)

+ CompactionDisable func()

+ CompactionEnable func()

+ CompactionEnabled func() bool

+ DropMeasurement func(ctx context.Context, name string) error

+ File func(name string, namePath string, isOrder bool) TSSPFile

+ FreeAllMemReader func()

+ FullCompact func(shid uint64) error

+ GetFilesRef func(measurement string, isOrder bool) []TSSPFile

+ GetFilesRefByAscending func(measurement string, isOrder bool, ascending bool, tr record.TimeRange) []TSSPFile

+ GetLastFlushTimeBySid func(measurement string, sid uint64) int64

+ GetMstFileStat func() *stats.FileStat

+ GetOutOfOrderFileNum func() int

+ GetRowCountsBySid func(measurement string, sid uint64) int64

+ LevelCompact func(level uint16, shid uint64) error

+ MergeDisable func()

+ MergeEnable func()

+ MergeEnabled func() bool

+ MergeOutOfOrder func(shId uint64) error

+ NextSequence func() uint64

+ Open func() (int64, int64, error)

+ ReplaceFiles func(name string, oldFiles, newFiles []TSSPFile, isOrder bool, log *Log.Logger) error

+ Sequencer func() *Sequencer

+ Tier func() uint64

+ type Time struct

+ func GetTimeCoder() *Time

+ func (enc *Time) Decoding(in []byte, out []byte) ([]byte, error)

+ func (enc *Time) Encoding(in []byte, out []byte) ([]byte, error)

+ func (enc *Time) SetEncodingType(ty int)

+ type TimePreAgg struct

+ func NewTimePreAgg() *TimePreAgg

+ type Tombstone struct

+ ID uint64

+ MaxTime int64

+ MinTime int64

+ type TombstoneFile struct

+ func (t *TombstoneFile) Path() string

+ func (t *TombstoneFile) TombstonesCount() int

+ type Trailer struct

+ func (t *Trailer) ContainsId(id uint64) bool

+ func (t *Trailer) ContainsTime(tm record.TimeRange) bool

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL