Hacktoberfest 2026:メンテナが10月に向けて印を付けた、オープンで初心者向けの issue。 Hacktoberfest の issue を見る

[Chore] Merge distributed Stream replies as batches in the liaison coordinator

オープン
#14,157 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

まだ誰も着手していません。

評価

難易度
4/5
見積もり時間
3〜5日
初心者へのやさしさ
58/100
issue の種類
リファクタリング
明瞭さ
明確に書かれている
活発さ
活発
技術スタック
go

調査の方向性

Read pkg/query/logical/stream/stream_plan_distributed.go, especially distributedPlan.Execute and decodeNodeElements, then inspect pkg/query/vectorized/stream for BuildStreamMergePipeline and the BatchSource and schema alignment work in #1426. Add a fake-broadcast test through distributedPlan.Execute for frame batches, traced replies, errors, and cancellation; done means the explicit oracle cases pass, frames are materialized as Elements only after merging, and reply batches are released on every exit path.

索引モデルが issue の本文から書いたものです。

説明

database

Summary (executable leaf)

The liaison's distributed Stream coordinator should merge data-node frame replies as columnar batches, instead of decoding each frame into []*streamv1.Element and merging rows. This is the distributed-coordinator half of #14070's original scope. #14070 was narrowed to the local multi-group path (apache/skywalking-banyandb#1426), and this half has no issue. Under #13998.

Checked against BanyanDB origin/main @ 450c73a3:

  • pkg/query/logical/stream/stream_plan_distributed.go distributedPlan.Execute (around lines 140–199) broadcasts the query. For each reply, decodeNodeElements (around 212–234) decodes a frame with streamframe.Decode and immediately calls vstream.BuildElementsFromBatch, turning the whole batch into Elements. The replies are then k-way merged as rows (newSortableElements → sort.NewItemIter, around 184–187), deduplicated by ElementId with a seen map, and sliced by distributedLimit (offset/limit, around 339–355).
  • Data nodes reply with a frame whenever the query is not traced. banyand/query/processor.go:366 streamVecEmitAsFrame returns distributed && wireModeRaw && !traced, and both svc_liaison.go:165 and svc_standalone.go:262 set the raw wire mode. So in a normal distributed query every reply is a frame, and the coordinator materializes all of them as rows.
  • The batch operators this needs already exist: pkg/query/vectorized/stream SortedMerge (time or OrderKey, asc/desc, stable ties) → Distinct (first-seen by element id) → Limit(offset, limit), composed by BuildStreamMergePipeline. apache/skywalking-banyandb#1426 adds a multi-source BatchSource and union-schema alignment for the local path, and this work should reuse them.

Required change

  1. In distributedPlan.Execute, decode each frame reply into a batch and feed every node's batches into one merge pipeline (reuse BuildStreamMergePipeline and #1426's BatchSource / schema alignment; no third merger). Replies with differing projected tags are aligned to one merged schema the same way #1426 aligns groups.
  2. Apply the request's offset/limit in the pipeline's Limit, and build Elements once, from the pipeline output, for the client response. distributedLimit then slices nothing for this path (keep it for the traced path below, or fold it in).
  3. Keep the per-node cap (queryRequest.Limit = maxElementSize) and the span tags (response_count, element_id_count) as today.
  4. Traced queries stay on the current path. When any reply is a *streamv1.QueryResponse (a traced query), keep today's Element merge for that broadcast, including span.AddSubTrace. Converting protobuf replies to batches is out of scope.

Concrete oracle

Two data nodes, frame replies, ordered by key ascending:

Node 1: A(key=1), B(key=3)
Node 2: A(key=1), C(key=2)

The two copies of A are the same element (same ElementID, a replica).

  • Ascending, offset 1, limit 2: C, B.
  • Descending, offset 0, limit 2: B, C.
  • Ordered by timestamp instead (A=1, C=2, B=3): the same results.
  • The ordering tag isn't in the client projection: same order, and the tag is absent from the response (it travels in the frame's OrderKey column).
  • Each node's rows split across two frames or batches: same results.
  • A node that returns an error: the merged result of the other nodes plus the error, as today (multierr).

Acceptance criteria

  • A test through distributedPlan.Execute with a fake broadcast of frame replies asserts every case of the oracle.
  • Path-sensitive: for frame replies, no Element is built before the merge. decodeNodeElements no longer calls BuildElementsFromBatch on the frame path, and a test fails if per-reply Element materialization is restored, e.g. by asserting Elements are built once from the merged output. Asserting the answers alone is insufficient, because the current row merge returns the same answers.
  • A traced query (protobuf replies) returns the same elements as before and still attaches every node's sub-trace.
  • Batches from every reply are released on completion, on a decode error, and on context cancellation; covered by tests.
  • Existing tests pass:
go test ./pkg/query/logical/stream/... ./pkg/query/vectorized/stream/... ./banyand/query
go test -tags=querymetrics ./test/integration/distributed/query/...

Dependencies

Builds on apache/skywalking-banyandb#1426 (BatchSource, schema alignment). Start after it merges, or base the branch on it.

Out of scope

  • Converting traced (protobuf) replies to batches; the Element path stays for them. That's a follow-up, after which newSortableElements and comparableElement can be deleted.
  • The Trace query merges (pkg/query/logical/trace), which apache/skywalking-banyandb#1423 is reworking.
  • Wire format, per-node cap semantics, filter-first semantics, and new memory-bound claims.

Compatibility and review policy

  • Preserve existing on-disk and wire formats.
  • Use explicit, hand-calculated expected results; do not use the removed row engine as the test oracle.
主要言語
Java
スター
25k
フォーク
6.6k
平均マージ
16時間 55分
マージ済み PR(30日)
31

環境構築

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

apache/skywalking のほかの issue

apache/skywalking の issue をすべて見る

似ている issue

Java の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。