events: store small terms as delta postings and query them un-materialized

cmd/stellar-rpc/internal/rpcv2/events/intersect.go

@@ -0,0 +1,175 @@
+package events
+
+import (
+	"cmp"
+	"slices"
+
+	"github.com/RoaringBitmap/roaring/v2"
+)
+
+// mergeRatio is the length ratio at which probing a sorted ID list switches
+// from a cursored sequential walk to binary search over the unconsumed tail.
+//
+// Per driver ID a walk costs O/D comparisons amortized and a search log2(O),
+// so on count alone they cross at O/D = log2(O). Both are cursored, so the
+// search skips ground it has covered too, and measured per comparison the walk
+// is the more expensive of the two (1.65ns against 1.02ns on scattered
+// postings, whose values mispredict the walk's data-dependent branch every
+// iteration). That puts the wall-clock crossover at 3 to 6 across driver sizes
+// from 4 to 256, and 4 sits on the safe side of a shallow optimum: the walk
+// gives up at most 1.28x there, against 4.1x to 4.7x at a ratio of 32.
+const mergeRatio = 4
+
+// Intersect returns the intersection of ps, or the zero Postings when any
+// input is absent or nothing survives. It may reorder ps.
+//
+// Where the postings come back as ID lists — every term at or below the index's
+// delta-coding threshold — intersecting them directly beats materializing them
+// into bitmaps first, by 4.3x at cardinality 4, 2.8x at 64 and 1.6x at 1024.
+// That is the whole reason the stores hand out un-materialized postings.
+//
+// The smallest input drives, so the work is bounded by the smallest side and
+// the result is at most that large. When every input is already a bitmap there
+// is nothing to save and roaring's container-wise FastAnd wins instead.
+func Intersect(ps []Postings) Postings {
+	if len(ps) == 0 {
+		return Postings{}
+	}
+	for _, p := range ps {
+		if !p.Present() {
+			return Postings{}
+		}
+	}
+	// See ConcurrentBitmaps.Get: FastAnd Clones a lone input.
+	if len(ps) == 1 {
+		return ps[0]
+	}
+
+	slices.SortFunc(ps, func(a, b Postings) int {
+		return cmp.Compare(a.Cardinality(), b.Cardinality())
+	})
+
+	if !slices.ContainsFunc(ps, func(p Postings) bool { return p.ids != nil }) {
+		bms := make([]*roaring.Bitmap, len(ps))
+		for i, p := range ps {
+			bms[i] = p.bm
+		}
+		return BitmapPostings(roaring.FastAnd(bms...))
+	}
+
+	// The driver is the smallest side, so reading a bitmap one out costs an
+	// allocation bounded by the smallest cardinality — cheap next to the
+	// materialization of every other side that this avoids.
+	driver := ps[0].ids
+	if driver == nil {
+		driver = ps[0].bm.ToArray()
+	}
+
+	probes := make([]prober, len(ps)-1)
+	for i, p := range ps[1:] {
+		probes[i] = prober{
+			ids:        p.ids,
+			bm:         p.bm,
+			sequential: len(p.ids) <= mergeRatio*len(driver),
+		}
+	}
+
+	hits := make([]uint32, 0, len(driver))
+next:
+	for _, id := range driver {
+		for i := range probes {
+			if !probes[i].contains(id) {
+				continue next
+			}
+		}
+		hits = append(hits, id)
+	}
+	return IDPostings(hits)
+}
+
+// prober tests membership in one intersect input. For an ID list it carries a
+// cursor, since the driver ascends and so no probe ever needs to look behind
+// the last ID it matched.
+type prober struct {
+	ids        []uint32
+	bm         *roaring.Bitmap
+	cursor     int
+	sequential bool
+}
+
+func (p *prober) contains(id uint32) bool {
+	if p.ids == nil {
+		return p.bm.Contains(id)
+	}
+	if p.sequential {
+		for p.cursor < len(p.ids) && p.ids[p.cursor] < id {
+			p.cursor++
+		}
+		return p.cursor < len(p.ids) && p.ids[p.cursor] == id
+	}
+	i, found := slices.BinarySearch(p.ids[p.cursor:], id)
+	p.cursor += i
+	return found
+}
+
+// Union returns the union of ps, or the zero Postings when none are present.
+// It compacts ps in place, dropping absent entries and zeroing the tail, so a
+// caller that reads ps afterwards sees neither its original order nor its
+// original length.
+//
+// All-list inputs are merged rather than materialized. Building a bitmap from a
+// scattered id list costs an allocation per container it touches, which is what
+// dominates: 2 filters of 16 postings measure 14.8us and 493 allocations
+// through FastOr against 114ns and 2 merged, and 5 filters of 1024 drop from
+// 3271 allocations to 9. Keeping the result a list is most of that win, since
+// the clip and the selection downstream are far cheaper on one.
+//
+// Anything else goes to FastOr, which also covers the region where it wins: a
+// bitmap input means that term is above the index's delta threshold.
+func Union(ps []Postings) Postings {
+	ps = slices.DeleteFunc(ps, func(p Postings) bool { return !p.Present() })
+	switch len(ps) {
+	case 0:
+		return Postings{}
+	case 1:
+		// See ConcurrentBitmaps.Get: FastOr Clones a lone input.
+		return ps[0]
+	}
+
+	if !slices.ContainsFunc(ps, func(p Postings) bool { return p.ids == nil }) {
+		merged := ps[0].ids
+		for _, p := range ps[1:] {
+			merged = mergeAscending(make([]uint32, 0, len(merged)+len(p.ids)), merged, p.ids)
+		}
+		return IDPostings(merged)
+	}
+
+	bms := make([]*roaring.Bitmap, len(ps))
+	for i, p := range ps {
+		bms[i] = p.Bitmap()
+	}
+	return BitmapPostings(roaring.FastOr(bms...))
+}
+
+// mergeAscending appends the union of two ascending deduped lists to dst.
+// Merging pairwise beats a k-way heap for the handful of inputs a filter set
+// produces: 5 lists of 1024 measure 46us pairwise against 96us through a heap.
+func mergeAscending(dst, a, b []uint32) []uint32 {
+	i, j := 0, 0
+	for i < len(a) && j < len(b) {
+		switch {
+		case a[i] < b[j]:
+			dst = append(dst, a[i])
+			i++
+		case a[i] > b[j]:
+			dst = append(dst, b[j])
+			j++
+		default:
+			dst = append(dst, a[i])
+			i++
+			j++
+		}
+	}
+	dst = append(dst, a[i:]...)
+	return append(dst, b[j:]...)
+}

cmd/stellar-rpc/internal/rpcv2/events/intersect_test.go

@@ -0,0 +1,186 @@
+package events
+
+import (
+	"math/rand"
+	"slices"
+	"testing"
+
+	"github.com/RoaringBitmap/roaring/v2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// makeTestPostings builds card postings scattered over span, in the requested
+// form, and returns the equivalent bitmap for cross-checking.
+func makeTestPostings(tb testing.TB, rng *rand.Rand, card int, span uint32, asIDs bool) (Postings, *roaring.Bitmap) {
+	tb.Helper()
+	// Without this the distinct-value loop below never terminates.
+	require.LessOrEqual(tb, uint32(card), span, "cardinality must fit in the span")
+	set := make(map[uint32]struct{}, card)
+	for len(set) < card {
+		set[uint32(rng.Int31n(int32(span)))] = struct{}{}
+	}
+	ids := make([]uint32, 0, len(set))
+	for id := range set {
+		ids = append(ids, id)
+	}
+	slices.Sort(ids)
+	bm := roaring.BitmapOf(ids...)
+	if asIDs {
+		return IDPostings(ids), bm
+	}
+	return BitmapPostings(bm), bm
+}
+
+// TestIntersectMatchesFastAnd cross-checks Intersect against roaring.FastAnd,
+// the operation it replaces. The span is kept tight relative to cardinality so
+// intersections are usually non-empty — with postings spread thinly, most
+// rounds would compare one empty result against another and prove nothing.
+func TestIntersectMatchesFastAnd(t *testing.T) {
+	rng := rand.New(rand.NewSource(7))
+
+	for round := range 300 {
+		inputs := make([]Postings, 2+rng.Intn(3))
+		want := make([]*roaring.Bitmap, len(inputs))
+		for i := range inputs {
+			// Mixed forms and a wide cardinality spread, so the driver is
+			// sometimes an ID list and sometimes a bitmap, and both probe
+			// strategies get exercised.
+			inputs[i], want[i] = makeTestPostings(t, rng, 1+rng.Intn(60), 100, rng.Intn(2) == 0)
+		}
+
+		got := Intersect(inputs)
+		expected := roaring.FastAnd(want...)
+		assert.Equal(t, expected.ToArray(), got.Bitmap().ToArray(), "round %d", round)
+		assert.Equal(t, !expected.IsEmpty(), got.Present(), "round %d presence", round)
+	}
+}
+
+// TestIntersectDriverStrategies covers the shapes that select each probe
+// strategy, including the ratio at which a cursored walk gives way to binary
+// search and the case where the smallest side is a bitmap.
+func TestIntersectDriverStrategies(t *testing.T) {
+	rng := rand.New(rand.NewSource(11))
+
+	for _, tc := range []struct {
+		name        string
+		driverCard  int
+		otherCard   int
+		driverAsIDs bool
+		otherAsIDs  bool
+	}{
+		{"equal lists take the walk", 400, 400, true, true},
+		{"lopsided lists take the search", 4, 400, true, true},
+		{"tiny list driver forces the search", 1, 2000, true, true},
+		{"list driver against a bitmap", 40, 2000, true, false},
+		{"bitmap driver against a list", 40, 2000, false, true},
+		{"all bitmaps fall back to FastAnd", 40, 2000, false, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			a, abm := makeTestPostings(t, rng, tc.driverCard, 4000, tc.driverAsIDs)
+			b, bbm := makeTestPostings(t, rng, tc.otherCard, 4000, tc.otherAsIDs)
+
+			// Pass largest-first: Intersect owns the ordering.
+			got := Intersect([]Postings{b, a})
+			assert.Equal(t, roaring.And(abm, bbm).ToArray(), got.Bitmap().ToArray())
+		})
+	}
+}
+
+func TestIntersectDegenerate(t *testing.T) {
+	present := IDPostings([]uint32{1, 2, 3})
+
+	assert.False(t, Intersect(nil).Present())
+	assert.False(t, Intersect([]Postings{}).Present())
+	assert.False(t, Intersect([]Postings{present, {}}).Present(),
+		"one absent term empties the intersection")
+	assert.False(t, Intersect([]Postings{IDPostings([]uint32{1}), IDPostings([]uint32{2})}).Present(),
+		"disjoint inputs are absent, not an empty present set")
+
+	// A single input passes through untouched, so a lone bitmap-backed term is
+	// not materialized on the way past.
+	bm := roaring.BitmapOf(1, 2, 3)
+	single := Intersect([]Postings{BitmapPostings(bm)})
+	require.True(t, single.Present())
+	assert.Same(t, bm, single.Bitmap())
+}
+
+// TestUnionMatchesFastOr cross-checks the merge path against roaring.FastOr,
+// which is both the operation it replaces and the fallback it still uses when
+// any input is bitmap-backed.
+func TestUnionMatchesFastOr(t *testing.T) {
+	rng := rand.New(rand.NewSource(13))
+
+	for round := range 400 {
+		inputs := make([]Postings, 2+rng.Intn(4))
+		want := make([]*roaring.Bitmap, len(inputs))
+		// Mixed forms so both the merge and the FastOr fallback run, and
+		// occasionally all-lists so the merge runs on every input.
+		allLists := rng.Intn(3) == 0
+		for i := range inputs {
+			asIDs := allLists || rng.Intn(2) == 0
+			inputs[i], want[i] = makeTestPostings(t, rng, 1+rng.Intn(80), 200, asIDs)
+		}
+
+		got := Union(inputs)
+		expected := roaring.FastOr(want...)
+		require.True(t, got.Present(), "round %d", round)
+		assert.Equal(t, expected.ToArray(), got.Bitmap().ToArray(), "round %d", round)
+		if allLists {
+			ids := got.IDs()
+			require.NotNil(t, ids, "round %d: all-list inputs must merge, not materialize", round)
+			// Assert on the raw slice, not through Bitmap(): that sorts and
+			// dedups, so it would launder a merge that emitted duplicates or
+			// went backwards. IDPostings does not check the precondition, and
+			// violating it answers wrongly rather than failing.
+			assert.Equal(t, expected.ToArray(), ids, "round %d: merged ids", round)
+		}
+	}
+}
+
+// TestUnionMergeBoundaries covers the shapes a randomized sweep reaches rarely:
+// full overlap, no overlap, one list a prefix or suffix of the other, and
+// duplicate values across every input.
+func TestUnionMergeBoundaries(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		in   [][]uint32
+		want []uint32
+	}{
+		{"identical", [][]uint32{{1, 2, 3}, {1, 2, 3}}, []uint32{1, 2, 3}},
+		{"disjoint interleaved", [][]uint32{{1, 3, 5}, {2, 4, 6}}, []uint32{1, 2, 3, 4, 5, 6}},
+		{"disjoint blocks", [][]uint32{{1, 2}, {8, 9}}, []uint32{1, 2, 8, 9}},
+		{"prefix", [][]uint32{{1, 2}, {1, 2, 3, 4}}, []uint32{1, 2, 3, 4}},
+		{"suffix", [][]uint32{{3, 4}, {1, 2, 3, 4}}, []uint32{1, 2, 3, 4}},
+		{"singletons", [][]uint32{{5}, {5}, {5}}, []uint32{5}},
+		{"three-way", [][]uint32{{1, 4}, {2, 5}, {3, 6}}, []uint32{1, 2, 3, 4, 5, 6}},
+		{"extremes", [][]uint32{{0}, {4294967295}}, []uint32{0, 4294967295}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ps := make([]Postings, len(tc.in))
+			for i, ids := range tc.in {
+				ps[i] = IDPostings(ids)
+			}
+			got := Union(ps)
+			require.NotNil(t, got.IDs(), "all-list inputs must merge")
+			assert.Equal(t, tc.want, got.SelectIDs(0, false))
+		})
+	}
+}
+
+func TestUnion(t *testing.T) {
+	a := IDPostings([]uint32{1, 5})
+	b := BitmapPostings(roaring.BitmapOf(5, 9))
+
+	assert.False(t, Union(nil).Present())
+	assert.False(t, Union([]Postings{{}, {}}).Present())
+
+	// A lone present input passes through, keeping its form.
+	single := Union([]Postings{{}, a, {}})
+	require.True(t, single.Present())
+	assert.NotNil(t, single.IDs(), "a passed-through input must keep its form")
+
+	both := Union([]Postings{a, b})
+	require.True(t, both.Present())
+	assert.Equal(t, []uint32{1, 5, 9}, both.SelectIDs(0, false))
+}