Split proto serialization to encapsulate private state (#21835)#21929
Conversation
9302fa9 to
5360d7d
Compare
e05de25 to
2a29583
Compare
|
@timsaucer wonder if you'd be willing to do an initial review on this change since we discussed it in the call yesterday? |
2a29583 to
0e6e8b0
Compare
0e6e8b0 to
32deeb5
Compare
|
I've started reviewing but this is a big one and it will take a bit |
222d25d to
8094825
Compare
I've tried to restack it to clean it up. But yes I know it's big, if you have any ideas for how to split it up or make it easier to review I'm all ears. |
8094825 to
c7d42ab
Compare
timsaucer
left a comment
There was a problem hiding this comment.
I found the description to have the answers I was looking for eventually, but the llm generated text had so much content I didn't want to read it all just to understand what is actually not a very complex refactor. This held me up more than anything. It would have been more helpful to have a more pithy description that got me right to the useful pieces.
I'm finding for my own PRs that I need to make a lot of edits to these generated descriptions because they're not always great at isolating what a reviewer really needs to understand vs cataloging all the work that was done.
| /// Takes the whole [`PhysicalExprNode`] — the exact inverse of what | ||
| /// [`PhysicalExpr::try_to_proto`] produces — so every expression's | ||
| /// `try_from_proto` shares one signature. The operator string is parsed | ||
| /// via the canonical [`Operator::from_proto_name`] mapping, so no `op` | ||
| /// argument needs to be threaded in by the caller. | ||
| /// | ||
| /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode | ||
| /// [`PhysicalExpr::try_to_proto`]: datafusion_physical_expr_common::physical_expr::PhysicalExpr::try_to_proto | ||
| /// [`PhysicalExprDecodeCtx::decode`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx::decode |
There was a problem hiding this comment.
Seems like some generic text that doesn't necessarily need to be in each implementation
| impl From<&CsvOptionsProto> for CsvOptions { | ||
| fn from(proto: &CsvOptionsProto) -> Self { |
There was a problem hiding this comment.
For this and also from_factory probably needs a line in the upgrade doc
There was a problem hiding this comment.
Same argument holds for json options, parquet, etc
There was a problem hiding this comment.
On second thought, from_factory wasn't pub so maybe this argument is moot.
| } | ||
| } | ||
|
|
||
| impl TryFrom<protobuf::WindowFrame> for WindowFrame { |
There was a problem hiding this comment.
TBH Since the methods are so similar it feels like keeping impl TryFrom is better.
There was a problem hiding this comment.
Ok, so I spent way more time than I wanted to understand the problem with the dependencies and why you had to introduce this.
|
LMK if you want me to take another look since you did another push since I reviewed |
|
Thanks so much @timsaucer !
It's the same content, I was just re-organizing the commits to make it easier to review 😄. I think it would be good for @alamb to take a look at this, especially since we can wait to ship it until after 54 has been released (so no big rush).
Sorry about this. I agree with you. I usually leave the LLM generated descriptions up for draft PRs or where they are "fine" and better than "I'm a lazy human so really didn't write much at all" but I do try to rewrite them where it would be helpful (e.g. incidentally I was just rewriting the one in #22300 by hand while you were reviewing this). Just bad timing / judgment on this one. I'll update it for the next reviewer and note the feedback. |
1d5a354 to
89d6cf7
Compare
| fn try_to_proto( | ||
| &self, | ||
| _ctx: &proto_encode::PhysicalExprEncodeCtx<'_>, | ||
| ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> { |
There was a problem hiding this comment.
@timsaucer do you think we should make this Result<...> instead of Result<Option<...>>? If we ever plan to remove the default impl, remove the codecs and force ourselves to implement this method on all expressions we produce then the default returning an error would make sense IMO. The price we pay is that in the meantime it would be hard to differentiate between "I called an expression with an implementation and it error, I should bubble that up to users so they can fix it instead of getting an unknown expression Column type error (by falling back to the match -> codec, etc.)" vs. "this expression hasn't been updated yet".
There was a problem hiding this comment.
I think what we have is fine. We can iterate on the API as we implement more expressions.
89d6cf7 to
0ff65cc
Compare
Part of apache#22435. Adds try_to_proto to HashTableLookupExpr so it participates in the expression-local serialization pattern introduced in apache#21929. HashTableLookupExpr holds a runtime Arc<Map> that cannot be serialized, so try_to_proto replaces it with lit(true). This is safe because the filter is a performance optimisation only — lit(true) passes all rows and the join produces correct results either way. The centralized arm in to_proto.rs remains as a fallback for now.
## Which issue does this PR close? Part of apache#22435 ## What changes are included? Adds `try_to_proto` to `HashTableLookupExpr` so it participates in the expression-local serialization pattern introduced in apache#21929. `HashTableLookupExpr` holds a runtime `Arc<Map>` that cannot be serialized, so `try_to_proto` replaces it with `lit(true)`. This is safe because the filter is a performance optimisation only — `lit(true)` passes all rows and the join produces correct results either way. The centralized arm in `to_proto.rs` remains as a fallback for now. Cleanup can follow in a separate PR once this lands. ## Are these changes tested? Yes — covered by the existing `roundtrip_hash_table_lookup_expr_to_lit` test in `datafusion/proto/tests/cases/roundtrip_physical_plan.rs`. ## Are there any user-facing changes? No. --------- Co-authored-by: Anurag Tryambak Raut <anuragtryambakraut@Anurags-MacBook-Air.local>
…expressions Ports the existing `try_to_proto` / `try_from_proto` implementations onto the helpers introduced in the previous commit (`expect_expr_variant!`) and the wider use of `decode_required_expression` / `decode_children_expressions` / `encode_children_expressions` from apache#22513. Covers every expression already migrated under apache#22418: - `Column`, `BinaryExpr` (originally apache#21929) - `LikeExpr` (apache#22471) - `InListExpr` (apache#22503) - `NegativeExpr` (apache#22483) `BinaryExpr` additionally switches its `l`/`r` legacy-decode arms to `decode_required_expression`, removing two more hand-rolled "missing required field" strings, and runs its `operands` encode/ decode loops through `encode_children_expressions` / `decode_children_expressions`. One existing test changes assertion text — `InListExpr`'s rejected- variant message was the only one using the article "an" instead of the macro's article-free "a"; updated to match. No wire-format change; `cargo test -p datafusion-proto --test proto_integration` is green (173 / 173).
## Which issue does this PR close? - Closes apache#22432. ## Rationale for this change `HashExpr` is part of the physical expression proto cleanup tracked by apache#22418. Its protobuf serialization should live with the expression implementation instead of in the central proto downcast chains, matching the hook pattern added in apache#21929. ## What changes are included in this PR? This adds `HashExpr`'s `PhysicalExpr::try_to_proto` override and a feature-gated inherent `HashExpr::try_from_proto`. The existing `PhysicalHashExprNode` wire shape is preserved, including `expr_id: None`, while `datafusion-proto` now routes `ExprType::HashExpr` decoding through the hook and no longer has a central `HashExpr` serialization arm. `datafusion-physical-plan` now exposes a `proto` feature so the hook code only compiles when proto support is requested. ## Are these changes tested? Yes. Added focused direct hook tests for encoding, decoding, and rejecting a wrong `expr_type`. Commands run: - `cargo fmt --all -- --check` - `cargo check -p datafusion-physical-plan` - `cargo test -p datafusion-physical-plan --features proto hash_expr_try` - `cargo test -p datafusion-proto --test proto_integration cases::roundtrip_physical_plan::roundtrip_hash_expr` - `cargo check -p datafusion-proto` - `cargo clippy -p datafusion-physical-plan --features proto --all-targets -- -D warnings` - `cargo clippy -p datafusion-proto --all-targets -- -D warnings` ## Are there any user-facing changes? No. This preserves the existing protobuf representation. --------- Signed-off-by: Nanook Claw <nanook@agentmail.to> Signed-off-by: Nanook <nanookclaw@users.noreply.github.com> Co-authored-by: Nanook <nanookclaw@users.noreply.github.com> Co-authored-by: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>
Part of apache#22434. Adds try_to_proto / try_from_proto to DynamicFilterPhysicalExpr so it participates in the expression-local serialization pattern introduced in apache#21929. The centralized arms in to_proto.rs / from_proto.rs remain as fallbacks for now. Cleanup of the pub-for-proto scaffolding (from_parts, inner, original_children, remapped_children) can follow once decode reads state directly via try_from_proto.
Part of apache#22434. Adds try_to_proto / try_from_proto to DynamicFilterPhysicalExpr so it participates in the expression-local serialization pattern introduced in apache#21929. The centralized arms in to_proto.rs / from_proto.rs remain as fallbacks for now. Cleanup of the pub-for-proto scaffolding (from_parts, inner, original_children, remapped_children) can follow once decode reads state directly via try_from_proto.
…pshot walk (#62) * [X-2935] proto: preserve nested DynamicFilterPhysicalExpr through snapshot walk `serialize_physical_expr_with_converter` only honored the top-level `DynamicFilterPhysicalExpr` special case; any nested wrapper inside a `BinaryExpr` / `CastExpr` / etc. was folded into its current() snapshot by the unconditional `snapshot_physical_expr(value)` call right after. When a `FilterPushdown(Post)` pass re-pushes a SortExec's self-filter into a FileScan that already carries a static WHERE, the resulting predicate is `BinaryExpr(AND, static_WHERE, dyn_filter)`. The old code serialized this as `BinaryExpr(AND, static_WHERE, lit(true))` on the wire (since `current()` returns `lit(true)` before TopK runs), and the data-server side decoded a fully static predicate with no live link to TopK -- the IncrementalRowGroupPruner saw no `snapshot_generation` movement and never tightened pruning. Replace the unconditional snapshot walker with a `transform_up` that skips `DynamicFilterPhysicalExpr` nodes. Other dynamic exprs (`HashTableLookupExpr`, etc.) still get snapshotted as before; the DynamicFilter wrapper survives to the downcast chain, where the existing line-318 branch serializes it as `PhysicalDynamicFilterNode` at whatever nesting depth. Adds `dynamic_filter_nested_in_binary_expr_survives_proto_roundtrip` covering the `BinaryExpr(static_WHERE, AND, dyn_filter)` shape. Upstream apache#21929 (commit 077f08a, merged 2026-05-22) fixed this structurally by removing the top-level snapshot call and routing every PhysicalExpr through `try_to_proto` hooks. This patch is the minimal port until the atlas DF fork adopts that refactor. * fmt: reorder imports per cargo fmt * address Copilot review: drop hardcoded line number, use Boolean update, drop env label
When I run `./datafusion/proto-models/regen.sh` from repository root off
main, I'm getting a dirty git state like so:
```sh
datafusion (main)$ ./datafusion/proto-models/regen.sh
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.07s
Running `/Users/jeffrey/.cargo_target_cache/debug/gen`
Copying datafusion/proto-models/src/datafusion.rs to datafusion/proto-models/src/generated/prost.rs
datafusion (main)$ git status
On branch main
Your branch is up to date with 'upstream/main'.
Untracked files:
(use "git add <file>..." to include in what will be committed)
datafusion/proto-models/proto/proto_descriptor.bin datafusion/proto-models/src/datafusion.rs datafusion/proto-models/src/datafusion.serde.rs datafusion/proto-models/src/datafusion_common.rs
nothing added to commit but untracked files present (use "git add" to track)
```
Copying over the `.gitignore` from `proto` to fix this
https://github.com/apache/datafusion/blob/2282d23d4ff0af91463b63aa99cd793635ecef8e/datafusion/proto/.gitignore#L1-L5
Related PR:
- apache#21929
…icalExpr` (apache#22452) ## Which issue does this PR close? Part of apache#22434 ## What changes are included? Adds `try_to_proto` and `try_from_proto` to `DynamicFilterPhysicalExpr` so it participates in the expression-local serialization pattern introduced in apache#21929. The centralized arms in `to_proto.rs` / `from_proto.rs` remain as fallbacks for now. Cleanup of the pub-for-proto scaffolding (`from_parts`, `inner`, `original_children`, `remapped_children`) can follow in a separate PR once decode reads state directly. ## Are these changes tested? Yes — all three existing dynamic filter roundtrip tests pass: - `test_dynamic_filter_roundtrip_dedupe` - `test_dynamic_filter_plan_roundtrip_dedupe` - `test_dynamic_filter_expression_id_is_stable_between_serializations` ## Are there any user-facing changes? No. --------- Co-authored-by: Anurag Tryambak Raut <anuragtryambakraut@Anurags-MacBook-Air.local>
Move ScalarSubqueryExpr proto encode/decode into the expression itself via the try_to_proto / try_from_proto hooks, removing the centralized downcast branch in to_proto.rs and the inline construction in from_proto.rs. Because the ScalarSubqueryResults container is runtime-only shared state and not part of the wire format, from_proto.rs still fetches it from the decode context and threads it into try_from_proto. Follows the pattern established in apache#21929.
…ern (apache#23130) Move ScalarSubqueryExpr proto encode/decode into the expression itself via the try_to_proto / try_from_proto hooks, removing the centralized downcast branch in to_proto.rs and the inline construction in from_proto.rs. Because the ScalarSubqueryResults container is runtime-only shared state and not part of the wire format, from_proto.rs still fetches it from the decode context and threads it into try_from_proto. Follows the pattern established in apache#21929. ## Which issue does this PR close? - Closes #[22433](apache#22433). ## Rationale for this change datafusion-proto serializes ScalarSubqueryExpr by downcasting in to_proto.rs and rebuilding it inline in from_proto.rs, which forces its internal fields to be exposed as public accessors just for the proto crate. PR #[21929](apache#21929) introduced a self-serialization pattern (try_to_proto / try_from_proto) where each expression owns its own wire format and the central dispatch just delegates. Other expressions like InListExpr and DynamicFilterPhysicalExpr already use it. This change moves ScalarSubqueryExpr to the same pattern, removing its bespoke proto branches and the proto-only public accessors. One wrinkle: ScalarSubqueryExpr's ScalarSubqueryResults is runtime-only shared state, not part of the wire format, so try_from_proto takes it as an extra argument that from_proto.rs threads in from the decode context. ## What changes are included in this PR? - [x] `try_to_proto` in `impl PhysicalExpr for ScalarSubqueryExpr` (not inherent), `#[cfg(feature = "proto")]` - [x] `ScalarSubqueryExpr::try_from_proto` wired into `from_proto.rs` - [x] central `to_proto.rs` + `from_proto.rs` arms deleted in this PR - [x] direct hook test incl. bad-input case; test module at end of file - [x] roundtrip tests pass; fmt + clippy `-D warnings` clean; PR template filled in ## Are these changes tested? - **Unit tests** (`scalar_subquery.rs`, `#[cfg(all(test, feature = "proto"))]` module at end of file) exercise the hooks directly: - `round_trips_through_proto` — encode via `try_to_proto`, decode via `try_from_proto`, asserting `data_type`/`nullable`/`index` and shared-results identity survive. - `rejects_non_scalar_subquery_node` — wrong `ExprType` variant errors cleanly. - `rejects_missing_data_type` — missing required field errors cleanly. - **Round-trip integration tests** (`datafusion-proto`, `--all-features`) cover the full plan path: `roundtrip_scalar_subquery_exec`, `roundtrip_nested_scalar_subquery_exec_scopes_results`, and `roundtrip_scalar_subquery_exec_with_default_converter_executes`. All pass, and `cargo fmt --all` + `cargo clippy --all-features --all-targets -- -D warnings` are clean. ## Are there any user-facing changes? No --------- Co-authored-by: Matthew Patton <matthewpatton@macbookpro.mynetworksettings.com>
…ern (apache#23130) Move ScalarSubqueryExpr proto encode/decode into the expression itself via the try_to_proto / try_from_proto hooks, removing the centralized downcast branch in to_proto.rs and the inline construction in from_proto.rs. Because the ScalarSubqueryResults container is runtime-only shared state and not part of the wire format, from_proto.rs still fetches it from the decode context and threads it into try_from_proto. Follows the pattern established in apache#21929. ## Which issue does this PR close? - Closes #[22433](apache#22433). ## Rationale for this change datafusion-proto serializes ScalarSubqueryExpr by downcasting in to_proto.rs and rebuilding it inline in from_proto.rs, which forces its internal fields to be exposed as public accessors just for the proto crate. PR #[21929](apache#21929) introduced a self-serialization pattern (try_to_proto / try_from_proto) where each expression owns its own wire format and the central dispatch just delegates. Other expressions like InListExpr and DynamicFilterPhysicalExpr already use it. This change moves ScalarSubqueryExpr to the same pattern, removing its bespoke proto branches and the proto-only public accessors. One wrinkle: ScalarSubqueryExpr's ScalarSubqueryResults is runtime-only shared state, not part of the wire format, so try_from_proto takes it as an extra argument that from_proto.rs threads in from the decode context. ## What changes are included in this PR? - [x] `try_to_proto` in `impl PhysicalExpr for ScalarSubqueryExpr` (not inherent), `#[cfg(feature = "proto")]` - [x] `ScalarSubqueryExpr::try_from_proto` wired into `from_proto.rs` - [x] central `to_proto.rs` + `from_proto.rs` arms deleted in this PR - [x] direct hook test incl. bad-input case; test module at end of file - [x] roundtrip tests pass; fmt + clippy `-D warnings` clean; PR template filled in ## Are these changes tested? - **Unit tests** (`scalar_subquery.rs`, `#[cfg(all(test, feature = "proto"))]` module at end of file) exercise the hooks directly: - `round_trips_through_proto` — encode via `try_to_proto`, decode via `try_from_proto`, asserting `data_type`/`nullable`/`index` and shared-results identity survive. - `rejects_non_scalar_subquery_node` — wrong `ExprType` variant errors cleanly. - `rejects_missing_data_type` — missing required field errors cleanly. - **Round-trip integration tests** (`datafusion-proto`, `--all-features`) cover the full plan path: `roundtrip_scalar_subquery_exec`, `roundtrip_nested_scalar_subquery_exec_scopes_results`, and `roundtrip_scalar_subquery_exec_with_default_converter_executes`. All pass, and `cargo fmt --all` + `cargo clippy --all-features --all-targets -- -D warnings` are clean. ## Are there any user-facing changes? No --------- Co-authored-by: Matthew Patton <matthewpatton@macbookpro.mynetworksettings.com>
Which issue does this PR close?
Rationale for this change
datafusion-protoserializes every built-inPhysicalExprthrough a single~300-line
downcast_refchain, with a mirrormatchon the decode side. Thatchain lives outside the crate where each expression is defined, so every field
an expression wants to round-trip has to be made
pub. #21807 is thecautionary tale: it had to add five
pub"proto-only, not stable" items toDynamicFilterPhysicalExprjust to serialize anRwLock-wrapped inner.This PR adds the infrastructure so a
PhysicalExprcan serialize itself andkeep its state private.
What changes are included in this PR?
A
PhysicalExprcan now opt into serializing itself, in both directions:try_to_protoreturningOk(None)(the default) means "fall through to theold downcast chain", so the change is purely additive — nothing is forced to
migrate.
ColumnandBinaryExprare migrated as working demos; everythingelse stays on the old path and migrates later, one expression at a time, with
no wire-format change.
Five stacked commits, each builds green on its own and is independently
reviewable (or splittable into its own PR):
datafusion-proto-modelscrate — move the.protofile andprost-generated types into a lightweight crate (mirrors the existing
datafusion-proto-commonsplit).try_to_protohook — feature-gated, off by default.Columnencode.Columndecode.BinaryExpr(both directions).A few design decisions worth flagging
FromProto/TryFromPrototraits instead of plainFrom/TryFrom.Once the prost types move into their own crate they are foreign to
datafusion-proto, and the orphan rule forbidsimpl From<&protobuf::X> for Ywhen both
XandYare foreign. So those conversions becomeFromProto/TryFromPrototraits indatafusion_proto::convert, and callers go from(&x).into()toY::from_proto(&x). This is a known workaround, not the endstate — see Future work.
&dyn.PhysicalExprEncodeCtx/PhysicalExprDecodeCtxwrap a sealed dispatch trait. Keeping them concretekeeps
&dynout of every expression's signature and gives a stable place toadd helpers (UDF encoding, registry hooks) later without churning a public
trait.
try_from_prototakes the wholePhysicalExprNode, not thepre-unwrapped variant payload, so every expression's decoder has the same
signature and can still see outer-node fields like
expr_id.Are these changes tested?
No new behavior, so no new tests.
ColumnandBinaryExprproduce and consumethe same wire format as before; the existing
roundtrip_physical_plan/roundtrip_physical_exprtests already cover both directions and now exercisethe new path.
Are there any user-facing changes?
Small API breaks in
datafusion-proto:try_from_physical_plan_with_converter/try_into_physical_plan_with_convertermove to a
PhysicalPlanNodeExttrait — callers adduse datafusion_proto::physical_plan::PhysicalPlanNodeExt;.From/TryFromconversions becomeFromProto/TryFromProto(see Design decisions above).datafusion_proto::generated::*is deprecated in favor ofdatafusion_proto::protobuf; it still works.The new
protofeature ondatafusion-physical-expr(-common)is off bydefault, so crates that don't serialize plans pay nothing.
Future work
DynamicFilterPhysicalExpr, the original motivation — one per follow-up PR.ExecutionPlanserialization.FromProto/TryFromProtoworkaround: collapsedatafusion-proto-commonintodatafusion-proto-modelsand push theconversion impls down to the target-type crates so callers use plain
From/TryFromagain. Full dep-graph analysis and a step-by-step plan are in#21835 (comment).
🤖 Generated with Claude Code