feat: serialize ASOF join plans - #23832
Conversation
github-actions
Bot
added
sql
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #23832 +/- ##
==========================================
- Coverage 81.93% 81.86% -0.08%
==========================================
Files 1133 1133
Lines 423529 424058 +529
Branches 423529 424058 +529
==========================================
+ Hits 347032 347148 +116
- Misses 55907 56296 +389
- Partials 20590 20614 +24 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
19940d8 to
6ae0de3
Compare
6ae0de3 to
54f3afb
Compare
github-actions
Bot
removed
physical-expr
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
github-actions
Bot
removed
sql
|
@2010YOUY01 & @jayzhan211, let's go! |
jayzhan211
left a comment
jayzhan211
left a comment
There was a problem hiding this comment.
Thanks @Xuanwo! One non-blocking suggestion:
The two new serialization hooks read their fields by direct access rather than
destructuring, which skips the guard every other self-serializing plan in
physical-plan uses. From nested_loop_join.rs:
Destructure exhaustively (no
..) so that a newly added proto field
is a compile error here instead of being silently ignored.
That pattern is in cross_join.rs, hash_join/exec.rs, nested_loop_join.rs,
piecewise_merge_join/exec.rs, sort_merge_join/exec.rs and
symmetric_hash_join.rs — 49 sites in total, on both the encode and decode side.
Nothing is dropped today: I checked all 12 AsOfJoinExec fields and all 7
AsOfJoinExecNode fields against the diff and the coverage is complete. The
concern is the next change. The ASOF stack is still landing, and when a filter
or null_equality is added to AsOfJoinExec, this code keeps compiling and
starts emitting plans that decode into a different — but still executable —
join on the remote node. Silent field loss is the failure mode this convention
exists to prevent, so I'd like it applied here before merge.
Encode side (asof_join.rs:557)
#[cfg(feature = "proto")]
fn try_to_proto(
&self,
ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
use datafusion_proto_models::protobuf;
+ // Destructure exhaustively (no `..`) so that a newly added field is a
+ // compile error here instead of being silently ignored.
+ let Self {
+ left,
+ right,
+ on,
+ match_condition,
+ projection,
+ // derived from the children's schemas by `try_new` on decode
+ join_schema: _,
+ // derived from the children's schemas by `try_new` on decode
+ column_indices: _,
+ // runtime metrics, not part of the plan
+ metrics: _,
+ // recomputed from `on` and `match_condition.op` by `try_new`
+ left_ordering: _,
+ // recomputed from `on` and `match_condition.op` by `try_new`
+ right_ordering: _,
+ // right input collected at execution time, not part of the plan
+ right_fut: _,
+ // recomputed by `try_new` on decode
+ cache: _,
+ } = self;
+
- let left = ctx.encode_child(&self.left)?;
- let right = ctx.encode_child(&self.right)?;
- let on = self
- .on
+ let left = ctx.encode_child(left)?;
+ let right = ctx.encode_child(right)?;
+ let on = on
.iter()…and the remaining self.match_condition / self.projection uses become
match_condition / projection.
Decode side (asof_join.rs:621), after the expect_plan_variant!
let asof_join = crate::expect_plan_variant!(
node,
protobuf::physical_plan_node::PhysicalPlanType::AsOfJoin,
"AsOfJoinExec",
);
+ // Destructure exhaustively (no `..`) so that a newly added proto field
+ // is a compile error here instead of being silently ignored.
+ let protobuf::AsOfJoinExecNode {
+ left,
+ right,
+ on,
+ left_match_expr,
+ right_match_expr,
+ match_operator,
+ projection,
+ } = &**asof_join;
+
- let left =
- ctx.decode_required_child(asof_join.left.as_deref(), "AsOfJoinExec", "left")?;
+ let left = ctx.decode_required_child(left.as_deref(), "AsOfJoinExec", "left")?;with the later asof_join.* accesses updated to the bound names.
|
Thank you @jayzhan211 for the review. I have updated this PR and fired #25214 🤓 |
3 participants
Which issue does this PR close?
Rationale for this change
This is the serialization layer of the ASOF JOIN stack. It gives logical and
physical ASOF plans explicit protobuf representations without coupling wire
format review to SQL or DataFrame APIs.
#23829 and its prerequisites are merged. After restacking onto the current
main, this PR now contains only the serialization layer.What changes are included in this PR?
join constraint, and right output indices.
ExecutionPlan::try_to_protoandAsOfJoinExec::try_from_proto, following the current self-serializingexecution-plan pattern.
prostandpbjsonsources with the repository generator. Thephysical-plan oneof uses the next append-only tag after
PiecewiseMergeJoinExec.AsOfJoinExecround trip.Are these changes tested?
Yes:
./datafusion/proto-models/regen.shcargo fmt --allcargo clippy --all-targets --all-features -- -D warnings./dev/rust_lint.shcargo test -p datafusion-proto roundtrip_asof_join --all-featuresAre there any user-facing changes?
Logical and physical ASOF join plans can be serialized through
datafusion-proto. The additions use new messages and append-only oneof/enumtags, so existing wire tags are not reused. Generated public Rust enums gain new
variants, however, so downstream exhaustive matches must add arms; this is a
Rust source-compatibility break even though the wire additions are compatible.
This PR can now be reviewed independently. It does not depend on the optional
floating-point follow-up #24375.