fix: let the shm transport absorb the coordinator's plan_proto encode#30
Closed
mdashti wants to merge 20 commits into
Closed
fix: let the shm transport absorb the coordinator's plan_proto encode#30mdashti wants to merge 20 commits into
plan_proto encode#30mdashti wants to merge 20 commits into
Conversation
The worker-protocol abstraction shipped only a gRPC implementation, so nothing exercised it without gRPC. This routes the three protocol methods straight to a co-located `Worker`, and is the seam a shared-memory transport plugs into. A worker resolves its own nested stage reads back through the same transport, so multi-stage queries stay in process.
A co-located transport (in-process worker, shared-memory mesh) speaks the same wire shape as gRPC, so the prost messages and the metrics codec have to build with the feature off and crate::proto must resolve either way. The plan-metrics collector and the receive-time stamp are the seams such a transport drives directly.
The benchmarks dev-dependency and a few `tonic`-importing test helpers re-unified gRPC into every test build, so nothing exercised the in-process transport with gRPC off. Moving the dataset suites into the benchmarks crate and gating the gRPC-coupled helpers lets `cargo test --no-default-features --lib` run the in-process transport, guarded by a new CI job.
Lets an embedder supply the dispatch bytes per stage instead of the coordinator encoding the plan it holds, whose exec-time state can be wrong to dispatch or not serializable through datafusion-proto.
An out-of-crate driver dispatches a query and drives the head stage itself, so it needs prepare_in_process_plan, the metrics store and a public insert, the partition-routing seam on NetworkBoundary, and a no-gRPC decode_task_metrics to file proto metric frames into the plain store.
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.
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.
The coordinator serializes every stage subplan before handing it to the channel, but the shm transport ships plans over DSM and discards that request. `ShmDiscardedPlanCodec` lets the throwaway encode succeed for nodes DataFusion's proto cannot represent, such as `SortMergeJoinExec`.
The shm fragment path has no task-level stamps, but the decoder requires the field present; emitting an empty set instead of `None` keeps every worker metrics frame decodable so the per-task EXPLAIN rewrite runs.
mdashti
force-pushed
the
moe/shm-on-protocol-grpc
branch
from
992673d to
ea63b5e
Compare
mdashti
force-pushed
the
moe/shm-plan-proto-discard
branch
from
a09c34a to
741f0c1
Compare
stuhood
force-pushed
the
moe/shm-on-protocol-grpc
branch
from
ea63b5e to
aa130c2
Compare
mdashti
force-pushed
the
moe/shm-on-protocol-grpc
branch
from
aa130c2 to
3f1137b
Compare
Collaborator
Author
|
Closing: the passive-worker rework removed both halves of this PR. Workers no longer run the in-process nested dispatch, so there is no discarded |
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.
This PR makes the shm embedder work under the channel-protocol coordinator, which serializes every stage subplan into
plan_protobefore dispatch even though the shm channel ships plans over DSM and drains-and-discards that request.Two changes, both under
src/shm/, so#28stays shm-only:src/shm/plan_codec.rsaddsShmDiscardedPlanCodec, aPhysicalExtensionCodecthe embedder registers viawith_distributed_user_codec. It sits behind the built-inDistributedCodecinnew_combined_with_user, so it is consulted only for nodes DataFusion's proto can't represent (custom scans,SortMergeJoinExec), and encodes each as empty. The bytes are never read, so it refuses to decode.collect_task_metricsreports an empty task-levelMetricsSetinstead ofNone, so the unchangeddecode_task_metricsaccepts the shm path's frames and the per-task EXPLAIN rewrite runs.Surfaced by the
pg_searchMPP regress in paradedb #5448: without the first, aSortMergeJoinExecquery errored mid-dispatch (DataFusion's proto can't encode it and no codec absorbed it); without the second, joins showed no workeroutput_rowsinEXPLAIN ANALYZE(every metrics frame failed to decode, so the rewrite timed out).Stacked on
#28. Draft for review.