perf: skip lid blocks

cmd/seq-db/seq-db.go

@@ -266,6 +266,7 @@ func startStore(
 			SealParams: common.SealParams{
 				IDsZstdLevel:           cfg.Compression.SealedZstdCompressionLevel,
 				LIDsZstdLevel:          cfg.Compression.SealedZstdCompressionLevel,
+				LIDBlockSize:           int(cfg.Compression.LIDBlockSize),
 				TokenListZstdLevel:     cfg.Compression.SealedZstdCompressionLevel,
 				DocsPositionsZstdLevel: cfg.Compression.SealedZstdCompressionLevel,
 				TokenTableZstdLevel:    cfg.Compression.SealedZstdCompressionLevel,
@@ -283,6 +284,7 @@ func startStore(
 				},
 				SkipSortDocs: !cfg.DocsSorting.Enabled,
 				KeepMetaFile: false,
+				LIDBlockSize: int(cfg.Compression.LIDBlockSize),
 			},
 			OffloadingEnabled:    cfg.Offloading.Enabled,
 			OffloadingRetention:  cfg.Offloading.Retention,

config/config.go

@@ -200,6 +200,8 @@ type Config struct {
 		MetasZstdCompressionLevel    int `config:"metas_zstd_compression_level" default:"1"`
 		SealedZstdCompressionLevel   int `config:"sealed_zstd_compression_level" default:"3"`
 		DocBlockZstdCompressionLevel int `config:"doc_block_zstd_compression_level" default:"3"`
+		// LIDBlockSize sets max lids (postings) saved per LIDs block.
+		LIDBlockSize Bytes `config:"lid_block_size" default:"64KiB"`
 	} `config:"compression"`
 
 	Indexing struct {

config/frac_version.go

@@ -18,6 +18,11 @@ const (
 	// Also in this version we've changed the binary layout of section storing
 	// info block. As a result we store info as a plain JSON without additional registry.
 	BinaryDataV3
+
+	// BinaryDataV4 - bitpack for LIDs/MIDs
+	BinaryDataV4
+	// BinaryDataV5 - LID blocks have firstLID/lastLID encoded in ext1, isContinued is not used, no legacy TID adjusting
+	BinaryDataV5
 )
 
-const CurrentFracVersion = BinaryDataV3
+const CurrentFracVersion = BinaryDataV5

config/validation.go

@@ -3,6 +3,8 @@ package config
 import (
 	"cmp"
 	"fmt"
+
+	"github.com/alecthomas/units"
 )
 
 type validateFn func() error
@@ -68,6 +70,8 @@ func (c *Config) storeValidations() []validateFn {
 
 		inRange("compression.sealed_zstd_compression_level", -7, 22, c.Compression.SealedZstdCompressionLevel),
 		inRange("compression.doc_block_zstd_compression_level", -7, 22, c.Compression.DocBlockZstdCompressionLevel),
+		greaterThan("compression.lid_block_cap", 0, c.Compression.LIDBlockSize),
+		lessOrEqThan("compression.lid_block_cap", int(64*units.KiB), int(c.Compression.LIDBlockSize)),
 		inRange("offloading.queue_size_percent", 0, 100, c.Offloading.QueueSizePercent),
 
 		greaterThan("experimental.max_regex_tokens_check", -1, c.Experimental.MaxRegexTokensCheck),
@@ -106,6 +110,18 @@ func greaterThan[T cmp.Ordered](field string, base, v T) validateFn {
 	}
 }
 
+func lessOrEqThan[T cmp.Ordered](field string, base, v T) validateFn {
+	return func() error {
+		if v > base {
+			return fmt.Errorf(
+				"field %q must be greater than %v",
+				field, base,
+			)
+		}
+		return nil
+	}
+}
+
 func inRange[T cmp.Ordered](field string, from, to, v T) validateFn {
 	return func() error {
 		if v < from || to < v {

config/validation_test.go

@@ -86,6 +86,18 @@ limits:
 			env:       map[string]string{"SEQDB_OFFLOADING_QUEUE_SIZE_PERCENT": "50"},
 			expectErr: false,
 		},
+		{
+			name:      "Invalid compression.lid_block_size",
+			cfg:       baseCfg,
+			env:       map[string]string{"SEQDB_COMPRESSION_LID_BLOCK_SIZE": "-1KiB"},
+			expectErr: true,
+		},
+		{
+			name:      "Valid compression.lid_block_size",
+			cfg:       baseCfg,
+			env:       map[string]string{"SEQDB_COMPRESSION_LID_BLOCK_SIZE": "8KiB"},
+			expectErr: false,
+		},
 	}
 
 	for _, tt := range tests {

consts/consts.go

@@ -11,9 +11,9 @@ const (
 	// DummyMID is used in aggregations when we do not need to build time series.
 	DummyMID = 0
 
-	IDsPerBlock      = int(4 * units.KiB)
-	LIDBlockCap      = int(64 * units.KiB)
-	RegularBlockSize = int(16 * units.KiB)
+	IDsPerBlock        = int(4 * units.KiB)
+	DefaultLIDBlockCap = int(64 * units.KiB)
+	RegularBlockSize   = int(16 * units.KiB)
 
 	DefaultMaintenanceDelay  = time.Second
 	DefaultCacheGCDelay      = 1 * time.Second

docs/en/02-configuration.md

@@ -104,12 +104,13 @@ Resource allocation settings.
 
 Compression level settings for various data types.
 
-| Field | Type | Default | Description |
-|-------|------|---------|-------------|
-| `compression.docs_zstd_compression_level` | int | `1` | Zstandard compression level for documents |
-| `compression.metas_zstd_compression_level` | int | `1` | Zstandard compression level for metadata |
-| `compression.sealed_zstd_compression_level` | int | `3` | Zstandard compression level for sealed fractions |
-| `compression.doc_block_zstd_compression_level` | int | `3` | Zstandard compression level for document blocks |
+| Field                                          | Type | Default | Description                                      |
+|------------------------------------------------|------|---------|--------------------------------------------------|
+| `compression.docs_zstd_compression_level`      | int | `1`     | Zstandard compression level for documents        |
+| `compression.metas_zstd_compression_level`     | int | `1`     | Zstandard compression level for metadata         |
+| `compression.sealed_zstd_compression_level`    | int | `3`     | Zstandard compression level for sealed fractions |
+| `compression.doc_block_zstd_compression_level` | int | `3`     | Zstandard compression level for document blocks  |
+| `compression.lid_block_size`                   | int | `64KiB` | Max lids (postings) saved per LIDs block         |
 
 ## Indexing Configuration
 

docs/ru/02-configuration.md

@@ -104,12 +104,13 @@ id: configuration
 
 Настройки уровня сжатия для различных типов данных.
 
-| Параметр | Тип | Значение по умолчанию | Описание |
-|----------|-----|----------------------|-----------|
-| `compression.docs_zstd_compression_level` | int | `1` | Уровень сжатия для документов |
-| `compression.metas_zstd_compression_level` | int | `1` | Уровень сжатия для метаданных |
-| `compression.sealed_zstd_compression_level` | int | `3` | Уровень сжатия для запечатанных фракций |
-| `compression.doc_block_zstd_compression_level` | int | `3` | Уровень сжатия для блоков документов |
+| Параметр                                       | Тип | Значение по умолчанию | Описание                                |
+|------------------------------------------------|-----|-----------------------|-----------------------------------------|
+| `compression.docs_zstd_compression_level`      | int | `1`                   | Уровень сжатия для документов           |
+| `compression.metas_zstd_compression_level`     | int | `1`                   | Уровень сжатия для метаданных           |
+| `compression.sealed_zstd_compression_level`    | int | `3`                   | Уровень сжатия для запечатанных фракций |
+| `compression.doc_block_zstd_compression_level` | int | `3`                   | Уровень сжатия для блоков документов    |
+| `compression.lid_block_size`                   | int | `64KiB`               | Максимальное количество лидов в блоках  |
 
 ## Конфигурация индексирования
 

frac/active.go

@@ -102,6 +102,9 @@ func NewActive(
 
 		skipMaskProvider: skipMaskProvider,
 	}
+	if cfg.LIDBlockSize > 0 {
+		f.info.ConstLIDBlockCap = cfg.LIDBlockSize
+	}
 
 	// use of 0 as keys in maps is prohibited – it's system key, so add first element
 	f.MIDs.Append(uint64(seq.SystemMID))

frac/common/info.go

@@ -52,7 +52,7 @@ func NewInfo(filename string, docsOnDisk, metaOnDisk uint64) *Info {
 		CreationTime:          uint64(time.Now().UnixMilli()),
 		ConstIDsPerBlock:      consts.IDsPerBlock,
 		ConstRegularBlockSize: consts.RegularBlockSize,
-		ConstLIDBlockCap:      consts.LIDBlockCap,
+		ConstLIDBlockCap:      consts.DefaultLIDBlockCap,
 		DocsOnDisk:            docsOnDisk,
 		MetaOnDisk:            metaOnDisk,
 	}

frac/common/seal_params.go

@@ -8,5 +8,6 @@ type SealParams struct {
 	TokenTableZstdLevel    int
 
 	DocBlocksZstdLevel int // DocBlocksZstdLevel is the zstd compress level of each document block.
+	LIDBlockSize       int
 	DocBlockSize       int // DocBlockSize is decompressed payload size of document block.
 }

frac/config.go

@@ -5,6 +5,7 @@ type Config struct {
 
 	SkipSortDocs bool
 	KeepMetaFile bool
+	LIDBlockSize int
 }
 
 type SearchConfig struct {

frac/fraction_concurrency_test.go

@@ -341,6 +341,7 @@ func seal(active *Active) (*Sealed, error) {
 		TokenTableZstdLevel:    1,
 		DocBlocksZstdLevel:     1,
 		DocBlockSize:           128 * int(units.KiB),
+		LIDBlockSize:           512,
 	}
 	activeSealingSource, err := NewActiveSealingSource(active, sealParams)
 	if err != nil {

frac/fraction_test.go

@@ -94,6 +94,7 @@ func (s *FractionTestSuite) SetupTestCommon() {
 		DocsPositionsZstdLevel: 1,
 		TokenTableZstdLevel:    1,
 		DocBlocksZstdLevel:     1,
+		LIDBlockSize:           256,
 		DocBlockSize:           128 * int(units.KiB),
 	}
 
@@ -1322,6 +1323,43 @@ func (s *FractionTestSuite) TestSearchLargeFrac() {
 			fromTime: fromTime,
 			toTime:   toTime,
 		},
+		// block skipping scenarios
+		{
+			name:  "service:gateway AND trace_id:trace-2026",
+			query: "service:gateway AND trace_id:trace-2026",
+			filter: func(doc *testDoc) bool {
+				return doc.service == gateway && doc.traceId == "trace-2026"
+			},
+			fromTime: fromTime,
+			toTime:   toTime,
+		},
+		{
+			name:  "service:gateway AND (trace_id:trace-0 OR trace_id:trace-2500 OR trace_id:trace-4999)",
+			query: "service:gateway AND (trace_id:trace-0 OR trace_id:trace-2500 OR trace_id:trace-4999)",
+			filter: func(doc *testDoc) bool {
+				return doc.service == gateway && (doc.traceId == "trace-0" || doc.traceId == "trace-2500" || doc.traceId == "trace-4999")
+			},
+			fromTime: fromTime,
+			toTime:   toTime,
+		},
+		{
+			name:  "service:gateway AND pod:pod-5",
+			query: "service:gateway AND pod:pod-5",
+			filter: func(doc *testDoc) bool {
+				return doc.service == gateway && doc.pod == "pod-5"
+			},
+			fromTime: fromTime,
+			toTime:   toTime,
+		},
+		{
+			name:  "service:gateway AND pod:pod-5 AND message:failed",
+			query: "service:gateway AND pod:pod-5 AND message:failed",
+			filter: func(doc *testDoc) bool {
+				return doc.service == gateway && doc.pod == "pod-5" && strings.Contains(doc.message, "failed")
+			},
+			fromTime: fromTime,
+			toTime:   toTime,
+		},
 		{
 			name:  "service:gateway AND message:processing AND message:retry AND level:5",
 			query: "service:gateway AND message:processing AND message:retry AND level:5",
@@ -1333,6 +1371,17 @@ func (s *FractionTestSuite) TestSearchLargeFrac() {
 			toTime:   toTime,
 		},
 		// OR operator queries
+		{
+			name:  "(service OR) AND (trace_id OR)",
+			query: "(service:bus OR service:kafka) AND (trace_id:trace-1000 OR trace_id:trace-1500 OR trace_id:trace-2000)",
+			filter: func(doc *testDoc) bool {
+				return (doc.service == bus || doc.service == kafka) && (doc.traceId == "trace-1000" ||
+					doc.traceId == "trace-1500" ||
+					doc.traceId == "trace-2000")
+			},
+			fromTime: fromTime,
+			toTime:   toTime,
+		},
 		{
 			name:  "trace_id OR",
 			query: "trace_id:trace-1000 OR trace_id:trace-1500 OR trace_id:trace-2000 OR trace_id:trace-2500 OR trace_id:trace-3000",

frac/sealed/lids/iterator_asc.go

@@ -81,13 +81,14 @@ func (it *IteratorAsc) NextGeq(nextID node.LID) node.LID {
 				return node.NullLID()
 			}
 
+			it.blockIndex = it.table.SeekBlockLeq(it.blockIndex, it.tid, nextID.Unpack())
+
 			it.loadNextLIDsBlock()
 			it.lids, it.tryNextBlock = it.narrowLIDsRange(it.lids, it.tryNextBlock)
 			it.counter.AddLIDsCount(len(it.lids))
 		}
 
 		// fast path: smallest remaining > nextID => skip entire block
-		// TODO(cheb0): We could also pass LID into narrowLIDsRange to perform block skipping once we add something like MinLID to LID block header
 		if it.lids[0] > nextID.Unpack() {
 			it.lids = it.lids[:0]
 			continue

frac/sealed/lids/iterator_desc.go

@@ -80,13 +80,14 @@ func (it *IteratorDesc) NextGeq(nextID node.LID) node.LID {
 				return node.NullLID()
 			}
 
+			it.blockIndex = it.table.SeekBlockGeq(it.blockIndex, it.tid, nextID.Unpack())
+
 			it.loadNextLIDsBlock() // last chunk in block but not last for tid; need load next block
 			it.lids, it.tryNextBlock = it.narrowLIDsRange(it.lids, it.tryNextBlock)
 			it.counter.AddLIDsCount(len(it.lids)) // inc loaded LIDs count
 		}
 
 		// fast path: last LID < nextID => skip the entire block
-		// TODO(cheb0): We could also pass LID into narrowLIDsRange to perform block skipping once we add something like MinLID to LID block header
 		if nextID.Unpack() > it.lids[len(it.lids)-1] {
 			it.lids = it.lids[:0]
 			continue

frac/sealed/lids/table.go

@@ -5,31 +5,44 @@ import (
 
 	"go.uber.org/zap"
 
+	"github.com/ozontech/seq-db/config"
+
 	"github.com/ozontech/seq-db/logger"
 )
 
 type Table struct {
 	StartBlockIndex uint32
 	MaxTIDs         []uint32 // defines last tid for each block
 	MinTIDs         []uint32 // defines first not continued tid for each block
+	FirstLIDs       []uint32
+	LastLIDs        []uint32
 
-	// TODO: We need fix MinTID issue that we have to compensate with DiskBlock.getAdjustedMinTID()
-	// TODO: After that we do not need store IsContinued flag, and able calc it as MaxTIDs[i] == MinTIDs[i+1]
-	IsContinued []bool
+	FracVer     config.BinaryDataVersion
+	IsContinued []bool // legacy field, only used in BinaryDataV0-BinaryDataV3 (inclusive)
 }
 
-func NewTable(startOfLIDsBlockIndex uint32, minTIDs, maxTIDs []uint32, isContinued []bool) *Table {
+func NewTable(
+	fracVer config.BinaryDataVersion,
+	startOfLIDsBlockIndex uint32,
+	minTIDs, maxTIDs []uint32,
+	firstLIDs, lastLIDs []uint32,
+	isContinued []bool) *Table {
 	return &Table{
 		StartBlockIndex: startOfLIDsBlockIndex,
 		MinTIDs:         minTIDs,
 		MaxTIDs:         maxTIDs,
+		FirstLIDs:       firstLIDs,
+		LastLIDs:        lastLIDs,
 		IsContinued:     isContinued,
+		FracVer:         fracVer,
 	}
 }
 
 func (t *Table) GetAdjustedMinTID(blockIndex uint32) uint32 {
-	if t.IsContinued[blockIndex] {
-		return t.MinTIDs[blockIndex] - 1
+	if t.FracVer < config.BinaryDataV5 {
+		if t.IsContinued[blockIndex] {
+			return t.MinTIDs[blockIndex] - 1
+		}
 	}
 	return t.MinTIDs[blockIndex]
 }
@@ -75,6 +88,46 @@ func (t *Table) GetLastBlockIndexForTID(tid uint32) uint32 {
 	return uint32(index)
 }
 
+// SeekBlockGeq finds next block for provided TID which contains
+// lid greater or equal to provided LID starting from provided index (inclusive).
+// - index: an index of block which is already suits and contains next portion of LIDs. Safe to return for old fractions.
+func (t *Table) SeekBlockGeq(index uint32, tid uint32, nextLID uint32) uint32 {
+	if t.FracVer < config.BinaryDataV5 {
+		// not supported for old frac versions
+		return index
+	}
+
+	res := index
+	for i := index + 1; i < uint32(len(t.MinTIDs)); i++ {
+		if t.MinTIDs[i] == tid && nextLID >= t.FirstLIDs[i] {
+			res = i
+			continue
+		}
+		break
+	}
+	return res
+}
+
+// SeekBlockLeq finds next block with lowest index for provided TID which contains LIDs
+// less or equal to provided LID starting from provided index (inclusive).
+// - index: an index of block which is already suits and contains next portion of LIDs. Safe to return for old fractions.
+func (t *Table) SeekBlockLeq(index uint32, tid uint32, nextLID uint32) uint32 {
+	if t.FracVer < config.BinaryDataV5 {
+		// not supported for old frac versions
+		return index
+	}
+
+	res := index
+	for i := int(index) - 1; i >= 0; i-- {
+		if t.MaxTIDs[i] == tid && nextLID <= t.LastLIDs[i] {
+			res = uint32(i)
+			continue
+		}
+		break
+	}
+	return res
+}
+
 func (t *Table) HasTIDInPrevBlock(blockIndex, tid uint32) bool {
 	if blockIndex == 0 { // it is no prev block
 		return false