apache / apache/skywalking

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

Open
#14,050 1 comment 0 reactions 0 assignees View on GitHub
database
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.