perf(invertedindex): Tier-1 in-place allocation / build-RSS reductions (byte-identical)

core/invertedindex/batch_close_test.go

@@ -0,0 +1,115 @@
+package invertedindex
+
+import (
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/codetrek/haystack/core/kv"
+	"github.com/codetrek/haystack/core/kv/pebblekv"
+	"github.com/codetrek/haystack/core/queue"
+)
+
+// recStore wraps a real kv.Store, delegating every method EXCEPT NewBatch, which
+// returns a recBatch so the test can observe Close() on every batch created.
+// Wrapping at the STORE captures both the three newBatch(idx.db) sites (flush
+// writes, flush deletes, merge) AND DeleteTable's direct idx.db.NewBatch(0),
+// which bypasses the newBatch package seam. batchCloses counts Close() across all
+// batches it hands out.
+type recStore struct {
+	kv.Store
+	batchCloses *atomic.Int64
+}
+
+func (s recStore) NewBatch(maxBatchSize int32) kv.Batch {
+	return &recBatch{Batch: s.Store.NewBatch(maxBatchSize), closes: s.batchCloses}
+}
+
+// recBatch embeds a real kv.Batch and overrides Close to increment the shared
+// counter (modelled on mockBatchWriteWithFuncs.Close) before delegating to the
+// embedded real Close.
+type recBatch struct {
+	kv.Batch
+	closes *atomic.Int64
+}
+
+func (b *recBatch) Close() error {
+	b.closes.Add(1)
+	return b.Batch.Close()
+}
+
+// TestBatchSitesCloseCommittedBatch proves every batch-creating site Closes its
+// committed batch (returning it to pebble's pool). Driving forceFlush (flush
+// writes + flush deletes), a synchronous merge, and DeleteTable exercises all
+// four sites; a store-level recorder observes Close on each. Before I5 no site
+// Closes (count 0 → fail); after I5 each op Closes at least once (count >= 4).
+func TestBatchSitesCloseCommittedBatch(t *testing.T) {
+	tempDir, err := os.MkdirTemp("", "haystack-ii-batchclose-*")
+	if err != nil {
+		t.Fatalf("mkdtemp: %v", err)
+	}
+	defer os.RemoveAll(tempDir)
+
+	real, err := pebblekv.Open(filepath.Join(tempDir, "data"), 0)
+	if err != nil {
+		t.Fatalf("open pebble: %v", err)
+	}
+
+	var closes atomic.Int64
+	store := recStore{Store: real, batchCloses: &closes}
+
+	q := queue.NewMpsc("TestBatchCloseQueue")
+	q.Start()
+
+	// Reset the package test-injection seams to their production defaults so the
+	// three newBatch sites actually reach recStore.NewBatch — prior tests
+	// (setupTestEnv, newBoundedIndex, the merger tests) reassign them, and a
+	// 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))
+	}
+
+	// Disable the periodic ticker (FlushTicker = 1h) so the only batches created
+	// are the ones this test drives explicitly — single-goroutine and race-free.
+	idx, err := New(store, q, Options{FlushTicker: time.Hour})
+	if err != nil {
+		q.Stop()
+		real.Close()
+		t.Fatalf("new index: %v", err)
+	}
+	defer func() {
+		idx.CloseAndWait()
+		q.Stop()
+		real.Close()
+	}()
+
+	const tableId = 7
+	idx.updateIndex(tableId, makeDocID("d1"), []string{"kw"})
+
+	// (a) forceFlush -> flushPendingWrites + flushPendingDeletes = 2 batches.
+	forceFlush(idx)
+
+	// (b) a synchronous merge via the queue = 1 batch. pendingWrites is now empty
+	// (drained by forceFlush), so mergeKeywordTask.Run does not early-return.
+	if err := idx.q.RunTask(&mergeKeywordTask{
+		idx:     idx,
+		merging: merging{NextIter: string(DefaultKeyTypeRow)},
+	}); err != nil {
+		t.Fatalf("merge RunTask: %v", err)
+	}
+
+	// (c) DeleteTable = 1 batch (idx.db.NewBatch(0)).
+	if err := idx.DeleteTable(tableId); err != nil {
+		t.Fatalf("DeleteTable: %v", err)
+	}
+
+	// Each of the four batch-creating ops must Close its committed batch. Before
+	// I5 the count is 0; after I5 it is >= 1 per op (>= 4 total).
+	if got := closes.Load(); got < 4 {
+		t.Fatalf("batch Close count = %d, want >= 4 (one per flush-writes, flush-deletes, merge, DeleteTable)", got)
+	}
+}

core/invertedindex/codec.go

@@ -6,7 +6,6 @@ import (
 	"slices"
 	"strconv"
 	"strings"
-	"time"
 )
 
 // Default on-disk key-type prefix bytes.  These values MUST NOT change after
@@ -69,18 +68,19 @@ func (idx *Index) encodeInvertedKeyPrefix(tableId int, keyword string) []byte {
 	return idx.appendInvertedKeyPrefix(b, tableId, keyword)
 }
 
-func (idx *Index) encodeInvertedKey(tableId int, keyword string, doccount int) []byte {
-	// "<prefix><doccount>|<tick>.<seq>" where tick is the current micros and seq
-	// is a per-Index monotonic counter. tick alone is not unique within a single
-	// microsecond, so two rows of the same (tableId,keyword,doccount) could get
-	// byte-identical keys and overwrite each other; the seq suffix makes every
-	// encoded key unique. decode treats everything after the last '|' as the
-	// opaque tick, so this does not change the on-disk format contract.
+func (idx *Index) encodeInvertedKey(tableId int, keyword string, doccount int, tick int64) []byte {
+	// "<prefix><doccount>|<tick>.<seq>" where tick is a decimal-micros timestamp
+	// SAMPLED ONCE PER flush/merge/delete batch by the caller and threaded in (not
+	// read per key), and seq is a per-Index monotonic counter. tick alone is not
+	// unique — two rows of the same (tableId,keyword,doccount) in one batch share
+	// it — so the seq suffix is the SOLE guarantor that every encoded key is
+	// distinct. decode treats everything after the last '|' as the opaque tick, so
+	// this does not change the on-disk format contract.
 	b := make([]byte, 0, 1+11+1+len(keyword)+1+11+1+19+1+20)
 	b = idx.appendInvertedKeyPrefix(b, tableId, keyword)
 	b = strconv.AppendInt(b, int64(doccount), 10)
 	b = append(b, '|')
-	b = strconv.AppendInt(b, time.Now().UnixMicro(), 10)
+	b = strconv.AppendInt(b, tick, 10)
 	b = append(b, '.')
 	b = strconv.AppendUint(b, idx.keySeq.Add(1), 10)
 	return b
@@ -146,11 +146,12 @@ func encodeTableValue(info TableInfo) []byte {
 // 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; deduplication
-// remains the caller's responsibility. 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.
+// 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,
+// 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 {
 	if len(docids) == 0 {
 		return []byte{}
@@ -160,17 +161,16 @@ func encodeInvertedValue(docids []int64) []byte {
 		us[i] = uint64(id)
 	}
 	slices.Sort(us)
+	us = slices.Compact(us) // adjacent-equal removal == set-dedup post-sort
 
 	buf := make([]byte, 0, len(us)+len(us)/2) // most ids encode to ~1 byte
-	var tmp [binary.MaxVarintLen64]byte
 	var prev uint64
 	for i, u := range us {
 		delta := u
 		if i > 0 {
 			delta = u - prev // non-negative: us is sorted ascending in uint64 space
 		}
-		n := binary.PutUvarint(tmp[:], delta)
-		buf = append(buf, tmp[:n]...)
+		buf = binary.AppendUvarint(buf, delta)
 		prev = u
 	}
 	return buf
@@ -234,7 +234,21 @@ func (idx *Index) encodeForwardKeyPrefix(tableId int) []byte {
 // doc-words value used. A keyword containing '|' splits the same lossy way it
 // always has (no behavior change vs. the old doc-words value).
 func encodeForwardValue(keywords []string) []byte {
-	return []byte(strings.Join(keywords, "|"))
+	if len(keywords) == 0 {
+		return []byte{}
+	}
+	n := len(keywords) - 1
+	for _, k := range keywords {
+		n += len(k)
+	}
+	b := make([]byte, 0, n)
+	for i, k := range keywords {
+		if i > 0 {
+			b = append(b, '|')
+		}
+		b = append(b, k...)
+	}
+	return b
 }
 
 // decodeForwardValue is the inverse of encodeForwardValue. An empty input decodes

core/invertedindex/codec_bench_test.go

@@ -27,7 +27,7 @@ func BenchmarkEncodeInvertedKey(b *testing.B) {
 	idx := benchIdx()
 	b.ReportAllocs()
 	for i := 0; i < b.N; i++ {
-		_ = idx.encodeInvertedKey(7, "handleRequest", 42)
+		_ = idx.encodeInvertedKey(7, "handleRequest", 42, 0)
 	}
 }
 
@@ -52,7 +52,7 @@ func BenchmarkEncodeInvertedKeyPrefix(b *testing.B) {
 // BenchmarkDecodeInvertedKey covers the per-row merge-scan key decoder (strings.Split).
 func BenchmarkDecodeInvertedKey(b *testing.B) {
 	idx := benchIdx()
-	key := string(idx.encodeInvertedKey(7, "handleRequest", 42))
+	key := string(idx.encodeInvertedKey(7, "handleRequest", 42, 0))
 	b.ReportAllocs()
 	for i := 0; i < b.N; i++ {
 		_, _, _, _ = idx.decodeInvertedKey(key)

core/invertedindex/codec_delimiter_test.go

@@ -32,7 +32,7 @@ func TestDecodeInvertedKey_KeywordWithDelimiter(t *testing.T) {
 		"a|b\xff|c", // delimiter + 0xff together
 	}
 	for _, kw := range cases {
-		key := idx.encodeInvertedKey(7, kw, 3)
+		key := idx.encodeInvertedKey(7, kw, 3, 0)
 		tid, gotKw, dc, tick := idx.decodeInvertedKey(string(key))
 		if tid != 7 || gotKw != kw || dc != 3 || tick == "" {
 			t.Errorf("decode(encode(kw=%q)) = (tableId=%d keyword=%q doccount=%d tick=%q); want (7, %q, 3, non-empty)",
@@ -53,8 +53,8 @@ func TestMerge_KeywordWithDelimiter_NoOrphan(t *testing.T) {
 	keywords := []string{"a|b", "c|d", "x\xffy", "plain"}
 
 	// Index several distinct docs per keyword across separate flushes (each flush
-	// writes a distinct tick'd row) so the merger's rewriteIndex path
-	// (len(Rows) >= 2) actually fires for every keyword.
+	// writes a distinct row, kept unique by the per-Index keySeq suffix) so the
+	// merger's rewriteIndex path (len(Rows) >= 2) actually fires for every keyword.
 	docsByKw := map[string][]int64{}
 	for round := 0; round < 3; round++ {
 		for ki, kw := range keywords {

core/invertedindex/codec_test.go

@@ -1,7 +1,9 @@
 package invertedindex
 
 import (
+	"bytes"
 	"encoding/json"
+	"math"
 	"slices"
 	"strings"
 	"testing"
@@ -86,7 +88,7 @@ func TestEncodeDecodeInvertedKeyRoundTrip(t *testing.T) {
 	keyword := "testing"
 	doccount := 42
 
-	key := testCodecIdx.encodeInvertedKey(tableId, keyword, doccount)
+	key := testCodecIdx.encodeInvertedKey(tableId, keyword, doccount, 0)
 	s := string(key)
 
 	// First byte is DefaultKeyTypeRow
@@ -327,3 +329,75 @@ func TestRemoveDuplicatesPreservesOrder(t *testing.T) {
 		}
 	}
 }
+
+// ---------------------------------------------------------------------------
+// encodeInvertedValue — set-dedup + byte-identity (I1)
+// ---------------------------------------------------------------------------
+
+// sortedUniqueUint64Space returns ids as the ascending-in-uint64-space set with
+// duplicates removed. It is an INDEPENDENT oracle for encodeInvertedValue's
+// sort+dedup contract — it never calls the function under test.
+func sortedUniqueUint64Space(ids []int64) []int64 {
+	us := make([]uint64, len(ids))
+	for i, id := range ids {
+		us[i] = uint64(id)
+	}
+	slices.Sort(us)
+	us = slices.Compact(us)
+	out := make([]int64, len(us))
+	for i, u := range us {
+		out[i] = int64(u)
+	}
+	return out
+}
+
+// TestEncodeInvertedValueGoldenBytes pins byte-identity against HAND-TYPED
+// delta-varint constants (never derived by calling encodeInvertedValue). The
+// encoder sorts the docids in uint64 space, drops duplicates, then writes the
+// first id followed by successive gaps as base-128 uvarints. This is the
+// load-bearing dedup guard: without slices.Compact the duplicate inputs encode
+// extra zero-gap varints and fail these goldens.
+func TestEncodeInvertedValueGoldenBytes(t *testing.T) {
+	tests := []struct {
+		name string
+		ids  []int64
+		want []byte
+	}{
+		// sort {1,1,2,3,3} -> unique {1,2,3} -> deltas 1,1,1
+		{"dups and unsorted", []int64{3, 1, 2, 1, 3}, []byte{0x01, 0x01, 0x01}},
+		// all identical -> {9} -> single absolute varint
+		{"all duplicates", []int64{9, 9, 9}, []byte{0x09}},
+		// already sorted, no dups -> deltas 1,1,2
+		{"sorted no dups", []int64{1, 2, 4}, []byte{0x01, 0x01, 0x02}},
+		// empty -> empty
+		{"empty", []int64{}, []byte{}},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got := encodeInvertedValue(tc.ids)
+			if !bytes.Equal(got, tc.want) {
+				t.Fatalf("encodeInvertedValue(%v) = %v, want %v", tc.ids, got, tc.want)
+			}
+		})
+	}
+}
+
+// TestEncodeInvertedValueRoundTripFullIntRange proves the encode/decode round
+// trip is exact for the full int64 range (negatives, MinInt64, MaxInt64) AND
+// that duplicates are dropped, comparing against the independent sorted-unique
+// oracle rather than hand-typed 10-byte varints.
+func TestEncodeInvertedValueRoundTripFullIntRange(t *testing.T) {
+	tests := [][]int64{
+		{math.MinInt64},
+		{-1},
+		{math.MaxInt64},
+		{5, -1, 5, math.MaxInt64},
+	}
+	for _, ids := range tests {
+		want := sortedUniqueUint64Space(ids)
+		got := decodeInvertedValue(encodeInvertedValue(ids))
+		if !slices.Equal(got, want) {
+			t.Fatalf("round-trip(%v): got %v, want %v", ids, got, want)
+		}
+	}
+}