frac

package
v0.73.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 25, 2026 License: Apache-2.0 Imports: 45 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Active 

type Active struct {
	Config *Config

	BaseFileName string

	MIDs *UInt64s
	RIDs *UInt64s

	DocBlocks *UInt64s

	TokenList *TokenList

	DocsPositions *DocsPositions
	IDsToLIDs     *ActiveLIDs
	// contains filtered or unexported fields
}

func NewActive 

func NewActive(
	baseFileName string,
	activeIndexer *ActiveIndexer,
	readLimiter *storage.ReadLimiter,
	docsCache *cache.Cache[[]byte],
	sortCache *cache.Cache[[]byte],
	cfg *Config,
	skipMaskProvider skipMaskProvider,
) *Active

func (*Active) Append 

func (f *Active) Append(docs storage.DocBlock, metas storage.WalBlock, wg *sync.WaitGroup) (err error)

Append causes data to be written on disk and sends metas to index workers

func (*Active) AppendIDs 

func (f *Active) AppendIDs(ids []seq.ID) []uint32

func (*Active) Contains 

func (f *Active) Contains(id seq.MID) bool

func (*Active) Fetch added in v0.62.1 

func (f *Active) Fetch(ctx context.Context, ids []seq.ID, noSkipMasks bool) ([][]byte, error)

func (*Active) FindLIDs added in v0.70.0 

func (f *Active) FindLIDs(ctx context.Context, ids []seq.ID) ([]seq.LID, error)

func (*Active) GetAllDocuments 

func (f *Active) GetAllDocuments() []uint32

func (*Active) Info 

func (f *Active) Info() *common.Info

func (*Active) IsIntersecting 

func (f *Active) IsIntersecting(from, to seq.MID) bool

func (*Active) Release 

func (f *Active) Release()

func (*Active) Replay 

func (f *Active) Replay(ctx context.Context) error

func (*Active) Search added in v0.62.1 

func (f *Active) Search(ctx context.Context, params processor.SearchParams) (*seq.QPR, error)

func (*Active) String 

func (f *Active) String() string

func (*Active) Suicide 

func (f *Active) Suicide()

func (*Active) UpdateStats 

func (f *Active) UpdateStats(minMID, maxMID seq.MID, docCount uint32, sizeCount uint64)

type ActiveIndexer 

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

func NewActiveIndexer 

func NewActiveIndexer(workerCount, chLen int) (*ActiveIndexer, func())

func (*ActiveIndexer) Index 

func (ai *ActiveIndexer) Index(frac *Active, metas storage.WalBlock, wg *sync.WaitGroup, sw *stopwatch.Stopwatch)

type ActiveLIDs added in v0.70.0 

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

func NewActiveLIDs added in v0.70.0 

func NewActiveLIDs() *ActiveLIDs

func (*ActiveLIDs) Get added in v0.70.0 

func (al *ActiveLIDs) Get(id seq.ID) (seq.LID, bool)

func (*ActiveLIDs) SetMultiple added in v0.70.0 

func (al *ActiveLIDs) SetMultiple(ids []seq.ID, lids []uint32)

type ActiveSealingSource added in v0.62.0 

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

func NewActiveSealingSource added in v0.62.0 

func NewActiveSealingSource(active *Active, params common.SealParams) (*ActiveSealingSource, error)

func (*ActiveSealingSource) BlockOffsets added in v0.71.0 

func (src *ActiveSealingSource) BlockOffsets() []uint64

func (*ActiveSealingSource) Docs added in v0.62.0 

func (src *ActiveSealingSource) Docs() iter.Seq2[Document, error]

Docs returns an iterator for documents with their IDs. Handles duplicate IDs (for nested indexes).

func (*ActiveSealingSource) ID added in v0.71.0 

func (src *ActiveSealingSource) ID() iter.Seq2[DocLocation, error]

func (*ActiveSealingSource) Info added in v0.62.0 

func (src *ActiveSealingSource) Info() *common.Info

func (*ActiveSealingSource) SortDocs added in v0.62.0 

func (src *ActiveSealingSource) SortDocs() error

SortDocs sorts documents and writes them in compressed form to disk. Creates a temporary file that is then renamed to the final one.

func (*ActiveSealingSource) TokenTriplet added in v0.71.0 

func (src *ActiveSealingSource) TokenTriplet() iter.Seq2[string, iter.Seq2[TokenPosting, error]]

type ActiveWriter 

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

func NewActiveWriter 

func NewActiveWriter(docsFile, walFile *os.File, docsOffset, walOffset int64, skipFsync bool) *ActiveWriter

NewActiveWriter creates a writer for *.wal files

func NewActiveWriterLegacy added in v0.69.0 

func NewActiveWriterLegacy(docsFile, metaFile *os.File, docsOffset, metaOffset int64, skipFsync bool) *ActiveWriter

NewActiveWriterLegacy creates a writer for *.meta files

func (*ActiveWriter) Stop 

func (a *ActiveWriter) Stop()

func (*ActiveWriter) Write 

func (a *ActiveWriter) Write(docs storage.DocBlock, meta storage.WalBlock, sw *stopwatch.Stopwatch) error

type AggLimits 

type AggLimits struct {
	MaxFieldTokens     int // MaxFieldTokens max AggQuery.Field uniq values to parse.
	MaxFieldValues     int // MaxFieldValues max AggQuery.Field uniq values to hold per aggregation request.
	MaxGroupTokens     int // MaxGroupTokens max AggQuery.GroupBy unique values.
	MaxTIDsPerFraction int // MaxTIDsPerFraction max number of tokens per fraction.
}

type BulkStats added in v0.65.0 

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

BulkStats holds statistics for bulk operations

type BulkStatsCollector added in v0.65.0 

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

BulkStatsCollector collects and periodically logs bulk operation statistics

func NewBulkStatsCollector added in v0.65.0 

func NewBulkStatsCollector(period time.Duration, queueSize int) *BulkStatsCollector

NewBulkStatsCollector creates a new stats collector with specified parameters

func (*BulkStatsCollector) Add added in v0.65.0 

func (c *BulkStatsCollector) Add(bs BulkStats)

func (*BulkStatsCollector) Stop added in v0.65.0 

func (c *BulkStatsCollector) Stop()

type Config 

type Config struct {
	Search SearchConfig

	SkipSortDocs bool
	KeepMetaFile bool
}

type DocLocation added in v0.71.0 

type DocLocation = util.Pair[seq.ID, seq.DocPos]

type DocsPositions 

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

func NewSyncDocsPositions 

func NewSyncDocsPositions() *DocsPositions

func (*DocsPositions) Get 

func (dp *DocsPositions) Get(id seq.ID) seq.DocPos

func (*DocsPositions) GetSync 

func (dp *DocsPositions) GetSync(id seq.ID) seq.DocPos

func (*DocsPositions) SetMultiple 

func (dp *DocsPositions) SetMultiple(ids []seq.ID, pos []seq.DocPos) []seq.ID

SetMultiple returns a slice of added ids

type Document added in v0.71.0 

type Document = util.Pair[seq.ID, []byte]

type Fraction 

type Fraction interface {
	Info() *common.Info
	IsIntersecting(from seq.MID, to seq.MID) bool
	Contains(mid seq.MID) bool
	Fetch(context.Context, []seq.ID, bool) ([][]byte, error)
	Search(context.Context, processor.SearchParams) (*seq.QPR, error)
	FindLIDs(context.Context, []seq.ID) ([]seq.LID, error)
}

type IndexCache 

type IndexCache struct {
	// Registry cache for legacy sealed fractions.
	LegacyRegistry *cache.Cache[[]byte]

	// Per-file registry caches (each IndexReader needs its own).
	TokenRegistry   *cache.Cache[[]byte]
	OffsetsRegistry *cache.Cache[[]byte]
	IDRegistry      *cache.Cache[[]byte]
	LIDRegistry     *cache.Cache[[]byte]

	// Block-level data caches shared across all readers.
	MIDs   *cache.Cache[[]byte]
	RIDs   *cache.Cache[seqids.BlockRIDs]
	Params *cache.Cache[seqids.BlockParams]

	Tokens     *cache.Cache[*token.Block]
	TokenTable *cache.Cache[token.Table]

	LIDs *cache.Cache[*lids.Block]
}

func (*IndexCache) Release 

func (s *IndexCache) Release()

type IndexReaders added in v0.71.0 

type IndexReaders struct {
	Token   storage.IndexReader
	Offsets storage.IndexReader
	ID      storage.IndexReader
	LID     storage.IndexReader
}

IndexReaders holds one IndexReader per split index file.

type IndexedDocBlock added in v0.71.0 

type IndexedDocBlock = util.Pair[[]byte, []seq.DocPos]

type LegacyLoader added in v0.71.0 

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

LegacyLoader reads the old single .index file format by scanning blocks sequentially. Block indices stored in lids.Table and seqids.Table are absolute within the .index file, so the same IndexReader can be passed to all sub-loaders unchanged.

func (*LegacyLoader) Load added in v0.71.0 

func (l *LegacyLoader) Load(blocksData *sealed.BlocksData, info *common.Info, reader storage.IndexReader)

Load populates blocksData from a single legacy .index file. It starts at block 1 (block 0 is the Info block, already read by loadHeader).

type LegacyMetaWriter added in v0.69.0 

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

LegacyMetaWriter is MetaWriter for the legacy *.meta files. Converts new storage.WalBlock block type to storage.DocBlock

func NewLegacyMetaWriter added in v0.69.0 

func NewLegacyMetaWriter(fw *storage.FileWriter) *LegacyMetaWriter

func (*LegacyMetaWriter) Stop added in v0.69.0 

func (l *LegacyMetaWriter) Stop()

func (*LegacyMetaWriter) Write added in v0.69.0 

func (l *LegacyMetaWriter) Write(data storage.WalBlock, sw *stopwatch.Stopwatch) (int64, error)

type Loader 

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

Loader reads the per-section index files to populate BlocksData. Token data is loaded lazily (BlockLoader / TableLoader use the Token reader directly). Info is loaded separately via loadHeader before Load is called.

func (*Loader) Load 

func (l *Loader) Load(blocksData *sealed.BlocksData, info *common.Info, readers IndexReaders)

Load populates blocksData from the .offsets, .id, and .lid files.

type MetaWriter added in v0.69.0 

type MetaWriter interface {
	Write(data storage.WalBlock, sw *stopwatch.Stopwatch) (int64, error)
	Stop()
}

type PSD 

type PSD int // emulates hard shutdown on different stages of fraction deletion, used for tests

const (
	Off PSD = iota
	HalfRename
	HalfRemove
)

type Remote added in v0.60.0 

type Remote struct {
	Config *Config

	BaseFileName string

	// IsLegacy is true for fractions that use the old single .index file format.
	IsLegacy bool
	// contains filtered or unexported fields
}

Remote fraction is a fraction that is backed by remote storage.

Structure of Remote fraction is almost identical to the Sealed one. In fact, they share the same on-disk binary layout, access methods and any other logic, but having Remote fraction allows us to easily distinguish between local and remote fractions.

func NewRemote added in v0.60.0 

func NewRemote(
	ctx context.Context,
	baseFile string,
	readLimiter *storage.ReadLimiter,
	indexCache *IndexCache,
	docsCache *cache.Cache[[]byte],
	info *common.Info,
	config *Config,
	s3cli *s3.Client,
	skipMaskProvider skipMaskProvider,
	isLegacy bool,
) *Remote

func (*Remote) Contains added in v0.60.0 

func (f *Remote) Contains(mid seq.MID) bool

func (*Remote) Fetch added in v0.62.1 

func (f *Remote) Fetch(ctx context.Context, ids []seq.ID, noSkipMasks bool) ([][]byte, error)

func (*Remote) FindLIDs added in v0.70.0 

func (f *Remote) FindLIDs(ctx context.Context, ids []seq.ID) ([]seq.LID, error)

func (*Remote) Info added in v0.60.0 

func (f *Remote) Info() *common.Info

func (*Remote) IsIntersecting added in v0.60.0 

func (f *Remote) IsIntersecting(from, to seq.MID) bool

func (*Remote) Search added in v0.62.1 

func (f *Remote) Search(ctx context.Context, params processor.SearchParams) (*seq.QPR, error)

func (*Remote) String added in v0.60.0 

func (f *Remote) String() string

func (*Remote) Suicide added in v0.60.0 

func (f *Remote) Suicide()

type Sealed 

type Sealed struct {
	Config *Config

	BaseFileName string

	// IsLegacy is true for fractions that use the old single .index file format.
	IsLegacy bool

	// shit for testing
	PartialSuicideMode PSD
	// contains filtered or unexported fields
}

func NewSealed 

func NewSealed(
	baseFile string,
	readLimiter *storage.ReadLimiter,
	indexCache *IndexCache,
	docsCache *cache.Cache[[]byte],
	info *common.Info,
	config *Config,
	skipMaskProvider skipMaskProvider,
	isLegacy bool,
) *Sealed

func NewSealedPreloaded 

func NewSealedPreloaded(
	baseFile string,
	preloaded *sealed.PreloadedData,
	rl *storage.ReadLimiter,
	indexCache *IndexCache,
	docsCache *cache.Cache[[]byte],
	config *Config,
	skipMaskProvider skipMaskProvider,
) *Sealed

func (*Sealed) Contains 

func (f *Sealed) Contains(id seq.MID) bool

func (*Sealed) Fetch added in v0.62.1 

func (f *Sealed) Fetch(ctx context.Context, ids []seq.ID, noSkipMasks bool) ([][]byte, error)

func (*Sealed) FindLIDs added in v0.70.0 

func (f *Sealed) FindLIDs(ctx context.Context, ids []seq.ID) ([]seq.LID, error)

func (*Sealed) Info 

func (f *Sealed) Info() *common.Info

func (*Sealed) IsIntersecting 

func (f *Sealed) IsIntersecting(from, to seq.MID) bool

func (*Sealed) Offload added in v0.60.0 

func (f *Sealed) Offload(ctx context.Context, u storage.Uploader) (bool, error)

Offload saves all index files and docs to remote storage.

func (*Sealed) Release added in v0.65.0 

func (f *Sealed) Release()

func (*Sealed) Search added in v0.62.1 

func (f *Sealed) Search(ctx context.Context, params processor.SearchParams) (*seq.QPR, error)

func (*Sealed) String 

func (f *Sealed) String() string

func (*Sealed) Suicide 

func (f *Sealed) Suicide()

type SearchConfig 

type SearchConfig struct {
	AggLimits AggLimits
}

type SeqIDCmp 

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

type TokenLIDs 

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

func (*TokenLIDs) GetLIDs 

func (tl *TokenLIDs) GetLIDs(mids, rids *UInt64s) []uint32

func (*TokenLIDs) PutLIDsInQueue 

func (tl *TokenLIDs) PutLIDsInQueue(lids []uint32) int

func (*TokenLIDs) SortedLIDsUnsafe added in v0.71.0 

func (tl *TokenLIDs) SortedLIDsUnsafe() []uint32

SortedLIDs returns pre-merged LIDs. Only safe to call after the fraction is frozen and lids queue was drained.

type TokenList 

type TokenList struct {
	FieldTIDs map[string][]uint32
	// contains filtered or unexported fields
}

func NewActiveTokenList 

func NewActiveTokenList(workers int) *TokenList

func (*TokenList) Append 

func (tl *TokenList) Append(tokens [][]byte, fieldsLengths []int, tokenLIDsPlaces []*TokenLIDs) []*TokenLIDs

func (*TokenList) FindPattern 

func (tl *TokenList) FindPattern(ctx context.Context, t parser.Token) ([]uint32, error)

func (*TokenList) GetAllTokenLIDs 

func (tl *TokenList) GetAllTokenLIDs() *TokenLIDs

func (*TokenList) GetFieldSizes 

func (tl *TokenList) GetFieldSizes() map[string]uint32

func (*TokenList) GetTIDsByField 

func (tl *TokenList) GetTIDsByField(f string) []uint32

func (*TokenList) GetValByTID 

func (tl *TokenList) GetValByTID(tid uint32) []byte

func (*TokenList) Provide 

func (tl *TokenList) Provide(tid uint32) *TokenLIDs

func (*TokenList) Stop 

func (tl *TokenList) Stop()

type TokenPosting added in v0.71.0 

type TokenPosting = util.Pair[[]byte, []uint32]

type UInt64s 

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

func NewIDs 

func NewIDs() *UInt64s

func (*UInt64s) Append 

func (l *UInt64s) Append(val uint64) uint32

func (*UInt64s) GetVals 

func (l *UInt64s) GetVals() []uint64

func (*UInt64s) Len 

func (l *UInt64s) Len() uint32

Source Files

View all Source files

Directories

Path Synopsis
lids
sealing
seqids
token

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.