fix(invertedindex): stamp flushed posting keys with the deduped doccount

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.

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

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")
+}

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")

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

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++
 	}
 

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 },

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)

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)
 

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)