From 7b0e0373af4496eff75836f4981b440d6cff2cd0 Mon Sep 17 00:00:00 2001
From: mailmindlin <mailmindlin@users.noreply.github.com>
Date: Tue, 12 May 2026 16:55:32 -0400
Subject: [PATCH 1/9] Add metrics to `FFI_ExecutionPlan`

---
 datafusion/ffi/src/lib.rs     |   1 +
 datafusion/ffi/src/metrics.rs | 743 ++++++++++++++++++++++++++++++++++
 2 files changed, 744 insertions(+)
 create mode 100644 datafusion/ffi/src/metrics.rs

diff --git a/datafusion/ffi/src/lib.rs b/datafusion/ffi/src/lib.rs
index de3caf8c17260..e9b703ddb4554 100644
--- a/datafusion/ffi/src/lib.rs
+++ b/datafusion/ffi/src/lib.rs
@@ -34,6 +34,7 @@ pub mod execution_plan;
 pub mod expr;
 pub mod ffi_option;
 pub mod insert_op;
+pub mod metrics;
 pub mod physical_expr;
 pub mod physical_optimizer;
 pub mod plan_properties;
diff --git a/datafusion/ffi/src/metrics.rs b/datafusion/ffi/src/metrics.rs
new file mode 100644
index 0000000000000..889065fe6a1db
--- /dev/null
+++ b/datafusion/ffi/src/metrics.rs
@@ -0,0 +1,743 @@
+// 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.
+
+//! FFI-stable mirrors of [`MetricsSet`] and related metric types.
+//!
+//! Metrics are passed across the FFI boundary as a **snapshot**: all
+//! atomic-backed counters/gauges/timers are read into plain integer fields
+//! at conversion time. Callers re-invoke [`ExecutionPlan::metrics()`] across
+//! the boundary to observe newer values. This matches the documented contract
+//! ("Once `self.execute()` has returned... metrics should be complete") and
+//! all in-tree consumers (`AnalyzeExec`, `DisplayableExecutionPlan`).
+//!
+//! The variant *order* of [`FFI_MetricValue`] is part of the stable ABI and
+//! must not be reordered. New variants must be appended at the end.
+//!
+//! [`ExecutionPlan::metrics()`]: datafusion_physical_plan::ExecutionPlan::metrics
+
+use std::any::Any;
+use std::borrow::Cow;
+use std::fmt::{self, Debug, Display};
+use std::sync::Arc;
+
+use chrono::{DateTime, Utc};
+use datafusion_common::format::{MetricCategory, MetricType};
+use datafusion_physical_expr_common::metrics::{
+    Count, CustomMetricValue, Gauge, MetricValue, MetricsSet, PruningMetrics,
+    RatioMergeStrategy, RatioMetrics, Time, Timestamp,
+};
+use datafusion_physical_expr_common::metrics::{Label, Metric};
+use stabby::string::String as SString;
+use stabby::vec::Vec as SVec;
+
+use crate::ffi_option::FFI_Option;
+
+/// FFI-stable mirror of [`MetricsSet`].
+#[repr(C)]
+#[derive(Debug, Clone)]
+pub struct FFI_MetricsSet {
+    pub metrics: SVec<FFI_Metric>,
+}
+
+/// FFI-stable mirror of [`Metric`].
+#[repr(C)]
+#[derive(Debug, Clone)]
+pub struct FFI_Metric {
+    pub value: FFI_MetricValue,
+    pub labels: SVec<FFI_Label>,
+    pub partition: FFI_Option<u64>,
+    pub metric_type: FFI_MetricType,
+    pub metric_category: FFI_Option<FFI_MetricCategory>,
+}
+
+/// FFI-stable mirror of [`Label`].
+#[repr(C)]
+#[derive(Debug, Clone)]
+pub struct FFI_Label {
+    pub name: SString,
+    pub value: SString,
+}
+
+/// FFI-stable mirror of [`MetricType`].
+#[expect(non_camel_case_types)]
+#[repr(u8)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FFI_MetricType {
+    Summary,
+    Dev,
+}
+
+/// FFI-stable mirror of [`MetricCategory`].
+#[expect(non_camel_case_types)]
+#[repr(u8)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FFI_MetricCategory {
+    Rows,
+    Bytes,
+    Timing,
+    Uncategorized,
+}
+
+/// FFI-stable mirror of [`PruningMetrics`]. All counts are snapshotted at
+/// conversion time.
+#[repr(C)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct FFI_PruningMetrics {
+    pub pruned: u64,
+    pub matched: u64,
+    pub fully_matched: u64,
+}
+
+/// FFI-stable mirror of [`RatioMergeStrategy`].
+#[expect(non_camel_case_types)]
+#[repr(u8)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FFI_RatioMergeStrategy {
+    AddPartAddTotal,
+    AddPartSetTotal,
+    SetPartAddTotal,
+}
+
+/// FFI-stable mirror of [`RatioMetrics`]. Numerator/denominator are
+/// snapshotted at conversion time.
+#[repr(C)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct FFI_RatioMetrics {
+    pub part: u64,
+    pub total: u64,
+    pub merge_strategy: FFI_RatioMergeStrategy,
+    pub display_raw_values: bool,
+}
+
+/// FFI-stable mirror of [`MetricValue`]. The variant order is load-bearing
+/// across the FFI boundary; new variants must only be appended.
+#[repr(C, u8)]
+#[derive(Debug, Clone)]
+pub enum FFI_MetricValue {
+    OutputRows(u64),
+    /// Nanoseconds.
+    ElapsedCompute(u64),
+    SpillCount(u64),
+    SpilledBytes(u64),
+    OutputBytes(u64),
+    OutputBatches(u64),
+    SpilledRows(u64),
+    CurrentMemoryUsage(u64),
+    Count {
+        name: SString,
+        count: u64,
+    },
+    Gauge {
+        name: SString,
+        gauge: u64,
+    },
+    /// Nanoseconds.
+    Time {
+        name: SString,
+        time: u64,
+    },
+    /// Unix nanoseconds (UTC).
+    StartTimestamp(FFI_Option<i64>),
+    /// Unix nanoseconds (UTC).
+    EndTimestamp(FFI_Option<i64>),
+    PruningMetrics {
+        name: SString,
+        pruning_metrics: FFI_PruningMetrics,
+    },
+    Ratio {
+        name: SString,
+        ratio_metrics: FFI_RatioMetrics,
+    },
+    /// Custom metrics are marshalled as their `Display` output plus the
+    /// `as_usize()` fallback. The underlying `dyn CustomMetricValue` type is
+    /// not preserved across the boundary, so `aggregate`/`as_any` downcasting
+    /// are lost; the reconstructed value uses [`FfiCustomMetricValue`].
+    Custom {
+        name: SString,
+        display: SString,
+        as_usize_value: u64,
+    },
+}
+
+// -----------------------------------------------------------------------------
+// MetricsSet <-> FFI_MetricsSet
+// -----------------------------------------------------------------------------
+
+impl From<&MetricsSet> for FFI_MetricsSet {
+    fn from(set: &MetricsSet) -> Self {
+        Self {
+            metrics: set.iter().map(|m| FFI_Metric::from(m.as_ref())).collect(),
+        }
+    }
+}
+
+impl From<FFI_MetricsSet> for MetricsSet {
+    fn from(set: FFI_MetricsSet) -> Self {
+        let mut out = MetricsSet::new();
+        for ffi_metric in set.metrics {
+            out.push(Arc::new(Metric::from(ffi_metric)));
+        }
+        out
+    }
+}
+
+// -----------------------------------------------------------------------------
+// Metric <-> FFI_Metric
+// -----------------------------------------------------------------------------
+
+impl From<&Metric> for FFI_Metric {
+    fn from(m: &Metric) -> Self {
+        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(),
+            metric_type: m.metric_type().into(),
+            metric_category: m.metric_category().map(FFI_MetricCategory::from).into(),
+        }
+    }
+}
+
+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 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());
+        if let Some(c) = category {
+            metric = metric.with_category(c.into());
+        }
+        metric
+    }
+}
+
+// -----------------------------------------------------------------------------
+// Label <-> FFI_Label
+// -----------------------------------------------------------------------------
+
+impl From<&Label> for FFI_Label {
+    fn from(l: &Label) -> Self {
+        Self {
+            name: SString::from(l.name()),
+            value: SString::from(l.value()),
+        }
+    }
+}
+
+impl From<FFI_Label> for Label {
+    fn from(l: FFI_Label) -> Self {
+        let name: String = l.name.into();
+        let value: String = l.value.into();
+        Label::new(name, value)
+    }
+}
+
+// -----------------------------------------------------------------------------
+// MetricType <-> FFI_MetricType
+// -----------------------------------------------------------------------------
+
+impl From<MetricType> for FFI_MetricType {
+    fn from(t: MetricType) -> Self {
+        match t {
+            MetricType::Summary => Self::Summary,
+            MetricType::Dev => Self::Dev,
+        }
+    }
+}
+
+impl From<FFI_MetricType> for MetricType {
+    fn from(t: FFI_MetricType) -> Self {
+        match t {
+            FFI_MetricType::Summary => Self::Summary,
+            FFI_MetricType::Dev => Self::Dev,
+        }
+    }
+}
+
+// -----------------------------------------------------------------------------
+// MetricCategory <-> FFI_MetricCategory
+// -----------------------------------------------------------------------------
+
+impl From<MetricCategory> for FFI_MetricCategory {
+    fn from(c: MetricCategory) -> Self {
+        match c {
+            MetricCategory::Rows => Self::Rows,
+            MetricCategory::Bytes => Self::Bytes,
+            MetricCategory::Timing => Self::Timing,
+            MetricCategory::Uncategorized => Self::Uncategorized,
+        }
+    }
+}
+
+impl From<FFI_MetricCategory> for MetricCategory {
+    fn from(c: FFI_MetricCategory) -> Self {
+        match c {
+            FFI_MetricCategory::Rows => Self::Rows,
+            FFI_MetricCategory::Bytes => Self::Bytes,
+            FFI_MetricCategory::Timing => Self::Timing,
+            FFI_MetricCategory::Uncategorized => Self::Uncategorized,
+        }
+    }
+}
+
+// -----------------------------------------------------------------------------
+// RatioMergeStrategy <-> FFI_RatioMergeStrategy
+// -----------------------------------------------------------------------------
+
+impl From<&RatioMergeStrategy> for FFI_RatioMergeStrategy {
+    fn from(s: &RatioMergeStrategy) -> Self {
+        match s {
+            RatioMergeStrategy::AddPartAddTotal => Self::AddPartAddTotal,
+            RatioMergeStrategy::AddPartSetTotal => Self::AddPartSetTotal,
+            RatioMergeStrategy::SetPartAddTotal => Self::SetPartAddTotal,
+        }
+    }
+}
+
+impl From<FFI_RatioMergeStrategy> for RatioMergeStrategy {
+    fn from(s: FFI_RatioMergeStrategy) -> Self {
+        match s {
+            FFI_RatioMergeStrategy::AddPartAddTotal => Self::AddPartAddTotal,
+            FFI_RatioMergeStrategy::AddPartSetTotal => Self::AddPartSetTotal,
+            FFI_RatioMergeStrategy::SetPartAddTotal => Self::SetPartAddTotal,
+        }
+    }
+}
+
+// -----------------------------------------------------------------------------
+// PruningMetrics <-> FFI_PruningMetrics
+// -----------------------------------------------------------------------------
+
+impl From<&PruningMetrics> for FFI_PruningMetrics {
+    fn from(p: &PruningMetrics) -> Self {
+        Self {
+            pruned: p.pruned() as u64,
+            matched: p.matched() as u64,
+            fully_matched: p.fully_matched() as u64,
+        }
+    }
+}
+
+impl From<FFI_PruningMetrics> for PruningMetrics {
+    fn from(p: FFI_PruningMetrics) -> Self {
+        let out = PruningMetrics::new();
+        out.add_pruned(p.pruned as usize);
+        out.add_matched(p.matched as usize);
+        out.add_fully_matched(p.fully_matched as usize);
+        out
+    }
+}
+
+// -----------------------------------------------------------------------------
+// RatioMetrics <-> FFI_RatioMetrics
+// -----------------------------------------------------------------------------
+
+impl From<&RatioMetrics> for FFI_RatioMetrics {
+    fn from(r: &RatioMetrics) -> Self {
+        Self {
+            part: r.part() as u64,
+            total: r.total() as u64,
+            merge_strategy: r.merge_strategy().into(),
+            display_raw_values: r.display_raw_values(),
+        }
+    }
+}
+
+impl From<FFI_RatioMetrics> for RatioMetrics {
+    fn from(r: FFI_RatioMetrics) -> Self {
+        let out = RatioMetrics::new()
+            .with_merge_strategy(r.merge_strategy.into())
+            .with_display_raw_values(r.display_raw_values);
+        out.set_part(r.part as usize);
+        out.set_total(r.total as usize);
+        out
+    }
+}
+
+// -----------------------------------------------------------------------------
+// MetricValue <-> FFI_MetricValue
+// -----------------------------------------------------------------------------
+
+fn timestamp_to_ffi(ts: &Timestamp) -> FFI_Option<i64> {
+    ts.value().and_then(|dt| dt.timestamp_nanos_opt()).into()
+}
+
+fn timestamp_from_ffi(nanos: FFI_Option<i64>) -> Timestamp {
+    let ts = Timestamp::new();
+    if let Some(n) = nanos.into_option() {
+        ts.set(DateTime::<Utc>::from_timestamp_nanos(n));
+    }
+    ts
+}
+
+fn count_from_value(v: u64) -> Count {
+    let c = Count::new();
+    c.add(v as usize);
+    c
+}
+
+fn gauge_from_value(v: u64) -> Gauge {
+    let g = Gauge::new();
+    g.add(v as usize);
+    g
+}
+
+fn time_from_nanos(v: u64) -> Time {
+    let t = Time::new();
+    t.add_duration(std::time::Duration::from_nanos(v));
+    t
+}
+
+impl From<&MetricValue> for FFI_MetricValue {
+    fn from(v: &MetricValue) -> Self {
+        match v {
+            MetricValue::OutputRows(c) => Self::OutputRows(c.value() as u64),
+            MetricValue::ElapsedCompute(t) => Self::ElapsedCompute(t.value() as u64),
+            MetricValue::SpillCount(c) => Self::SpillCount(c.value() as u64),
+            MetricValue::SpilledBytes(c) => Self::SpilledBytes(c.value() as u64),
+            MetricValue::OutputBytes(c) => Self::OutputBytes(c.value() as u64),
+            MetricValue::OutputBatches(c) => Self::OutputBatches(c.value() as u64),
+            MetricValue::SpilledRows(c) => Self::SpilledRows(c.value() as u64),
+            MetricValue::CurrentMemoryUsage(g) => {
+                Self::CurrentMemoryUsage(g.value() as u64)
+            }
+            MetricValue::Count { name, count } => Self::Count {
+                name: SString::from(name.as_ref()),
+                count: count.value() as u64,
+            },
+            MetricValue::Gauge { name, gauge } => Self::Gauge {
+                name: SString::from(name.as_ref()),
+                gauge: gauge.value() as u64,
+            },
+            MetricValue::Time { name, time } => Self::Time {
+                name: SString::from(name.as_ref()),
+                time: time.value() as u64,
+            },
+            MetricValue::StartTimestamp(ts) => Self::StartTimestamp(timestamp_to_ffi(ts)),
+            MetricValue::EndTimestamp(ts) => Self::EndTimestamp(timestamp_to_ffi(ts)),
+            MetricValue::PruningMetrics {
+                name,
+                pruning_metrics,
+            } => Self::PruningMetrics {
+                name: SString::from(name.as_ref()),
+                pruning_metrics: pruning_metrics.into(),
+            },
+            MetricValue::Ratio {
+                name,
+                ratio_metrics,
+            } => Self::Ratio {
+                name: SString::from(name.as_ref()),
+                ratio_metrics: ratio_metrics.into(),
+            },
+            MetricValue::Custom { name, value } => Self::Custom {
+                name: SString::from(name.as_ref()),
+                display: SString::from(value.to_string().as_str()),
+                as_usize_value: value.as_usize() as u64,
+            },
+        }
+    }
+}
+
+impl From<FFI_MetricValue> for MetricValue {
+    fn from(v: FFI_MetricValue) -> Self {
+        match v {
+            FFI_MetricValue::OutputRows(n) => Self::OutputRows(count_from_value(n)),
+            FFI_MetricValue::ElapsedCompute(n) => {
+                Self::ElapsedCompute(time_from_nanos(n))
+            }
+            FFI_MetricValue::SpillCount(n) => Self::SpillCount(count_from_value(n)),
+            FFI_MetricValue::SpilledBytes(n) => Self::SpilledBytes(count_from_value(n)),
+            FFI_MetricValue::OutputBytes(n) => Self::OutputBytes(count_from_value(n)),
+            FFI_MetricValue::OutputBatches(n) => Self::OutputBatches(count_from_value(n)),
+            FFI_MetricValue::SpilledRows(n) => Self::SpilledRows(count_from_value(n)),
+            FFI_MetricValue::CurrentMemoryUsage(n) => {
+                Self::CurrentMemoryUsage(gauge_from_value(n))
+            }
+            FFI_MetricValue::Count { name, count } => Self::Count {
+                name: Cow::Owned(name.into()),
+                count: count_from_value(count),
+            },
+            FFI_MetricValue::Gauge { name, gauge } => Self::Gauge {
+                name: Cow::Owned(name.into()),
+                gauge: gauge_from_value(gauge),
+            },
+            FFI_MetricValue::Time { name, time } => Self::Time {
+                name: Cow::Owned(name.into()),
+                time: time_from_nanos(time),
+            },
+            FFI_MetricValue::StartTimestamp(nanos) => {
+                Self::StartTimestamp(timestamp_from_ffi(nanos))
+            }
+            FFI_MetricValue::EndTimestamp(nanos) => {
+                Self::EndTimestamp(timestamp_from_ffi(nanos))
+            }
+            FFI_MetricValue::PruningMetrics {
+                name,
+                pruning_metrics,
+            } => Self::PruningMetrics {
+                name: Cow::Owned(name.into()),
+                pruning_metrics: pruning_metrics.into(),
+            },
+            FFI_MetricValue::Ratio {
+                name,
+                ratio_metrics,
+            } => Self::Ratio {
+                name: Cow::Owned(name.into()),
+                ratio_metrics: ratio_metrics.into(),
+            },
+            FFI_MetricValue::Custom {
+                name,
+                display,
+                as_usize_value,
+            } => Self::Custom {
+                name: Cow::Owned(name.into()),
+                value: Arc::new(FfiCustomMetricValue {
+                    display: display.into(),
+                    as_usize_value: as_usize_value as usize,
+                }),
+            },
+        }
+    }
+}
+
+/// [`CustomMetricValue`] shim used when reconstructing a `MetricValue::Custom`
+/// on the consumer side of the FFI boundary.
+///
+/// The original `dyn CustomMetricValue` is not preserved across FFI — only its
+/// `Display` output and `as_usize()` fallback. As a result:
+/// - `Display` returns the producer-side rendered string.
+/// - `as_usize` returns the snapshot value captured at marshal time.
+/// - `aggregate` is a no-op: snapshots from FFI are not mergeable.
+/// - `as_any` only downcasts to `FfiCustomMetricValue` itself.
+#[derive(Debug)]
+pub struct FfiCustomMetricValue {
+    display: String,
+    as_usize_value: usize,
+}
+
+impl Display for FfiCustomMetricValue {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(&self.display)
+    }
+}
+
+impl CustomMetricValue for FfiCustomMetricValue {
+    fn new_empty(&self) -> Arc<dyn CustomMetricValue> {
+        Arc::new(FfiCustomMetricValue {
+            display: String::new(),
+            as_usize_value: 0,
+        })
+    }
+
+    fn aggregate(&self, _other: Arc<dyn CustomMetricValue>) {
+        // FFI snapshots are immutable and not mergeable; aggregation across
+        // the boundary is intentionally a no-op.
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn as_usize(&self) -> usize {
+        self.as_usize_value
+    }
+
+    fn is_eq(&self, other: &Arc<dyn CustomMetricValue>) -> bool {
+        other
+            .as_any()
+            .downcast_ref::<Self>()
+            .map(|o| o.display == self.display && o.as_usize_value == self.as_usize_value)
+            .unwrap_or(false)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use datafusion_physical_expr_common::metrics::Label;
+
+    fn assert_value_roundtrip(value: MetricValue) {
+        let ffi: FFI_MetricValue = (&value).into();
+        let back: MetricValue = ffi.into();
+        assert_eq!(value, back, "round-trip mismatch for {value:?}");
+    }
+
+    #[test]
+    fn roundtrip_named_variants() {
+        let c = Count::new();
+        c.add(7);
+        assert_value_roundtrip(MetricValue::OutputRows(c.clone()));
+        assert_value_roundtrip(MetricValue::SpillCount(c.clone()));
+        assert_value_roundtrip(MetricValue::SpilledBytes(c.clone()));
+        assert_value_roundtrip(MetricValue::OutputBytes(c.clone()));
+        assert_value_roundtrip(MetricValue::OutputBatches(c.clone()));
+        assert_value_roundtrip(MetricValue::SpilledRows(c.clone()));
+
+        let g = Gauge::new();
+        g.add(123);
+        assert_value_roundtrip(MetricValue::CurrentMemoryUsage(g));
+
+        let t = Time::new();
+        t.add_duration(std::time::Duration::from_nanos(456));
+        assert_value_roundtrip(MetricValue::ElapsedCompute(t));
+    }
+
+    #[test]
+    fn roundtrip_keyed_count_gauge_time() {
+        let count = Count::new();
+        count.add(11);
+        assert_value_roundtrip(MetricValue::Count {
+            name: Cow::Borrowed("custom_count"),
+            count,
+        });
+
+        let gauge = Gauge::new();
+        gauge.add(22);
+        assert_value_roundtrip(MetricValue::Gauge {
+            name: Cow::Borrowed("custom_gauge"),
+            gauge,
+        });
+
+        let time = Time::new();
+        time.add_duration(std::time::Duration::from_nanos(33));
+        assert_value_roundtrip(MetricValue::Time {
+            name: Cow::Borrowed("custom_time"),
+            time,
+        });
+    }
+
+    #[test]
+    fn roundtrip_timestamps() {
+        let unset = Timestamp::new();
+        assert_value_roundtrip(MetricValue::StartTimestamp(unset.clone()));
+        assert_value_roundtrip(MetricValue::EndTimestamp(unset));
+
+        let set = Timestamp::new();
+        set.set(DateTime::<Utc>::from_timestamp_nanos(
+            1_700_000_000_000_000_000,
+        ));
+        assert_value_roundtrip(MetricValue::StartTimestamp(set.clone()));
+        assert_value_roundtrip(MetricValue::EndTimestamp(set));
+    }
+
+    #[test]
+    fn roundtrip_pruning_and_ratio() {
+        let pruning = PruningMetrics::new();
+        pruning.add_pruned(3);
+        pruning.add_matched(5);
+        pruning.add_fully_matched(2);
+        assert_value_roundtrip(MetricValue::PruningMetrics {
+            name: Cow::Borrowed("file_prune"),
+            pruning_metrics: pruning,
+        });
+
+        let ratio = RatioMetrics::new()
+            .with_merge_strategy(RatioMergeStrategy::SetPartAddTotal)
+            .with_display_raw_values(false);
+        ratio.set_part(20);
+        ratio.set_total(100);
+        assert_value_roundtrip(MetricValue::Ratio {
+            name: Cow::Borrowed("selectivity"),
+            ratio_metrics: ratio,
+        });
+    }
+
+    #[test]
+    fn custom_metric_value_stringifies() {
+        #[derive(Debug)]
+        struct DummyCustom;
+        impl Display for DummyCustom {
+            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+                write!(f, "dummy:42")
+            }
+        }
+        impl CustomMetricValue for DummyCustom {
+            fn new_empty(&self) -> Arc<dyn CustomMetricValue> {
+                Arc::new(DummyCustom)
+            }
+            fn aggregate(&self, _other: Arc<dyn CustomMetricValue>) {}
+            fn as_any(&self) -> &dyn Any {
+                self
+            }
+            fn as_usize(&self) -> usize {
+                42
+            }
+            fn is_eq(&self, other: &Arc<dyn CustomMetricValue>) -> bool {
+                other.as_any().downcast_ref::<Self>().is_some()
+            }
+        }
+
+        let original = MetricValue::Custom {
+            name: Cow::Borrowed("dummy"),
+            value: Arc::new(DummyCustom),
+        };
+        let ffi: FFI_MetricValue = (&original).into();
+        let back: MetricValue = ffi.into();
+
+        match back {
+            MetricValue::Custom { name, value } => {
+                assert_eq!(name.as_ref(), "dummy");
+                assert_eq!(value.to_string(), "dummy:42");
+                assert_eq!(value.as_usize(), 42);
+            }
+            other => panic!("expected Custom, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn roundtrip_full_metric_with_labels_partition_type_category() {
+        let count = Count::new();
+        count.add(99);
+        let metric = Metric::new_with_labels(
+            MetricValue::OutputRows(count),
+            Some(3),
+            vec![Label::new("file", "a.parquet")],
+        )
+        .with_type(MetricType::Summary)
+        .with_category(MetricCategory::Rows);
+
+        let ffi: FFI_Metric = (&metric).into();
+        let back: Metric = ffi.into();
+
+        assert_eq!(back.value(), metric.value());
+        assert_eq!(back.partition(), Some(3));
+        assert_eq!(back.labels().len(), 1);
+        assert_eq!(back.labels()[0].name(), "file");
+        assert_eq!(back.labels()[0].value(), "a.parquet");
+        assert_eq!(back.metric_type(), MetricType::Summary);
+        assert_eq!(back.metric_category(), Some(MetricCategory::Rows));
+    }
+
+    #[test]
+    fn roundtrip_metrics_set() {
+        let mut set = MetricsSet::new();
+        let c = Count::new();
+        c.add(1);
+        set.push(Arc::new(Metric::new(MetricValue::OutputRows(c), Some(0))));
+        let c2 = Count::new();
+        c2.add(2);
+        set.push(Arc::new(Metric::new(MetricValue::OutputRows(c2), Some(1))));
+
+        let ffi: FFI_MetricsSet = (&set).into();
+        let back: MetricsSet = ffi.into();
+        assert_eq!(back.output_rows(), Some(3));
+    }
+}

From c4fa4896276347e857100cbfe06858009c99494d Mon Sep 17 00:00:00 2001
From: mailmindlin <mailmindlin@users.noreply.github.com>
Date: Wed, 13 May 2026 12:37:59 -0400
Subject: [PATCH 2/9] Fix off-by-one bug

---
 datafusion/ffi/src/metrics.rs | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/datafusion/ffi/src/metrics.rs b/datafusion/ffi/src/metrics.rs
index 889065fe6a1db..35bb8d9202fc5 100644
--- a/datafusion/ffi/src/metrics.rs
+++ b/datafusion/ffi/src/metrics.rs
@@ -400,9 +400,12 @@ fn gauge_from_value(v: u64) -> Gauge {
     g
 }
 
-fn time_from_nanos(v: u64) -> Time {
+fn time_from_nanos(nanos: u64) -> Time {
     let t = Time::new();
-    t.add_duration(std::time::Duration::from_nanos(v));
+    if nanos != 0 {
+        // add_duration always adds at least one
+        t.add_duration(std::time::Duration::from_nanos(nanos));
+    }
     t
 }
 

From 7b66b0bcab8135c0fdedfab0d61eff5913e48dc0 Mon Sep 17 00:00:00 2001
From: mailmindlin <mailmindlin@users.noreply.github.com>
Date: Wed, 13 May 2026 12:38:12 -0400
Subject: [PATCH 3/9] Fix roundtrip tests

---
 datafusion/ffi/src/metrics.rs | 60 +++++++++++++++++++++++++++++++++--
 1 file changed, 57 insertions(+), 3 deletions(-)

diff --git a/datafusion/ffi/src/metrics.rs b/datafusion/ffi/src/metrics.rs
index 35bb8d9202fc5..30a31e22448df 100644
--- a/datafusion/ffi/src/metrics.rs
+++ b/datafusion/ffi/src/metrics.rs
@@ -577,10 +577,16 @@ mod tests {
     use super::*;
     use datafusion_physical_expr_common::metrics::Label;
 
-    fn assert_value_roundtrip(value: MetricValue) {
+    /// Round-trips `value` through the FFI representation, asserts the
+    /// `PartialEq`-based equality, and returns the reconstructed value so
+    /// callers can additionally assert fields that `MetricValue`'s
+    /// `PartialEq` impl does not compare (e.g. `PruningMetrics::fully_matched`,
+    /// `RatioMetrics::merge_strategy`).
+    fn assert_value_roundtrip(value: MetricValue) -> MetricValue {
         let ffi: FFI_MetricValue = (&value).into();
         let back: MetricValue = ffi.into();
         assert_eq!(value, back, "round-trip mismatch for {value:?}");
+        back
     }
 
     #[test]
@@ -647,20 +653,50 @@ mod tests {
         pruning.add_pruned(3);
         pruning.add_matched(5);
         pruning.add_fully_matched(2);
-        assert_value_roundtrip(MetricValue::PruningMetrics {
+        let back = assert_value_roundtrip(MetricValue::PruningMetrics {
             name: Cow::Borrowed("file_prune"),
             pruning_metrics: pruning,
         });
+        match back {
+            MetricValue::PruningMetrics {
+                name,
+                pruning_metrics,
+            } => {
+                assert_eq!(name.as_ref(), "file_prune");
+                assert_eq!(pruning_metrics.pruned(), 3);
+                assert_eq!(pruning_metrics.matched(), 5);
+                assert_eq!(pruning_metrics.fully_matched(), 2);
+            }
+            other => panic!("expected PruningMetrics, got {other:?}"),
+        }
 
         let ratio = RatioMetrics::new()
             .with_merge_strategy(RatioMergeStrategy::SetPartAddTotal)
             .with_display_raw_values(false);
         ratio.set_part(20);
         ratio.set_total(100);
-        assert_value_roundtrip(MetricValue::Ratio {
+        // `RatioMetrics`'s `PartialEq` does not compare `merge_strategy`,
+        // so assert it explicitly after the round-trip.
+        let back = assert_value_roundtrip(MetricValue::Ratio {
             name: Cow::Borrowed("selectivity"),
             ratio_metrics: ratio,
         });
+        match back {
+            MetricValue::Ratio {
+                name,
+                ratio_metrics,
+            } => {
+                assert_eq!(name.as_ref(), "selectivity");
+                assert_eq!(ratio_metrics.part(), 20);
+                assert_eq!(ratio_metrics.total(), 100);
+                assert!(matches!(
+                    ratio_metrics.merge_strategy(),
+                    RatioMergeStrategy::SetPartAddTotal
+                ));
+                assert!(!ratio_metrics.display_raw_values());
+            }
+            other => panic!("expected Ratio, got {other:?}"),
+        }
     }
 
     #[test]
@@ -741,6 +777,24 @@ mod tests {
 
         let ffi: FFI_MetricsSet = (&set).into();
         let back: MetricsSet = ffi.into();
+
+        // Aggregate check.
         assert_eq!(back.output_rows(), Some(3));
+
+        // Per-metric check: ordering, partition, and per-partition value must
+        // all survive the round-trip (the aggregate above can't catch a lost
+        // partition or a metric count mismatch).
+        let metrics: Vec<_> = back.iter().collect();
+        assert_eq!(metrics.len(), 2);
+        assert_eq!(metrics[0].partition(), Some(0));
+        assert_eq!(
+            metrics[0].value(),
+            &MetricValue::OutputRows(count_from_value(1))
+        );
+        assert_eq!(metrics[1].partition(), Some(1));
+        assert_eq!(
+            metrics[1].value(),
+            &MetricValue::OutputRows(count_from_value(2))
+        );
     }
 }

From 1f6dcb0a9b202987cdf2f348e9118e92fe3c68e3 Mon Sep 17 00:00:00 2001
From: mailmindlin <mailmindlin@users.noreply.github.com>
Date: Wed, 13 May 2026 13:10:38 -0400
Subject: [PATCH 4/9] Remove awkward docs

---
 datafusion/ffi/src/metrics.rs | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/datafusion/ffi/src/metrics.rs b/datafusion/ffi/src/metrics.rs
index 30a31e22448df..beeeceebddfb0 100644
--- a/datafusion/ffi/src/metrics.rs
+++ b/datafusion/ffi/src/metrics.rs
@@ -123,8 +123,7 @@ pub struct FFI_RatioMetrics {
     pub display_raw_values: bool,
 }
 
-/// FFI-stable mirror of [`MetricValue`]. The variant order is load-bearing
-/// across the FFI boundary; new variants must only be appended.
+/// FFI-stable mirror of [`MetricValue`].
 #[repr(C, u8)]
 #[derive(Debug, Clone)]
 pub enum FFI_MetricValue {

From 73a495cd5c19a19273d0b29e9f6c07f788a46e7d Mon Sep 17 00:00:00 2001
From: mailmindlin <mailmindlin@users.noreply.github.com>
Date: Wed, 13 May 2026 13:10:49 -0400
Subject: [PATCH 5/9] Explicit units

---
 datafusion/ffi/src/metrics.rs | 30 +++++++++++++-----------------
 1 file changed, 13 insertions(+), 17 deletions(-)

diff --git a/datafusion/ffi/src/metrics.rs b/datafusion/ffi/src/metrics.rs
index beeeceebddfb0..b58fba0430d48 100644
--- a/datafusion/ffi/src/metrics.rs
+++ b/datafusion/ffi/src/metrics.rs
@@ -128,8 +128,7 @@ pub struct FFI_RatioMetrics {
 #[derive(Debug, Clone)]
 pub enum FFI_MetricValue {
     OutputRows(u64),
-    /// Nanoseconds.
-    ElapsedCompute(u64),
+    ElapsedComputeNs(u64),
     SpillCount(u64),
     SpilledBytes(u64),
     OutputBytes(u64),
@@ -144,15 +143,12 @@ pub enum FFI_MetricValue {
         name: SString,
         gauge: u64,
     },
-    /// Nanoseconds.
     Time {
         name: SString,
-        time: u64,
+        time_ns: u64,
     },
-    /// Unix nanoseconds (UTC).
-    StartTimestamp(FFI_Option<i64>),
-    /// Unix nanoseconds (UTC).
-    EndTimestamp(FFI_Option<i64>),
+    StartTimestampNsUTC(FFI_Option<i64>),
+    EndTimestampNsUTC(FFI_Option<i64>),
     PruningMetrics {
         name: SString,
         pruning_metrics: FFI_PruningMetrics,
@@ -412,7 +408,7 @@ impl From<&MetricValue> for FFI_MetricValue {
     fn from(v: &MetricValue) -> Self {
         match v {
             MetricValue::OutputRows(c) => Self::OutputRows(c.value() as u64),
-            MetricValue::ElapsedCompute(t) => Self::ElapsedCompute(t.value() as u64),
+            MetricValue::ElapsedCompute(t) => Self::ElapsedComputeNs(t.value() as u64),
             MetricValue::SpillCount(c) => Self::SpillCount(c.value() as u64),
             MetricValue::SpilledBytes(c) => Self::SpilledBytes(c.value() as u64),
             MetricValue::OutputBytes(c) => Self::OutputBytes(c.value() as u64),
@@ -431,10 +427,10 @@ impl From<&MetricValue> for FFI_MetricValue {
             },
             MetricValue::Time { name, time } => Self::Time {
                 name: SString::from(name.as_ref()),
-                time: time.value() as u64,
+                time_ns: time.value() as u64,
             },
-            MetricValue::StartTimestamp(ts) => Self::StartTimestamp(timestamp_to_ffi(ts)),
-            MetricValue::EndTimestamp(ts) => Self::EndTimestamp(timestamp_to_ffi(ts)),
+            MetricValue::StartTimestamp(ts) => Self::StartTimestampNsUTC(timestamp_to_ffi(ts)),
+            MetricValue::EndTimestamp(ts) => Self::EndTimestampNsUTC(timestamp_to_ffi(ts)),
             MetricValue::PruningMetrics {
                 name,
                 pruning_metrics,
@@ -462,7 +458,7 @@ impl From<FFI_MetricValue> for MetricValue {
     fn from(v: FFI_MetricValue) -> Self {
         match v {
             FFI_MetricValue::OutputRows(n) => Self::OutputRows(count_from_value(n)),
-            FFI_MetricValue::ElapsedCompute(n) => {
+            FFI_MetricValue::ElapsedComputeNs(n) => {
                 Self::ElapsedCompute(time_from_nanos(n))
             }
             FFI_MetricValue::SpillCount(n) => Self::SpillCount(count_from_value(n)),
@@ -481,14 +477,14 @@ impl From<FFI_MetricValue> for MetricValue {
                 name: Cow::Owned(name.into()),
                 gauge: gauge_from_value(gauge),
             },
-            FFI_MetricValue::Time { name, time } => Self::Time {
+            FFI_MetricValue::Time { name, time_ns } => Self::Time {
                 name: Cow::Owned(name.into()),
-                time: time_from_nanos(time),
+                time: time_from_nanos(time_ns),
             },
-            FFI_MetricValue::StartTimestamp(nanos) => {
+            FFI_MetricValue::StartTimestampNsUTC(nanos) => {
                 Self::StartTimestamp(timestamp_from_ffi(nanos))
             }
-            FFI_MetricValue::EndTimestamp(nanos) => {
+            FFI_MetricValue::EndTimestampNsUTC(nanos) => {
                 Self::EndTimestamp(timestamp_from_ffi(nanos))
             }
             FFI_MetricValue::PruningMetrics {

From 1b00bbd9bf6be45b52b1dc7bcab8e59fdb9bcf23 Mon Sep 17 00:00:00 2001
From: mailmindlin <mailmindlin@users.noreply.github.com>
Date: Wed, 13 May 2026 13:22:54 -0400
Subject: [PATCH 6/9] Fix clippy lint

Not sure why it shows up on the FFI enum but not the original
---
 datafusion/ffi/src/metrics.rs               | 795 --------------------
 datafusion/ffi/src/physical_expr/metrics.rs |   8 +-
 2 files changed, 2 insertions(+), 801 deletions(-)
 delete mode 100644 datafusion/ffi/src/metrics.rs

diff --git a/datafusion/ffi/src/metrics.rs b/datafusion/ffi/src/metrics.rs
deleted file mode 100644
index b58fba0430d48..0000000000000
--- a/datafusion/ffi/src/metrics.rs
+++ /dev/null
@@ -1,795 +0,0 @@
-// 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.
-
-//! FFI-stable mirrors of [`MetricsSet`] and related metric types.
-//!
-//! Metrics are passed across the FFI boundary as a **snapshot**: all
-//! atomic-backed counters/gauges/timers are read into plain integer fields
-//! at conversion time. Callers re-invoke [`ExecutionPlan::metrics()`] across
-//! the boundary to observe newer values. This matches the documented contract
-//! ("Once `self.execute()` has returned... metrics should be complete") and
-//! all in-tree consumers (`AnalyzeExec`, `DisplayableExecutionPlan`).
-//!
-//! The variant *order* of [`FFI_MetricValue`] is part of the stable ABI and
-//! must not be reordered. New variants must be appended at the end.
-//!
-//! [`ExecutionPlan::metrics()`]: datafusion_physical_plan::ExecutionPlan::metrics
-
-use std::any::Any;
-use std::borrow::Cow;
-use std::fmt::{self, Debug, Display};
-use std::sync::Arc;
-
-use chrono::{DateTime, Utc};
-use datafusion_common::format::{MetricCategory, MetricType};
-use datafusion_physical_expr_common::metrics::{
-    Count, CustomMetricValue, Gauge, MetricValue, MetricsSet, PruningMetrics,
-    RatioMergeStrategy, RatioMetrics, Time, Timestamp,
-};
-use datafusion_physical_expr_common::metrics::{Label, Metric};
-use stabby::string::String as SString;
-use stabby::vec::Vec as SVec;
-
-use crate::ffi_option::FFI_Option;
-
-/// FFI-stable mirror of [`MetricsSet`].
-#[repr(C)]
-#[derive(Debug, Clone)]
-pub struct FFI_MetricsSet {
-    pub metrics: SVec<FFI_Metric>,
-}
-
-/// FFI-stable mirror of [`Metric`].
-#[repr(C)]
-#[derive(Debug, Clone)]
-pub struct FFI_Metric {
-    pub value: FFI_MetricValue,
-    pub labels: SVec<FFI_Label>,
-    pub partition: FFI_Option<u64>,
-    pub metric_type: FFI_MetricType,
-    pub metric_category: FFI_Option<FFI_MetricCategory>,
-}
-
-/// FFI-stable mirror of [`Label`].
-#[repr(C)]
-#[derive(Debug, Clone)]
-pub struct FFI_Label {
-    pub name: SString,
-    pub value: SString,
-}
-
-/// FFI-stable mirror of [`MetricType`].
-#[expect(non_camel_case_types)]
-#[repr(u8)]
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum FFI_MetricType {
-    Summary,
-    Dev,
-}
-
-/// FFI-stable mirror of [`MetricCategory`].
-#[expect(non_camel_case_types)]
-#[repr(u8)]
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum FFI_MetricCategory {
-    Rows,
-    Bytes,
-    Timing,
-    Uncategorized,
-}
-
-/// FFI-stable mirror of [`PruningMetrics`]. All counts are snapshotted at
-/// conversion time.
-#[repr(C)]
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub struct FFI_PruningMetrics {
-    pub pruned: u64,
-    pub matched: u64,
-    pub fully_matched: u64,
-}
-
-/// FFI-stable mirror of [`RatioMergeStrategy`].
-#[expect(non_camel_case_types)]
-#[repr(u8)]
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum FFI_RatioMergeStrategy {
-    AddPartAddTotal,
-    AddPartSetTotal,
-    SetPartAddTotal,
-}
-
-/// FFI-stable mirror of [`RatioMetrics`]. Numerator/denominator are
-/// snapshotted at conversion time.
-#[repr(C)]
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub struct FFI_RatioMetrics {
-    pub part: u64,
-    pub total: u64,
-    pub merge_strategy: FFI_RatioMergeStrategy,
-    pub display_raw_values: bool,
-}
-
-/// FFI-stable mirror of [`MetricValue`].
-#[repr(C, u8)]
-#[derive(Debug, Clone)]
-pub enum FFI_MetricValue {
-    OutputRows(u64),
-    ElapsedComputeNs(u64),
-    SpillCount(u64),
-    SpilledBytes(u64),
-    OutputBytes(u64),
-    OutputBatches(u64),
-    SpilledRows(u64),
-    CurrentMemoryUsage(u64),
-    Count {
-        name: SString,
-        count: u64,
-    },
-    Gauge {
-        name: SString,
-        gauge: u64,
-    },
-    Time {
-        name: SString,
-        time_ns: u64,
-    },
-    StartTimestampNsUTC(FFI_Option<i64>),
-    EndTimestampNsUTC(FFI_Option<i64>),
-    PruningMetrics {
-        name: SString,
-        pruning_metrics: FFI_PruningMetrics,
-    },
-    Ratio {
-        name: SString,
-        ratio_metrics: FFI_RatioMetrics,
-    },
-    /// Custom metrics are marshalled as their `Display` output plus the
-    /// `as_usize()` fallback. The underlying `dyn CustomMetricValue` type is
-    /// not preserved across the boundary, so `aggregate`/`as_any` downcasting
-    /// are lost; the reconstructed value uses [`FfiCustomMetricValue`].
-    Custom {
-        name: SString,
-        display: SString,
-        as_usize_value: u64,
-    },
-}
-
-// -----------------------------------------------------------------------------
-// MetricsSet <-> FFI_MetricsSet
-// -----------------------------------------------------------------------------
-
-impl From<&MetricsSet> for FFI_MetricsSet {
-    fn from(set: &MetricsSet) -> Self {
-        Self {
-            metrics: set.iter().map(|m| FFI_Metric::from(m.as_ref())).collect(),
-        }
-    }
-}
-
-impl From<FFI_MetricsSet> for MetricsSet {
-    fn from(set: FFI_MetricsSet) -> Self {
-        let mut out = MetricsSet::new();
-        for ffi_metric in set.metrics {
-            out.push(Arc::new(Metric::from(ffi_metric)));
-        }
-        out
-    }
-}
-
-// -----------------------------------------------------------------------------
-// Metric <-> FFI_Metric
-// -----------------------------------------------------------------------------
-
-impl From<&Metric> for FFI_Metric {
-    fn from(m: &Metric) -> Self {
-        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(),
-            metric_type: m.metric_type().into(),
-            metric_category: m.metric_category().map(FFI_MetricCategory::from).into(),
-        }
-    }
-}
-
-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 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());
-        if let Some(c) = category {
-            metric = metric.with_category(c.into());
-        }
-        metric
-    }
-}
-
-// -----------------------------------------------------------------------------
-// Label <-> FFI_Label
-// -----------------------------------------------------------------------------
-
-impl From<&Label> for FFI_Label {
-    fn from(l: &Label) -> Self {
-        Self {
-            name: SString::from(l.name()),
-            value: SString::from(l.value()),
-        }
-    }
-}
-
-impl From<FFI_Label> for Label {
-    fn from(l: FFI_Label) -> Self {
-        let name: String = l.name.into();
-        let value: String = l.value.into();
-        Label::new(name, value)
-    }
-}
-
-// -----------------------------------------------------------------------------
-// MetricType <-> FFI_MetricType
-// -----------------------------------------------------------------------------
-
-impl From<MetricType> for FFI_MetricType {
-    fn from(t: MetricType) -> Self {
-        match t {
-            MetricType::Summary => Self::Summary,
-            MetricType::Dev => Self::Dev,
-        }
-    }
-}
-
-impl From<FFI_MetricType> for MetricType {
-    fn from(t: FFI_MetricType) -> Self {
-        match t {
-            FFI_MetricType::Summary => Self::Summary,
-            FFI_MetricType::Dev => Self::Dev,
-        }
-    }
-}
-
-// -----------------------------------------------------------------------------
-// MetricCategory <-> FFI_MetricCategory
-// -----------------------------------------------------------------------------
-
-impl From<MetricCategory> for FFI_MetricCategory {
-    fn from(c: MetricCategory) -> Self {
-        match c {
-            MetricCategory::Rows => Self::Rows,
-            MetricCategory::Bytes => Self::Bytes,
-            MetricCategory::Timing => Self::Timing,
-            MetricCategory::Uncategorized => Self::Uncategorized,
-        }
-    }
-}
-
-impl From<FFI_MetricCategory> for MetricCategory {
-    fn from(c: FFI_MetricCategory) -> Self {
-        match c {
-            FFI_MetricCategory::Rows => Self::Rows,
-            FFI_MetricCategory::Bytes => Self::Bytes,
-            FFI_MetricCategory::Timing => Self::Timing,
-            FFI_MetricCategory::Uncategorized => Self::Uncategorized,
-        }
-    }
-}
-
-// -----------------------------------------------------------------------------
-// RatioMergeStrategy <-> FFI_RatioMergeStrategy
-// -----------------------------------------------------------------------------
-
-impl From<&RatioMergeStrategy> for FFI_RatioMergeStrategy {
-    fn from(s: &RatioMergeStrategy) -> Self {
-        match s {
-            RatioMergeStrategy::AddPartAddTotal => Self::AddPartAddTotal,
-            RatioMergeStrategy::AddPartSetTotal => Self::AddPartSetTotal,
-            RatioMergeStrategy::SetPartAddTotal => Self::SetPartAddTotal,
-        }
-    }
-}
-
-impl From<FFI_RatioMergeStrategy> for RatioMergeStrategy {
-    fn from(s: FFI_RatioMergeStrategy) -> Self {
-        match s {
-            FFI_RatioMergeStrategy::AddPartAddTotal => Self::AddPartAddTotal,
-            FFI_RatioMergeStrategy::AddPartSetTotal => Self::AddPartSetTotal,
-            FFI_RatioMergeStrategy::SetPartAddTotal => Self::SetPartAddTotal,
-        }
-    }
-}
-
-// -----------------------------------------------------------------------------
-// PruningMetrics <-> FFI_PruningMetrics
-// -----------------------------------------------------------------------------
-
-impl From<&PruningMetrics> for FFI_PruningMetrics {
-    fn from(p: &PruningMetrics) -> Self {
-        Self {
-            pruned: p.pruned() as u64,
-            matched: p.matched() as u64,
-            fully_matched: p.fully_matched() as u64,
-        }
-    }
-}
-
-impl From<FFI_PruningMetrics> for PruningMetrics {
-    fn from(p: FFI_PruningMetrics) -> Self {
-        let out = PruningMetrics::new();
-        out.add_pruned(p.pruned as usize);
-        out.add_matched(p.matched as usize);
-        out.add_fully_matched(p.fully_matched as usize);
-        out
-    }
-}
-
-// -----------------------------------------------------------------------------
-// RatioMetrics <-> FFI_RatioMetrics
-// -----------------------------------------------------------------------------
-
-impl From<&RatioMetrics> for FFI_RatioMetrics {
-    fn from(r: &RatioMetrics) -> Self {
-        Self {
-            part: r.part() as u64,
-            total: r.total() as u64,
-            merge_strategy: r.merge_strategy().into(),
-            display_raw_values: r.display_raw_values(),
-        }
-    }
-}
-
-impl From<FFI_RatioMetrics> for RatioMetrics {
-    fn from(r: FFI_RatioMetrics) -> Self {
-        let out = RatioMetrics::new()
-            .with_merge_strategy(r.merge_strategy.into())
-            .with_display_raw_values(r.display_raw_values);
-        out.set_part(r.part as usize);
-        out.set_total(r.total as usize);
-        out
-    }
-}
-
-// -----------------------------------------------------------------------------
-// MetricValue <-> FFI_MetricValue
-// -----------------------------------------------------------------------------
-
-fn timestamp_to_ffi(ts: &Timestamp) -> FFI_Option<i64> {
-    ts.value().and_then(|dt| dt.timestamp_nanos_opt()).into()
-}
-
-fn timestamp_from_ffi(nanos: FFI_Option<i64>) -> Timestamp {
-    let ts = Timestamp::new();
-    if let Some(n) = nanos.into_option() {
-        ts.set(DateTime::<Utc>::from_timestamp_nanos(n));
-    }
-    ts
-}
-
-fn count_from_value(v: u64) -> Count {
-    let c = Count::new();
-    c.add(v as usize);
-    c
-}
-
-fn gauge_from_value(v: u64) -> Gauge {
-    let g = Gauge::new();
-    g.add(v as usize);
-    g
-}
-
-fn time_from_nanos(nanos: u64) -> Time {
-    let t = Time::new();
-    if nanos != 0 {
-        // add_duration always adds at least one
-        t.add_duration(std::time::Duration::from_nanos(nanos));
-    }
-    t
-}
-
-impl From<&MetricValue> for FFI_MetricValue {
-    fn from(v: &MetricValue) -> Self {
-        match v {
-            MetricValue::OutputRows(c) => Self::OutputRows(c.value() as u64),
-            MetricValue::ElapsedCompute(t) => Self::ElapsedComputeNs(t.value() as u64),
-            MetricValue::SpillCount(c) => Self::SpillCount(c.value() as u64),
-            MetricValue::SpilledBytes(c) => Self::SpilledBytes(c.value() as u64),
-            MetricValue::OutputBytes(c) => Self::OutputBytes(c.value() as u64),
-            MetricValue::OutputBatches(c) => Self::OutputBatches(c.value() as u64),
-            MetricValue::SpilledRows(c) => Self::SpilledRows(c.value() as u64),
-            MetricValue::CurrentMemoryUsage(g) => {
-                Self::CurrentMemoryUsage(g.value() as u64)
-            }
-            MetricValue::Count { name, count } => Self::Count {
-                name: SString::from(name.as_ref()),
-                count: count.value() as u64,
-            },
-            MetricValue::Gauge { name, gauge } => Self::Gauge {
-                name: SString::from(name.as_ref()),
-                gauge: gauge.value() as u64,
-            },
-            MetricValue::Time { name, time } => Self::Time {
-                name: SString::from(name.as_ref()),
-                time_ns: time.value() as u64,
-            },
-            MetricValue::StartTimestamp(ts) => Self::StartTimestampNsUTC(timestamp_to_ffi(ts)),
-            MetricValue::EndTimestamp(ts) => Self::EndTimestampNsUTC(timestamp_to_ffi(ts)),
-            MetricValue::PruningMetrics {
-                name,
-                pruning_metrics,
-            } => Self::PruningMetrics {
-                name: SString::from(name.as_ref()),
-                pruning_metrics: pruning_metrics.into(),
-            },
-            MetricValue::Ratio {
-                name,
-                ratio_metrics,
-            } => Self::Ratio {
-                name: SString::from(name.as_ref()),
-                ratio_metrics: ratio_metrics.into(),
-            },
-            MetricValue::Custom { name, value } => Self::Custom {
-                name: SString::from(name.as_ref()),
-                display: SString::from(value.to_string().as_str()),
-                as_usize_value: value.as_usize() as u64,
-            },
-        }
-    }
-}
-
-impl From<FFI_MetricValue> for MetricValue {
-    fn from(v: FFI_MetricValue) -> Self {
-        match v {
-            FFI_MetricValue::OutputRows(n) => Self::OutputRows(count_from_value(n)),
-            FFI_MetricValue::ElapsedComputeNs(n) => {
-                Self::ElapsedCompute(time_from_nanos(n))
-            }
-            FFI_MetricValue::SpillCount(n) => Self::SpillCount(count_from_value(n)),
-            FFI_MetricValue::SpilledBytes(n) => Self::SpilledBytes(count_from_value(n)),
-            FFI_MetricValue::OutputBytes(n) => Self::OutputBytes(count_from_value(n)),
-            FFI_MetricValue::OutputBatches(n) => Self::OutputBatches(count_from_value(n)),
-            FFI_MetricValue::SpilledRows(n) => Self::SpilledRows(count_from_value(n)),
-            FFI_MetricValue::CurrentMemoryUsage(n) => {
-                Self::CurrentMemoryUsage(gauge_from_value(n))
-            }
-            FFI_MetricValue::Count { name, count } => Self::Count {
-                name: Cow::Owned(name.into()),
-                count: count_from_value(count),
-            },
-            FFI_MetricValue::Gauge { name, gauge } => Self::Gauge {
-                name: Cow::Owned(name.into()),
-                gauge: gauge_from_value(gauge),
-            },
-            FFI_MetricValue::Time { name, time_ns } => Self::Time {
-                name: Cow::Owned(name.into()),
-                time: time_from_nanos(time_ns),
-            },
-            FFI_MetricValue::StartTimestampNsUTC(nanos) => {
-                Self::StartTimestamp(timestamp_from_ffi(nanos))
-            }
-            FFI_MetricValue::EndTimestampNsUTC(nanos) => {
-                Self::EndTimestamp(timestamp_from_ffi(nanos))
-            }
-            FFI_MetricValue::PruningMetrics {
-                name,
-                pruning_metrics,
-            } => Self::PruningMetrics {
-                name: Cow::Owned(name.into()),
-                pruning_metrics: pruning_metrics.into(),
-            },
-            FFI_MetricValue::Ratio {
-                name,
-                ratio_metrics,
-            } => Self::Ratio {
-                name: Cow::Owned(name.into()),
-                ratio_metrics: ratio_metrics.into(),
-            },
-            FFI_MetricValue::Custom {
-                name,
-                display,
-                as_usize_value,
-            } => Self::Custom {
-                name: Cow::Owned(name.into()),
-                value: Arc::new(FfiCustomMetricValue {
-                    display: display.into(),
-                    as_usize_value: as_usize_value as usize,
-                }),
-            },
-        }
-    }
-}
-
-/// [`CustomMetricValue`] shim used when reconstructing a `MetricValue::Custom`
-/// on the consumer side of the FFI boundary.
-///
-/// The original `dyn CustomMetricValue` is not preserved across FFI — only its
-/// `Display` output and `as_usize()` fallback. As a result:
-/// - `Display` returns the producer-side rendered string.
-/// - `as_usize` returns the snapshot value captured at marshal time.
-/// - `aggregate` is a no-op: snapshots from FFI are not mergeable.
-/// - `as_any` only downcasts to `FfiCustomMetricValue` itself.
-#[derive(Debug)]
-pub struct FfiCustomMetricValue {
-    display: String,
-    as_usize_value: usize,
-}
-
-impl Display for FfiCustomMetricValue {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str(&self.display)
-    }
-}
-
-impl CustomMetricValue for FfiCustomMetricValue {
-    fn new_empty(&self) -> Arc<dyn CustomMetricValue> {
-        Arc::new(FfiCustomMetricValue {
-            display: String::new(),
-            as_usize_value: 0,
-        })
-    }
-
-    fn aggregate(&self, _other: Arc<dyn CustomMetricValue>) {
-        // FFI snapshots are immutable and not mergeable; aggregation across
-        // the boundary is intentionally a no-op.
-    }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
-    fn as_usize(&self) -> usize {
-        self.as_usize_value
-    }
-
-    fn is_eq(&self, other: &Arc<dyn CustomMetricValue>) -> bool {
-        other
-            .as_any()
-            .downcast_ref::<Self>()
-            .map(|o| o.display == self.display && o.as_usize_value == self.as_usize_value)
-            .unwrap_or(false)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use datafusion_physical_expr_common::metrics::Label;
-
-    /// Round-trips `value` through the FFI representation, asserts the
-    /// `PartialEq`-based equality, and returns the reconstructed value so
-    /// callers can additionally assert fields that `MetricValue`'s
-    /// `PartialEq` impl does not compare (e.g. `PruningMetrics::fully_matched`,
-    /// `RatioMetrics::merge_strategy`).
-    fn assert_value_roundtrip(value: MetricValue) -> MetricValue {
-        let ffi: FFI_MetricValue = (&value).into();
-        let back: MetricValue = ffi.into();
-        assert_eq!(value, back, "round-trip mismatch for {value:?}");
-        back
-    }
-
-    #[test]
-    fn roundtrip_named_variants() {
-        let c = Count::new();
-        c.add(7);
-        assert_value_roundtrip(MetricValue::OutputRows(c.clone()));
-        assert_value_roundtrip(MetricValue::SpillCount(c.clone()));
-        assert_value_roundtrip(MetricValue::SpilledBytes(c.clone()));
-        assert_value_roundtrip(MetricValue::OutputBytes(c.clone()));
-        assert_value_roundtrip(MetricValue::OutputBatches(c.clone()));
-        assert_value_roundtrip(MetricValue::SpilledRows(c.clone()));
-
-        let g = Gauge::new();
-        g.add(123);
-        assert_value_roundtrip(MetricValue::CurrentMemoryUsage(g));
-
-        let t = Time::new();
-        t.add_duration(std::time::Duration::from_nanos(456));
-        assert_value_roundtrip(MetricValue::ElapsedCompute(t));
-    }
-
-    #[test]
-    fn roundtrip_keyed_count_gauge_time() {
-        let count = Count::new();
-        count.add(11);
-        assert_value_roundtrip(MetricValue::Count {
-            name: Cow::Borrowed("custom_count"),
-            count,
-        });
-
-        let gauge = Gauge::new();
-        gauge.add(22);
-        assert_value_roundtrip(MetricValue::Gauge {
-            name: Cow::Borrowed("custom_gauge"),
-            gauge,
-        });
-
-        let time = Time::new();
-        time.add_duration(std::time::Duration::from_nanos(33));
-        assert_value_roundtrip(MetricValue::Time {
-            name: Cow::Borrowed("custom_time"),
-            time,
-        });
-    }
-
-    #[test]
-    fn roundtrip_timestamps() {
-        let unset = Timestamp::new();
-        assert_value_roundtrip(MetricValue::StartTimestamp(unset.clone()));
-        assert_value_roundtrip(MetricValue::EndTimestamp(unset));
-
-        let set = Timestamp::new();
-        set.set(DateTime::<Utc>::from_timestamp_nanos(
-            1_700_000_000_000_000_000,
-        ));
-        assert_value_roundtrip(MetricValue::StartTimestamp(set.clone()));
-        assert_value_roundtrip(MetricValue::EndTimestamp(set));
-    }
-
-    #[test]
-    fn roundtrip_pruning_and_ratio() {
-        let pruning = PruningMetrics::new();
-        pruning.add_pruned(3);
-        pruning.add_matched(5);
-        pruning.add_fully_matched(2);
-        let back = assert_value_roundtrip(MetricValue::PruningMetrics {
-            name: Cow::Borrowed("file_prune"),
-            pruning_metrics: pruning,
-        });
-        match back {
-            MetricValue::PruningMetrics {
-                name,
-                pruning_metrics,
-            } => {
-                assert_eq!(name.as_ref(), "file_prune");
-                assert_eq!(pruning_metrics.pruned(), 3);
-                assert_eq!(pruning_metrics.matched(), 5);
-                assert_eq!(pruning_metrics.fully_matched(), 2);
-            }
-            other => panic!("expected PruningMetrics, got {other:?}"),
-        }
-
-        let ratio = RatioMetrics::new()
-            .with_merge_strategy(RatioMergeStrategy::SetPartAddTotal)
-            .with_display_raw_values(false);
-        ratio.set_part(20);
-        ratio.set_total(100);
-        // `RatioMetrics`'s `PartialEq` does not compare `merge_strategy`,
-        // so assert it explicitly after the round-trip.
-        let back = assert_value_roundtrip(MetricValue::Ratio {
-            name: Cow::Borrowed("selectivity"),
-            ratio_metrics: ratio,
-        });
-        match back {
-            MetricValue::Ratio {
-                name,
-                ratio_metrics,
-            } => {
-                assert_eq!(name.as_ref(), "selectivity");
-                assert_eq!(ratio_metrics.part(), 20);
-                assert_eq!(ratio_metrics.total(), 100);
-                assert!(matches!(
-                    ratio_metrics.merge_strategy(),
-                    RatioMergeStrategy::SetPartAddTotal
-                ));
-                assert!(!ratio_metrics.display_raw_values());
-            }
-            other => panic!("expected Ratio, got {other:?}"),
-        }
-    }
-
-    #[test]
-    fn custom_metric_value_stringifies() {
-        #[derive(Debug)]
-        struct DummyCustom;
-        impl Display for DummyCustom {
-            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-                write!(f, "dummy:42")
-            }
-        }
-        impl CustomMetricValue for DummyCustom {
-            fn new_empty(&self) -> Arc<dyn CustomMetricValue> {
-                Arc::new(DummyCustom)
-            }
-            fn aggregate(&self, _other: Arc<dyn CustomMetricValue>) {}
-            fn as_any(&self) -> &dyn Any {
-                self
-            }
-            fn as_usize(&self) -> usize {
-                42
-            }
-            fn is_eq(&self, other: &Arc<dyn CustomMetricValue>) -> bool {
-                other.as_any().downcast_ref::<Self>().is_some()
-            }
-        }
-
-        let original = MetricValue::Custom {
-            name: Cow::Borrowed("dummy"),
-            value: Arc::new(DummyCustom),
-        };
-        let ffi: FFI_MetricValue = (&original).into();
-        let back: MetricValue = ffi.into();
-
-        match back {
-            MetricValue::Custom { name, value } => {
-                assert_eq!(name.as_ref(), "dummy");
-                assert_eq!(value.to_string(), "dummy:42");
-                assert_eq!(value.as_usize(), 42);
-            }
-            other => panic!("expected Custom, got {other:?}"),
-        }
-    }
-
-    #[test]
-    fn roundtrip_full_metric_with_labels_partition_type_category() {
-        let count = Count::new();
-        count.add(99);
-        let metric = Metric::new_with_labels(
-            MetricValue::OutputRows(count),
-            Some(3),
-            vec![Label::new("file", "a.parquet")],
-        )
-        .with_type(MetricType::Summary)
-        .with_category(MetricCategory::Rows);
-
-        let ffi: FFI_Metric = (&metric).into();
-        let back: Metric = ffi.into();
-
-        assert_eq!(back.value(), metric.value());
-        assert_eq!(back.partition(), Some(3));
-        assert_eq!(back.labels().len(), 1);
-        assert_eq!(back.labels()[0].name(), "file");
-        assert_eq!(back.labels()[0].value(), "a.parquet");
-        assert_eq!(back.metric_type(), MetricType::Summary);
-        assert_eq!(back.metric_category(), Some(MetricCategory::Rows));
-    }
-
-    #[test]
-    fn roundtrip_metrics_set() {
-        let mut set = MetricsSet::new();
-        let c = Count::new();
-        c.add(1);
-        set.push(Arc::new(Metric::new(MetricValue::OutputRows(c), Some(0))));
-        let c2 = Count::new();
-        c2.add(2);
-        set.push(Arc::new(Metric::new(MetricValue::OutputRows(c2), Some(1))));
-
-        let ffi: FFI_MetricsSet = (&set).into();
-        let back: MetricsSet = ffi.into();
-
-        // Aggregate check.
-        assert_eq!(back.output_rows(), Some(3));
-
-        // Per-metric check: ordering, partition, and per-partition value must
-        // all survive the round-trip (the aggregate above can't catch a lost
-        // partition or a metric count mismatch).
-        let metrics: Vec<_> = back.iter().collect();
-        assert_eq!(metrics.len(), 2);
-        assert_eq!(metrics[0].partition(), Some(0));
-        assert_eq!(
-            metrics[0].value(),
-            &MetricValue::OutputRows(count_from_value(1))
-        );
-        assert_eq!(metrics[1].partition(), Some(1));
-        assert_eq!(
-            metrics[1].value(),
-            &MetricValue::OutputRows(count_from_value(2))
-        );
-    }
-}
diff --git a/datafusion/ffi/src/physical_expr/metrics.rs b/datafusion/ffi/src/physical_expr/metrics.rs
index 6c29bd0ea6095..a6ee0dc457590 100644
--- a/datafusion/ffi/src/physical_expr/metrics.rs
+++ b/datafusion/ffi/src/physical_expr/metrics.rs
@@ -433,12 +433,8 @@ impl From<&MetricValue> for FFI_MetricValue {
                 name: SString::from(name.as_ref()),
                 time_ns: time.value() as u64,
             },
-            MetricValue::StartTimestamp(ts) => {
-                Self::StartTimestampNsUTC(timestamp_to_ffi(ts))
-            }
-            MetricValue::EndTimestamp(ts) => {
-                Self::EndTimestampNsUTC(timestamp_to_ffi(ts))
-            }
+            MetricValue::StartTimestamp(ts) => Self::StartTimestampNsUTC(timestamp_to_ffi(ts)),
+            MetricValue::EndTimestamp(ts) => Self::EndTimestampNsUTC(timestamp_to_ffi(ts)),
             MetricValue::PruningMetrics {
                 name,
                 pruning_metrics,

From e828d7dbd2b1b68a7a4c9a0ab57f42516a0d2085 Mon Sep 17 00:00:00 2001
From: mailmindlin <mailmindlin@users.noreply.github.com>
Date: Wed, 13 May 2026 13:23:02 -0400
Subject: [PATCH 7/9] fmt

---
 datafusion/ffi/src/physical_expr/metrics.rs | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/datafusion/ffi/src/physical_expr/metrics.rs b/datafusion/ffi/src/physical_expr/metrics.rs
index a6ee0dc457590..6c29bd0ea6095 100644
--- a/datafusion/ffi/src/physical_expr/metrics.rs
+++ b/datafusion/ffi/src/physical_expr/metrics.rs
@@ -433,8 +433,12 @@ impl From<&MetricValue> for FFI_MetricValue {
                 name: SString::from(name.as_ref()),
                 time_ns: time.value() as u64,
             },
-            MetricValue::StartTimestamp(ts) => Self::StartTimestampNsUTC(timestamp_to_ffi(ts)),
-            MetricValue::EndTimestamp(ts) => Self::EndTimestampNsUTC(timestamp_to_ffi(ts)),
+            MetricValue::StartTimestamp(ts) => {
+                Self::StartTimestampNsUTC(timestamp_to_ffi(ts))
+            }
+            MetricValue::EndTimestamp(ts) => {
+                Self::EndTimestampNsUTC(timestamp_to_ffi(ts))
+            }
             MetricValue::PruningMetrics {
                 name,
                 pruning_metrics,

From 0c214132fbed4a884a4c394288600a86ca49f9ba Mon Sep 17 00:00:00 2001
From: mailmindlin <mailmindlin@users.noreply.github.com>
Date: Wed, 13 May 2026 13:23:35 -0400
Subject: [PATCH 8/9] move metrics.rs

---
 datafusion/ffi/src/lib.rs | 1 -
 1 file changed, 1 deletion(-)

diff --git a/datafusion/ffi/src/lib.rs b/datafusion/ffi/src/lib.rs
index e9b703ddb4554..de3caf8c17260 100644
--- a/datafusion/ffi/src/lib.rs
+++ b/datafusion/ffi/src/lib.rs
@@ -34,7 +34,6 @@ pub mod execution_plan;
 pub mod expr;
 pub mod ffi_option;
 pub mod insert_op;
-pub mod metrics;
 pub mod physical_expr;
 pub mod physical_optimizer;
 pub mod plan_properties;

From a171b9ed9643d73ebc889fde02f344e8c0bbaa9d Mon Sep 17 00:00:00 2001
From: mailmindlin <mailmindlin@users.noreply.github.com>
Date: Wed, 13 May 2026 16:29:34 -0400
Subject: [PATCH 9/9] FFI statistics

---
 datafusion/ffi/src/execution_plan.rs | 100 ++++++++++++++++++-
 datafusion/ffi/src/lib.rs            |   1 +
 datafusion/ffi/src/statistics.rs     | 124 ++++++++++++++++++++++++
 datafusion/ffi/src/table_provider.rs | 140 +++++++++++++++++++++++++++
 4 files changed, 364 insertions(+), 1 deletion(-)
 create mode 100644 datafusion/ffi/src/statistics.rs

diff --git a/datafusion/ffi/src/execution_plan.rs b/datafusion/ffi/src/execution_plan.rs
index 1a8c9767fbed3..1df5f0d410b25 100644
--- a/datafusion/ffi/src/execution_plan.rs
+++ b/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<u64>,
+    ) -> 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<u64>,
+) -> FFI_Result<SVec<u8>> {
+    let partition: Option<usize> = Option::<u64>::from(partition).map(|p| p as usize);
+    plan.inner()
+        .partition_statistics(partition)
+        .map(|stats| serialize_statistics(stats.as_ref()).into_iter().collect())
+        .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,10 +476,24 @@ 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.map(|p| p as u64).into(),
+            )
+        })?;
+        Ok(Arc::new(deserialize_statistics(bytes.as_slice())?))
+    }
 }
 
 #[cfg(any(test, feature = "integration-tests"))]
 pub mod tests {
+    #[cfg(test)]
+    use datafusion_common::stats::Precision;
+    #[cfg(test)]
+    use datafusion_common::{ColumnStatistics, ScalarValue};
     use datafusion_physical_plan::Partitioning;
     use datafusion_physical_plan::execution_plan::{Boundedness, EmissionType};
 
@@ -468,6 +504,7 @@ pub mod tests {
         props: Arc<PlanProperties>,
         children: Vec<Arc<dyn ExecutionPlan>>,
         metrics: Option<MetricsSet>,
+        statistics: Option<Statistics>,
     }
 
     impl EmptyExec {
@@ -481,6 +518,7 @@ pub mod tests {
                 )),
                 children: Vec::default(),
                 metrics: None,
+                statistics: None,
             }
         }
 
@@ -488,6 +526,11 @@ pub mod tests {
             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 +564,7 @@ pub mod tests {
                 props: Arc::clone(&self.props),
                 children,
                 metrics: self.metrics.clone(),
+                statistics: self.statistics.clone(),
             }))
         }
 
@@ -536,6 +580,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 +712,51 @@ pub mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_ffi_execution_plan_partition_statistics_round_trip() -> Result<()> {
+        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![
diff --git a/datafusion/ffi/src/lib.rs b/datafusion/ffi/src/lib.rs
index de3caf8c17260..4df6c4b570f34 100644
--- a/datafusion/ffi/src/lib.rs
+++ b/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;
diff --git a/datafusion/ffi/src/statistics.rs b/datafusion/ffi/src/statistics.rs
new file mode 100644
index 0000000000000..019a8b2f22fab
--- /dev/null
+++ b/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);
+    }
+}
diff --git a/datafusion/ffi/src/table_provider.rs b/datafusion/ffi/src/table_provider.rs
index b6c077526c5f2..b23f22f13dcb6 100644
--- a/datafusion/ffi/src/table_provider.rs
+++ b/datafusion/ffi/src/table_provider.rs
@@ -22,6 +22,7 @@ use arrow::datatypes::SchemaRef;
 use async_ffi::{FfiFuture, FutureExt};
 use async_trait::async_trait;
 use datafusion_catalog::{Session, TableProvider};
+use datafusion_common::Statistics;
 use datafusion_common::error::{DataFusionError, Result};
 use datafusion_execution::TaskContext;
 use datafusion_expr::dml::InsertOp;
@@ -44,6 +45,7 @@ use crate::arrow_wrappers::WrappedSchema;
 use crate::execution::FFI_TaskContextProvider;
 use crate::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
 use crate::session::{FFI_SessionRef, ForeignSession};
+use crate::statistics::{deserialize_statistics, serialize_statistics};
 use crate::table_source::{FFI_TableProviderFilterPushDown, FFI_TableType};
 use crate::util::{FFI_Option, FFI_Result};
 use crate::{df_result, sresult_return};
@@ -133,6 +135,13 @@ pub struct FFI_TableProvider {
         insert_op: FFI_InsertOp,
     ) -> FfiFuture<FFI_Result<FFI_ExecutionPlan>>,
 
+    /// Snapshot the provider's table-level statistics. The inner
+    /// [`FFI_Option::None`] corresponds to [`TableProvider::statistics`]
+    /// returning `None`; `Some(bytes)` is a prost-encoded
+    /// `datafusion_proto_common::Statistics`.
+    pub statistics:
+        unsafe extern "C" fn(provider: &Self) -> FFI_Result<FFI_Option<SVec<u8>>>,
+
     pub logical_codec: FFI_LogicalExtensionCodec,
 
     /// Used to create a clone on the provider of the execution plan. This should
@@ -179,6 +188,16 @@ unsafe extern "C" fn schema_fn_wrapper(provider: &FFI_TableProvider) -> WrappedS
     provider.inner().schema().into()
 }
 
+unsafe extern "C" fn statistics_fn_wrapper(
+    provider: &FFI_TableProvider,
+) -> FFI_Result<FFI_Option<SVec<u8>>> {
+    let serialized: Option<SVec<u8>> = provider
+        .inner()
+        .statistics()
+        .map(|s| serialize_statistics(&s).into_iter().collect());
+    FFI_Result::Ok(serialized.into())
+}
+
 unsafe extern "C" fn table_type_fn_wrapper(
     provider: &FFI_TableProvider,
 ) -> FFI_TableType {
@@ -344,6 +363,7 @@ unsafe extern "C" fn clone_fn_wrapper(provider: &FFI_TableProvider) -> FFI_Table
         table_type: table_type_fn_wrapper,
         supports_filters_pushdown: provider.supports_filters_pushdown,
         insert_into: provider.insert_into,
+        statistics: statistics_fn_wrapper,
         logical_codec: provider.logical_codec.clone(),
         clone: clone_fn_wrapper,
         release: release_fn_wrapper,
@@ -404,6 +424,7 @@ impl FFI_TableProvider {
                 false => None,
             },
             insert_into: insert_into_fn_wrapper,
+            statistics: statistics_fn_wrapper,
             logical_codec,
             clone: clone_fn_wrapper,
             release: release_fn_wrapper,
@@ -451,6 +472,28 @@ impl TableProvider for ForeignTableProvider {
         unsafe { (self.0.table_type)(&self.0).into() }
     }
 
+    fn statistics(&self) -> Option<Statistics> {
+        let ffi_result = df_result!(unsafe { (self.0.statistics)(&self.0) });
+        let ffi_opt = match ffi_result {
+            Ok(v) => v,
+            Err(e) => {
+                log::warn!("FFI TableProvider::statistics failed: {e}");
+                debug_assert!(false, "FFI TableProvider::statistics failed: {e}");
+                return None;
+            }
+        };
+        let bytes: Option<SVec<u8>> = ffi_opt.into();
+        let bytes = bytes?;
+        match deserialize_statistics(bytes.as_slice()) {
+            Ok(stats) => Some(stats),
+            Err(e) => {
+                log::warn!("Failed to deserialize FFI statistics: {e}");
+                debug_assert!(false, "Failed to deserialize FFI statistics: {e}");
+                None
+            }
+        }
+    }
+
     async fn scan(
         &self,
         session: &dyn Session,
@@ -772,4 +815,101 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_ffi_table_provider_statistics_round_trip() -> Result<()> {
+        use arrow::datatypes::{DataType, Field};
+        use datafusion::arrow::array::Int32Array;
+        use datafusion::arrow::record_batch::RecordBatch;
+        use datafusion::datasource::MemTable;
+        use datafusion_common::stats::Precision;
+        use datafusion_common::{ColumnStatistics, ScalarValue};
+
+        // A thin wrapper that lets us inject statistics onto any TableProvider.
+        #[derive(Debug)]
+        struct TableWithStats {
+            inner: Arc<dyn TableProvider>,
+            stats: Option<Statistics>,
+        }
+
+        #[async_trait]
+        impl TableProvider for TableWithStats {
+            fn schema(&self) -> SchemaRef {
+                self.inner.schema()
+            }
+            fn table_type(&self) -> TableType {
+                self.inner.table_type()
+            }
+            fn statistics(&self) -> Option<Statistics> {
+                self.stats.clone()
+            }
+            async fn scan(
+                &self,
+                session: &dyn Session,
+                projection: Option<&Vec<usize>>,
+                filters: &[Expr],
+                limit: Option<usize>,
+            ) -> Result<Arc<dyn ExecutionPlan>> {
+                self.inner.scan(session, projection, filters, limit).await
+            }
+        }
+
+        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
+
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
+        )?;
+
+        let ctx = Arc::new(SessionContext::new());
+        let task_ctx_provider = Arc::clone(&ctx) as Arc<dyn TaskContextProvider>;
+        let task_ctx_provider = FFI_TaskContextProvider::from(&task_ctx_provider);
+
+        // Provider without statistics should cross the boundary as None.
+        let no_stats_inner = Arc::new(MemTable::try_new(
+            Arc::clone(&schema),
+            vec![vec![batch.clone()]],
+        )?);
+        let no_stats_provider = Arc::new(TableWithStats {
+            inner: no_stats_inner,
+            stats: None,
+        });
+        let mut ffi_provider = FFI_TableProvider::new(
+            no_stats_provider,
+            true,
+            None,
+            task_ctx_provider.clone(),
+            None,
+        );
+        ffi_provider.library_marker_id = crate::mock_foreign_marker_id;
+        let foreign: Arc<dyn TableProvider> = (&ffi_provider).into();
+        assert!(foreign.statistics().is_none());
+
+        // Provider with statistics should round-trip faithfully.
+        let original_stats = Statistics {
+            num_rows: Precision::Exact(3),
+            total_byte_size: Precision::Inexact(12),
+            column_statistics: vec![ColumnStatistics {
+                null_count: Precision::Exact(0),
+                max_value: Precision::Exact(ScalarValue::Int32(Some(3))),
+                min_value: Precision::Exact(ScalarValue::Int32(Some(1))),
+                sum_value: Precision::Exact(ScalarValue::Int64(Some(6))),
+                distinct_count: Precision::Exact(3),
+                byte_size: Precision::Exact(12),
+            }],
+        };
+        let stats_inner =
+            Arc::new(MemTable::try_new(Arc::clone(&schema), vec![vec![batch]])?);
+        let stats_provider = Arc::new(TableWithStats {
+            inner: stats_inner,
+            stats: Some(original_stats.clone()),
+        });
+        let mut ffi_provider =
+            FFI_TableProvider::new(stats_provider, true, None, task_ctx_provider, None);
+        ffi_provider.library_marker_id = crate::mock_foreign_marker_id;
+        let foreign: Arc<dyn TableProvider> = (&ffi_provider).into();
+        assert_eq!(foreign.statistics().as_ref(), Some(&original_stats));
+
+        Ok(())
+    }
 }