[Performance] Bounded-memory Top-N for vectorized Stream queries

[hanahmily]

Summary

Vectorized Stream queries use scan → batch → SortedMerge → Distinct → Limit, but SortedMerge currently buffers every row reference from every input batch, performs a full stable sort (O(N log N)), and only then applies the in-order top-N cap (by distinct ElementID). For a query with LIMIT 20 over a large time range, memory and CPU still scale with N (all scanned rows), not with the limit.

The row path uses an incremental heap merge (MergeStreamResults / blockHeap.merge) that stops once it has collected top-N distinct elements without sorting the full set. The vec path is correct for capped queries (unit tests and row-parity logic confirm cap-after-merge semantics) but is not bounded in work or memory.

Filtered (criteria) Stream queries are worse: for index-order plans the merge runs uncapped (mergeCap = 0) to preserve row-path parity, then results are materialized to protobuf Element and filtered per row at egress. This is the highest-cost path and was identified during analysis of the vectorized Stream Top-N execution model (see related discussion in the BanyanDB vec query workstream).

Boundary

Replace or augment pkg/query/vectorized/stream.SortedMerge (and its use in BuildStreamMergePipeline) so that top-N queries bound memory and sort/merge work by limit + offset (distinct ElementID aware), without changing observable query results relative to the row path.

Callers that must continue to behave identically:

• pkg/query/logical/stream.(*localIndexScan).ExecuteVectorized — sets mergeCap from maxElementSize or 0 when deferLimitToEgress
• banyand/query/processor.go — vec dispatch, optional applyStreamTagFilter egress
• Distributed liaison frame merge path

Requirements

R1. For a criteria-less Stream query with limit = L and offset = O, vec execution must retain the same element set and order as the row path, while using O(L + O) (or better) auxiliary memory for merge state — not O(N) over all scanned rows.

R2. SortedMerge must not perform a full sort of N rows when mergeCap > 0 and N ≫ mergeCap; merge work must be bounded by the cap (e.g. k-way heap merge or streaming top-N heap during consume), preserving stable tie-breaking and distinct-ElementID semantics documented in sorted_merge.go.

R3. Filtered (criteria) index-order queries currently set mergeCap = 0 (stream_plan_indexscan_local_vectorized.go). Document whether bounded merge is feasible without starving the tag filter; if not, state the parity constraint explicitly in code and issue follow-ups. Any improvement that reduces materialized row count before filter without changing results is in scope.

R4. Regressions for existing correctness must remain green: TestSortedMergeCapKeepsInOrderTopN, TestSortedMergeCapCountsDistinctElementIDs, TestSortedMergeHugeCapDoesNotOverAllocate, and stream vec parity / integration tests.

Acceptance criteria

• Benchmark or test demonstrating that a LIMIT 20 query over ≥100k matching rows uses substantially less merge memory than current SortedMerge (e.g. heap growth bounded by cap, not row count).
• All existing tests in pkg/query/vectorized/stream/ pass.
• test/integration/standalone/query/vectorized_stream_test.go and test/integration/distributed/query/vectorized_stream_test.go pass (vec == row parity).
• No change to on-disk format or wire protocol (bydb file compatible change not required).
• Filtered-query behaviour unchanged unless a bounded approach is proven row-equivalent (with new parity tests).

Scope

Packages: pkg/query/vectorized/stream/ (sorted_merge.go, pipeline.go, distinct.go), pkg/query/logical/stream/stream_plan_indexscan_local_vectorized.go, banyand/stream/query_vectorized.go, banyand/query/processor.go.

Out of scope: Measure/Trace vec top-N (separate pipelines); removing row-based Stream query path (#13998); columnar tag-filter pushdown (separate follow-up); ANN/embedding vector search.

References

• Current merge: pkg/query/vectorized/stream/sorted_merge.go
• Pipeline: pkg/query/vectorized/stream/pipeline.go
• Planner cap / defer logic: pkg/query/logical/stream/stream_plan_indexscan_local_vectorized.go
• Row-path oracle: pkg/query/model/model.go (MergeStreamResults)
• Stream vec introduction: PR TraceId is invalid in Logback AsyncAppender #1241

Size audit (for implementers)

Classification: tracking parent (first deliverable should be R1/R2 for criteria-less queries; R3 is a follow-up leaf).

Boundary: SortedMerge / BuildStreamMergePipeline.

RED test: memory or row-count assertion on a synthetic pipeline with N ≫ cap fails today because len(SortedMerge.rows) == N before cap.

Focused suites: go test ./pkg/query/vectorized/stream/..., integration vec stream tests above.

[hanahmily]

R3 follow-up filed as #14056 (sub-issue of this tracking parent).

PR #1322 delivered R1/R2/R4 and documented why filtered index-order merges stay uncapped. Remaining work is columnar / pre-merge tag-filter pushdown so mergeCap can be applied without starving the filter.