[Chore] Merge Stream results as batches across groups and data nodes
- Dominant language
- Java
- Stars
- 25k
- Forks
- 6.6k
- Avg merge
- 10h 5m
- Merged PRs (30d)
- 16
Description
## Summary
Remove intermediate `Element` materialization from Stream result merging and deduplication. Follow-up to #14066 under #13998.
## Size audit and dependencies
Classification: **tracking parent**, not an executable leaf or automated implementation task.
Boundaries: local multi-group dispatch using `MergeGroupElements`, and the distributed coordinator that currently decodes frames into Elements before merging.
These are separate production cutovers and must become separate executable tasks.
Blocked by: #14066 via BanyanDB PR apache/skywalking-banyandb#1326.
Related dependency: #14067. Coordinate with that workstream for full filtered/hidden-tag coverage; basic frame-input coordinator merging can be sliced independently. Do not start a child whose required projection boundary is not yet merged.
## Requirements
R1. Merge and deduplicate batches directly for both local multi-group and distributed callers.
R2. Preserve ordering, selected duplicate values, and global offset/limit behavior. Retain internal sort keys even when clients do not project ordering tags.
R3. Accept frame and protobuf node responses for tracing and mixed-version compatibility. Convert protobuf input at the compatibility boundary, rather than forcing frame inputs back into Elements.
R4. Materialize final Elements only where the client output requires them; release batch resources on completion, cancellation, and malformed responses.
## Concrete oracle
```text
Source 1: A(key=1), B(key=3)
Source 2: A(key=1), C(key=2)
```
The two copies of A are identical. Ascending order with offset 1 and limit 2 returns **C, B**. Exercise the same oracle across batches, across local groups, and across data-node replies. Specify duplicate-version and equal-key fixtures separately when defining each executable child; do not invent a new tie rule as a side effect of the cutover.
## Completion criteria and RED-test direction
- [ ] Real local and distributed callers use batch merge/deduplication, including cross-batch duplicates and pagination.
- [ ] Mixed frame/protobuf replies preserve the oracle and tracing information.
- [ ] Path-sensitive tests fail if frame inputs are materialized into Elements before merging. Output assertions alone are not RED because the current row merge can return the same answers.
- [ ] Cancellation, malformed replies, and batch release are covered in each caller cutover.
- [ ] Benchmark comparisons quantify allocation changes; no new wire format or unproved bounded-memory guarantee is required.
Relevant code: `banyand/query/processor.go`, `pkg/query/logical/stream/stream_plan_merge.go`, `pkg/query/logical/stream/stream_plan_distributed.go`, and `pkg/query/vectorized/stream`.
```sh
go test ./banyand/query ./pkg/query/logical/stream ./pkg/query/vectorized/stream/...
go test ./test/integration/standalone/query/... ./test/integration/distributed/query/...
```
## Out of scope
Changing filter-first semantics, timestamp under-fill behavior, or wire formats; implementing Stream filtering/projection itself (separate workstream).
Parent: #13998.
## Compatibility and review policy
- Preserve existing on-disk and wire formats and mixed-version compatibility. Any necessary protocol change requires a separately reviewed compatibility contract.
- Use explicit, hand-calculated expected results; do not depend on the removed row engine as the test oracle.
- Human review enforces the no-new-row-path policy. A row-call-site lint guard is not required.
Contributor guide
Research direction
This is a tracking parent rather than an executable leaf; first read banyand/query/processor.go, pkg/query/logical/stream/stream_plan_merge.go, stream_plan_distributed.go, and pkg/query/vectorized/stream. Run the listed Go unit and integration tests, then split the local and distributed cutovers into executable tasks after #14066 and the projection boundary are available. Done requires explicit batch merge tests for ordering, deduplication, pagination, compatibility, cancellation, malformed replies, and resource release.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- backend, distributed-systems, observability-sre
- Issue type
- Refactor
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100