From be6aa9b2312aa8f4723a7086d92226daa523c6eb Mon Sep 17 00:00:00 2001
From: JingsongLi <jingsonglee0@gmail.com>
Date: Mon, 6 Apr 2026 18:56:22 +0800
Subject: [PATCH 1/2] feat: Implement statistics for data fusion scan

---
 .../datafusion/src/physical_plan/scan.rs      | 30 +++++++++++++++++++
 1 file changed, 30 insertions(+)

diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs b/crates/integrations/datafusion/src/physical_plan/scan.rs
index d7dfc7aa..00257eae 100644
--- a/crates/integrations/datafusion/src/physical_plan/scan.rs
+++ b/crates/integrations/datafusion/src/physical_plan/scan.rs
@@ -19,6 +19,8 @@ use std::any::Any;
 use std::sync::Arc;
 
 use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::common::stats::Precision;
+use datafusion::common::Statistics;
 use datafusion::error::Result as DFResult;
 use datafusion::execution::{SendableRecordBatchStream, TaskContext};
 use datafusion::physical_expr::EquivalenceProperties;
@@ -165,6 +167,34 @@ impl ExecutionPlan for PaimonTableScan {
             futures::stream::once(fut).try_flatten(),
         )))
     }
+
+    fn statistics(&self) -> DFResult<Statistics> {
+        self.partition_statistics(None)
+    }
+
+    fn partition_statistics(&self, partition: Option<usize>) -> DFResult<Statistics> {
+        let partitions: &[Arc<[DataSplit]>] = match partition {
+            Some(idx) => std::slice::from_ref(&self.planned_partitions[idx]),
+            None => &self.planned_partitions,
+        };
+
+        let mut total_rows: usize = 0;
+        let mut total_bytes: usize = 0;
+        for splits in partitions {
+            for split in splits.iter() {
+                total_rows += split.merged_row_count().unwrap_or(split.row_count()) as usize;
+                for file in split.data_files() {
+                    total_bytes += file.file_size as usize;
+                }
+            }
+        }
+
+        Ok(Statistics {
+            num_rows: Precision::Inexact(total_rows),
+            total_byte_size: Precision::Inexact(total_bytes),
+            column_statistics: Statistics::unknown_column(&self.schema()),
+        })
+    }
 }
 
 impl DisplayAs for PaimonTableScan {

From aff692b95d809aff77ab55d6dc9e52536f3e02e9 Mon Sep 17 00:00:00 2001
From: JingsongLi <jingsonglee0@gmail.com>
Date: Mon, 6 Apr 2026 22:06:22 +0800
Subject: [PATCH 2/2] Enrich DisplayAs

---
 .../datafusion/src/physical_plan/scan.rs       | 18 +++++++++++++++++-
 1 file changed, 17 insertions(+), 1 deletion(-)

diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs b/crates/integrations/datafusion/src/physical_plan/scan.rs
index 00257eae..1a181eba 100644
--- a/crates/integrations/datafusion/src/physical_plan/scan.rs
+++ b/crates/integrations/datafusion/src/physical_plan/scan.rs
@@ -203,11 +203,27 @@ impl DisplayAs for PaimonTableScan {
         _t: datafusion::physical_plan::DisplayFormatType,
         f: &mut std::fmt::Formatter,
     ) -> std::fmt::Result {
+        write!(f, "PaimonTableScan: table={}", self.table.identifier())?;
+
+        let total_splits: usize = self.planned_partitions.iter().map(|p| p.len()).sum();
+        let total_files: usize = self
+            .planned_partitions
+            .iter()
+            .flat_map(|p| p.iter())
+            .map(|s| s.data_files().len())
+            .sum();
         write!(
             f,
-            "PaimonTableScan: partitions={}",
+            ", partitions={}, splits={total_splits}, files={total_files}",
             self.planned_partitions.len()
         )?;
+
+        if let Some(ref columns) = self.projected_columns {
+            write!(f, ", projection=[{}]", columns.join(", "))?;
+        }
+        if let Some(ref predicate) = self.pushed_predicate {
+            write!(f, ", predicate={predicate}")?;
+        }
         if let Some(limit) = self.limit {
             write!(f, ", limit={limit}")?;
         }