[Performance] Bounded-memory Top-N for vectorized Stream queries
- Dominant language
- Java
- Stars
- 25k
- Forks
- 6.6k
- Avg merge
- 10h 5m
- Merged PRs (30d)
- 16
Description
## 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 #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.
Contributor guide
Research direction
Start with pkg/query/vectorized/stream/sorted_merge.go and pipeline.go, then compare their cap and distinct handling with model.MergeStreamResults in pkg/query/model/model.go. Run the focused vectorized stream tests, especially the four named SortedMerge tests, before checking the integration parity suites. Done means capped criteria-less merges avoid full-N buffering while preserving row-path results; filtered-query behavior remains unchanged unless parity is demonstrated.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- backend, performance
- Issue type
- Refactor
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100