Restore pluggable left-join accumulator seam (CollectLeftAccumulator trait + recreate_with_accumulator) on DF53.1

datafusion/common/src/config.rs

@@ -1115,6 +1115,21 @@ config_namespace! {
         /// See: <https://trino.io/docs/current/admin/dynamic-filtering.html#dynamic-filter-collection-thresholds>
         pub hash_join_inlist_pushdown_max_distinct_values: usize, default = 150
 
+        /// Maximum size in bytes of the exact join filter (e.g. an exact `InList` of
+        /// build-side keys) that a custom build-side
+        /// [`CollectLeftAccumulator`](https://docs.rs/datafusion-physical-plan/latest/datafusion_physical_plan/joins/trait.CollectLeftAccumulator.html)
+        /// may materialize for dynamic filter pushdown.
+        ///
+        /// This is an additional budget consulted by pluggable exact-membership
+        /// accumulators (the accumulator seam); build sides whose exact filter would
+        /// exceed this size are expected to fall back to a coarser strategy. It does
+        /// not affect the native min/max accumulator, whose InList budget is governed
+        /// by `hash_join_inlist_pushdown_max_size`.
+        ///
+        /// The default (`usize::MAX`) imposes no additional limit, preserving existing
+        /// behavior.
+        pub exact_join_filter_max_bytes: usize, default = usize::MAX
+
         /// The default filter selectivity used by Filter Statistics
         /// when an exact selectivity cannot be determined. Valid values are
         /// between 0 (no selectivity) and 100 (all rows are selected).

datafusion/datasource/src/test_util.rs

@@ -88,6 +88,17 @@ impl FileSource for MockSource {
         Arc::new(Self { ..self.clone() })
     }
 
+    fn with_metadata_cols(
+        &self,
+        metadata_cols: Vec<crate::metadata::MetadataColumn>,
+    ) -> Option<Arc<dyn FileSource>> {
+        let mut source = self.clone();
+        source.table_schema = source.table_schema.with_metadata_cols(metadata_cols);
+        source.projection =
+            crate::projection::SplitProjection::unprojected(&source.table_schema);
+        Some(Arc::new(source))
+    }
+
     fn metrics(&self) -> &ExecutionPlanMetricsSet {
         &self.metrics
     }
@@ -106,8 +117,8 @@ impl FileSource for MockSource {
     ) -> Result<Option<Arc<dyn FileSource>>> {
         let mut source = self.clone();
         let new_projection = self.projection.source.try_merge(projection)?;
-        let split_projection = crate::projection::SplitProjection::new(
-            self.table_schema.file_schema(),
+        let split_projection = crate::projection::SplitProjection::new_with_table_schema(
+            &self.table_schema,
             &new_projection,
         );
         source.projection = split_projection;

datafusion/physical-expr/src/projection.rs

@@ -661,7 +661,15 @@ impl ProjectionExprs {
         for proj_expr in self.exprs.iter() {
             let expr = &proj_expr.expr;
             let col_stats = if let Some(col) = expr.as_any().downcast_ref::<Column>() {
-                std::mem::take(&mut stats.column_statistics[col.index()])
+                // Spice metadata columns (and projected partition columns) reference
+                // table-schema indices that fall beyond the upstream file-level
+                // statistics, which only cover the file (and partition) columns. Those
+                // injected columns have no known statistics, so fall back to unknown
+                if col.index() < stats.column_statistics.len() {
+                    std::mem::take(&mut stats.column_statistics[col.index()])
+                } else {
+                    ColumnStatistics::new_unknown()
+                }
             } else if let Some(literal) = expr.as_any().downcast_ref::<Literal>() {
                 // Handle literal expressions (constants) by calculating proper statistics
                 let data_type = expr.data_type(output_schema)?;
@@ -2737,6 +2745,54 @@ pub(crate) mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_project_statistics_column_index_beyond_input_stats() -> Result<()> {
+        // File-level statistics only cover the three file columns.
+        let input_stats = get_stats();
+        assert_eq!(input_stats.column_statistics.len(), 3);
+
+        // The table schema additionally exposes a metadata column at index 3.
+        let table_schema = Schema::new(vec![
+            Field::new("col0", DataType::Int64, false),
+            Field::new("col1", DataType::Utf8, false),
+            Field::new("col2", DataType::Float32, false),
+            Field::new("_location", DataType::Utf8, true),
+        ]);
+
+        // SELECT col0, _location — the metadata column (index 3) is beyond the
+        // input statistics width (3).
+        let projection = ProjectionExprs::new(vec![
+            ProjectionExpr {
+                expr: Arc::new(Column::new("col0", 0)),
+                alias: "col0".to_string(),
+            },
+            ProjectionExpr {
+                expr: Arc::new(Column::new("_location", 3)),
+                alias: "_location".to_string(),
+            },
+        ]);
+
+        let output_stats = projection
+            .project_statistics(input_stats, &projection.project_schema(&table_schema)?)?;
+
+        assert_eq!(output_stats.num_rows, Precision::Exact(5));
+        assert_eq!(output_stats.column_statistics.len(), 2);
+
+        // col0 keeps its known statistics.
+        assert_eq!(
+            output_stats.column_statistics[0].max_value,
+            Precision::Exact(ScalarValue::Int64(Some(21)))
+        );
+
+        // The metadata column has no upstream statistics -> unknown, not a panic.
+        assert_eq!(
+            output_stats.column_statistics[1],
+            ColumnStatistics::new_unknown()
+        );
+
+        Ok(())
+    }
+
     #[test]
     fn test_project_statistics_primitive_width_only() -> Result<()> {
         let input_stats = get_stats();