Expose ExecutionPlan statistics across the FFI boundary (#22157)

## Which issue does this PR close?

- Closes #22152 

## Rationale for this change

`ExecutionPlan::partition_statistics` and `TableProvider::statistics`
are not currently transported across the DataFusion FFI boundary, so
foreign plans and providers always report `Statistics::new_unknown` /
`None`. This blocks optimizer rules that depend on statistics (e.g. join
reordering, partition pruning) from working with out-of-process plugins,
which defeats the point of exposing those hooks to plugin authors.

`Statistics` contains `Precision<ScalarValue>` for column min/max/sum.
`ScalarValue` is a large enum that's impractical to mirror in
`#[repr(C)]`, so I reuse the existing
`datafusion_proto_common::Statistics` prost encoding — the same pattern
this crate already uses for filter expressions.

## What changes are included in this PR?

- New `datafusion_ffi::statistics` module with
`[de]serialize_statistics` helpers wrapping
the`datafusion_proto_common::Statistics` round-trip.
- New `partition_statistics` field on `FFI_ExecutionPlan` and
corresponding `ExecutionPlan::partition_statistics` impl on
`ForeignExecutionPlan`
- New `statistics` field on `FFI_TableProvider` and corresponding
`TableProvider::statistics` impl on `ForeignTableProvider`. Since the
trait returns `Option<Statistics>`, the implementation cannot propagate
decode errors, it logs a `log::warn!` and triggers a `debug_assert!`.

This PR is expected to be merged after #22136 so it includes those
changes.

## Are these changes tested?

Yes:

- Unit tests in `statistics.rs` cover three round-trip cases:
`Statistics::new_unknown`, fully-exact statistics with
`ScalarValue::Int32`/`Int64`/`Utf8` min/max/sum, and mixed
`Precision::Exact`/`Inexact`/`Absent` values.
- A new round-trip integration test in `execution_plan.rs` exercises
`ForeignExecutionPlan::partition_statistics` with both `None` and
`Some(idx)` partitions, against a plan with no statistics (returns
`Statistics::new_unknown`) and a plan with concrete statistics.
- A new round-trip integration test in `table_provider.rs` uses a thin
`TableWithStats` wrapper over `MemTable` to verify both the `None` path
and the concrete `Statistics` path through
`ForeignTableProvider::statistics`.

## Are there any user-facing changes?

This is a breaking ABI change for the `datafusion-ffi` crate:

- `FFI_ExecutionPlan` gains a `partition_statistics` field.
- `FFI_TableProvider` gains a `statistics` field.

Plugins compiled against earlier versions of `datafusion-ffi` will need
to be recompiled. There are no breaking changes to the Rust trait
surface or to `Statistics` itself; downstream `ExecutionPlan` /
`TableProvider` implementations require no changes.

Author: mailmindlin

datafusion/ffi/src/execution_plan.rs

@@ -21,7 +21,7 @@ use std::sync::Arc;
 
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::TreeNodeRecursion;
-use datafusion_common::{DataFusionError, Result};
+use datafusion_common::{DataFusionError, Result, Statistics};
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
 use datafusion_physical_expr_common::metrics::MetricsSet;
 use datafusion_physical_plan::{
@@ -36,6 +36,7 @@ use crate::execution::FFI_TaskContext;
 use crate::physical_expr::metrics::FFI_MetricsSet;
 use crate::plan_properties::FFI_PlanProperties;
 use crate::record_batch_stream::FFI_RecordBatchStream;
+use crate::statistics::{deserialize_statistics, serialize_statistics};
 use crate::util::{FFI_Option, FFI_Result};
 use crate::{df_result, sresult, sresult_return};
 
@@ -74,6 +75,15 @@ pub struct FFI_ExecutionPlan {
     /// underlying [`ExecutionPlan::metrics`] returned `None`.
     pub metrics: unsafe extern "C" fn(plan: &Self) -> FFI_Option<FFI_MetricsSet>,
 
+    /// Snapshot partition statistics. `partition == None` corresponds to
+    /// statistics over all partitions; `Some(idx)` corresponds to a specific
+    /// partition. The returned bytes are a prost-encoded
+    /// `datafusion_proto_common::Statistics`.
+    pub partition_statistics: unsafe extern "C" fn(
+        plan: &Self,
+        partition: FFI_Option<usize>,
+    ) -> FFI_Result<SVec<u8>>,
+
     /// Used to create a clone on the provider of the execution plan. This should
     /// only need to be called by the receiver of the plan.
     pub clone: unsafe extern "C" fn(plan: &Self) -> Self,
@@ -195,6 +205,17 @@ unsafe extern "C" fn metrics_fn_wrapper(
         .into()
 }
 
+unsafe extern "C" fn partition_statistics_fn_wrapper(
+    plan: &FFI_ExecutionPlan,
+    partition: FFI_Option<usize>,
+) -> FFI_Result<SVec<u8>> {
+    let partition: Option<usize> = partition.into();
+    plan.inner()
+        .partition_statistics(partition)
+        .map(|stats| SVec::from(serialize_statistics(stats.as_ref()).as_slice()))
+        .into()
+}
+
 unsafe extern "C" fn release_fn_wrapper(plan: &mut FFI_ExecutionPlan) {
     unsafe {
         debug_assert!(!plan.private_data.is_null());
@@ -287,6 +308,7 @@ impl FFI_ExecutionPlan {
             execute: execute_fn_wrapper,
             repartitioned: repartitioned_fn_wrapper,
             metrics: metrics_fn_wrapper,
+            partition_statistics: partition_statistics_fn_wrapper,
             clone: clone_fn_wrapper,
             release: release_fn_wrapper,
             private_data: Box::into_raw(private_data) as *mut c_void,
@@ -454,6 +476,13 @@ impl ExecutionPlan for ForeignExecutionPlan {
             unsafe { (self.plan.metrics)(&self.plan) }.into();
         ffi.map(MetricsSet::from)
     }
+
+    fn partition_statistics(&self, partition: Option<usize>) -> Result<Arc<Statistics>> {
+        let bytes = df_result!(unsafe {
+            (self.plan.partition_statistics)(&self.plan, partition.into())
+        })?;
+        Ok(Arc::new(deserialize_statistics(bytes.as_slice())?))
+    }
 }
 
 #[cfg(any(test, feature = "integration-tests"))]
@@ -468,6 +497,7 @@ pub mod tests {
         props: Arc<PlanProperties>,
         children: Vec<Arc<dyn ExecutionPlan>>,
         metrics: Option<MetricsSet>,
+        statistics: Option<Statistics>,
     }
 
     impl EmptyExec {
@@ -481,13 +511,19 @@ pub mod tests {
                 )),
                 children: Vec::default(),
                 metrics: None,
+                statistics: None,
             }
         }
 
         pub fn with_metrics(mut self, metrics: MetricsSet) -> Self {
             self.metrics = Some(metrics);
             self
         }
+
+        pub fn with_statistics(mut self, statistics: Statistics) -> Self {
+            self.statistics = Some(statistics);
+            self
+        }
     }
 
     impl DisplayAs for EmptyExec {
@@ -521,6 +557,7 @@ pub mod tests {
                 props: Arc::clone(&self.props),
                 children,
                 metrics: self.metrics.clone(),
+                statistics: self.statistics.clone(),
             }))
         }
 
@@ -536,6 +573,15 @@ pub mod tests {
             self.metrics.clone()
         }
 
+        fn partition_statistics(
+            &self,
+            _partition: Option<usize>,
+        ) -> Result<Arc<Statistics>> {
+            Ok(Arc::new(self.statistics.clone().unwrap_or_else(|| {
+                Statistics::new_unknown(self.props.eq_properties.schema())
+            })))
+        }
+
         fn apply_expressions(
             &self,
             f: &mut dyn FnMut(
@@ -659,6 +705,54 @@ pub mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_ffi_execution_plan_partition_statistics_round_trip() -> Result<()> {
+        use datafusion_common::stats::Precision;
+        use datafusion_common::{ColumnStatistics, ScalarValue};
+
+        let schema = Arc::new(arrow::datatypes::Schema::new(vec![
+            arrow::datatypes::Field::new("a", arrow::datatypes::DataType::Int32, true),
+        ]));
+
+        // Plans without explicit statistics return Statistics::new_unknown across
+        // the boundary.
+        let bare_plan = Arc::new(EmptyExec::new(Arc::clone(&schema)));
+        let mut bare_local = FFI_ExecutionPlan::new(bare_plan, None);
+        bare_local.library_marker_id = crate::mock_foreign_marker_id;
+        let bare_foreign: Arc<dyn ExecutionPlan> = (&bare_local).try_into()?;
+        let bare_stats = bare_foreign.partition_statistics(None)?;
+        assert_eq!(bare_stats.as_ref(), &Statistics::new_unknown(&schema));
+
+        // Plans with statistics round-trip them faithfully, including
+        // ScalarValue-typed min/max.
+        let original_stats = Statistics {
+            num_rows: Precision::Exact(7),
+            total_byte_size: Precision::Inexact(128),
+            column_statistics: vec![ColumnStatistics {
+                null_count: Precision::Exact(1),
+                max_value: Precision::Exact(ScalarValue::Int32(Some(10))),
+                min_value: Precision::Exact(ScalarValue::Int32(Some(-3))),
+                sum_value: Precision::Absent,
+                distinct_count: Precision::Inexact(6),
+                byte_size: Precision::Exact(28),
+            }],
+        };
+        let stats_plan = Arc::new(
+            EmptyExec::new(Arc::clone(&schema)).with_statistics(original_stats.clone()),
+        );
+        let mut stats_local = FFI_ExecutionPlan::new(stats_plan, None);
+        stats_local.library_marker_id = crate::mock_foreign_marker_id;
+        let stats_foreign: Arc<dyn ExecutionPlan> = (&stats_local).try_into()?;
+
+        let observed = stats_foreign.partition_statistics(None)?;
+        assert_eq!(observed.as_ref(), &original_stats);
+
+        let observed_partition = stats_foreign.partition_statistics(Some(1))?;
+        assert_eq!(observed_partition.as_ref(), &original_stats);
+
+        Ok(())
+    }
+
     #[test]
     fn test_ffi_execution_plan_local_bypass() {
         let schema = Arc::new(arrow::datatypes::Schema::new(vec![

datafusion/ffi/src/lib.rs

@@ -41,6 +41,7 @@ pub mod proto;
 pub mod record_batch_stream;
 pub mod schema_provider;
 pub mod session;
+pub mod statistics;
 pub mod table_provider;
 pub mod table_provider_factory;
 pub mod table_source;

datafusion/ffi/src/physical_expr/metrics.rs

@@ -59,7 +59,7 @@ pub struct FFI_MetricsSet {
 pub struct FFI_Metric {
     pub value: FFI_MetricValue,
     pub labels: SVec<FFI_Label>,
-    pub partition: FFI_Option<u64>,
+    pub partition: FFI_Option<usize>,
     pub metric_type: FFI_MetricType,
     pub metric_category: FFI_Option<FFI_MetricCategory>,
 }
@@ -203,7 +203,7 @@ impl From<&Metric> for FFI_Metric {
         Self {
             value: FFI_MetricValue::from(m.value()),
             labels: m.labels().iter().map(FFI_Label::from).collect(),
-            partition: m.partition().map(|p| p as u64).into(),
+            partition: m.partition().into(),
             metric_type: m.metric_type().into(),
             metric_category: m.metric_category().map(FFI_MetricCategory::from).into(),
         }
@@ -213,14 +213,10 @@ impl From<&Metric> for FFI_Metric {
 impl From<FFI_Metric> for Metric {
     fn from(m: FFI_Metric) -> Self {
         let labels: Vec<Label> = m.labels.into_iter().map(Label::from).collect();
-        let partition: Option<u64> = m.partition.into();
+        let partition: Option<usize> = m.partition.into();
         let category: Option<FFI_MetricCategory> = m.metric_category.into();
-        let mut metric = Metric::new_with_labels(
-            m.value.into(),
-            partition.map(|p| p as usize),
-            labels,
-        )
-        .with_type(m.metric_type.into());
+        let mut metric = Metric::new_with_labels(m.value.into(), partition, labels)
+            .with_type(m.metric_type.into());
         if let Some(c) = category {
             metric = metric.with_category(c.into());
         }

datafusion/ffi/src/statistics.rs

@@ -0,0 +1,124 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Helpers for moving [`Statistics`] across the FFI boundary as prost-encoded
+//! `datafusion_proto_common::Statistics` bytes.
+//!
+//! [`Statistics`] contains [`Precision<ScalarValue>`] for column min/max/sum,
+//! and `ScalarValue` is a large enum that's impractical to mirror in
+//! `#[repr(C)]`. The proto round-trip already exists in `datafusion-proto-common`
+//! and is the same pattern used to ship filter expressions across the FFI
+//! boundary, so we reuse it here.
+//!
+//! [`Precision<ScalarValue>`]: datafusion_common::stats::Precision
+
+use datafusion_common::{DataFusionError, Result, Statistics};
+use prost::Message;
+
+/// Serialize [`Statistics`] to prost-encoded
+/// `datafusion_proto_common::Statistics` bytes.
+pub(crate) fn serialize_statistics(stats: &Statistics) -> Vec<u8> {
+    datafusion_proto_common::Statistics::from(stats).encode_to_vec()
+}
+
+/// Decode prost-encoded `datafusion_proto_common::Statistics` bytes back into
+/// [`Statistics`].
+pub(crate) fn deserialize_statistics(bytes: &[u8]) -> Result<Statistics> {
+    let proto = datafusion_proto_common::Statistics::decode(bytes).map_err(|e| {
+        DataFusionError::Plan(format!("failed to decode Statistics: {e}"))
+    })?;
+    Statistics::try_from(&proto)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use arrow::datatypes::{DataType, Field, Schema};
+    use datafusion_common::ScalarValue;
+    use datafusion_common::stats::Precision;
+    use datafusion_common::{ColumnStatistics, Statistics};
+
+    use super::*;
+
+    #[test]
+    fn round_trip_unknown_statistics() {
+        let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
+        let original = Statistics::new_unknown(&Arc::new(schema));
+
+        let bytes = serialize_statistics(&original);
+        let observed = deserialize_statistics(&bytes).expect("decode");
+
+        assert_eq!(observed, original);
+    }
+
+    #[test]
+    fn round_trip_exact_statistics_with_scalar_values() {
+        let original = Statistics {
+            num_rows: Precision::Exact(100),
+            total_byte_size: Precision::Exact(4096),
+            column_statistics: vec![
+                ColumnStatistics {
+                    null_count: Precision::Exact(2),
+                    max_value: Precision::Exact(ScalarValue::Int32(Some(50))),
+                    min_value: Precision::Exact(ScalarValue::Int32(Some(-10))),
+                    sum_value: Precision::Exact(ScalarValue::Int64(Some(1234))),
+                    distinct_count: Precision::Exact(40),
+                    byte_size: Precision::Exact(800),
+                },
+                ColumnStatistics {
+                    null_count: Precision::Exact(0),
+                    max_value: Precision::Exact(ScalarValue::Utf8(Some(
+                        "zebra".to_string(),
+                    ))),
+                    min_value: Precision::Exact(ScalarValue::Utf8(Some(
+                        "ant".to_string(),
+                    ))),
+                    sum_value: Precision::Absent,
+                    distinct_count: Precision::Inexact(95),
+                    byte_size: Precision::Inexact(2048),
+                },
+            ],
+        };
+
+        let bytes = serialize_statistics(&original);
+        let observed = deserialize_statistics(&bytes).expect("decode");
+
+        assert_eq!(observed, original);
+    }
+
+    #[test]
+    fn round_trip_mixed_precision() {
+        let original = Statistics {
+            num_rows: Precision::Inexact(42),
+            total_byte_size: Precision::Absent,
+            column_statistics: vec![ColumnStatistics {
+                null_count: Precision::Absent,
+                max_value: Precision::Inexact(ScalarValue::Float64(Some(1.5))),
+                min_value: Precision::Absent,
+                sum_value: Precision::Inexact(ScalarValue::Float64(Some(63.0))),
+                distinct_count: Precision::Absent,
+                byte_size: Precision::Absent,
+            }],
+        };
+
+        let bytes = serialize_statistics(&original);
+        let observed = deserialize_statistics(&bytes).expect("decode");
+
+        assert_eq!(observed, original);
+    }
+}