fix(shm): split oversized batches so every frame fits its ring by mdashti · Pull Request #37 · paradedb/datafusion-distributed

mdashti · 2026-07-15T07:52:52Z

What

This PR splits an oversized batch at the shm send path so every encoded frame fits its ring.

Why

Operators emit offset slices of their accumulated state, and arrow-ipc writes a sliced variable-length array's whole values buffer, so one frame balloons to the state's size no matter the batch size. Small rings then fail wide GROUP BY and DISTINCT shapes with MessageTooLarge, and the only workaround was sizing every ring for the widest state any query might carry.

How

BatchChannelSender gains max_frame_bytes() (the ring's combined slot payload; None for unbounded in-proc channels). When an encoded frame exceeds it, the send path compacts the rows with take (dropping the shared buffers that cause the ballooning) and halves by rows until each frame fits, left-to-right so row order survives for sorted streams. The common case pays one length check. A single row larger than the ring still errors with the raise-the-knob message.

Tests

Unit: an oversized sliced batch splits into under-cap frames that decode back to the identical rows in order; a single oversized row still errors. End-to-end: a wide max(s) GROUP BY over 64KiB rings, whose raw partial-aggregate emits are ~10x the ring, matches the serial reference exactly.

The prost message types carry no tonic dependency, so a transport that is not gRPC can speak the same wire shape without pulling in the gRPC stack. Only the tonic client and server stay gated; the generator emits those gates so a regeneration cannot drop them. tonic-prost feeds only the generated client and server, so it moves behind the feature too. Co-authored-by: Stu Hood <stuhood@gmail.com>

The benchmarks crate's dev-dependency on the lib re-unified grpc into every test build, so a genuine no-gRPC test run was impossible; the dataset suites move into the benchmarks crate and the gRPC-coupled test utilities gate behind grpc. A unit-test-no-grpc job then runs the whole lib suite with the feature off. Co-authored-by: Stu Hood <stuhood@gmail.com>

InProcessChannelResolver routes the three protocol methods straight to a co-located Worker, with no gRPC, no IPC, and no serialization round-trip: the reference implementation of the protocol for a co-located worker, and the first transport that exercises the abstraction with grpc off. Its end-to-end test (a distributed GROUP BY across tasks) runs under the no-gRPC CI job. Co-authored-by: Stu Hood <stuhood@gmail.com>

A transport that returns worker metrics out-of-band, rather than over the coordinator_channel return stream, needs its driver to decode the frames and file them into the executed plan's store before the per-task EXPLAIN rewrite reads it: metrics_store() with a public insert, a no-gRPC decode_task_metrics, and the frame builders collect_plan_metrics_protos and set_received_time. The metrics codec moves out from behind grpc with them, and the gRPC client drops its private copy of the decode. The test pins the frame-to-store path. Co-authored-by: Stu Hood <stuhood@gmail.com>

An embedder whose plan nodes the coordinator's codec cannot represent, or whose serialization needs embedder-side handling the codec extension point cannot express, serializes the dispatch bytes itself. The coordinator hands it the TaskKey and the ready-to-run per-task plan it would otherwise encode; returning None falls back to the coordinator's own encode. The tests pin the contract from all three sides: consultation with nested stages already Remote, source-provided bytes running the query, and a source error failing the dispatch. Co-authored-by: Stu Hood <stuhood@gmail.com>

A pull-based transport never places a produced partition: the consumer computes its own slice inside the boundary's execute. A push-based transport places every partition before any consumer asks, so it reads the boundary's consumer layout through route_partition instead of re-deriving it from node properties and drifting when the layout changes. NetworkCoalesceExec overrides it with an error: its consumers read whole per-producer-task groups, not slices. Co-authored-by: Stu Hood <stuhood@gmail.com>

The shm ring/DSM/mesh/framing core moves verbatim from the fork; the old WorkerTransport umbrella is gone, so the resolver hands out a channel and execute_task reads the rings per partition. The mesh is the no-gRPC transport, so it builds in both the grpc-on and grpc-off configs. Workers are passive executors: a fragment arrives with its nested stages already Remote, so the worker runs it as-is and its boundary leaves read the mesh; nothing on the worker converts or dispatches. collect_task_metrics reports an empty task-level metric set instead of a missing one, since one failed decode starves the whole store.

ShmWorkerChannel::coordinator_channel now ships each stage's SetPlanRequest to the worker proc that owns its task, as a SetPlan frame the worker takes with take_set_plan, so the embedder no longer routes plans out of band.

Run CI on `target-patch-*` branches. (#35)

It deploys the docs site to GitHub Pages on every push to `main`; the fork does not publish that site, so the job only fails on the fork.

The `paradedb/paradedb` repo consumes this repo as a library, rather than as a binary: consequently, it needs stable SHAs to consume (which won't be wiped out by the rebase process). This change adds a job for creating these snapshots with standardized tags.

Incorporate paradedb/actions#7 to fix Slack mentions by always mentioning at-most the approvers: never `@here`.

## What Bump the three `paradedb/actions` references used by the upstream sync from `v7` to `v8`: - `.github/workflows/sync-upstream-rebase.yml` — `reusable-rebase.yml@v7` → `@v8` - `.github/workflows/sync-promote-branch.yml` — `reusable-promote.yml@v7` → `@v8` - `scripts/sync-upstream.sh` — `.../actions/v7/scripts/sync-core.sh` → `.../v8/...` ## Why The scheduled **Upstream Rebase** was failing when the runner hit `No space left on device`: the reusable workflow checked out with `fetch-depth: 0` and then did a bare `git fetch`, pulling *every* branch and its unrelated history and filling the runner disk. `v8` fixes this with `fetch-depth: 1` plus a `fetch_branch` helper that fetches only the target and upstream `main` branches (paradedb/actions#8). It also adds a `range-diff` subcommand (paradedb/actions#11). ## Note All three references must move together — `sync-upstream.sh` downloads `sync-core.sh` from the version path, so bumping just the workflow refs would still execute the old core script. The `v8` tag is already published (signed/verified) on `paradedb/actions`. https://claude.ai/code/session_016i7usbUvUuBwHsv13uaa4S

`sync-upstream-rebase.yml` and `sync-promote-branch.yml` are the only two workflows unique to this repo — every other workflow arrives via sync from `paradedb/paradedb`, so this is the one place the hardening has to be applied directly. Both call reusable workflows in `paradedb/actions@v8`, which declare job-level permissions: | Reusable workflow | Declares | |---|---| | `reusable-rebase.yml` | `actions: write`, `checks: read`, `contents: write`, `issues: write`, `pull-requests: read` | | `reusable-promote.yml` | `actions: write`, `contents: write`, `issues: write`, `pull-requests: read` | **A called workflow's permissions are capped by the calling job's token.** The calling jobs here declare nothing, so today they inherit the repository default (write) and the reusable workflows get what they ask for. If the org default is switched to read-only — which grants only `contents: read` and `packages: read` — the callers would silently cap the reusable workflows below what they need, and both sync workflows would break: - `sync-core.sh` reads check runs via `gh api .../check-runs` — needs `checks: read` - `sync-core.sh` restarts failed jobs via `gh api -X POST .../actions/jobs/{id}/rerun` — needs `actions: write` - `reusable-promote.yml` uses `trstringer/manual-approval` with `github.token`, which opens an approval issue — needs `issues: write` This mirrors each reusable workflow's declared permissions onto its caller so the flip is safe. Note `sync-upstream-rebase.yml` runs on a schedule, so a regression here would surface as a silent failure rather than a red PR. https://claude.ai/code/session_01Rg3UBFSnmKMPQLNpNKkn8d