refactor(proto): destructure plan and proto structs in core plan serde hooks

[adriangb]

Which issue does this PR close?

• Follow-up cleanup for EPIC [EPIC] Port ExecutionPlan serialization to try_to_proto / try_from_proto hooks #23494; does not close an issue on its own.

Rationale for this change

EPIC #23494 moved every built-in ExecutionPlan off the central downcast_ref
chain in datafusion-proto onto per-plan hooks: a try_to_proto override on
the ExecutionPlan trait and an inherent FooExec::try_from_proto, both living
in the plan's own module.

Those hooks currently read plan state through getters (self.expr(),
self.fetch(), ...). That means adding a field to a plan struct is invisible to
serialization: nothing breaks, nothing warns, the field is just silently
dropped on round-trip. This is not hypothetical — it has already cost us a real
bug: HashJoinExec.fetch is not serialized and is lost on round-trip (being
fixed separately).

Starting each hook with an exhaustive destructure turns that class of mistake
into a compile error:

• Encode side — destructure self naming every field, no ... Add a field
  to SortExec and SortExec::try_to_proto stops compiling until you either
  serialize it or bind it to _ with a comment saying why it does not need to
  be (derived at construction, runtime state, recomputed on decode).
• Decode side — destructure the prost node struct the same way. The
  generated structs are plain, all-pub and not #[non_exhaustive], so adding
  a field to the .proto becomes a compile error in every decoder rather than a
  silent omission.

What changes are included in this PR?

A mechanical, behavior-preserving refactor of the serde hooks for the core
(non-join, non-aggregate, non-window) plans in datafusion/physical-plan/src/:

SortExec, SortPreservingMergeExec, GlobalLimitExec, LocalLimitExec,
FilterExec, ProjectionExec, RepartitionExec, UnionExec,
InterleaveExec, CoalesceBatchesExec, CoalescePartitionsExec,
CooperativeExec, BufferExec, EmptyExec, PlaceholderRowExec,
ExplainExec, ScalarSubqueryExec.

The wire format is unchanged — byte for byte — and no behavior changes. The
same values are written to and read from the same proto fields; only the way
they are reached in Rust changes.

Split into four commits, one per group of plans, each of which builds green.

Fields the refactor documented as not serialized

Writing the destructures surfaced and pinned down, in comments at the point of
use, which fields do not survive a round-trip today. All of these are
pre-existing; none of them are changed here, because fixing any of them
means changing the wire format, which is a separate decision:

• SortPreservingMergeExec::enable_round_robin_repartition — no field on
  SortPreservingMergeExecNode; decoding always restores the true default
  from SortPreservingMergeExec::new.
• GlobalLimitExec::required_ordering and LocalLimitExec::required_ordering —
  no field on the limit nodes; they are set by the enforce_sorting optimizer
  rule, so a decoded plan starts with None.

Everything else that binds to _ is genuinely not plan shape: runtime metrics
(metrics / metrics_set), execution-time state (RepartitionExec::state,
ScalarSubqueryExec::subquery_future and results), values derived at
construction from other serialized fields (SortExec::common_sort_prefix), the
positional ScalarSubqueryLink::index, and the cache / properties plan
properties that are recomputed on decode.

Two small notes for reviewers:

• RepartitionExec keeps its output partitioning inside cache
  (partitioning() is &self.cache.partitioning), so cache is bound rather
  than _-ed there and is read for the serialized partitioning field.
• EmptyExec / PlaceholderRowExec now read the encoded partition count from
  the partitions field instead of going through
  properties().output_partitioning().partition_count(). The two are kept in
  sync by with_partitions, the only writer besides new, so the encoded value
  is identical.

Are these changes tested?

Covered by the existing round-trip tests, which are the real proof that the wire
format did not move. Locally on this branch:

• cargo test -p datafusion-proto --test proto_integration — 214 passed, 0 failed
• cargo test -p datafusion-physical-plan — 1640 passed, 0 failed
• cargo clippy -p datafusion-physical-plan --all-targets --all-features -- -D warnings — clean
• cargo fmt --all

No new tests are added here: the change adds no new behavior to test, and the
compile-error property it introduces is enforced by rustc rather than by a test.

On the comprehensiveness of that existing coverage (thanks @andygrove for
pushing on this — the original wording oversold it):

• SortPreservingMergeExec, updated by this PR, has no round-trip test at
  all — SortPreservingMerge does not appear anywhere in
  roundtrip_physical_plan.rs. The other 16 plans touched here each have at
  least one dedicated roundtrip_* test.
• Separately, roundtrip_test asserts on format!("{plan:?}"), so it cannot
  catch a dropped field that a plan's Debug impl doesn't print. That is
  precisely how the HashJoinExec::fetch bug (fix(proto): preserve HashJoinExec fetch across serialization #24165) went unnoticed.

Both are now tracked in #24171, with the test work split out into its own PR so
this mechanical refactor and the coverage improvements can be reviewed
independently. Sequencing is up to reviewers — happy to land the tests first if
that's preferred.

Are there any user-facing changes?

No. No public API changes, no wire format changes, no behavior changes.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 83.73206% with 34 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.05%. Comparing base (92f4e8f) to head (f150f91).

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/limit.rs	81.81%	0 Missing and 4 partials ⚠️
datafusion/physical-plan/src/filter.rs	90.32%	0 Missing and 3 partials ⚠️
datafusion/physical-plan/src/projection.rs	80.00%	0 Missing and 3 partials ⚠️
datafusion/physical-plan/src/repartition/mod.rs	80.00%	0 Missing and 3 partials ⚠️
datafusion/physical-plan/src/scalar_subquery.rs	70.00%	1 Missing and 2 partials ⚠️
datafusion/physical-plan/src/buffer.rs	75.00%	0 Missing and 2 partials ⚠️
datafusion/physical-plan/src/coalesce_batches.rs	86.66%	0 Missing and 2 partials ⚠️
datafusion/physical-plan/src/coop.rs	66.66%	0 Missing and 2 partials ⚠️
datafusion/physical-plan/src/empty.rs	75.00%	0 Missing and 2 partials ⚠️
datafusion/physical-plan/src/explain.rs	85.71%	0 Missing and 2 partials ⚠️
... and 5 more

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24167      +/-   ##
==========================================
- Coverage   81.05%   81.05%   -0.01%     
==========================================
  Files        1107     1107              
  Lines      381555   381617      +62     
  Branches   381555   381617      +62     
==========================================
+ Hits       309277   309316      +39     
- Misses      54020    54037      +17     
- Partials    18258    18264       +6     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[adriangb]

@andygrove might I bother you to help evaluate this change since you had to deal with this bug before and this should prevent it going forward?

Not tracked yet but I also want to look into running our SLT tests under a proto roundtrip like we do for substrait, which I think will close the loop on proto serde bugs.

[andygrove]

Thanks @adriangb. This looks good overall. The PR description says that the changes in this PR are tested by existing roundtrip tests, but I'm not sure that those are comprehensive? For example, I don't see a roundtrip test for SortPreservingMergeExec whcih is updated in this PR.

[adriangb]

Thanks @adriangb. This looks good overall. The PR description says that the changes in this PR are tested by existing roundtrip tests, but I'm not sure that those are comprehensive? For example, I don't see a roundtrip test for SortPreservingMergeExec whcih is updated in this PR.

Good point. I'll check and add tests that are missing.

[andygrove]

(LLM-assisted review — I used one to trace the required_ordering → preserve_order chain, so worth a sanity check on the specifics.)

The two fields you documented as not-serialized aren't equal in severity, and I think required_ordering deserves its own issue rather than just a comment.

GlobalLimitExec/LocalLimitExec::required_ordering is load-bearing. enforce_sorting (enforce_sorting/mod.rs:518,522) sets it precisely when it deletes a redundant SortExec with a fetch — it's the only surviving record that the limit is order-sensitive. limit_pushdown.rs:420,428 reads it into preserve_order, which reaches FileScanConfig.preserve_order, whose doc says files must be read in exact order "to produce correct results (e.g. for ORDER BY ... LIMIT)".

It doesn't bite today because FileScanConfigBuilder::build() re-derives preserve_order = preserve_order || !output_ordering.is_empty() (file_scan_config/mod.rs:548) and output_ordering is serialized, so a plan with the limit already pushed into the scan self-heals. But FileScanConfig::with_preserve_order (:1153) is a struct literal that bypasses the builder, so a false actively clears it. Decode a plan that still has the limit node, re-run the optimizer, and ORDER BY a LIMIT 10 can return the wrong ten rows. That's a plausible path for anything that re-plans per stage.

enable_round_robin_repartition is milder but also not just perf — per its own docs, false means a stable merge, and the lossy direction restores the true default. Nothing in-tree sets it today, so low priority, but "results differ only among tied keys, only after a round-trip" is a nasty one to debug if a downstream consumer starts using it.

Nothing blocking — the getter→field substitutions elsewhere look equivalent, including the dynamic_expressions_produced and EmptyExec.partitions rewrites.

[adriangb]

Edit: Claude posted a sloppy comment (even though I keep telling it not to do that). I edited it by hand now. Sorry about that.

Thanks @andygrove.

SortPreservingMergeExec has no round-trip test at all. The other 16 plans in this PR do seem to have coverage, so it seems isolated, but the claim in the PR description was wrong (edited now).

That said there is a more structural problem: roundtrip_test asserts on format!("{plan:?}"), so it can't detect a dropped field that a plan's Debug impl doesn't print. That's exactly how the HashJoinExec::fetch bug in #24165 survived. I've filed #24171 for both, and I'm splitting the test work into its own PR based on main rather than adding it here, so this mechanical destructuring change stays reviewable on its own.

Up to you if we should move forward with the work in parallel or if you'd rather gate this refactor on improved testing.

[andygrove]

Up to you if we should move forward with the work in parallel or if you'd rather gate this refactor on improved testing.

I'm fine with filing follow-on issues and keep moving forward

[adriangb]

@andygrove I opened #24173 to track required_ordering missing, lots of bugs being uncovered 😅. It seems then nothing should block this PR and we can move forward with it and then do the followups.