Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

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

Aperta
  • Qualcosa non va?
#14,157 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
58/100
Tipo di issue
Refactoring
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
go
Ambito
backend, distributed-systems, observability

Direzione di ricerca

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.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

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.
Lingua principale
Java
Stelle
25k
Fork
6.6k
Merge medio
15h 37m
PR unite (30g)
32

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/skywalking

Tutte le issue di apache/skywalking

Issue simili

Altre issue su Java

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.