Hacktoberfest 2026: los issues que los mantenedores marcaron para octubre, abiertos y aptos para principiantes. Explorar issues de Hacktoberfest

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

Abierto
  • ¿Algo no está bien?
#14,157 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
58/100
Tipo de issue
Refactorización
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
go
Área
backend, distributed-systems, observability

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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.
Lenguaje dominante
Java
Estrellas
25k
Forks
6.6k
Merge medio
15 h 4 min
PR fusionados (30 d)
32

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/skywalking

Todos los issues de apache/skywalking

Issues similares

Más issues de Java

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.