From 5cad9f94fe04592625f8db3317ce2be9db0e8ded Mon Sep 17 00:00:00 2001 From: sandudb Date: Wed, 26 Aug 2026 00:10:41 +0300 Subject: [PATCH 1/5] Added runtime statistics rom snapshot --- iceberg/src/data_source.rs | 102 +++++++++++++++++++++++++++++++++++-- 1 file changed, 98 insertions(+), 4 deletions(-) diff --git a/iceberg/src/data_source.rs b/iceberg/src/data_source.rs index ed15343dc..0dc31d1a7 100644 --- a/iceberg/src/data_source.rs +++ b/iceberg/src/data_source.rs @@ -1,6 +1,7 @@ use std::sync::Arc; use datafusion::arrow::datatypes::SchemaRef; +use datafusion::common::stats::Precision; use datafusion::common::{Statistics, exec_datafusion_err}; use datafusion::config::ConfigOptions; use datafusion::datasource::source::DataSource; @@ -18,6 +19,7 @@ use datafusion::prelude::Expr; use datafusion_distributed::WorkUnitFeed; use futures::{StreamExt, TryStreamExt}; use iceberg::arrow::ArrowReaderBuilder; +use iceberg::spec::SnapshotRef; use crate::common::{convert_filters_to_predicate, df_err, iceberg_err}; use crate::{IcebergConfig, IcebergWorkUnitFeed}; @@ -249,10 +251,7 @@ impl DataSource for IcebergDataSource { } fn partition_statistics(&self, _partition: Option) -> Result> { - // TODO: Implement planning time statistics for this DataSource. - // At this point, we have information about the iceberg::table::Table which we are about - // to read, so maybe there's something we can get from there. - Ok(Arc::new(Statistics::new_unknown(&self.schema))) + stats_from_snapshot(self.current_snapshot.as_ref(), &self.schema) } fn with_fetch(&self, fetch: Option) -> Option> { @@ -297,3 +296,98 @@ impl DataSource for IcebergDataSource { Ok(SortOrderPushdownResult::Unsupported) } } + +/// Getting statistics from the provided snapshot. +fn stats_from_snapshot( + snapshot: Option<&SnapshotRef>, + schema: &SchemaRef, +) -> Result> { + let Some(snap) = snapshot else { + return Ok(Arc::new(Statistics::new_unknown(schema))); + }; + let props = &snap.summary().additional_properties; + + let num_rows = props + .get("total-records") + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + let total_byte_size = props + .get("total-files-size") + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + + Ok(Arc::new(Statistics { + num_rows: Precision::Exact(num_rows), + total_byte_size: Precision::Exact(total_byte_size), + column_statistics: vec![], + })) +} + +#[cfg(test)] +mod tests { + use super::*; + use datafusion::arrow::datatypes::{DataType, Field, Schema}; + use iceberg::spec::{Operation, Snapshot, Summary}; + + fn schema() -> SchemaRef { + Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])) + } + + fn make_snapshot(extras: &[(&str, &str)]) -> SnapshotRef { + let props = extras + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + Arc::new( + Snapshot::builder() + .with_snapshot_id(1) + .with_timestamp_ms(0) + .with_sequence_number(0) + .with_schema_id(0) + .with_manifest_list("") + .with_summary(Summary { + operation: Operation::Append, + additional_properties: props, + }) + .build(), + ) + } + + #[test] + fn no_snapshot_returns_unknown_stats() { + let s = stats_from_snapshot(None, &schema()).unwrap(); + assert!(matches!(s.num_rows, Precision::Absent)); + assert!(matches!(s.total_byte_size, Precision::Absent)); + } + + #[test] + fn valid_props_are_parsed_exactly_stats() { + let snap = make_snapshot(&[("total-records", "100"), ("total-files-size", "4096")]); + let s = stats_from_snapshot(Some(&snap), &schema()).unwrap(); + assert_eq!(s.num_rows, Precision::Exact(100)); + assert_eq!(s.total_byte_size, Precision::Exact(4096)); + } + + #[test] + fn missing_props_default_to_zero_stats() { + let snap = make_snapshot(&[]); + let s = stats_from_snapshot(Some(&snap), &schema()).unwrap(); + assert_eq!(s.num_rows, Precision::Exact(0)); + assert_eq!(s.total_byte_size, Precision::Exact(0)); + } + + #[test] + fn unparseable_props_default_to_zero_stats() { + let snap = make_snapshot(&[("total-records", "3.14"), ("total-files-size", "-1")]); + let s = stats_from_snapshot(Some(&snap), &schema()).unwrap(); + assert_eq!(s.num_rows, Precision::Exact(0)); + assert_eq!(s.total_byte_size, Precision::Exact(0)); + } + + #[test] + fn column_statistics_are_empty_stats() { + let snap = make_snapshot(&[("total-records", "10"), ("total-files-size", "512")]); + let s = stats_from_snapshot(Some(&snap), &schema()).unwrap(); + assert!(s.column_statistics.is_empty()); + } +} From 63bd0293d6b8a28b280a27fc4f0351b96ed24dc7 Mon Sep 17 00:00:00 2001 From: sandudb Date: Wed, 26 Aug 2026 00:29:35 +0300 Subject: [PATCH 2/5] Refactored compilation error part --- iceberg/src/data_source.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/iceberg/src/data_source.rs b/iceberg/src/data_source.rs index 0dc31d1a7..03e1f1ef3 100644 --- a/iceberg/src/data_source.rs +++ b/iceberg/src/data_source.rs @@ -117,6 +117,7 @@ pub struct IcebergDataSource { partitioning: Partitioning, fetch: Option, metrics: ExecutionPlanMetricsSet, + current_snapshot: Option, iceberg_file_io: iceberg::io::FileIO, iceberg_runtime: iceberg::Runtime, feed: WorkUnitFeed, @@ -150,6 +151,8 @@ impl IcebergDataSource { .collect::>() }); + let current_snapshot = table.metadata().current_snapshot().cloned(); + let predicates = convert_filters_to_predicate(opts.filters); Self { @@ -169,6 +172,7 @@ impl IcebergDataSource { partitioning, sync_manager: Default::default(), }), + current_snapshot, } } } From 7d3b5cdbcb9a5edc6d153e30a7f2695f5d22134f Mon Sep 17 00:00:00 2001 From: sandudb Date: Wed, 26 Aug 2026 14:58:58 +0300 Subject: [PATCH 3/5] Resolved comments, added Iceberg spec info link + iceberg-rust naming link --- iceberg/src/data_source.rs | 35 +++++++++++++++++++++-------------- 1 file changed, 21 insertions(+), 14 deletions(-) diff --git a/iceberg/src/data_source.rs b/iceberg/src/data_source.rs index 03e1f1ef3..c9949392e 100644 --- a/iceberg/src/data_source.rs +++ b/iceberg/src/data_source.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use datafusion::arrow::datatypes::SchemaRef; use datafusion::common::stats::Precision; -use datafusion::common::{Statistics, exec_datafusion_err}; +use datafusion::common::{ColumnStatistics, Statistics, exec_datafusion_err}; use datafusion::config::ConfigOptions; use datafusion::datasource::source::DataSource; use datafusion::error::Result; @@ -24,6 +24,14 @@ use iceberg::spec::SnapshotRef; use crate::common::{convert_filters_to_predicate, df_err, iceberg_err}; use crate::{IcebergConfig, IcebergWorkUnitFeed}; +/// Snapshot summary keys defined by the Iceberg table spec: +/// https://iceberg.apache.org/spec/#optional-snapshot-summary-fields +/// +/// iceberg-rust defines them privately: +/// https://github.com/apache/iceberg-rust/blob/4168a0b2950dc5f85588e5cb3ab6796e5228b309/crates/iceberg/src/spec/snapshot_summary.rs#L46-L47 +const TOTAL_RECORDS: &str = "total-records"; +const TOTAL_FILE_SIZE: &str = "total-files-size"; + /// Consumes a stream of [iceberg::scan::FileScanTask]s per partition and reads the underlying /// files into an Arrow stream. /// @@ -307,23 +315,28 @@ fn stats_from_snapshot( schema: &SchemaRef, ) -> Result> { let Some(snap) = snapshot else { - return Ok(Arc::new(Statistics::new_unknown(schema))); + // A table with no current snapshot has never had a commit. It was created, but zero data files were added + return Ok(Arc::new(Statistics { + num_rows: Precision::Exact(0), + total_byte_size: Precision::Exact(0), + column_statistics: vec![ColumnStatistics::new_unknown(); schema.fields().len()], + })); }; let props = &snap.summary().additional_properties; let num_rows = props - .get("total-records") + .get(TOTAL_RECORDS) .and_then(|v| v.parse().ok()) .unwrap_or(0); let total_byte_size = props - .get("total-files-size") + .get(TOTAL_FILE_SIZE) .and_then(|v| v.parse().ok()) .unwrap_or(0); Ok(Arc::new(Statistics { num_rows: Precision::Exact(num_rows), total_byte_size: Precision::Exact(total_byte_size), - column_statistics: vec![], + column_statistics: vec![ColumnStatistics::new_unknown(); schema.fields().len()], })) } @@ -357,11 +370,12 @@ mod tests { ) } + /// Table was created, but not committed => means no snapshot and everything is set to zero #[test] fn no_snapshot_returns_unknown_stats() { let s = stats_from_snapshot(None, &schema()).unwrap(); - assert!(matches!(s.num_rows, Precision::Absent)); - assert!(matches!(s.total_byte_size, Precision::Absent)); + assert!(matches!(s.num_rows, Precision::Exact(0))); + assert!(matches!(s.total_byte_size, Precision::Exact(0))); } #[test] @@ -387,11 +401,4 @@ mod tests { assert_eq!(s.num_rows, Precision::Exact(0)); assert_eq!(s.total_byte_size, Precision::Exact(0)); } - - #[test] - fn column_statistics_are_empty_stats() { - let snap = make_snapshot(&[("total-records", "10"), ("total-files-size", "512")]); - let s = stats_from_snapshot(Some(&snap), &schema()).unwrap(); - assert!(s.column_statistics.is_empty()); - } } From 5d9bf9f9c4ae7c2ad0064810baed11c59e0b0b75 Mon Sep 17 00:00:00 2001 From: sandudb Date: Wed, 26 Aug 2026 15:25:36 +0300 Subject: [PATCH 4/5] Added physical plan extraction to test harness + separate integration test file for stats --- iceberg/src/data_source.rs | 63 -------------- iceberg/src/test_utils/harness.rs | 6 +- iceberg/tests/statistics.rs | 133 ++++++++++++++++++++++++++++++ 3 files changed, 138 insertions(+), 64 deletions(-) create mode 100644 iceberg/tests/statistics.rs diff --git a/iceberg/src/data_source.rs b/iceberg/src/data_source.rs index c9949392e..6a3666822 100644 --- a/iceberg/src/data_source.rs +++ b/iceberg/src/data_source.rs @@ -339,66 +339,3 @@ fn stats_from_snapshot( column_statistics: vec![ColumnStatistics::new_unknown(); schema.fields().len()], })) } - -#[cfg(test)] -mod tests { - use super::*; - use datafusion::arrow::datatypes::{DataType, Field, Schema}; - use iceberg::spec::{Operation, Snapshot, Summary}; - - fn schema() -> SchemaRef { - Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])) - } - - fn make_snapshot(extras: &[(&str, &str)]) -> SnapshotRef { - let props = extras - .iter() - .map(|(k, v)| (k.to_string(), v.to_string())) - .collect(); - Arc::new( - Snapshot::builder() - .with_snapshot_id(1) - .with_timestamp_ms(0) - .with_sequence_number(0) - .with_schema_id(0) - .with_manifest_list("") - .with_summary(Summary { - operation: Operation::Append, - additional_properties: props, - }) - .build(), - ) - } - - /// Table was created, but not committed => means no snapshot and everything is set to zero - #[test] - fn no_snapshot_returns_unknown_stats() { - let s = stats_from_snapshot(None, &schema()).unwrap(); - assert!(matches!(s.num_rows, Precision::Exact(0))); - assert!(matches!(s.total_byte_size, Precision::Exact(0))); - } - - #[test] - fn valid_props_are_parsed_exactly_stats() { - let snap = make_snapshot(&[("total-records", "100"), ("total-files-size", "4096")]); - let s = stats_from_snapshot(Some(&snap), &schema()).unwrap(); - assert_eq!(s.num_rows, Precision::Exact(100)); - assert_eq!(s.total_byte_size, Precision::Exact(4096)); - } - - #[test] - fn missing_props_default_to_zero_stats() { - let snap = make_snapshot(&[]); - let s = stats_from_snapshot(Some(&snap), &schema()).unwrap(); - assert_eq!(s.num_rows, Precision::Exact(0)); - assert_eq!(s.total_byte_size, Precision::Exact(0)); - } - - #[test] - fn unparseable_props_default_to_zero_stats() { - let snap = make_snapshot(&[("total-records", "3.14"), ("total-files-size", "-1")]); - let s = stats_from_snapshot(Some(&snap), &schema()).unwrap(); - assert_eq!(s.num_rows, Precision::Exact(0)); - assert_eq!(s.total_byte_size, Precision::Exact(0)); - } -} diff --git a/iceberg/src/test_utils/harness.rs b/iceberg/src/test_utils/harness.rs index c8be724c3..b837b2ca1 100644 --- a/iceberg/src/test_utils/harness.rs +++ b/iceberg/src/test_utils/harness.rs @@ -7,7 +7,7 @@ use datafusion::arrow::util::pretty::pretty_format_batches; use datafusion::dataframe::DataFrame; use datafusion::error::Result; use datafusion::execution::SessionStateBuilder; -use datafusion::physical_plan::displayable; +use datafusion::physical_plan::{ExecutionPlan, displayable}; use datafusion::prelude::{SessionConfig, SessionContext}; use futures::StreamExt; use futures::stream::BoxStream; @@ -58,6 +58,10 @@ impl IcebergTestHarness { pretty_format_batches(&batches)?.to_string(), )) } + + pub async fn physical_plan(&self, sql: &str) -> Result> { + self.ctx.sql(sql).await?.create_physical_plan().await + } } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/iceberg/tests/statistics.rs b/iceberg/tests/statistics.rs new file mode 100644 index 000000000..0f847c07b --- /dev/null +++ b/iceberg/tests/statistics.rs @@ -0,0 +1,133 @@ +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use datafusion::common::stats::Precision; + use datafusion::common::Statistics; + use datafusion::datasource::source::DataSourceExec; + use datafusion::error::Result; + use datafusion::physical_plan::{displayable, ExecutionPlan}; + use datafusion_distributed_iceberg::test_utils::IcebergTestHarness; + use datafusion_distributed_iceberg::IcebergDataSource; + + // Values from testdata/iceberg/taxi/metadata/v1.metadata.json snapshot summary. + const TAXI_ROWS: usize = 175_000; + const TAXI_BYTES: usize = 4_480_382; + const TAXI_COLUMNS: usize = 13; + + #[tokio::test] + async fn reports_exact_row_count_and_byte_size_for_full_scan() -> Result<()> { + let harness = IcebergTestHarness::new().await?; + let stats = source_statistics(&harness, "SELECT * FROM taxi").await?; + + assert_eq!(stats.num_rows, Precision::Exact(TAXI_ROWS)); + assert_eq!(stats.total_byte_size, Precision::Exact(TAXI_BYTES)); + Ok(()) + } + + #[tokio::test] + async fn column_statistics_match_full_schema() -> Result<()> { + let harness = IcebergTestHarness::new().await?; + let stats = source_statistics(&harness, "SELECT * FROM taxi").await?; + + assert_eq!(stats.column_statistics.len(), TAXI_COLUMNS); + Ok(()) + } + + #[tokio::test] + async fn column_statistics_match_projected_schema() -> Result<()> { + // Regression: a column_statistics vec shorter than the output schema + // makes DataFusion panic while propagating statistics upstream. + let harness = IcebergTestHarness::new().await?; + let stats = source_statistics(&harness, "SELECT vendor_id, pickup_date FROM taxi").await?; + + assert_eq!(stats.column_statistics.len(), 2); + assert_eq!(stats.num_rows, Precision::Exact(TAXI_ROWS)); + Ok(()) + } + + #[tokio::test] + async fn statistics_propagate_through_filter() -> Result<()> { + let harness = IcebergTestHarness::new().await?; + let plan = harness + .physical_plan("SELECT vendor_id FROM taxi WHERE pickup_date = DATE '2024-01-10'") + .await?; + let stats = plan.partition_statistics(None)?; + + // The filter cannot keep the count exact, but it must not lose it. + assert!(matches!(stats.num_rows, Precision::Inexact(_))); + assert_eq!(stats.column_statistics.len(), 1); + Ok(()) + } + + #[tokio::test] + async fn statistics_propagate_through_projection_and_sort() -> Result<()> { + let harness = IcebergTestHarness::new().await?; + let plan = harness + .physical_plan("SELECT vendor_id, trip_distance FROM taxi ORDER BY pickup_at") + .await?; + let stats = plan.partition_statistics(None)?; + + assert_eq!(stats.num_rows, Precision::Exact(TAXI_ROWS)); + assert_eq!(stats.column_statistics.len(), 2); + Ok(()) + } + + #[tokio::test] + async fn explain_shows_statistics_on_the_iceberg_source() -> Result<()> { + let harness = IcebergTestHarness::new().await?; + let plan = harness.physical_plan("SELECT vendor_id FROM taxi").await?; + let display = displayable(plan.as_ref()) + .set_show_statistics(true) + .indent(true) + .to_string(); + + insta::assert_snapshot!(display, @" + CooperativeExec, statistics=[Rows=Exact(175000), Bytes=Exact(4480382), [(Col[0]:)]] + DataSourceExec: format=iceberg, projection=[vendor_id], statistics=[Rows=Exact(175000), Bytes=Exact(4480382), [(Col[0]:)]] + "); + Ok(()) + } + + #[tokio::test] + async fn exact_row_count_lets_count_star_skip_the_scan() -> Result<()> { + // With Precision::Exact(num_rows) the AggregateStatistics optimizer + // rule answers COUNT(*) from metadata without reading any data file. + let harness = IcebergTestHarness::new().await?; + let (plan, batches) = harness.query("SELECT count(*) FROM taxi").await?; + + insta::assert_snapshot!(plan, @" + ProjectionExec: expr=[175000 as count(*)] + PlaceholderRowExec + "); + insta::assert_snapshot!(batches, @" + +----------+ + | count(*) | + +----------+ + | 175000 | + +----------+ + "); + Ok(()) + } + + /// Finds the single Iceberg `DataSourceExec` in the plan and returns the + /// statistics reported by the `IcebergDataSource` itself. + async fn source_statistics(harness: &IcebergTestHarness, sql: &str) -> Result { + let plan = harness.physical_plan(sql).await?; + let exec = find_iceberg_exec(&plan).expect("plan contains an Iceberg DataSourceExec"); + Ok(Arc::unwrap_or_clone(exec.partition_statistics(None)?)) + } + + fn find_iceberg_exec(plan: &Arc) -> Option> { + if let Some(exec) = plan.downcast_ref::() { + if exec + .data_source() + .downcast_ref::() + .is_some() + { + return Some(Arc::new(exec.clone())); + } + } + plan.children().into_iter().find_map(find_iceberg_exec) + } +} From fa023263c7aa88fb146e9722fdaee639379c8783 Mon Sep 17 00:00:00 2001 From: sandudb Date: Wed, 26 Aug 2026 15:35:27 +0300 Subject: [PATCH 5/5] Fixed formatting --- iceberg/tests/statistics.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/iceberg/tests/statistics.rs b/iceberg/tests/statistics.rs index 0f847c07b..87ffa131b 100644 --- a/iceberg/tests/statistics.rs +++ b/iceberg/tests/statistics.rs @@ -2,13 +2,13 @@ mod tests { use std::sync::Arc; - use datafusion::common::stats::Precision; use datafusion::common::Statistics; + use datafusion::common::stats::Precision; use datafusion::datasource::source::DataSourceExec; use datafusion::error::Result; - use datafusion::physical_plan::{displayable, ExecutionPlan}; - use datafusion_distributed_iceberg::test_utils::IcebergTestHarness; + use datafusion::physical_plan::{ExecutionPlan, displayable}; use datafusion_distributed_iceberg::IcebergDataSource; + use datafusion_distributed_iceberg::test_utils::IcebergTestHarness; // Values from testdata/iceberg/taxi/metadata/v1.metadata.json snapshot summary. const TAXI_ROWS: usize = 175_000;