[X-2935] proto: preserve nested DynamicFilterPhysicalExpr through snapshot walk

[zhuqi-lucas]

Summary

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 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. IncrementalRowGroupPruner saw no snapshot_generation movement and never tightened pruning.

What this PR changes

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 branch serializes it as PhysicalDynamicFilterNode at whatever nesting depth.

Adds a regression test dynamic_filter_nested_in_binary_expr_survives_proto_roundtrip covering the BinaryExpr(static_WHERE, AND, dyn_filter) shape. Pins that the decoded right-side stays a live DynamicFilterPhysicalExpr (not a Literal) and that an update() on it bumps the tree-wide snapshot_generation.

Affects

• Atlas distributed pipeline once we bump the fork rev. The X-2935 regression was observed and debugged on the ny2 staging cluster; primary clusters carrying the same code path would hit the same issue.
• Anyone serializing a PhysicalExpr tree that contains a nested DynamicFilterPhysicalExpr — wire output now carries PhysicalDynamicFilterNode instead of a Literal snapshot.

Upstream reference

apache#21929 (commit 077f08a9a, merged 2026-05-22) fixed this structurally by removing the top-level snapshot call entirely and routing every PhysicalExpr through try_to_proto hooks. This patch is the minimal port until the atlas DF fork adopts that refactor.

Validation

• cargo test -p datafusion-proto --test proto_integration dynamic_filter → 5/5 pass (4 existing + 1 new)
• Verified end-to-end against the atlas tree locally via a [patch] block: the atlas-side test_dedup_post_filterpushdown_rerun_preserves_shared_inner_e2e flips from FAIL to PASS purely with this patch applied (no atlas-local workaround needed).

[Copilot]

Pull request overview

This PR fixes distributed physical-expression protobuf serialization so that DynamicFilterPhysicalExpr wrappers are preserved even when nested inside other expressions (e.g., BinaryExpr(AND, static_where, dyn_filter)), rather than being snapshotted into their current() literal and losing the live update link required for pruning improvements downstream.

Changes:

• Update serialize_physical_expr_with_converter to snapshot dynamic expressions via a transform_up walk that skips DynamicFilterPhysicalExpr nodes, allowing them to reach the dedicated PhysicalDynamicFilterNode serialization path at any depth.
• Add a regression test ensuring a nested DynamicFilterPhysicalExpr survives a proto roundtrip and that update() bumps the tree-wide snapshot_generation.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.

File	Description
datafusion/proto/src/physical_plan/to_proto.rs	Reworks snapshotting during serialization to preserve nested DynamicFilterPhysicalExpr while still snapshotting other dynamic exprs.
datafusion/proto/tests/cases/roundtrip_physical_plan.rs	Adds a regression test for the nested dynamic-filter-in-BinaryExpr roundtrip and generation bump behavior.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.