diff --git a/config/frac_version.go b/config/frac_version.go index 73c3261a..6d83123e 100644 --- a/config/frac_version.go +++ b/config/frac_version.go @@ -21,6 +21,10 @@ const ( // BinaryDataV4 - delta bitpack encoded MIDs and LIDs BinaryDataV4 + + // BinaryDataV5 - LID blocks have firstLID/lastLID encoded in ext1, + // isContinued is not used, no legacy TID adjusting + BinaryDataV5 ) -const CurrentFracVersion = BinaryDataV4 +const CurrentFracVersion = BinaryDataV5 diff --git a/frac/fraction_test.go b/frac/fraction_test.go index 9a4f0b4c..7b809dc1 100644 --- a/frac/fraction_test.go +++ b/frac/fraction_test.go @@ -102,7 +102,7 @@ func (s *FractionTestSuite) SetupTestCommon() { DocsPositionsZstdLevel: 1, TokenTableZstdLevel: 1, DocBlocksZstdLevel: 1, - LIDBlockSize: 512, + LIDBlockSize: 256, DocBlockSize: 128 * int(units.KiB), } @@ -1366,6 +1366,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", @@ -1377,6 +1414,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", diff --git a/frac/sealed/lids/iterator_asc.go b/frac/sealed/lids/iterator_asc.go index 5f7a5f03..3f1adf05 100644 --- a/frac/sealed/lids/iterator_asc.go +++ b/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 @@ -114,13 +115,15 @@ func (it *IteratorAsc) NextBatchGeq(nextID node.LID) node.LIDBatch { if !it.tryNextBlock { return node.NewAscBatch(nil) } + + 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 diff --git a/frac/sealed/lids/iterator_desc.go b/frac/sealed/lids/iterator_desc.go index cb2551f4..76030494 100644 --- a/frac/sealed/lids/iterator_desc.go +++ b/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 @@ -114,6 +115,9 @@ func (it *IteratorDesc) NextBatchGeq(nextID node.LID) node.LIDBatch { if !it.tryNextBlock { return node.NewDescBatch(nil) } + + it.blockIndex = it.table.SeekBlockGeq(it.blockIndex, it.tid, nextID.Unpack()) + it.loadNextLIDsBlock() it.lids, it.tryNextBlock = it.narrowLIDsRange(it.lids, it.tryNextBlock) it.counter.AddLIDsCount(len(it.lids)) diff --git a/frac/sealed/lids/table.go b/frac/sealed/lids/table.go index 5a04acdd..e40d5610 100644 --- a/frac/sealed/lids/table.go +++ b/frac/sealed/lids/table.go @@ -5,6 +5,7 @@ import ( "go.uber.org/zap" + "github.com/ozontech/seq-db/config" "github.com/ozontech/seq-db/logger" ) @@ -12,24 +13,36 @@ 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-BinaryDataV4 (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, tid, 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, tid, 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 diff --git a/frac/sealed_loader.go b/frac/sealed_loader.go index 10d95a2d..9b6a694d 100644 --- a/frac/sealed_loader.go +++ b/frac/sealed_loader.go @@ -41,7 +41,7 @@ func (l *LegacyLoader) Load(blocksData *sealed.BlocksData, info *common.Info, re logger.Fatal("legacy load ids error", zap.Error(err)) } - blocksData.LIDsTable, err = l.loadLIDsTable() + blocksData.LIDsTable, err = l.loadLIDsTable(info.BinaryDataVer) if err != nil { logger.Fatal("legacy load lids error", zap.Error(err)) } @@ -126,12 +126,14 @@ func (l *LegacyLoader) loadIDs(info *common.Info) (seqids.Table, []uint64, error } // loadLIDsTable scans LID block headers, recording the absolute start index for lids.Table. -func (l *LegacyLoader) loadLIDsTable() (*lids.Table, error) { +func (l *LegacyLoader) loadLIDsTable(fracVer config.BinaryDataVersion) (*lids.Table, error) { startIndex := l.blockIndex // absolute index of first LID block in .index var ( maxTIDs []uint32 minTIDs []uint32 + firstLIDs []uint32 + lastLIDs []uint32 isContinued []bool ) @@ -149,10 +151,15 @@ func (l *LegacyLoader) loadLIDsTable() (*lids.Table, error) { maxTIDs = append(maxTIDs, uint32(h.GetExt2()>>32)) minTIDs = append(minTIDs, uint32(h.GetExt2()&0xFFFFFFFF)) - isContinued = append(isContinued, h.GetExt1() == 1) + if fracVer >= config.BinaryDataV5 { + lastLIDs = append(lastLIDs, uint32(h.GetExt1()>>32)) + firstLIDs = append(firstLIDs, uint32(h.GetExt1()&0xFFFFFFFF)) + } else { + isContinued = append(isContinued, h.GetExt1() == 1) + } } - return lids.NewTable(startIndex, minTIDs, maxTIDs, isContinued), nil + return lids.NewTable(fracVer, startIndex, minTIDs, maxTIDs, firstLIDs, lastLIDs, isContinued), nil } // IndexReaders holds one IndexReader per split index file. @@ -186,7 +193,7 @@ func (l *Loader) Load(blocksData *sealed.BlocksData, info *common.Info, readers blocksData.BlocksOffsets = blockOffsets.Offsets blocksData.IDsTable = l.loadIDsTable(readers.ID, info) - blocksData.LIDsTable, err = l.loadLIDsTable(readers.LID) + blocksData.LIDsTable, err = l.loadLIDsTable(readers.LID, info.BinaryDataVer) if err != nil { logger.Fatal("load lids error", zap.Error(err)) } @@ -262,10 +269,12 @@ func (l *Loader) loadIDsTable(r storage.IndexReader, info *common.Info) seqids.T } // loadLIDsTable scans block headers in the .lid file to build lids.Table. -func (l *Loader) loadLIDsTable(r storage.IndexReader) (*lids.Table, error) { +func (l *Loader) loadLIDsTable(r storage.IndexReader, fracVer config.BinaryDataVersion) (*lids.Table, error) { var ( maxTIDs []uint32 minTIDs []uint32 + firstLIDs []uint32 + lastLIDs []uint32 isContinued []bool ) @@ -287,8 +296,14 @@ func (l *Loader) loadLIDsTable(r storage.IndexReader) (*lids.Table, error) { maxTIDs = append(maxTIDs, uint32(ext2>>32)) minTIDs = append(minTIDs, uint32(ext2&0xFFFFFFFF)) - isContinued = append(isContinued, header.GetExt1() == 1) + ext1 := header.GetExt1() + if fracVer >= config.BinaryDataV5 { + lastLIDs = append(lastLIDs, uint32(ext1>>32)) + firstLIDs = append(firstLIDs, uint32(ext1&0xFFFFFFFF)) + } else { + isContinued = append(isContinued, ext1 == 1) + } } - return lids.NewTable(0, minTIDs, maxTIDs, isContinued), nil + return lids.NewTable(fracVer, 0, minTIDs, maxTIDs, firstLIDs, lastLIDs, isContinued), nil } diff --git a/indexwriter/blocks.go b/indexwriter/blocks.go index 7b2d6c0c..b8a39eee 100644 --- a/indexwriter/blocks.go +++ b/indexwriter/blocks.go @@ -25,11 +25,12 @@ type unpackedTokenBlock struct { payload token.Block // Actual token data payload } -// lidExt represents the range and continuation status of LID blocks. +// lidExt represents the range of LID blocks. type lidExt struct { - minTID uint32 // First token ID in the LID block - maxTID uint32 // Last token ID in the LID block - isContinued bool // Whether LID sequence continues in next block + minTID uint32 // First token ID in the LID block + maxTID uint32 // Last token ID in the LID block + firstLID uint32 // First LID value in the block + lastLID uint32 // Last LID value in the block } // unpackedLIDBlock represents a sealed block containing LID (Local ID) data. diff --git a/indexwriter/blocks_test.go b/indexwriter/blocks_test.go index 546d08a7..6062d6b9 100644 --- a/indexwriter/blocks_test.go +++ b/indexwriter/blocks_test.go @@ -240,27 +240,27 @@ func TestBlocksBuilder_BuildTokenBlocks(t *testing.T) { expectedLIDBlocks := []unpackedLIDBlock{ { - ext: lidExt{minTID: 1, maxTID: 1, isContinued: false}, + ext: lidExt{minTID: 1, maxTID: 1, firstLID: 10, lastLID: 30}, payload: lids.Block{LIDs: []uint32{10, 20, 30}, Offsets: []uint32{0, 3}, IsLastLID: false}, }, { - ext: lidExt{minTID: 1, maxTID: 3, isContinued: true}, + ext: lidExt{minTID: 1, maxTID: 3, firstLID: 40, lastLID: 3}, payload: lids.Block{LIDs: []uint32{40, 2, 3}, Offsets: []uint32{0, 1, 2, 3}, IsLastLID: true}, }, { - ext: lidExt{minTID: 4, maxTID: 6, isContinued: false}, + ext: lidExt{minTID: 4, maxTID: 6, firstLID: 4, lastLID: 6}, payload: lids.Block{LIDs: []uint32{4, 5, 6}, Offsets: []uint32{0, 1, 2, 3}, IsLastLID: true}, }, { - ext: lidExt{minTID: 7, maxTID: 9, isContinued: false}, + ext: lidExt{minTID: 7, maxTID: 9, firstLID: 7, lastLID: 9}, payload: lids.Block{LIDs: []uint32{7, 8, 9}, Offsets: []uint32{0, 1, 2, 3}, IsLastLID: true}, }, { - ext: lidExt{minTID: 10, maxTID: 12, isContinued: false}, + ext: lidExt{minTID: 10, maxTID: 12, firstLID: 10, lastLID: 12}, payload: lids.Block{LIDs: []uint32{10, 11, 12}, Offsets: []uint32{0, 1, 2, 3}, IsLastLID: true}, }, { - ext: lidExt{minTID: 13, maxTID: 14, isContinued: false}, + ext: lidExt{minTID: 13, maxTID: 14, firstLID: 13, lastLID: 14}, payload: lids.Block{LIDs: []uint32{13, 14}, Offsets: []uint32{0, 1, 2}, IsLastLID: true}, }, } diff --git a/indexwriter/index.go b/indexwriter/index.go index cdf6d1c5..ef0d74d4 100644 --- a/indexwriter/index.go +++ b/indexwriter/index.go @@ -4,6 +4,7 @@ import ( "io" "iter" + "github.com/ozontech/seq-db/config" "github.com/ozontech/seq-db/consts" "github.com/ozontech/seq-db/frac/common" "github.com/ozontech/seq-db/frac/sealed" @@ -290,21 +291,17 @@ func (s *IndexWriter) packPosBlock(block unpackedIDBlock) indexBlock { // packLIDsBlock packs Local IDs (LIDs) into a compressed index block. // Also updates LIDs table for preloaded data access. func (s *IndexWriter) packLIDsBlock(block unpackedLIDBlock) indexBlock { - var ext1 uint64 - if block.ext.isContinued { // todo: Legacy continuation flag - ext1 = 1 - block.ext.minTID++ // Adjust for legacy format - } - // Update LIDs table for PreloadedData + s.lidsTable.FracVer = config.CurrentFracVersion + s.lidsTable.FirstLIDs = append(s.lidsTable.FirstLIDs, block.ext.firstLID) + s.lidsTable.LastLIDs = append(s.lidsTable.LastLIDs, block.ext.lastLID) s.lidsTable.MinTIDs = append(s.lidsTable.MinTIDs, block.ext.minTID) s.lidsTable.MaxTIDs = append(s.lidsTable.MaxTIDs, block.ext.maxTID) - s.lidsTable.IsContinued = append(s.lidsTable.IsContinued, block.ext.isContinued) // Packing block s.buf1 = block.payload.Pack(s.buf1[:0], s.buf32[:0]) b := s.newIndexBlockZSTD(s.buf1, s.params.LIDsZstdLevel) - b.ext1 = ext1 // Legacy continuation flag + b.ext1 = uint64(block.ext.lastLID)<<32 | uint64(block.ext.firstLID) b.ext2 = uint64(block.ext.maxTID)<<32 | uint64(block.ext.minTID) // TID range return b diff --git a/indexwriter/lid_accumulator.go b/indexwriter/lid_accumulator.go index d2b1b97a..81e798cc 100644 --- a/indexwriter/lid_accumulator.go +++ b/indexwriter/lid_accumulator.go @@ -13,7 +13,6 @@ type lidAccumulator struct { currentBlock unpackedLIDBlock isEndOfToken bool - isContinued bool } func newLIDAccumulator( @@ -84,9 +83,11 @@ func (a *lidAccumulator) finalizeBlock() unpackedLIDBlock { } result := a.currentBlock + if blockLIDs := result.payload.LIDs; len(blockLIDs) > 0 { + result.ext.firstLID = blockLIDs[0] + result.ext.lastLID = blockLIDs[len(blockLIDs)-1] + } result.payload.IsLastLID = a.isEndOfToken - result.ext.isContinued = a.isContinued - a.isContinued = !a.isEndOfToken return result }