Skip to content

[Chore] Merge Stream results as batches across groups and data nodes #14070

Description

@hanahmily

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

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.

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.

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    databaseBanyanDB - SkyWalking native database

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions