[X-2935] Cherry-pick missing dedup with_new_children block from apache#21807

When this fork ported apache#21807 in commit 47f263d, the
'with_new_children on cache hit' block in DeduplicatingDeserializer
was accidentally omitted. The simpler 'return Ok(Arc::clone(cached))'
version we landed is correct only when every occurrence of a given
Inner has the same outer wrapper shape.

In production this fails: FilterPushdown clones a SortExec's
DynamicFilterPhysicalExpr and rewrites its children's column refs to
match the downstream FileScan's file schema. Both outer wrappers
share the same Inner Arc (same expression_id, same wire expr_id) but
DIFFERENT children. On decode the simpler cache hit returns the
first decode's outer wrapper, silently discarding the second
occurrence's children. prune_by_statistics then resolves column refs
against the wrong positions and pruning becomes a no-op.

Observed end-to-end on ny2 staging 2026-06-15 after walker removal:
  row_groups_pruned_statistics=0 total
  bytes_scanned=91 MB
  time_elapsed_processing=31 s
  (was 1.99K pruned / 136 KB / ~100ms with the walker still active)

Cherry-picks the missing block: parse the proto body first, then on
cache hit return Arc::clone(cached).with_new_children(parsed.children()).
This keeps the cached Inner (so TopK heap-max updates propagate) but
installs the proto body's children on a fresh outer wrapper (so each
occurrence keeps its own column refs).

Adds a regression test that fails without the fix and passes with it,
with assertion messages pointing at the exact root cause.

Author: zhuqi-lucas

datafusion/proto/src/physical_plan/mod.rs

@@ -4072,24 +4072,49 @@ impl PhysicalProtoConverterExtension for DeduplicatingDeserializer {
     where
         Self: Sized,
     {
-        if let Some(expr_id) = proto.expr_id {
-            // Check cache first
-            if let Some(cached) = self.cache.borrow().get(&expr_id) {
-                return Ok(Arc::clone(cached));
-            }
-            // Deserialize and cache
-            let expr = parse_physical_expr_with_converter(
+        let Some(expr_id) = proto.expr_id else {
+            return parse_physical_expr_with_converter(
                 proto,
                 ctx,
                 input_schema,
                 codec,
                 self,
-            )?;
-            self.cache.borrow_mut().insert(expr_id, Arc::clone(&expr));
-            Ok(expr)
-        } else {
-            parse_physical_expr_with_converter(proto, ctx, input_schema, codec, self)
+            );
+        };
+
+        // Always parse the proto body first. The cache hit path below uses
+        // `parsed.children()` (which carry this occurrence's column refs as
+        // they appeared in the proto -- e.g. file-schema indices after
+        // FilterPushdown rewrote them), so we can't short-circuit before
+        // parsing.
+        let parsed =
+            parse_physical_expr_with_converter(proto, ctx, input_schema, codec, self)?;
+
+        let mut cache = self.cache.borrow_mut();
+        if let Some(cached) = cache.get(&expr_id) {
+            // Since expressions may manage their own internal state when
+            // deriving expressions via `with_new_children`, we use
+            // `with_new_children` to opt into the same behavior.
+            //
+            // For example, one `DynamicFilterPhysicalExpr` may be derived
+            // from another resulting in shared references (e.g. a SortExec
+            // dynamic filter at the parent schema and a pushed-down FileScan
+            // predicate at the file schema, sharing the same `Inner` Arc
+            // but with different children). Using `with_new_children`
+            // preserves those references: the cached `Arc<Inner>` is kept,
+            // while this occurrence's children (parsed from the proto body)
+            // are installed onto a fresh outer wrapper.
+            //
+            // Ported from apache/datafusion#21807; without this the cache
+            // hit path returned the entire cached outer Arc and silently
+            // discarded the proto body's children, breaking column ref
+            // alignment in plans where FilterPushdown rewrote them.
+            let children: Vec<_> = parsed.children().into_iter().cloned().collect();
+            return Arc::clone(cached).with_new_children(children);
         }
+
+        cache.insert(expr_id, Arc::clone(&parsed));
+        Ok(parsed)
     }
 
     fn physical_expr_to_proto(

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -3215,22 +3215,32 @@ fn dynamic_filter_dedup_with_deduplicating_codec() -> Result<()> {
     let d1 = deserializer.proto_to_physical_expr(&proto1, &ctx, &schema, &codec)?;
     let d2 = deserializer.proto_to_physical_expr(&proto2, &ctx, &schema, &codec)?;
 
-    assert!(
-        Arc::ptr_eq(&d1, &d2),
-        "DeduplicatingDeserializer must return the same Arc for two refs with the same expr_id"
-    );
-
+    // d1 and d2 may be distinct OUTER Arcs because the cache-hit path
+    // rewraps via `with_new_children` to preserve each occurrence's
+    // children. The invariant that matters is shared INNER (same
+    // expression_id, TopK updates visible to both).
     let d1_df = d1
         .as_any()
         .downcast_ref::<DynamicFilterPhysicalExpr>()
         .expect(
             "decoded expr must be DynamicFilterPhysicalExpr; snapshot path is bypassed",
         );
+    let d2_df = d2
+        .as_any()
+        .downcast_ref::<DynamicFilterPhysicalExpr>()
+        .expect(
+            "decoded expr must be DynamicFilterPhysicalExpr; snapshot path is bypassed",
+        );
     assert_eq!(
         d1_df.inner().expression_id,
         id_before,
         "expression_id must survive proto roundtrip"
     );
+    assert_eq!(
+        d1_df.inner().expression_id,
+        d2_df.inner().expression_id,
+        "DeduplicatingDeserializer must share Inner for two refs with the same expr_id"
+    );
 
     Ok(())
 }
@@ -3300,9 +3310,152 @@ fn dynamic_filter_dedup_distinct_outer_arcs_same_inner() -> Result<()> {
     let deserializer = DeduplicatingDeserializer::new();
     let d1 = deserializer.proto_to_physical_expr(&proto1, &ctx, &schema, &codec)?;
     let d2 = deserializer.proto_to_physical_expr(&proto2, &ctx, &schema, &codec)?;
-    assert!(
-        Arc::ptr_eq(&d1, &d2),
-        "Distinct-outer same-Inner refs must reconstruct to one Arc"
+    // d1 and d2 may be distinct OUTER Arcs because the cache-hit path
+    // rewraps via `with_new_children` to preserve each occurrence's
+    // children. The invariant that matters is shared INNER.
+    let d1_inner_id = d1
+        .as_any()
+        .downcast_ref::<DynamicFilterPhysicalExpr>()
+        .unwrap()
+        .inner()
+        .expression_id;
+    let d2_inner_id = d2
+        .as_any()
+        .downcast_ref::<DynamicFilterPhysicalExpr>()
+        .unwrap()
+        .inner()
+        .expression_id;
+    assert_eq!(
+        d1_inner_id, d2_inner_id,
+        "Distinct-outer same-Inner refs must share Inner after decode"
+    );
+    assert_eq!(d1_inner_id, expected_id);
+
+    Ok(())
+}
+
+/// Two outer `Arc<DynamicFilterPhysicalExpr>` sharing the same Inner but
+/// carrying DIFFERENT children must roundtrip to two outer Arcs whose
+/// children match their original occurrence's children. This is the shape
+/// `FilterPushdown` actually produces: it clones a SortExec's dyn filter and
+/// rewrites the children column refs to match the FileScan's schema (e.g.
+/// `ticker@0` at SortExec input → `ticker@12` at file schema). Both outer
+/// wrappers share the same `Inner` Arc (so TopK heap-max updates propagate)
+/// but the column-ref indices differ.
+///
+/// Without `DeduplicatingDeserializer::proto_to_physical_expr` applying
+/// `with_new_children(parsed.children())` on cache hits, the second decode
+/// silently returns the FIRST occurrence's children and discards the proto
+/// body's children -- `prune_by_statistics` then resolves column refs against
+/// the wrong file-schema positions and pruning becomes a no-op. Observed on
+/// X-2935 ny2 staging 2026-06-15: `row_groups_pruned_statistics=0`,
+/// `bytes_scanned=91 MB`, `time_elapsed_processing=31 s`.
+#[test]
+fn dynamic_filter_dedup_distinct_children_via_with_new_children() -> Result<()> {
+    use datafusion::physical_plan::expressions::Column;
+    use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr;
+    use datafusion_proto::physical_plan::{
+        DeduplicatingDeserializer, DeduplicatingSerializer,
+        PhysicalProtoConverterExtension,
+    };
+
+    // Schema with the SAME column name at TWO positions, mimicking a
+    // FilterPushdown step where SortExec sees `a@0` (input schema) but the
+    // FileScan below it has `a@1` (file schema after some projection).
+    let schema = Schema::new(vec![
+        Field::new("extra", DataType::Int64, false), // physical position 0
+        Field::new("a", DataType::Int64, false),     // physical position 1
+    ]);
+
+    let initial: Arc<dyn PhysicalExpr> = lit(true);
+    let col_a_idx0: Arc<dyn PhysicalExpr> = Arc::new(Column::new("a", 0));
+    let col_a_idx1: Arc<dyn PhysicalExpr> = Arc::new(Column::new("a", 1));
+
+    // outer A: SortExec.filter-style (children at index 0).
+    let df_arc1: Arc<dyn PhysicalExpr> = Arc::new(DynamicFilterPhysicalExpr::new(
+        vec![Arc::clone(&col_a_idx0)],
+        Arc::clone(&initial),
+    ));
+    // outer B: FileScan.predicate-style (children at index 1) sharing Inner.
+    let df_arc2 =
+        Arc::clone(&df_arc1).with_new_children(vec![Arc::clone(&col_a_idx1)])?;
+    assert_eq!(
+        df_arc1
+            .as_any()
+            .downcast_ref::<DynamicFilterPhysicalExpr>()
+            .unwrap()
+            .inner()
+            .expression_id,
+        df_arc2
+            .as_any()
+            .downcast_ref::<DynamicFilterPhysicalExpr>()
+            .unwrap()
+            .inner()
+            .expression_id,
+        "shared Inner must produce identical expression_id"
+    );
+
+    let codec = DefaultPhysicalExtensionCodec {};
+    let serializer = DeduplicatingSerializer::new();
+    let proto1 = serializer.physical_expr_to_proto(&df_arc1, &codec)?;
+    let proto2 = serializer.physical_expr_to_proto(&df_arc2, &codec)?;
+    assert_eq!(
+        proto1.expr_id, proto2.expr_id,
+        "same Inner must stamp same wire expr_id"
+    );
+
+    let ctx = SessionContext::new().task_ctx();
+    let deserializer = DeduplicatingDeserializer::new();
+    let d1 = deserializer.proto_to_physical_expr(&proto1, &ctx, &schema, &codec)?;
+    let d2 = deserializer.proto_to_physical_expr(&proto2, &ctx, &schema, &codec)?;
+
+    // Inner sharing: an update via d1 must be visible from d2 (and vice
+    // versa). Without inner-level dedup, d1 and d2 would each have their
+    // own Inner and updates wouldn't propagate.
+    let d1_df = d1
+        .as_any()
+        .downcast_ref::<DynamicFilterPhysicalExpr>()
+        .expect("d1 must remain a DynamicFilterPhysicalExpr after roundtrip");
+    let d2_df = d2
+        .as_any()
+        .downcast_ref::<DynamicFilterPhysicalExpr>()
+        .expect("d2 must remain a DynamicFilterPhysicalExpr after roundtrip");
+    assert_eq!(
+        d1_df.inner().expression_id,
+        d2_df.inner().expression_id,
+        "Inner must be shared (same expression_id) after deduplicating decode"
+    );
+
+    // The bug: each outer wrapper must carry its OCCURRENCE's children, not
+    // get clobbered by whichever side decoded first. d1's child Column index
+    // must be 0; d2's child Column index must be 1.
+    fn child_column_index(expr: &Arc<dyn PhysicalExpr>) -> usize {
+        let dyn_filter = expr
+            .as_any()
+            .downcast_ref::<DynamicFilterPhysicalExpr>()
+            .unwrap();
+        dyn_filter
+            .children()
+            .first()
+            .unwrap()
+            .as_any()
+            .downcast_ref::<Column>()
+            .unwrap()
+            .index()
+    }
+    assert_eq!(
+        child_column_index(&d1),
+        0,
+        "decoded d1 must carry its OCCURRENCE's children (Column at index 0); \
+         if this is 1, dedup-by-Inner is missing and the cache hit clobbered \
+         the children with d2's"
+    );
+    assert_eq!(
+        child_column_index(&d2),
+        1,
+        "decoded d2 must carry its OCCURRENCE's children (Column at index 1); \
+         if this is 0, dedup-by-Inner is missing and the cache hit returned \
+         d1's children unchanged"
     );
 
     Ok(())