fix(shm): split oversized batches so every frame fits its ring#37
Draft
mdashti wants to merge 11 commits into
Draft
fix(shm): split oversized batches so every frame fits its ring#37mdashti wants to merge 11 commits into
mdashti wants to merge 11 commits into
Conversation
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.
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. Compacting and halving on the oversized path bounds frames by the ring instead of asking embedders to size rings for their widest state.
paradedb-github-bot
Bot
force-pushed
the
main
branch
from
f35ee8e to
34933e5
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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
BatchChannelSendergainsmax_frame_bytes()(the ring's combined slot payload;Nonefor unbounded in-proc channels). When an encoded frame exceeds it, the send path compacts the rows withtake(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.