feat: use exact integer statistics for ordering inference

datafusion/datasource/src/file_scan_config/mod.rs

@@ -198,15 +198,16 @@ pub struct FileScanConfig {
     /// Expression adapter used to adapt filters and projections that are pushed down into the scan
     /// from the logical schema to the physical schema of the file.
     pub expr_adapter_factory: Option<Arc<dyn PhysicalExprAdapterFactory>>,
-    /// Statistics for the entire table (file schema + partition columns).
+    /// Statistics for the entire table (file schema + partition columns), before
+    /// projection or filtering.
     /// See [`FileScanConfigBuilder::with_statistics`] for more details.
     ///
     /// The effective statistics are computed on-demand via
     /// [`ProjectionExprs::project_statistics`].
     ///
-    /// Note that this field is pub(crate) because accessing it directly from outside
-    /// would be incorrect if there are filters being applied, thus this should be accessed
-    /// via [`FileScanConfig::statistics`].
+    /// Retained for serialization and conservative bounds used by ordering inference.
+    /// Use [`FileScanConfig::statistics`] for filter-adjusted statistics; these raw
+    /// values may no longer be exact for the scan output.
     pub(crate) statistics: Statistics,
     /// Declared physical output partitioning for this scan.
     ///
@@ -948,7 +949,9 @@ impl DataSource for FileScanConfig {
             Arc::clone(schema),
             self.validated_output_ordering(),
         )
-        .with_constraints(self.constraints.clone());
+        .with_constraints(self.constraints.clone())
+        // Filters only remove rows, so original exact extrema remain valid bounds.
+        .with_column_statistics(&self.statistics.column_statistics);
 
         // Only a filter that the source applies exactly gives valid equivalences.
         // A filter that is used only for pruning lets non-matching rows through.

datafusion/datasource/src/file_scan_config/proto.rs

@@ -71,9 +71,7 @@ impl FileScanConfig {
             file_source,
             batch_size,
             expr_adapter_factory,
-            // Serialized through `statistics()` so its pushed-filter policy
-            // remains centralized.
-            statistics: _,
+            statistics,
             output_partitioning,
         } = self;
 
@@ -139,11 +137,11 @@ impl FileScanConfig {
                     (*file_compression_type.get_variant()).into();
                 compression as i32
             });
-        let statistics = self.statistics();
-
+        // Serialize the original statistics so decoded scans retain trusted bounds.
+        // The decoded scan still applies the filter policy in `statistics()`.
         Ok(protobuf::FileScanExecConf {
             file_groups: proto_file_groups,
-            statistics: Some((&statistics).into()),
+            statistics: Some(statistics.into()),
             limit: limit
                 .map(|limit| usize_to_wire::<u32>(limit, "FileScanConfig", "limit"))
                 .transpose()?

datafusion/physical-expr/src/equivalence/properties/dependency.rs

@@ -1807,6 +1807,75 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn project_shares_ranges_with_trusted_integer_bounds() -> Result<()> {
+        use datafusion_common::stats::Precision;
+        use datafusion_common::{ColumnStatistics, ScalarValue};
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int64, true),
+            Field::new(
+                "s",
+                DataType::Struct(Fields::from(vec![Field::new(
+                    "x",
+                    DataType::Int64,
+                    true,
+                )])),
+                true,
+            ),
+        ]));
+        let bounds =
+            Interval::try_new(ScalarValue::Int64(Some(1)), ScalarValue::Int64(Some(3)))?;
+        let input = EquivalenceProperties::new_with_orderings(
+            Arc::clone(&schema),
+            ["a", "s"].map(|name| {
+                vec![PhysicalSortExpr::new_default(col(name, &schema).unwrap())]
+            }),
+        )
+        .with_column_statistics(&[
+            ColumnStatistics {
+                min_value: Precision::Exact(bounds.lower().clone()),
+                max_value: Precision::Exact(bounds.upper().clone()),
+                ..ColumnStatistics::new_unknown()
+            },
+            ColumnStatistics::new_unknown(),
+        ]);
+        let (mapping, seen) = probe_mapping(
+            vec![
+                (col("a", &schema)?, 0, false),
+                (col("a", &schema)?, 1, false),
+                (col("s", &schema)?, 2, false),
+                (col("s", &schema)?, 3, false),
+            ],
+            &schema,
+        )?;
+        let projected = input.project(&mapping, output_schema(&mapping, &schema)?);
+        assert_eq!(
+            sorted_orderings(&projected),
+            vec!["p0@0 ASC", "p1@1 ASC", "p2@2 ASC", "p3@3 ASC"]
+        );
+        for probe in &seen[..2] {
+            let ranges = probe.lock().unwrap();
+            assert!(ranges.len() >= 3);
+            assert!(ranges.iter().all(|range| range == &bounds));
+        }
+        // Bound propagation, ordering substitution and ordering derivation all
+        // reuse the same unbounded struct range, even with trusted integer bounds.
+        let mut arrays = vec![];
+        for probe in &seen[2..] {
+            let ranges = probe.lock().unwrap();
+            assert!(ranges.len() >= 3);
+            for range in ranges.iter() {
+                let ScalarValue::Struct(array) = range.lower() else {
+                    panic!("{range:?}");
+                };
+                arrays.push(Arc::clone(array));
+            }
+        }
+        assert!(arrays.iter().all(|array| Arc::ptr_eq(array, &arrays[0])));
+        Ok(())
+    }
+
     #[test]
     fn project_does_not_reuse_a_column_range_across_calls() -> Result<()> {
         for data_type in [DataType::Utf8, DataType::Int64] {