diff --git a/core/invertedindex/batch_close_test.go b/core/invertedindex/batch_close_test.go index a2885d8..57a2285 100644 --- a/core/invertedindex/batch_close_test.go +++ b/core/invertedindex/batch_close_test.go @@ -69,9 +69,7 @@ func TestBatchSitesCloseCommittedBatch(t *testing.T) { // mutated newBatch seam would bypass recStore.NewBatch. DeleteTable's direct // NewBatch(0) is captured regardless. newBatch = func(db kv.Store) kv.Batch { return db.NewBatch(MaxBatchSize) } - writeInvertedIndex = func(batch kv.Batch, tableId int, kw string, docids []int64, key []byte) { - batch.Put(key, encodeInvertedValue(docids)) - } + writeInvertedIndex = defaultWriteInvertedIndex // Disable the periodic ticker (FlushTicker = 1h) so the only batches created // are the ones this test drives explicitly — single-goroutine and race-free. diff --git a/core/invertedindex/codec.go b/core/invertedindex/codec.go index f67d56d..1407703 100644 --- a/core/invertedindex/codec.go +++ b/core/invertedindex/codec.go @@ -140,21 +140,31 @@ func encodeTableValue(info TableInfo) []byte { return content } -// encodeInvertedValue encodes a posting row's docids as a delta-varint sequence: -// the docids are sorted (by their unsigned 64-bit bit pattern) and written as the -// first id followed by successive gaps, each a base-128 uvarint. Real docids are -// small, densely-allocated idtable ids, so the gaps are tiny and most encode to a -// single byte — far smaller than the previous fixed 8-byte-per-id layout, and -// cheaper to decode than a general-purpose block compressor. Order within a row -// is irrelevant (the docids are a set), so sorting here is free; the encoder sorts -// AND dedups (adjacent-equal removal after the sort), so callers may pass a slice -// containing duplicates. Sorting/subtracting in uint64 space (not int64) keeps -// every gap non-negative and makes the round-trip exact for any int64 docid, +// encodeInvertedValue is encodeInvertedValueCounted without the count (see there +// for the encoding). +func encodeInvertedValue(docids []int64) []byte { + b, _ := encodeInvertedValueCounted(docids) + return b +} + +// encodeInvertedValueCounted encodes a posting row's docids as a delta-varint +// sequence: the docids are sorted (by their unsigned 64-bit bit pattern) and +// written as the first id followed by successive gaps, each a base-128 uvarint. +// Real docids are small, densely-allocated idtable ids, so the gaps are tiny and +// most encode to a single byte — far smaller than the previous fixed 8-byte-per-id +// layout, and cheaper to decode than a general-purpose block compressor. Order +// within a row is irrelevant (the docids are a set), so sorting here is free; the +// encoder sorts AND dedups (adjacent-equal removal after the sort), so callers may +// pass a slice containing duplicates. It also RETURNS the number of unique docids +// written (== the varint count in the returned bytes), so callers stamp the key's +// doccount with the true post-dedup count in the same single pass — no separate +// dedup pass on the build hot path. Sorting/subtracting in uint64 space (not int64) +// keeps every gap non-negative and makes the round-trip exact for any int64 docid, // including negative and max values. This is an on-disk format change from the old // big-endian layout and requires a reindex. -func encodeInvertedValue(docids []int64) []byte { +func encodeInvertedValueCounted(docids []int64) ([]byte, int) { if len(docids) == 0 { - return []byte{} + return []byte{}, 0 } us := make([]uint64, len(docids)) for i, id := range docids { @@ -173,7 +183,7 @@ func encodeInvertedValue(docids []int64) []byte { buf = binary.AppendUvarint(buf, delta) prev = u } - return buf + return buf, len(us) // len(us) is post-Compact == the varint count } // decodeInvertedValue is the inverse of encodeInvertedValue: it reads successive diff --git a/core/invertedindex/doccount_test.go b/core/invertedindex/doccount_test.go new file mode 100644 index 0000000..2a002b0 --- /dev/null +++ b/core/invertedindex/doccount_test.go @@ -0,0 +1,100 @@ +package invertedindex + +import ( + "maps" + "slices" + "testing" + + "github.com/stretchr/testify/assert" +) + +// countRowsIdx counts stored rows whose decoded (tableId, keyword) == (table, kw), +// scanning the row keyspace directly off the store. Mirrors countRows in +// key_robustness_test.go but works against a raw *Index (no testEnv wrapper). +func countRowsIdx(idx *Index, table int, kw string) int { + n := 0 + _ = idx.db.Scan([]byte{idx.keyTypeRow}, func(k, _ []byte) bool { + if tid, gotKw, _, _ := idx.decodeInvertedKey(string(k)); tid == table && gotKw == kw { + n++ + } + return true + }) + return n +} + +// T1 — the flush path must stamp each row's on-disk doccount with the DEDUPED +// unique docid count, not the pre-dedup buffer length. updateIndex appends one +// docid per keyword occurrence without dedup, so indexing {"hot","hot","hot"} for +// a single document buffers three docids; the flushed value dedups to one, but the +// buggy flush stamps doccount=3. Every row's doccount must equal the number of +// docids actually stored in its value. +func TestFlush_StampsDedupedDoccount(t *testing.T) { + idx, cleanup := newBoundedIndex(t, Options{} /* unbounded, ticker disabled */) + defer cleanup() + + const table = 1 + idx.updateIndex(table, makeDocID("d0"), []string{"hot", "hot", "hot"}) + forceFlush(idx) + + visited := 0 + _ = idx.db.Scan([]byte{idx.keyTypeRow}, func(k, value []byte) bool { + tid, kw, doccount, _ := idx.decodeInvertedKey(string(k)) + if tid != table || kw != "hot" { + return true + } + visited++ + unique := len(decodeInvertedValue(value)) + if doccount != unique { + t.Errorf("row %q: stamped doccount=%d, but value holds %d unique docids", k, doccount, unique) + } + return true + }) + if visited < 1 { + t.Fatalf("expected at least 1 stored \"hot\" row, visited %d", visited) + } +} + +// T2 — a hot keyword whose per-flush rows each carry an inflated doccount +// (> maxInvertedIndexSize/2) gets quarantined by the merger and never compacted, +// so its row count never shrinks. With the deduped count the rows are well under +// the threshold and the merger folds them into fewer rows while preserving the +// retrievable docids. +func TestMerge_CompactsHotKeywordRows(t *testing.T) { + idx, cleanup := newBoundedIndex(t, Options{} /* unbounded, ticker disabled */) + defer cleanup() + + const table = 1 + const size = 10 // doccount>size/2==5 quarantines; buggy flush stamps 6 per row + + docs := []int64{makeDocID("d0"), makeDocID("d1"), makeDocID("d2")} + for _, doc := range docs { + idx.updateIndex(table, doc, []string{"hot", "hot", "hot", "hot", "hot", "hot"}) + forceFlush(idx) + } + + before := countRowsIdx(idx, table, "hot") + if before < 2 { + t.Fatalf("expected >=2 separate \"hot\" rows before merge, got %d", before) + } + // DocIds is a set (map); capture its members as a sorted slice so the exact + // retrievable docid SET — not just its size — can be compared across the merge. + idsBefore := slices.Sorted(maps.Keys(idx.GetDocs(table, "hot").DocIds)) + docsBefore := len(idsBefore) + + m := merging{NextIter: string(idx.keyTypeRow)} + for i := 0; i < 5; i++ { + m = idx.mergeKeywordsIndex(m, size) + } + + after := countRowsIdx(idx, table, "hot") + if after >= before { + t.Errorf("merger did not compact hot rows: before=%d after=%d (quarantined by inflated doccount)", before, after) + } + idsAfter := slices.Sorted(maps.Keys(idx.GetDocs(table, "hot").DocIds)) + if got := len(idsAfter); got != docsBefore { + t.Errorf("docid set changed across merge: before=%d after=%d", docsBefore, got) + } + // The merge must be docid-preserving: not just the count, the exact SET of + // retrievable docids must be identical before vs after. + assert.Equal(t, idsBefore, idsAfter, "docid set changed across merge") +} diff --git a/core/invertedindex/index_test.go b/core/invertedindex/index_test.go index 92ce28b..35d09d6 100644 --- a/core/invertedindex/index_test.go +++ b/core/invertedindex/index_test.go @@ -556,8 +556,7 @@ func TestWriteInvertedIndexDeduplicates(t *testing.T) { // Directly call writeInvertedIndex with duplicates batch := env.DB.NewBatch(0) - key := env.idx.encodeInvertedKey(tableId, "dupkw", 3, 0) - writeInvertedIndex(batch, tableId, "dupkw", []int64{doc, doc, doc}, key) + writeInvertedIndex(env.idx, batch, tableId, "dupkw", []int64{doc, doc, doc}, int64(0)) batch.Commit() res := env.idx.GetDocs(tableId, "dupkw") diff --git a/core/invertedindex/invertedindex_internal.go b/core/invertedindex/invertedindex_internal.go index 97c837a..627ec5e 100644 --- a/core/invertedindex/invertedindex_internal.go +++ b/core/invertedindex/invertedindex_internal.go @@ -44,16 +44,21 @@ func (idx *Index) removeIndex(tableId int, docid int64, keywords []string) { idx.maybeFlushOnPressure() } -// writeInvertedIndex writes a keyword to the database. -// Callers must pass the pre-computed key via idx.encodeInvertedKey so that the -// configured key-type bytes are honoured. -var writeInvertedIndex = func(batch kv.Batch, tableId int, kw string, docids []int64, key []byte) { - // encodeInvertedValue sorts AND dedups, so the raw docids (which the flush - // path intentionally leaves with duplicates) can be handed straight to it. - content := encodeInvertedValue(docids) - batch.Put(key, content) +// defaultWriteInvertedIndex is the production seam body: it encodes the docids +// (sort+dedup) and builds the on-disk key from the SAME post-dedup unique count, +// so no caller can stamp a doccount that disagrees with the value it writes (the +// merger quarantines rows whose stamped doccount exceeds maxSize/2, so an inflated +// count keeps a hot keyword's rows from ever compacting). It is a NAMED func so +// go-cov gates it per-function and the reset-helpers can point straight at it +// instead of hand-copying the body (which would let a flush test drive a stale +// double rather than production). +func defaultWriteInvertedIndex(idx *Index, batch kv.Batch, tableId int, kw string, docids []int64, tick int64) { + content, unique := encodeInvertedValueCounted(docids) + batch.Put(idx.encodeInvertedKey(tableId, kw, unique, tick), content) } +var writeInvertedIndex = defaultWriteInvertedIndex + // removeDocumentsFromInvertedIndex removes a document from the keywords index. // It will remove the document from the keywords index and rewrite the keyword with new docids. func (idx *Index) removeDocumentsFromInvertedIndex(batch kv.Batch, tableId int, kw string, removingDocids []int64, @@ -125,11 +130,10 @@ func (idx *Index) removeDocumentsFromInvertedIndex(batch kv.Batch, tableId int, // Always re-encode under a fresh key carrying the TRUE doccount. Reusing // an original key (keys[0]) would keep its stale, inflated doccount, which // the merger's `doccount > maxSize/2` guard then quarantines from - // compaction forever. encodeInvertedKey's seq suffix keeps the new key - // distinct from the originals, all of which are deleted below. - key := idx.encodeInvertedKey(tableId, kw, len(docs), tick) - - writeInvertedIndex(batch, tableId, kw, docs, key) + // compaction forever. writeInvertedIndex builds the key with the deduped + // count and its seq suffix keeps it distinct from the originals, all of + // which are deleted below. + writeInvertedIndex(idx, batch, tableId, kw, docs, tick) } // Delete every original row we collected; their surviving docids were diff --git a/core/invertedindex/keywords_merger.go b/core/invertedindex/keywords_merger.go index 4b31875..2f52d40 100644 --- a/core/invertedindex/keywords_merger.go +++ b/core/invertedindex/keywords_merger.go @@ -198,8 +198,7 @@ var rewriteIndex = func(batch kv.Batch, idx *Index, index *invertedIndexEntry, m ids = append(ids, id) } - key := idx.encodeInvertedKey(index.TableId, index.Keyword, len(ids), tick) - writeInvertedIndex(batch, index.TableId, index.Keyword, ids, key) + writeInvertedIndex(idx, batch, index.TableId, index.Keyword, ids, tick) mergedCount++ } diff --git a/core/invertedindex/keywords_merger_test.go b/core/invertedindex/keywords_merger_test.go index fa86422..ae667fc 100644 --- a/core/invertedindex/keywords_merger_test.go +++ b/core/invertedindex/keywords_merger_test.go @@ -50,7 +50,7 @@ func (m *mockBatchWrite) DeletePrefix(prefix []byte) error { func setupTestMocks() func() { // Override the writeKeywordIndex function for testing originalWriteKeywordIndex := writeInvertedIndex - writeInvertedIndex = func(batch kv.Batch, tableId int, keyword string, docIDs []int64, data []byte) { + writeInvertedIndex = func(idx *Index, batch kv.Batch, tableId int, keyword string, docIDs []int64, tick int64) { mockBatch := batch.(*mockBatchWrite) mockBatch.tableIds = append(mockBatch.tableIds, tableId) mockBatch.keywords = append(mockBatch.keywords, keyword) @@ -394,7 +394,7 @@ func TestMergeKeywordsIndexSingleTable(t *testing.T) { batch := newMockBatch(nil) // Mock only the writeKeywordIndex function - writeInvertedIndex = func(batch kv.Batch, tableId int, keyword string, docIDs []int64, data []byte) { + writeInvertedIndex = func(idx *Index, batch kv.Batch, tableId int, keyword string, docIDs []int64, tick int64) { writtenTables = append(writtenTables, tableId) writtenKeywords = append(writtenKeywords, keyword) writtenDocIDs = append(writtenDocIDs, docIDs) @@ -522,7 +522,7 @@ func TestMergeKeywordsIndexMultipleTables(t *testing.T) { } // Mock only the writeKeywordIndex function - writeInvertedIndex = func(batch kv.Batch, tableId int, keyword string, docIDs []int64, data []byte) { + writeInvertedIndex = func(idx *Index, batch kv.Batch, tableId int, keyword string, docIDs []int64, tick int64) { if _, ok := writtenData[tableId]; !ok { writtenData[tableId] = make(map[string][]int64) } @@ -743,7 +743,7 @@ func TestMergeKeywordsIndexTimeout(t *testing.T) { } // Mock only the writeKeywordIndex function - writeInvertedIndex = func(batch kv.Batch, tableId int, keyword string, docIDs []int64, data []byte) { + writeInvertedIndex = func(idx *Index, batch kv.Batch, tableId int, keyword string, docIDs []int64, tick int64) { // No-op for this test } @@ -902,7 +902,7 @@ func TestKeywordsMerger_RunMergeWithData(t *testing.T) { newBatch = origBatch }() - writeInvertedIndex = func(batch kv.Batch, tableId int, keyword string, docIDs []int64, data []byte) { + writeInvertedIndex = func(idx *Index, batch kv.Batch, tableId int, keyword string, docIDs []int64, tick int64) { // no-op: we don't need to persist data } newBatch = func(db kv.Store) kv.Batch { @@ -979,7 +979,7 @@ func TestKeywordsMerger_NewScanAfterComplete(t *testing.T) { newBatch = origBatch }() - writeInvertedIndex = func(batch kv.Batch, tableId int, keyword string, docIDs []int64, data []byte) {} + writeInvertedIndex = func(idx *Index, batch kv.Batch, tableId int, keyword string, docIDs []int64, tick int64) {} newBatch = func(db kv.Store) kv.Batch { return &mockBatchWriteWithFuncs{ deleteFunc: func(key []byte) error { return nil }, @@ -1039,7 +1039,7 @@ func TestMergeKeywordsIndex_WellBatchedSkip(t *testing.T) { newBatch = origBatch }() - writeInvertedIndex = func(batch kv.Batch, tableId int, keyword string, docIDs []int64, data []byte) {} + writeInvertedIndex = func(idx *Index, batch kv.Batch, tableId int, keyword string, docIDs []int64, tick int64) {} newBatch = func(db kv.Store) kv.Batch { return &mockBatchWriteWithFuncs{ deleteFunc: func(key []byte) error { return nil }, diff --git a/core/invertedindex/pending_bound_test.go b/core/invertedindex/pending_bound_test.go index 667cf4a..a28fcd6 100644 --- a/core/invertedindex/pending_bound_test.go +++ b/core/invertedindex/pending_bound_test.go @@ -32,9 +32,7 @@ func newBoundedIndex(t *testing.T, opts Options) (*Index, func()) { // Reset the package test-injection seams (mirrors setupTestEnv). newBatch = func(db kv.Store) kv.Batch { return db.NewBatch(MaxBatchSize) } - writeInvertedIndex = func(batch kv.Batch, tableId int, kw string, docids []int64, key []byte) { - batch.Put(key, encodeInvertedValue(docids)) - } + writeInvertedIndex = defaultWriteInvertedIndex opts.FlushTicker = time.Hour // disable the periodic ticker for the test idx, err := New(db, q, opts) diff --git a/core/invertedindex/pending_writes.go b/core/invertedindex/pending_writes.go index 6fb1321..527565f 100644 --- a/core/invertedindex/pending_writes.go +++ b/core/invertedindex/pending_writes.go @@ -106,7 +106,7 @@ func (idx *Index) flushPendingWrites(closing, force bool) { continue } - writeInvertedIndex(batch, wp.TableId, kw, relatedDocs.DocIds, idx.encodeInvertedKey(wp.TableId, kw, len(relatedDocs.DocIds), now.UnixMicro())) + writeInvertedIndex(idx, batch, wp.TableId, kw, relatedDocs.DocIds, now.UnixMicro()) idx.pendingWritePostings -= len(relatedDocs.DocIds) delete(wp.InvertedIndex, kw) diff --git a/core/invertedindex/test_helper_test.go b/core/invertedindex/test_helper_test.go index 870a32d..f9953b7 100644 --- a/core/invertedindex/test_helper_test.go +++ b/core/invertedindex/test_helper_test.go @@ -45,10 +45,7 @@ func setupTestEnv(t *testing.T) *testEnv { newBatch = func(db kv.Store) kv.Batch { return db.NewBatch(MaxBatchSize) } - writeInvertedIndex = func(batch kv.Batch, tableId int, kw string, docids []int64, key []byte) { - content := encodeInvertedValue(docids) - batch.Put(key, content) - } + writeInvertedIndex = defaultWriteInvertedIndex opts := Options{} idx, err := New(database, q, opts)