fix: adapt cherry-picked proto serialization PRs for branch-52

Fixups for the cherry-picked commits from PRs apache#19437, apache#20037, apache#20416,
and #2 to work with branch-52's partition-index APIs:

- Update remap_children callers to use instance method signature
- Adapt DynamicFilterUpdate::Global enum for new code paths
- Add missing partitioned_exprs/runtime_partition fields to new constructors
- Remove null_aware field (not on branch-52)
- Replace FilterExecBuilder with FilterExec::try_new
- Remove non-compiling tests that depend on upstream-only APIs
- Fix duplicate imports in roundtrip test file

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: jayshrivastava

datafusion/physical-expr/src/expressions/dynamic_filters.rs

@@ -223,10 +223,12 @@ impl From<DynamicFilterSnapshot> for DynamicFilterPhysicalExpr {
                 generation,
                 expr: inner_expr,
                 is_complete,
+                partitioned_exprs: vec![],
             })),
             state_watch,
             data_type: Arc::new(RwLock::new(None)),
             nullable: Arc::new(RwLock::new(None)),
+            runtime_partition: None,
         }
     }
 }
@@ -373,6 +375,7 @@ impl DynamicFilterPhysicalExpr {
             state_watch: self.state_watch.clone(),
             data_type: Arc::clone(&self.data_type),
             nullable: Arc::clone(&self.nullable),
+            runtime_partition: None,
         })
     }
 
@@ -414,7 +417,7 @@ impl DynamicFilterPhysicalExpr {
             self.current_for_partition(partition)
         } else {
             let expr = Arc::clone(self.inner.read().expr());
-            Self::remap_children(&self.children, self.remapped_children.as_ref(), expr)
+            self.remap_children(expr)
         }
     }
 
@@ -439,11 +442,7 @@ impl DynamicFilterPhysicalExpr {
             // and the same externally facing `PhysicalExpr` is used for both
             // `with_new_children` and `update()`.
             DynamicFilterUpdate::Global(new_expr) => {
-                DynamicFilterUpdate::Global(Self::remap_children(
-                    &self.children,
-                    self.remapped_children.as_ref(),
-                    new_expr,
-                )?)
+                DynamicFilterUpdate::Global(self.remap_children(new_expr)?)
             }
             DynamicFilterUpdate::Partitioned(partition_exprs) => {
                 DynamicFilterUpdate::Partitioned(partition_exprs)
@@ -502,21 +501,13 @@ impl DynamicFilterPhysicalExpr {
         if guard.partitioned_exprs.is_empty() {
             let expr = Arc::clone(guard.expr());
             drop(guard);
-            return Self::remap_children(
-                &self.children,
-                self.remapped_children.as_ref(),
-                expr,
-            );
+            return self.remap_children(expr);
         }
         match guard.partitioned_exprs.get(partition) {
             Some(Some(expr)) => {
                 let expr = Arc::clone(expr);
                 drop(guard);
-                Self::remap_children(
-                    &self.children,
-                    self.remapped_children.as_ref(),
-                    expr,
-                )
+                self.remap_children(expr)
             }
             Some(None) => Ok(lit(false) as Arc<dyn PhysicalExpr>),
             None => Ok(lit(true) as Arc<dyn PhysicalExpr>),

datafusion/physical-plan/src/aggregates/mod.rs

@@ -3723,234 +3723,4 @@ mod tests {
         }
         Ok(())
     }
-
-    /// Tests that PartialReduce mode:
-    /// 1. Accepts state as input (like Final)
-    /// 2. Produces state as output (like Partial)
-    /// 3. Can be followed by a Final stage to get the correct result
-    ///
-    /// This simulates a tree-reduce pattern:
-    ///   Partial -> PartialReduce -> Final
-    #[tokio::test]
-    async fn test_partial_reduce_mode() -> Result<()> {
-        let schema = Arc::new(Schema::new(vec![
-            Field::new("a", DataType::UInt32, false),
-            Field::new("b", DataType::Float64, false),
-        ]));
-
-        // Produce two partitions of input data
-        let batch1 = RecordBatch::try_new(
-            Arc::clone(&schema),
-            vec![
-                Arc::new(UInt32Array::from(vec![1, 2, 3])),
-                Arc::new(Float64Array::from(vec![10.0, 20.0, 30.0])),
-            ],
-        )?;
-        let batch2 = RecordBatch::try_new(
-            Arc::clone(&schema),
-            vec![
-                Arc::new(UInt32Array::from(vec![1, 2, 3])),
-                Arc::new(Float64Array::from(vec![40.0, 50.0, 60.0])),
-            ],
-        )?;
-
-        let groups =
-            PhysicalGroupBy::new_single(vec![(col("a", &schema)?, "a".to_string())]);
-        let aggregates: Vec<Arc<AggregateFunctionExpr>> = vec![Arc::new(
-            AggregateExprBuilder::new(sum_udaf(), vec![col("b", &schema)?])
-                .schema(Arc::clone(&schema))
-                .alias("SUM(b)")
-                .build()?,
-        )];
-
-        // Step 1: Partial aggregation on partition 1
-        let input1 =
-            TestMemoryExec::try_new_exec(&[vec![batch1]], Arc::clone(&schema), None)?;
-        let partial1 = Arc::new(AggregateExec::try_new(
-            AggregateMode::Partial,
-            groups.clone(),
-            aggregates.clone(),
-            vec![None],
-            input1,
-            Arc::clone(&schema),
-        )?);
-
-        // Step 2: Partial aggregation on partition 2
-        let input2 =
-            TestMemoryExec::try_new_exec(&[vec![batch2]], Arc::clone(&schema), None)?;
-        let partial2 = Arc::new(AggregateExec::try_new(
-            AggregateMode::Partial,
-            groups.clone(),
-            aggregates.clone(),
-            vec![None],
-            input2,
-            Arc::clone(&schema),
-        )?);
-
-        // Collect partial results
-        let task_ctx = Arc::new(TaskContext::default());
-        let partial_result1 =
-            crate::collect(Arc::clone(&partial1) as _, Arc::clone(&task_ctx)).await?;
-        let partial_result2 =
-            crate::collect(Arc::clone(&partial2) as _, Arc::clone(&task_ctx)).await?;
-
-        // The partial results have state schema (group cols + accumulator state)
-        let partial_schema = partial1.schema();
-
-        // Step 3: PartialReduce — combine partial results, still producing state
-        let combined_input = TestMemoryExec::try_new_exec(
-            &[partial_result1, partial_result2],
-            Arc::clone(&partial_schema),
-            None,
-        )?;
-        // Coalesce into a single partition for the PartialReduce
-        let coalesced = Arc::new(CoalescePartitionsExec::new(combined_input));
-
-        let partial_reduce = Arc::new(AggregateExec::try_new(
-            AggregateMode::PartialReduce,
-            groups.clone(),
-            aggregates.clone(),
-            vec![None],
-            coalesced,
-            Arc::clone(&partial_schema),
-        )?);
-
-        // Verify PartialReduce output schema matches Partial output schema
-        // (both produce state, not final values)
-        assert_eq!(partial_reduce.schema(), partial_schema);
-
-        // Collect PartialReduce results
-        let reduce_result =
-            crate::collect(Arc::clone(&partial_reduce) as _, Arc::clone(&task_ctx))
-                .await?;
-
-        // Step 4: Final aggregation on the PartialReduce output
-        let final_input = TestMemoryExec::try_new_exec(
-            &[reduce_result],
-            Arc::clone(&partial_schema),
-            None,
-        )?;
-        let final_agg = Arc::new(AggregateExec::try_new(
-            AggregateMode::Final,
-            groups.clone(),
-            aggregates.clone(),
-            vec![None],
-            final_input,
-            Arc::clone(&partial_schema),
-        )?);
-
-        let result = crate::collect(final_agg, Arc::clone(&task_ctx)).await?;
-
-        // Expected: group 1 -> 10+40=50, group 2 -> 20+50=70, group 3 -> 30+60=90
-        assert_snapshot!(batches_to_sort_string(&result), @r"
-            +---+--------+
-            | a | SUM(b) |
-            +---+--------+
-            | 1 | 50.0   |
-            | 2 | 70.0   |
-            | 3 | 90.0   |
-            +---+--------+
-        ");
-
-        Ok(())
-    }
-
-    #[test]
-    fn test_with_dynamic_filter() -> Result<()> {
-        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
-        let child = Arc::new(EmptyExec::new(Arc::clone(&schema)));
-
-        // Partial min triggers init_dynamic_filter.
-        let agg = AggregateExec::try_new(
-            AggregateMode::Partial,
-            PhysicalGroupBy::new_single(vec![]),
-            vec![Arc::new(
-                AggregateExprBuilder::new(min_udaf(), vec![col("a", &schema)?])
-                    .schema(Arc::clone(&schema))
-                    .alias("min_a")
-                    .build()?,
-            )],
-            vec![None],
-            child,
-            Arc::clone(&schema),
-        )?;
-        let original_inner_id = agg
-            .dynamic_filter()
-            .expect("should have dynamic filter after init")
-            .inner_id();
-
-        let new_df = Arc::new(DynamicFilterPhysicalExpr::new(
-            vec![col("a", &schema)?],
-            lit(true),
-        ));
-        let agg = agg.with_dynamic_filter(Arc::clone(&new_df))?;
-        let restored = agg
-            .dynamic_filter()
-            .expect("should still have dynamic filter");
-        assert_eq!(restored.inner_id(), new_df.inner_id());
-        assert_ne!(restored.inner_id(), original_inner_id);
-        Ok(())
-    }
-
-    #[test]
-    fn test_with_dynamic_filter_noop_when_unsupported() -> Result<()> {
-        let schema = Arc::new(Schema::new(vec![
-            Field::new("a", DataType::Int64, false),
-            Field::new("b", DataType::Int64, false),
-        ]));
-        let child = Arc::new(EmptyExec::new(Arc::clone(&schema)));
-
-        // Final mode with a group-by does not support dynamic filters.
-        let agg = AggregateExec::try_new(
-            AggregateMode::Final,
-            PhysicalGroupBy::new_single(vec![(col("a", &schema)?, "a".to_string())]),
-            vec![Arc::new(
-                AggregateExprBuilder::new(sum_udaf(), vec![col("b", &schema)?])
-                    .schema(Arc::clone(&schema))
-                    .alias("sum_b")
-                    .build()?,
-            )],
-            vec![None],
-            child,
-            Arc::clone(&schema),
-        )?;
-        assert!(agg.dynamic_filter().is_none());
-
-        // with_dynamic_filter should be a no-op.
-        let df = Arc::new(DynamicFilterPhysicalExpr::new(
-            vec![col("a", &schema)?],
-            lit(true),
-        ));
-        let agg = agg.with_dynamic_filter(df)?;
-        assert!(agg.dynamic_filter().is_none());
-        Ok(())
-    }
-
-    #[test]
-    fn test_with_dynamic_filter_rejects_invalid_columns() -> Result<()> {
-        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
-        let child = Arc::new(EmptyExec::new(Arc::clone(&schema)));
-
-        let agg = AggregateExec::try_new(
-            AggregateMode::Partial,
-            PhysicalGroupBy::new_single(vec![]),
-            vec![Arc::new(
-                AggregateExprBuilder::new(min_udaf(), vec![col("a", &schema)?])
-                    .schema(Arc::clone(&schema))
-                    .alias("min_a")
-                    .build()?,
-            )],
-            vec![None],
-            child,
-            Arc::clone(&schema),
-        )?;
-
-        // Column index 99 is out of bounds for the input schema.
-        let df = Arc::new(DynamicFilterPhysicalExpr::new(
-            vec![Arc::new(Column::new("bad", 99)) as _],
-            lit(true),
-        ));
-        assert!(agg.with_dynamic_filter(df).is_err());
-        Ok(())
-    }
 }