frac

package
v0.69.0 Latest Latest
Warning

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

 Go to latest Published: Mar 24, 2026 License: Apache-2.0 Imports: 43 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
	// 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,
) *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) ([][]byte, 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 ActiveSealingSource added in v0.62.0 

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

ActiveSealingSource transforms data from in-memory (frac.Active) storage into a format suitable for disk writing during index creation.

The main purpose of this type is to provide access to sorted data through a set of iterators that allow sequential processing of data in sized blocks for disk writing:

  • TokenBlocks() - iterator for token blocks, sorted by fields and values
  • Fields() - iterator for sorted fields with maximum TIDs
  • IDsBlocks() - iterator for document ID blocks and their positions
  • TokenLIDs() - iterator for LID lists for each token
  • Docs() - iterator for documents themselves with duplicate handling

All iterators work with pre-sorted data and return information in an order optimal for creating disk index structures.

func NewActiveSealingSource added in v0.62.0 

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

NewActiveSealingSource creates a new data source for sealing based on an active in-memory index.

func (*ActiveSealingSource) BlocksOffsets added in v0.62.0 

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

BlocksOffsets returns document block offsets.

func (*ActiveSealingSource) Docs added in v0.62.0 

func (src *ActiveSealingSource) Docs() iter.Seq2[seq.ID, []byte]

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

func (*ActiveSealingSource) Fields added in v0.62.0 

func (src *ActiveSealingSource) Fields() iter.Seq2[string, uint32]

Fields returns an iterator for sorted fields and their maximum TIDs. Fields are sorted lexicographically, ensuring predictable order when building disk index structures.

func (*ActiveSealingSource) IDsBlocks added in v0.62.0 

func (src *ActiveSealingSource) IDsBlocks(blockSize int) iter.Seq2[[]seq.ID, []seq.DocPos]

IDsBlocks returns an iterator for document ID blocks and corresponding positions. IDs are sorted. Block size is controlled by blockSize parameter for balance between performance and memory usage.

func (*ActiveSealingSource) Info added in v0.62.0 

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

Info returns index metadata information.

func (*ActiveSealingSource) LastError added in v0.62.0 

func (src *ActiveSealingSource) LastError() error

LastError returns the last error that occurred during processing.

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) TokenBlocks added in v0.62.0 

func (src *ActiveSealingSource) TokenBlocks(blockSize int) iter.Seq[[][]byte]

TokenBlocks returns an iterator for token blocks for disk writing. Tokens are pre-sorted: first by fields, then lexicographically within each field. Each block contains up to blockSize bytes of data for efficient writing.

func (*ActiveSealingSource) TokenLIDs added in v0.62.0 

func (src *ActiveSealingSource) TokenLIDs() iter.Seq[[]uint32]

TokenLIDs returns an iterator for LID lists for each token. LIDs are converted to new numbering after document sorting. Each iterator call returns a list of documents containing a specific token, in sorted order.

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 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 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) ([][]byte, error)
	Search(context.Context, processor.SearchParams) (*seq.QPR, error)
}

type IndexCache 

type IndexCache struct {
	Registry   *cache.Cache[[]byte]
	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 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
}

func (*Loader) Load 

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

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
	// 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,
) *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) ([][]byte, 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

	// 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,
) *Sealed

func NewSealedPreloaded 

func NewSealedPreloaded(
	baseFile string,
	preloaded *sealed.PreloadedData,
	rl *storage.ReadLimiter,
	indexCache *IndexCache,
	docsCache *cache.Cache[[]byte],
	config *Config,
) *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) ([][]byte, 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 `.docs` (or `.sdocs`) and `.index` files into remote storage. It does not free any of the occupied memory (nor on disk nor in memory).

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

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 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.