From 1f14ba1b8d3e84882e2169a67310be2d28488c80 Mon Sep 17 00:00:00 2001 From: Anurag Mantripragada Date: Mon, 12 Oct 2026 15:27:34 -0700 Subject: [PATCH 1/4] feat(scan): optionally retain column statistics on FileScanTask Add TableScanBuilder::with_column_stats, mirroring Java's Scan.includeColumnStats. The named columns resolve to field ids at build time with the scan's case sensitivity, and each planned FileScanTask carries a FileScanTaskColumnStats holding the data file's value, null and NaN counts and lower and upper bounds for only those field ids. Retention is opt-in: by default column_stats is None, nothing is copied from the manifest entry, and the field is omitted when a task is serialized, so serialized tasks are unchanged. --- crates/iceberg/public-api.txt | 26 +++- crates/iceberg/src/scan/context.rs | 15 ++- crates/iceberg/src/scan/mod.rs | 182 +++++++++++++++++++++++++- crates/iceberg/src/scan/task.rs | 108 ++++++++++++++- crates/iceberg/src/test_utils/scan.rs | 72 ++++++++++ 5 files changed, 396 insertions(+), 7 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 4666147573..36e14a92c6 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1331,6 +1331,7 @@ pub mod iceberg::scan pub struct iceberg::scan::FileScanTask impl iceberg::scan::FileScanTask pub fn iceberg::scan::FileScanTask::case_sensitive(&self) -> bool +pub fn iceberg::scan::FileScanTask::column_stats(&self) -> core::option::Option<&iceberg::scan::FileScanTaskColumnStats> pub fn iceberg::scan::FileScanTask::data_file_format(&self) -> iceberg::spec::DataFileFormat pub fn iceberg::scan::FileScanTask::data_file_path(&self) -> &str pub fn iceberg::scan::FileScanTask::data_sequence_number(&self) -> core::option::Option @@ -1361,11 +1362,33 @@ impl core::fmt::Debug for iceberg::scan::FileScanTask pub fn iceberg::scan::FileScanTask::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTask impl iceberg::scan::FileScanTask -pub fn iceberg::scan::FileScanTask::builder() -> FileScanTaskBuilder<((), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), ())> +pub fn iceberg::scan::FileScanTask::builder() -> FileScanTaskBuilder<((), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), ())> impl serde_core::ser::Serialize for iceberg::scan::FileScanTask pub fn iceberg::scan::FileScanTask::serialize(&self, serializer: S) -> core::result::Result<::Ok, ::Error> where S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTask pub fn iceberg::scan::FileScanTask::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg::scan::FileScanTaskColumnStats +impl iceberg::scan::FileScanTaskColumnStats +pub fn iceberg::scan::FileScanTaskColumnStats::lower_bounds(&self) -> &std::collections::hash::map::HashMap +pub fn iceberg::scan::FileScanTaskColumnStats::nan_value_counts(&self) -> &std::collections::hash::map::HashMap +pub fn iceberg::scan::FileScanTaskColumnStats::null_value_counts(&self) -> &std::collections::hash::map::HashMap +pub fn iceberg::scan::FileScanTaskColumnStats::upper_bounds(&self) -> &std::collections::hash::map::HashMap +pub fn iceberg::scan::FileScanTaskColumnStats::value_counts(&self) -> &std::collections::hash::map::HashMap +impl core::clone::Clone for iceberg::scan::FileScanTaskColumnStats +pub fn iceberg::scan::FileScanTaskColumnStats::clone(&self) -> iceberg::scan::FileScanTaskColumnStats +impl core::cmp::PartialEq for iceberg::scan::FileScanTaskColumnStats +pub fn iceberg::scan::FileScanTaskColumnStats::eq(&self, other: &iceberg::scan::FileScanTaskColumnStats) -> bool +impl core::default::Default for iceberg::scan::FileScanTaskColumnStats +pub fn iceberg::scan::FileScanTaskColumnStats::default() -> iceberg::scan::FileScanTaskColumnStats +impl core::fmt::Debug for iceberg::scan::FileScanTaskColumnStats +pub fn iceberg::scan::FileScanTaskColumnStats::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTaskColumnStats +impl iceberg::scan::FileScanTaskColumnStats +pub fn iceberg::scan::FileScanTaskColumnStats::builder() -> FileScanTaskColumnStatsBuilder<((), (), (), (), ())> +impl serde_core::ser::Serialize for iceberg::scan::FileScanTaskColumnStats +pub fn iceberg::scan::FileScanTaskColumnStats::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTaskColumnStats +pub fn iceberg::scan::FileScanTaskColumnStats::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::scan::FileScanTaskDeleteFile impl iceberg::scan::FileScanTaskDeleteFile pub fn iceberg::scan::FileScanTaskDeleteFile::content_offset(&self) -> core::option::Option @@ -1423,6 +1446,7 @@ pub fn iceberg::scan::TableScanBuilder<'a>::snapshot_id(self, snapshot_id: i64) pub fn iceberg::scan::TableScanBuilder<'a>::with_batch_size(self, batch_size: core::option::Option) -> Self pub fn iceberg::scan::TableScanBuilder<'a>::with_bloom_filter_enabled(self, bloom_filter_enabled: bool) -> Self pub fn iceberg::scan::TableScanBuilder<'a>::with_case_sensitive(self, case_sensitive: bool) -> Self +pub fn iceberg::scan::TableScanBuilder<'a>::with_column_stats(self, column_names: impl core::iter::traits::collect::IntoIterator) -> Self pub fn iceberg::scan::TableScanBuilder<'a>::with_concurrency_limit(self, limit: usize) -> Self pub fn iceberg::scan::TableScanBuilder<'a>::with_data_file_concurrency_limit(self, limit: usize) -> Self pub fn iceberg::scan::TableScanBuilder<'a>::with_filter(self, predicate: iceberg::expr::Predicate) -> Self diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index d17835afa5..c04c3601ce 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -25,8 +25,8 @@ use crate::delete_file_index::DeleteFileIndex; use crate::expr::{Bind, BoundPredicate, Predicate}; use crate::io::object_cache::ObjectCache; use crate::scan::{ - BoundPredicates, ExpressionEvaluatorCache, FileScanTask, ManifestEvaluatorCache, - PartitionFilterCache, + BoundPredicates, ExpressionEvaluatorCache, FileScanTask, FileScanTaskColumnStats, + ManifestEvaluatorCache, PartitionFilterCache, }; use crate::spec::{ ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, NameMapping, @@ -52,6 +52,7 @@ pub(crate) struct ManifestFileContext { partition_spec: Option, unified_partition_type: Option>, sort_orders: Arc>, + column_stats_field_ids: Option>>, } /// Wraps a [`ManifestEntryRef`] alongside the objects that are needed @@ -71,6 +72,7 @@ pub(crate) struct ManifestEntryContext { pub unified_partition_type: Option>, pub sort_order_id: Option, pub sort_order: Option, + pub column_stats_field_ids: Option>>, } impl ManifestFileContext { @@ -91,6 +93,7 @@ impl ManifestFileContext { partition_spec, unified_partition_type, sort_orders, + column_stats_field_ids, } = self; let manifest = object_cache.get_manifest(&manifest_file).await?; @@ -122,6 +125,7 @@ impl ManifestFileContext { unified_partition_type: unified_partition_type.clone(), sort_order_id, sort_order, + column_stats_field_ids: column_stats_field_ids.clone(), }; sender @@ -170,6 +174,9 @@ impl ManifestEntryContext { .with_key_metadata(self.manifest_entry.data_file.key_metadata().map(Box::from)) .with_sort_order_id(self.sort_order_id) .with_sort_order(self.sort_order) + .with_column_stats(self.column_stats_field_ids.map(|field_ids| { + FileScanTaskColumnStats::from_data_file(self.manifest_entry.data_file(), &field_ids) + })) .build() } @@ -216,6 +223,9 @@ pub(crate) struct PlanContext { /// [`ManifestFileContext`] carries only this narrow map rather than the full table /// metadata. Mirrors how `unified_partition_type` carries a precomputed value. pub sort_orders: Arc>, + + /// Field ids whose column statistics each [`FileScanTask`] retains, or `None` to retain none. + pub column_stats_field_ids: Option>>, } impl PlanContext { @@ -347,6 +357,7 @@ impl PlanContext { .cloned(), unified_partition_type: self.unified_partition_type.clone(), sort_orders: self.sort_orders.clone(), + column_stats_field_ids: self.column_stats_field_ids.clone(), } } } diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 7e81900a5b..71390a23a8 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -103,6 +103,26 @@ fn collect_scan_field_ids( .collect() } +fn collect_column_stats_field_ids( + schema: &Schema, + column_names: &[String], + case_sensitive: bool, +) -> Result>>> { + if column_names.is_empty() { + return Ok(None); + } + + column_names + .iter() + .map(|column_name| { + resolve_field_id(schema, column_name, case_sensitive)?.ok_or_else(|| { + invalid_data!("Column {column_name} not found in table. Schema: {schema}") + }) + }) + .collect::>>() + .map(|field_ids| Some(Arc::new(field_ids))) +} + fn bind_scan_predicate( schema: &SchemaRef, predicate: Option<&Predicate>, @@ -135,6 +155,7 @@ pub struct TableScanBuilder<'a> { table: &'a Table, // Defaults to none which means select all columns column_names: Option>, + column_stats_names: Vec, snapshot_id: Option, batch_size: Option, case_sensitive: bool, @@ -154,6 +175,7 @@ impl<'a> TableScanBuilder<'a> { Self { table, column_names: None, + column_stats_names: vec![], snapshot_id: None, batch_size: None, case_sensitive: true, @@ -211,6 +233,25 @@ impl<'a> TableScanBuilder<'a> { self } + /// Retains the data file's column statistics for the named columns on each planned + /// [`FileScanTask`], exposed as [`FileScanTask::column_stats`]. + /// + /// Names resolve against the scan's schema with the scan's case sensitivity and may name + /// nested fields. An empty list retains no statistics, which is the default. + /// + /// [`build`](Self::build) fails if a name matches no field, or more than one field when the + /// scan is case-insensitive. + pub fn with_column_stats( + mut self, + column_names: impl IntoIterator, + ) -> Self { + self.column_stats_names = column_names + .into_iter() + .map(|item| item.to_string()) + .collect(); + self + } + /// Set the snapshot to scan. When not set, it uses current snapshot. pub fn snapshot_id(mut self, snapshot_id: i64) -> Self { self.snapshot_id = Some(snapshot_id); @@ -316,6 +357,8 @@ impl<'a> TableScanBuilder<'a> { let schema = snapshot.schema(self.table.metadata())?; let field_ids = collect_scan_field_ids(&schema, self.column_names.as_deref(), self.case_sensitive)?; + let column_stats_field_ids = + collect_column_stats_field_ids(&schema, &self.column_stats_names, self.case_sensitive)?; let snapshot_bound_predicate = bind_scan_predicate(&schema, self.filter.as_ref(), self.case_sensitive)?; let name_mapping = self @@ -351,6 +394,7 @@ impl<'a> TableScanBuilder<'a> { expression_evaluator_cache: Arc::new(ExpressionEvaluatorCache::new()), unified_partition_type, sort_orders, + column_stats_field_ids, }; Ok(TableScan { @@ -703,7 +747,7 @@ mod tests { RESERVED_COL_NAME_FILE, RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER, RESERVED_COL_NAME_POS, RESERVED_COL_NAME_SPEC_ID, RESERVED_FIELD_ID_POS, }; - use crate::scan::{FileScanTask, FileScanTaskDeleteFile}; + use crate::scan::{FileScanTask, FileScanTaskColumnStats, FileScanTaskDeleteFile, TableScan}; use crate::spec::{ DataContentType, DataFileFormat, Datum, Literal, MAIN_BRANCH, MappedField, NameMapping, NestedField, NullOrder, Operation, PartitionSpec, PrimitiveType, Schema, Snapshot, @@ -1106,6 +1150,112 @@ mod tests { ); } + /// One column's statistics: field id, value count, null count, NaN count, lower bound + /// and upper bound. + type ColumnStat = (i32, u64, u64, Option, Datum, Datum); + + fn column_stats(columns: &[ColumnStat]) -> FileScanTaskColumnStats { + FileScanTaskColumnStats::builder() + .with_value_counts(columns.iter().map(|c| (c.0, c.1)).collect()) + .with_null_value_counts(columns.iter().map(|c| (c.0, c.2)).collect()) + .with_nan_value_counts(columns.iter().filter_map(|c| Some((c.0, c.3?))).collect()) + .with_lower_bounds(columns.iter().map(|c| (c.0, c.4.clone())).collect()) + .with_upper_bounds(columns.iter().map(|c| (c.0, c.5.clone())).collect()) + .build() + } + + async fn column_stats_by_file(scan: TableScan) -> Vec> { + let mut tasks: Vec<_> = scan + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + tasks.sort_by(|a, b| a.data_file_path().cmp(b.data_file_path())); + tasks + .into_iter() + .map(|task| task.column_stats().cloned()) + .collect() + } + + #[tokio::test] + async fn test_plan_files_retains_requested_column_stats() { + let mut fixture = TableTestFixture::new(); + + let x: ColumnStat = (1, 1, 0, None, Datum::long(1), Datum::long(10)); + let a: ColumnStat = (4, 1, 0, None, Datum::string("apple"), Datum::string("pear")); + let dbl: ColumnStat = (5, 1, 0, Some(1), Datum::double(1.5), Datum::double(2.5)); + + // The second file has no statistics for `x`. + fixture + .setup_manifest_files_with_column_stats(&[ + column_stats(&[x.clone(), a.clone(), dbl.clone()]), + column_stats(&[a.clone(), dbl.clone()]), + ]) + .await; + let table = &fixture.table; + + assert_eq!( + column_stats_by_file(table.scan().build().unwrap()).await, + vec![None, None] + ); + assert_eq!( + column_stats_by_file( + table + .scan() + .with_column_stats(Vec::::new()) + .build() + .unwrap() + ) + .await, + vec![None, None] + ); + assert_eq!( + column_stats_by_file( + table + .scan() + .with_column_stats(["x", "dbl"]) + .build() + .unwrap() + ) + .await, + vec![ + Some(column_stats(&[x, dbl.clone()])), + Some(column_stats(&[dbl])), + ] + ); + assert_eq!( + column_stats_by_file( + table + .scan() + .with_case_sensitive(false) + .with_column_stats(["A"]) + .build() + .unwrap() + ) + .await, + vec![ + Some(column_stats(std::slice::from_ref(&a))), + Some(column_stats(&[a])), + ] + ); + + for column in ["unknown", "A"] { + let err = table + .scan() + .with_column_stats([column]) + .build() + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.message() + .starts_with(&format!("Column {column} not found in table")), + "unexpected error: {err}" + ); + } + } + #[tokio::test] async fn test_plan_files_on_table_without_any_snapshots() { let table = TableTestFixture::new_empty().table; @@ -1950,6 +2100,21 @@ mod tests { .with_unified_partition_type(Some(unified_partition_type)) .with_sort_order_id(Some(1)) .with_sort_order(Some(sort_order)) + .with_column_stats(Some( + FileScanTaskColumnStats::builder() + .with_value_counts(HashMap::from([(1, 10), (2, 10)])) + .with_null_value_counts(HashMap::from([(1, 0), (2, 1)])) + .with_nan_value_counts(HashMap::from([(3, 2)])) + .with_lower_bounds(HashMap::from([ + (1, Datum::string("apple")), + (2, Datum::decimal_from_str("-12.34").unwrap()), + ])) + .with_upper_bounds(HashMap::from([ + (1, Datum::string("pear")), + (2, Datum::decimal_from_str("56.78").unwrap()), + ])) + .build(), + )) .with_case_sensitive(true) .with_key_metadata(Some(vec![1, 2, 3].into_boxed_slice())) .build() @@ -1957,6 +2122,21 @@ mod tests { assert_file_scan_task_serde_round_trip(task); } + #[test] + fn test_file_scan_task_serde_omits_absent_column_stats() { + let task = file_scan_task_with_partition( + PrimitiveType::Long, + Transform::Identity, + Literal::long(42), + ); + + let serialized = serde_json::to_value(&task).unwrap(); + assert!(serialized.get("column_stats").is_none()); + + let deserialized: FileScanTask = serde_json::from_value(serialized).unwrap(); + assert_eq!(deserialized.column_stats(), None); + } + #[test] fn test_file_scan_task_serde_with_date_partition() { let task = file_scan_task_with_partition( diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 45940b0dfd..7f439089c3 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use std::collections::HashMap; use std::sync::Arc; use futures::stream::BoxStream; @@ -24,8 +25,8 @@ use typed_builder::TypedBuilder; use crate::error::invalid_data; use crate::expr::BoundPredicate; use crate::spec::{ - DataContentType, DataFileFormat, ManifestEntryRef, NameMapping, PartitionSpec, Schema, - SchemaRef, SortOrderRef, Struct, StructType, + DataContentType, DataFile, DataFileFormat, Datum, ManifestEntryRef, NameMapping, PartitionSpec, + Schema, SchemaRef, SortOrderRef, Struct, StructType, }; use crate::{Error, Result}; @@ -137,6 +138,12 @@ pub struct FileScanTask { #[builder(default)] sort_order: Option, + /// The data file's column statistics for the columns requested with + /// [`TableScanBuilder::with_column_stats`](crate::scan::TableScanBuilder::with_column_stats). + /// `None` if no statistics were requested. + #[builder(default)] + column_stats: Option, + /// Whether this scan task should treat column names as case-sensitive when binding predicates. case_sensitive: bool, @@ -258,6 +265,13 @@ impl FileScanTask { self.sort_order.as_ref() } + /// Returns the data file's column statistics for the columns requested with + /// [`TableScanBuilder::with_column_stats`](crate::scan::TableScanBuilder::with_column_stats), + /// or `None` if none were requested. + pub fn column_stats(&self) -> Option<&FileScanTaskColumnStats> { + self.column_stats.as_ref() + } + /// Returns whether names are treated as case-sensitive. pub fn case_sensitive(&self) -> bool { self.case_sensitive @@ -434,7 +448,7 @@ mod _serde { use serde::{Deserialize, Serialize}; - use super::{FileScanTask, FileScanTaskDeleteFile}; + use super::{FileScanTask, FileScanTaskColumnStats, FileScanTaskDeleteFile}; use crate::error::invalid_data; use crate::expr::BoundPredicate; use crate::spec::{ @@ -469,6 +483,8 @@ mod _serde { sort_order_id: Option, #[serde(default)] sort_order: Option, + #[serde(default)] + column_stats: Option, case_sensitive: bool, #[serde(default)] key_metadata: Option>, @@ -504,6 +520,8 @@ mod _serde { sort_order_id: Option, #[serde(skip_serializing_if = "Option::is_none")] sort_order: Option<&'a SortOrderRef>, + #[serde(skip_serializing_if = "Option::is_none")] + column_stats: Option<&'a FileScanTaskColumnStats>, case_sensitive: bool, #[serde(skip_serializing_if = "Option::is_none")] key_metadata: Option<&'a [u8]>, @@ -553,6 +571,7 @@ mod _serde { unified_partition_type: value.unified_partition_type.as_ref(), sort_order_id: value.sort_order_id, sort_order: value.sort_order.as_ref(), + column_stats: value.column_stats.as_ref(), case_sensitive: value.case_sensitive, key_metadata: value.key_metadata.as_deref(), }) @@ -603,6 +622,7 @@ mod _serde { .with_unified_partition_type(value.unified_partition_type) .with_sort_order_id(value.sort_order_id) .with_sort_order(value.sort_order) + .with_column_stats(value.column_stats) .with_case_sensitive(value.case_sensitive) .with_key_metadata(value.key_metadata) .build() @@ -760,6 +780,88 @@ impl From for Result { } } +/// Column statistics of a [`FileScanTask`]'s data file, from its manifest entry. +/// +/// Each map is keyed by field id and holds only the columns requested for the scan. A +/// requested column missing from a map has unknown statistics of that kind: metrics mode +/// `none` records nothing for the column, while metrics mode `counts` and an all-null column +/// record counts but no bounds. +/// +/// Bounds are inclusive. A truncated lower bound is still less than or equal to every value in +/// the file, and a truncated upper bound is still greater than or equal to every value. Bounds +/// are typed with the schema the manifest was written with, so after type promotion a bound +/// can be an `int` while the scan schema says `long`. +/// +/// The statistics describe the whole data file: they are not adjusted for delete files, the +/// scan's filter, or the byte range of a split task. +#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize, TypedBuilder)] +#[builder(field_defaults(default, setter(prefix = "with_")))] +pub struct FileScanTaskColumnStats { + /// Number of values per field id, including nulls and NaNs. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + value_counts: HashMap, + + /// Number of null values per field id. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + null_value_counts: HashMap, + + /// Number of NaN values per field id. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + nan_value_counts: HashMap, + + /// Inclusive lower bound per field id. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + lower_bounds: HashMap, + + /// Inclusive upper bound per field id. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + upper_bounds: HashMap, +} + +impl FileScanTaskColumnStats { + pub(crate) fn from_data_file(data_file: &DataFile, field_ids: &[i32]) -> Self { + fn retain(map: &HashMap, field_ids: &[i32]) -> HashMap { + field_ids + .iter() + .filter_map(|id| map.get(id).map(|value| (*id, value.clone()))) + .collect() + } + + Self { + value_counts: retain(data_file.value_counts(), field_ids), + null_value_counts: retain(data_file.null_value_counts(), field_ids), + nan_value_counts: retain(data_file.nan_value_counts(), field_ids), + lower_bounds: retain(data_file.lower_bounds(), field_ids), + upper_bounds: retain(data_file.upper_bounds(), field_ids), + } + } + + /// Returns the number of values per field id, including nulls and NaNs. + pub fn value_counts(&self) -> &HashMap { + &self.value_counts + } + + /// Returns the number of null values per field id. + pub fn null_value_counts(&self) -> &HashMap { + &self.null_value_counts + } + + /// Returns the number of NaN values per field id. + pub fn nan_value_counts(&self) -> &HashMap { + &self.nan_value_counts + } + + /// Returns the inclusive lower bound per field id. + pub fn lower_bounds(&self) -> &HashMap { + &self.lower_bounds + } + + /// Returns the inclusive upper bound per field id. + pub fn upper_bounds(&self) -> &HashMap { + &self.upper_bounds + } +} + #[cfg(test)] mod tests { use std::collections::HashMap; diff --git a/crates/iceberg/src/test_utils/scan.rs b/crates/iceberg/src/test_utils/scan.rs index 0c4a7e9dd7..3b940ef596 100644 --- a/crates/iceberg/src/test_utils/scan.rs +++ b/crates/iceberg/src/test_utils/scan.rs @@ -41,6 +41,7 @@ use crate::metadata_columns::{ RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER, RESERVED_FIELD_ID_DELETE_FILE_PATH, RESERVED_FIELD_ID_DELETE_FILE_POS, }; +use crate::scan::FileScanTaskColumnStats; use crate::spec::{ DataContentType, DataFileBuilder, DataFileFormat, FormatVersion, Literal, ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, PartitionSpec, Struct, StructType, @@ -1074,6 +1075,77 @@ impl TableTestFixture { manifest_list_write.close().await.unwrap(); } + /// Writes a manifest with one live "Added" data-file entry per element of + /// `column_stats` (`1.parquet`, `2.parquet`, ..., partitioned on `x` = 100, 200, ...), + /// each carrying that element's value, null and NaN counts and lower and upper bounds. + pub(crate) async fn setup_manifest_files_with_column_stats( + &mut self, + column_stats: &[FileScanTaskColumnStats], + ) { + let current_snapshot = self.table.metadata().current_snapshot().unwrap(); + let current_schema = current_snapshot.schema(self.table.metadata()).unwrap(); + let current_partition_spec = self.table.metadata().default_partition_spec(); + let parquet_file_size = self.write_parquet_data_files(); + + let mut writer = ManifestWriterBuilder::new( + self.next_manifest_file(), + Some(current_snapshot.snapshot_id()), + current_schema.clone(), + current_partition_spec.as_ref().clone(), + ) + .build_v2_data(); + + for (i, stats) in column_stats.iter().enumerate() { + let data_file = DataFileBuilder::default() + .partition_spec_id(0) + .content(DataContentType::Data) + .file_path(format!("{}/{}.parquet", &self.table_location, i + 1)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(parquet_file_size) + .record_count(1) + .partition(Struct::from_iter([Some(Literal::long( + 100 * (i as i64 + 1), + ))])) + .value_counts(stats.value_counts().clone()) + .null_value_counts(stats.null_value_counts().clone()) + .nan_value_counts(stats.nan_value_counts().clone()) + .lower_bounds(stats.lower_bounds().clone()) + .upper_bounds(stats.upper_bounds().clone()) + .build() + .unwrap(); + + writer + .add_entry( + ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file(data_file) + .build(), + ) + .unwrap(); + } + + let data_file_manifest = writer.write_manifest_file().await.unwrap(); + + let manifest_list_writer = self + .table + .file_io() + .new_output(current_snapshot.manifest_list()) + .unwrap() + .writer() + .await + .unwrap(); + let mut manifest_list_write = ManifestListWriter::v2( + manifest_list_writer, + current_snapshot.snapshot_id(), + current_snapshot.parent_snapshot_id(), + current_snapshot.sequence_number(), + ); + manifest_list_write + .add_manifests(std::iter::once(data_file_manifest)) + .unwrap(); + manifest_list_write.close().await.unwrap(); + } + /// Writes `mrg.parquet` with three 100-row row groups. Columns `x` (field /// id `1`) and `y` (field id `2`) both run 1000..1300, so row position `p` /// carries `x = y = 1000 + p`. Returns `(path, file_size_in_bytes)`. From e0f391941e1abaf35b75c5f595c29a0d7228720f Mon Sep 17 00:00:00 2001 From: Anurag Mantripragada Date: Tue, 13 Oct 2026 13:30:06 -0700 Subject: [PATCH 2/4] fix(scan): serialize column bounds as spec binary values Bounds went through Datum's JSON encoding, which writes non-finite floats as null and rejects many finite float values when reading them back, so a task with such a bound serialized but failed to deserialize. Serialize each bound as its type and the spec's single-value binary encoding, as the DataFile JSON serde already does, so every bound round-trips exactly. --- crates/iceberg/src/scan/mod.rs | 4 +++ crates/iceberg/src/scan/task.rs | 58 +++++++++++++++++++++++++++++++-- 2 files changed, 60 insertions(+), 2 deletions(-) diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 71390a23a8..69241fd235 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -2108,10 +2108,14 @@ mod tests { .with_lower_bounds(HashMap::from([ (1, Datum::string("apple")), (2, Datum::decimal_from_str("-12.34").unwrap()), + (3, Datum::float(100.1_f32)), + (4, Datum::double(f64::NEG_INFINITY)), ])) .with_upper_bounds(HashMap::from([ (1, Datum::string("pear")), (2, Datum::decimal_from_str("56.78").unwrap()), + (3, Datum::float(f32::INFINITY)), + (4, Datum::double(f64::INFINITY)), ])) .build(), )) diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 7f439089c3..d6560cf5cd 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -668,6 +668,52 @@ mod _serde { .build() } } + + /// Serializes column bounds as their type and the spec's single-value binary encoding, + /// because the JSON encoding of [`RawLiteral`] does not round-trip every float value. + pub(super) mod bounds { + use std::collections::HashMap; + + use serde::ser::SerializeMap; + use serde::{Deserialize, Deserializer, Serialize, Serializer}; + use serde_bytes::ByteBuf; + + use crate::spec::{Datum, PrimitiveType}; + + #[derive(Serialize, Deserialize)] + struct BoundSerde { + r#type: PrimitiveType, + bytes: ByteBuf, + } + + pub(in crate::scan::task) fn serialize( + bounds: &HashMap, + serializer: S, + ) -> Result { + let mut map = serializer.serialize_map(Some(bounds.len()))?; + for (field_id, datum) in bounds { + let bound = BoundSerde { + r#type: datum.data_type().clone(), + bytes: datum.to_bytes().map_err(serde::ser::Error::custom)?, + }; + map.serialize_entry(field_id, &bound)?; + } + map.end() + } + + pub(in crate::scan::task) fn deserialize<'de, D: Deserializer<'de>>( + deserializer: D, + ) -> Result, D::Error> { + HashMap::::deserialize(deserializer)? + .into_iter() + .map(|(field_id, bound)| { + Datum::try_from_bytes(&bound.bytes, bound.r#type) + .map(|datum| (field_id, datum)) + .map_err(serde::de::Error::custom) + }) + .collect() + } + } } impl FileScanTaskDeleteFile { @@ -810,11 +856,19 @@ pub struct FileScanTaskColumnStats { nan_value_counts: HashMap, /// Inclusive lower bound per field id. - #[serde(default, skip_serializing_if = "HashMap::is_empty")] + #[serde( + default, + skip_serializing_if = "HashMap::is_empty", + with = "crate::scan::task::_serde::bounds" + )] lower_bounds: HashMap, /// Inclusive upper bound per field id. - #[serde(default, skip_serializing_if = "HashMap::is_empty")] + #[serde( + default, + skip_serializing_if = "HashMap::is_empty", + with = "crate::scan::task::_serde::bounds" + )] upper_bounds: HashMap, } From 5345a45c8b5509d1a0bbd9345e2444b5c885d7c7 Mon Sep 17 00:00:00 2001 From: Anurag Mantripragada Date: Tue, 13 Oct 2026 13:48:54 -0700 Subject: [PATCH 3/4] refactor(scan): address review of column statistics retention Retain column_sizes for the requested ids too, matching Java's BaseFile(copy, requestedColumnIds). Share the column-name lookup and its error with column selection, keep the requested names as an Option like column_names so an empty list yields empty statistics, and pass the ids as Arc<[i32]>. Skip work for an empty map when filtering, trim the struct docs, and cover nested field names. --- crates/iceberg/public-api.txt | 3 +- crates/iceberg/src/scan/context.rs | 6 +- crates/iceberg/src/scan/mod.rs | 114 +++++++++++++++++--------- crates/iceberg/src/scan/task.rs | 44 ++++++---- crates/iceberg/src/test_utils/scan.rs | 3 +- 5 files changed, 109 insertions(+), 61 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 36e14a92c6..3f1c020ce0 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1369,6 +1369,7 @@ impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTask pub fn iceberg::scan::FileScanTask::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::scan::FileScanTaskColumnStats impl iceberg::scan::FileScanTaskColumnStats +pub fn iceberg::scan::FileScanTaskColumnStats::column_sizes(&self) -> &std::collections::hash::map::HashMap pub fn iceberg::scan::FileScanTaskColumnStats::lower_bounds(&self) -> &std::collections::hash::map::HashMap pub fn iceberg::scan::FileScanTaskColumnStats::nan_value_counts(&self) -> &std::collections::hash::map::HashMap pub fn iceberg::scan::FileScanTaskColumnStats::null_value_counts(&self) -> &std::collections::hash::map::HashMap @@ -1384,7 +1385,7 @@ impl core::fmt::Debug for iceberg::scan::FileScanTaskColumnStats pub fn iceberg::scan::FileScanTaskColumnStats::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTaskColumnStats impl iceberg::scan::FileScanTaskColumnStats -pub fn iceberg::scan::FileScanTaskColumnStats::builder() -> FileScanTaskColumnStatsBuilder<((), (), (), (), ())> +pub fn iceberg::scan::FileScanTaskColumnStats::builder() -> FileScanTaskColumnStatsBuilder<((), (), (), (), (), ())> impl serde_core::ser::Serialize for iceberg::scan::FileScanTaskColumnStats pub fn iceberg::scan::FileScanTaskColumnStats::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTaskColumnStats diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index c04c3601ce..4582b57474 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -52,7 +52,7 @@ pub(crate) struct ManifestFileContext { partition_spec: Option, unified_partition_type: Option>, sort_orders: Arc>, - column_stats_field_ids: Option>>, + column_stats_field_ids: Option>, } /// Wraps a [`ManifestEntryRef`] alongside the objects that are needed @@ -72,7 +72,7 @@ pub(crate) struct ManifestEntryContext { pub unified_partition_type: Option>, pub sort_order_id: Option, pub sort_order: Option, - pub column_stats_field_ids: Option>>, + pub column_stats_field_ids: Option>, } impl ManifestFileContext { @@ -225,7 +225,7 @@ pub(crate) struct PlanContext { pub sort_orders: Arc>, /// Field ids whose column statistics each [`FileScanTask`] retains, or `None` to retain none. - pub column_stats_field_ids: Option>>, + pub column_stats_field_ids: Option>, } impl PlanContext { diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 69241fd235..8664484236 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -66,6 +66,12 @@ fn resolve_field_id( } } +/// Resolves a column name to its field ID, failing if no field matches. +fn require_field_id(schema: &Schema, column_name: &str, case_sensitive: bool) -> Result { + resolve_field_id(schema, column_name, case_sensitive)? + .ok_or_else(|| invalid_data!("Column {column_name} not found in table. Schema: {schema}")) +} + fn collect_scan_field_ids( schema: &Schema, column_names: Option<&[String]>, @@ -82,9 +88,7 @@ fn collect_scan_field_ids( return get_metadata_field_id(column_name); } - let field_id = resolve_field_id(schema, column_name, case_sensitive)?.ok_or_else(|| { - invalid_data!("Column {column_name} not found in table. Schema: {schema}") - })?; + let field_id = require_field_id(schema, column_name, case_sensitive)?; schema .as_struct() @@ -105,22 +109,17 @@ fn collect_scan_field_ids( fn collect_column_stats_field_ids( schema: &Schema, - column_names: &[String], + column_names: Option<&[String]>, case_sensitive: bool, -) -> Result>>> { - if column_names.is_empty() { - return Ok(None); - } - +) -> Result>> { column_names - .iter() - .map(|column_name| { - resolve_field_id(schema, column_name, case_sensitive)?.ok_or_else(|| { - invalid_data!("Column {column_name} not found in table. Schema: {schema}") - }) + .map(|column_names| { + column_names + .iter() + .map(|column_name| require_field_id(schema, column_name, case_sensitive)) + .collect() }) - .collect::>>() - .map(|field_ids| Some(Arc::new(field_ids))) + .transpose() } fn bind_scan_predicate( @@ -155,7 +154,7 @@ pub struct TableScanBuilder<'a> { table: &'a Table, // Defaults to none which means select all columns column_names: Option>, - column_stats_names: Vec, + column_stats_names: Option>, snapshot_id: Option, batch_size: Option, case_sensitive: bool, @@ -175,7 +174,7 @@ impl<'a> TableScanBuilder<'a> { Self { table, column_names: None, - column_stats_names: vec![], + column_stats_names: None, snapshot_id: None, batch_size: None, case_sensitive: true, @@ -237,7 +236,7 @@ impl<'a> TableScanBuilder<'a> { /// [`FileScanTask`], exposed as [`FileScanTask::column_stats`]. /// /// Names resolve against the scan's schema with the scan's case sensitivity and may name - /// nested fields. An empty list retains no statistics, which is the default. + /// nested fields. By default, tasks carry no statistics. /// /// [`build`](Self::build) fails if a name matches no field, or more than one field when the /// scan is case-insensitive. @@ -245,10 +244,12 @@ impl<'a> TableScanBuilder<'a> { mut self, column_names: impl IntoIterator, ) -> Self { - self.column_stats_names = column_names - .into_iter() - .map(|item| item.to_string()) - .collect(); + self.column_stats_names = Some( + column_names + .into_iter() + .map(|item| item.to_string()) + .collect(), + ); self } @@ -357,8 +358,11 @@ impl<'a> TableScanBuilder<'a> { let schema = snapshot.schema(self.table.metadata())?; let field_ids = collect_scan_field_ids(&schema, self.column_names.as_deref(), self.case_sensitive)?; - let column_stats_field_ids = - collect_column_stats_field_ids(&schema, &self.column_stats_names, self.case_sensitive)?; + let column_stats_field_ids = collect_column_stats_field_ids( + &schema, + self.column_stats_names.as_deref(), + self.case_sensitive, + )?; let snapshot_bound_predicate = bind_scan_predicate(&schema, self.filter.as_ref(), self.case_sensitive)?; let name_mapping = self @@ -739,7 +743,7 @@ mod tests { use futures::{TryStreamExt, stream}; use uuid::Uuid; - use super::resolve_field_id; + use super::{collect_column_stats_field_ids, resolve_field_id}; use crate::arrow::ArrowReaderBuilder; use crate::expr::{BoundPredicate, Reference}; use crate::io::FileIO; @@ -751,7 +755,7 @@ mod tests { use crate::spec::{ DataContentType, DataFileFormat, Datum, Literal, MAIN_BRANCH, MappedField, NameMapping, NestedField, NullOrder, Operation, PartitionSpec, PrimitiveType, Schema, Snapshot, - SortDirection, SortField, SortOrder, Struct, Summary, TableMetadataBuilder, + SortDirection, SortField, SortOrder, Struct, StructType, Summary, TableMetadataBuilder, TableProperties, Transform, Type, UnboundPartitionSpec, }; use crate::table::Table; @@ -809,6 +813,28 @@ mod tests { ); } + #[test] + fn test_collect_column_stats_field_ids_resolves_nested_field() { + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required( + 1, + "s", + Type::Struct(StructType::new(vec![ + NestedField::optional(2, "f", PrimitiveType::Long.into()).into(), + ])), + ) + .into(), + ]) + .build() + .unwrap(); + + assert_eq!( + collect_column_stats_field_ids(&schema, Some(&["s.f".to_string()]), true).unwrap(), + Some(Arc::<[i32]>::from([2])) + ); + } + #[test] fn test_resolve_field_id_case_insensitive_rejects_ambiguous_name() { let schema = Schema::builder() @@ -1150,17 +1176,18 @@ mod tests { ); } - /// One column's statistics: field id, value count, null count, NaN count, lower bound - /// and upper bound. - type ColumnStat = (i32, u64, u64, Option, Datum, Datum); + /// One column's statistics: field id, column size, value count, null count, NaN count, + /// lower bound and upper bound. + type ColumnStat = (i32, u64, u64, u64, Option, Datum, Datum); fn column_stats(columns: &[ColumnStat]) -> FileScanTaskColumnStats { FileScanTaskColumnStats::builder() - .with_value_counts(columns.iter().map(|c| (c.0, c.1)).collect()) - .with_null_value_counts(columns.iter().map(|c| (c.0, c.2)).collect()) - .with_nan_value_counts(columns.iter().filter_map(|c| Some((c.0, c.3?))).collect()) - .with_lower_bounds(columns.iter().map(|c| (c.0, c.4.clone())).collect()) - .with_upper_bounds(columns.iter().map(|c| (c.0, c.5.clone())).collect()) + .with_column_sizes(columns.iter().map(|c| (c.0, c.1)).collect()) + .with_value_counts(columns.iter().map(|c| (c.0, c.2)).collect()) + .with_null_value_counts(columns.iter().map(|c| (c.0, c.3)).collect()) + .with_nan_value_counts(columns.iter().filter_map(|c| Some((c.0, c.4?))).collect()) + .with_lower_bounds(columns.iter().map(|c| (c.0, c.5.clone())).collect()) + .with_upper_bounds(columns.iter().map(|c| (c.0, c.6.clone())).collect()) .build() } @@ -1183,9 +1210,17 @@ mod tests { async fn test_plan_files_retains_requested_column_stats() { let mut fixture = TableTestFixture::new(); - let x: ColumnStat = (1, 1, 0, None, Datum::long(1), Datum::long(10)); - let a: ColumnStat = (4, 1, 0, None, Datum::string("apple"), Datum::string("pear")); - let dbl: ColumnStat = (5, 1, 0, Some(1), Datum::double(1.5), Datum::double(2.5)); + let x: ColumnStat = (1, 8, 1, 0, None, Datum::long(1), Datum::long(10)); + let a: ColumnStat = ( + 4, + 9, + 1, + 0, + None, + Datum::string("apple"), + Datum::string("pear"), + ); + let dbl: ColumnStat = (5, 8, 1, 0, Some(1), Datum::double(1.5), Datum::double(2.5)); // The second file has no statistics for `x`. fixture @@ -1209,7 +1244,7 @@ mod tests { .unwrap() ) .await, - vec![None, None] + vec![Some(column_stats(&[])), Some(column_stats(&[]))] ); assert_eq!( column_stats_by_file( @@ -2102,6 +2137,7 @@ mod tests { .with_sort_order(Some(sort_order)) .with_column_stats(Some( FileScanTaskColumnStats::builder() + .with_column_sizes(HashMap::from([(1, 80), (2, 96)])) .with_value_counts(HashMap::from([(1, 10), (2, 10)])) .with_null_value_counts(HashMap::from([(1, 0), (2, 1)])) .with_nan_value_counts(HashMap::from([(3, 2)])) diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index d6560cf5cd..9ba9627309 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -826,23 +826,21 @@ impl From for Result { } } -/// Column statistics of a [`FileScanTask`]'s data file, from its manifest entry. +/// Column statistics of a [`FileScanTask`]'s data file, keyed by field id and limited to the +/// columns requested for the scan. /// -/// Each map is keyed by field id and holds only the columns requested for the scan. A -/// requested column missing from a map has unknown statistics of that kind: metrics mode -/// `none` records nothing for the column, while metrics mode `counts` and an all-null column -/// record counts but no bounds. -/// -/// Bounds are inclusive. A truncated lower bound is still less than or equal to every value in -/// the file, and a truncated upper bound is still greater than or equal to every value. Bounds -/// are typed with the schema the manifest was written with, so after type promotion a bound -/// can be an `int` while the scan schema says `long`. -/// -/// The statistics describe the whole data file: they are not adjusted for delete files, the -/// scan's filter, or the byte range of a split task. +/// A requested column missing from a map has unknown statistics of that kind, for example when +/// the table's metrics mode records no bounds. Bounds are inclusive even when truncated, and are +/// typed with the schema the manifest was written with, so after type promotion an `int` bound +/// can describe a `long` column. All statistics cover the whole file, before delete files, the +/// scan's filter or task splits. #[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize, TypedBuilder)] #[builder(field_defaults(default, setter(prefix = "with_")))] pub struct FileScanTaskColumnStats { + /// Size in bytes of each column's data in the file, per field id. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + column_sizes: HashMap, + /// Number of values per field id, including nulls and NaNs. #[serde(default, skip_serializing_if = "HashMap::is_empty")] value_counts: HashMap, @@ -875,13 +873,20 @@ pub struct FileScanTaskColumnStats { impl FileScanTaskColumnStats { pub(crate) fn from_data_file(data_file: &DataFile, field_ids: &[i32]) -> Self { fn retain(map: &HashMap, field_ids: &[i32]) -> HashMap { - field_ids - .iter() - .filter_map(|id| map.get(id).map(|value| (*id, value.clone()))) - .collect() + if map.is_empty() { + return HashMap::new(); + } + let mut retained = HashMap::with_capacity(field_ids.len().min(map.len())); + retained.extend( + field_ids + .iter() + .filter_map(|id| map.get(id).map(|value| (*id, value.clone()))), + ); + retained } Self { + column_sizes: retain(data_file.column_sizes(), field_ids), value_counts: retain(data_file.value_counts(), field_ids), null_value_counts: retain(data_file.null_value_counts(), field_ids), nan_value_counts: retain(data_file.nan_value_counts(), field_ids), @@ -890,6 +895,11 @@ impl FileScanTaskColumnStats { } } + /// Returns the size in bytes of each column's data in the file, per field id. + pub fn column_sizes(&self) -> &HashMap { + &self.column_sizes + } + /// Returns the number of values per field id, including nulls and NaNs. pub fn value_counts(&self) -> &HashMap { &self.value_counts diff --git a/crates/iceberg/src/test_utils/scan.rs b/crates/iceberg/src/test_utils/scan.rs index 3b940ef596..45b51d5d27 100644 --- a/crates/iceberg/src/test_utils/scan.rs +++ b/crates/iceberg/src/test_utils/scan.rs @@ -1077,7 +1077,7 @@ impl TableTestFixture { /// Writes a manifest with one live "Added" data-file entry per element of /// `column_stats` (`1.parquet`, `2.parquet`, ..., partitioned on `x` = 100, 200, ...), - /// each carrying that element's value, null and NaN counts and lower and upper bounds. + /// each carrying that element's column sizes, value, null and NaN counts and bounds. pub(crate) async fn setup_manifest_files_with_column_stats( &mut self, column_stats: &[FileScanTaskColumnStats], @@ -1106,6 +1106,7 @@ impl TableTestFixture { .partition(Struct::from_iter([Some(Literal::long( 100 * (i as i64 + 1), ))])) + .column_sizes(stats.column_sizes().clone()) .value_counts(stats.value_counts().clone()) .null_value_counts(stats.null_value_counts().clone()) .nan_value_counts(stats.nan_value_counts().clone()) From fe1bf907ab0a7b041a26526120221e0c6e1ca8cd Mon Sep 17 00:00:00 2001 From: Anurag Mantripragada Date: Fri, 16 Oct 2026 15:34:18 -0700 Subject: [PATCH 4/4] fix(scan): reject column names that cannot have statistics Struct, list and map fields have no statistics of their own, and metadata columns are not stored in data files, so requesting stats for them silently returned empty maps or failed with a misleading "not found" error. Fail build() with an error that says why. Also cover bounds that keep the manifest's type after a type promotion. --- crates/iceberg/src/scan/mod.rs | 116 ++++++++++++++++++++++---- crates/iceberg/src/test_utils/scan.rs | 12 ++- 2 files changed, 111 insertions(+), 17 deletions(-) diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 6b3d19fccf..4679dc58fc 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -116,7 +116,23 @@ fn collect_column_stats_field_ids( .map(|column_names| { column_names .iter() - .map(|column_name| require_field_id(schema, column_name, case_sensitive)) + .map(|column_name| { + if is_metadata_column_name(column_name) { + return Err(invalid_data!( + "Metadata column {column_name} has no column statistics" + )); + } + let field_id = require_field_id(schema, column_name, case_sensitive)?; + if !schema + .field_by_id(field_id) + .is_some_and(|field| field.field_type.is_primitive()) + { + return Err(invalid_data!( + "Column {column_name} is not a primitive field and has no column statistics" + )); + } + Ok(field_id) + }) .collect() }) .transpose() @@ -235,11 +251,11 @@ impl<'a> TableScanBuilder<'a> { /// Retains the data file's column statistics for the named columns on each planned /// [`FileScanTask`], exposed as [`FileScanTask::column_stats`]. /// - /// Names resolve against the scan's schema with the scan's case sensitivity and may name - /// nested fields. By default, tasks carry no statistics. + /// Names resolve against the scan's schema with the scan's case sensitivity and must name + /// primitive fields, which may be nested. By default, tasks carry no statistics. /// - /// [`build`](Self::build) fails if a name matches no field, or more than one field when the - /// scan is case-insensitive. + /// [`build`](Self::build) fails if a name matches no field, matches more than one field when + /// the scan is case-insensitive, or names a struct, list, map or metadata column. pub fn with_column_stats( mut self, column_names: impl IntoIterator, @@ -820,9 +836,9 @@ mod tests { ); } - #[test] - fn test_collect_column_stats_field_ids_resolves_nested_field() { - let schema = Schema::builder() + /// A schema with a struct `s` (field id 1) whose only child is `f` (field id 2). + fn struct_schema() -> Schema { + Schema::builder() .with_fields(vec![ NestedField::required( 1, @@ -834,14 +850,30 @@ mod tests { .into(), ]) .build() - .unwrap(); + .unwrap() + } + #[test] + fn test_collect_column_stats_field_ids_resolves_nested_field() { assert_eq!( - collect_column_stats_field_ids(&schema, Some(&["s.f".to_string()]), true).unwrap(), + collect_column_stats_field_ids(&struct_schema(), Some(&["s.f".to_string()]), true) + .unwrap(), Some(Arc::<[i32]>::from([2])) ); } + #[test] + fn test_collect_column_stats_field_ids_rejects_struct_field() { + let err = collect_column_stats_field_ids(&struct_schema(), Some(&["s".to_string()]), true) + .unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert_eq!( + err.message(), + "Column s is not a primitive field and has no column statistics" + ); + } + #[test] fn test_resolve_field_id_case_insensitive_rejects_ambiguous_name() { let schema = Schema::builder() @@ -1264,10 +1296,13 @@ mod tests { // The second file has no statistics for `x`. fixture - .setup_manifest_files_with_column_stats(&[ - column_stats(&[x.clone(), a.clone(), dbl.clone()]), - column_stats(&[a.clone(), dbl.clone()]), - ]) + .setup_manifest_files_with_column_stats( + &[ + column_stats(&[x.clone(), a.clone(), dbl.clone()]), + column_stats(&[a.clone(), dbl.clone()]), + ], + None, + ) .await; let table = &fixture.table; @@ -1329,6 +1364,59 @@ mod tests { "unexpected error: {err}" ); } + + let err = table + .scan() + .with_column_stats([RESERVED_COL_NAME_FILE]) + .build() + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert_eq!( + err.message(), + format!("Metadata column {RESERVED_COL_NAME_FILE} has no column statistics") + ); + } + + #[tokio::test] + async fn test_plan_files_keeps_bounds_typed_with_manifest_schema() { + let mut fixture = TableTestFixture::new(); + + // `y` is a `long` in the table schema but was an `int` when the manifest was written. + let manifest_schema = Arc::new( + Schema::builder() + .with_fields( + fixture + .table + .metadata() + .current_schema() + .as_struct() + .fields() + .iter() + .map(|field| match field.id { + 2 => NestedField::required(2, "y", PrimitiveType::Int.into()).into(), + _ => field.clone(), + }), + ) + .build() + .unwrap(), + ); + let y = column_stats(&[(2, 4, 1, 0, None, Datum::int(5), Datum::int(7))]); + fixture + .setup_manifest_files_with_column_stats(std::slice::from_ref(&y), Some(manifest_schema)) + .await; + + assert_eq!( + column_stats_by_file( + fixture + .table + .scan() + .with_column_stats(["y"]) + .build() + .unwrap() + ) + .await, + vec![Some(y)] + ); } #[tokio::test] diff --git a/crates/iceberg/src/test_utils/scan.rs b/crates/iceberg/src/test_utils/scan.rs index a4ce7488e4..aa662bfe0e 100644 --- a/crates/iceberg/src/test_utils/scan.rs +++ b/crates/iceberg/src/test_utils/scan.rs @@ -45,7 +45,8 @@ use crate::scan::FileScanTaskColumnStats; use crate::spec::{ DataContentType, DataFileBuilder, DataFileFormat, FormatVersion, Literal, ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, NestedField, PartitionSpec, - PrimitiveType, Schema, Struct, StructType, TableMetadata, TableMetadataBuilder, Type, + PrimitiveType, Schema, SchemaRef, Struct, StructType, TableMetadata, TableMetadataBuilder, + Type, }; use crate::table::Table; use crate::test_utils::test_runtime; @@ -1103,13 +1104,18 @@ impl TableTestFixture { /// Writes a manifest with one live "Added" data-file entry per element of /// `column_stats` (`1.parquet`, `2.parquet`, ..., partitioned on `x` = 100, 200, ...), - /// each carrying that element's column sizes, value, null and NaN counts and bounds. + /// each carrying that element's column sizes, value, null and NaN counts and bounds. The + /// manifest is written with `manifest_schema`, or the current snapshot's schema if `None`. pub(crate) async fn setup_manifest_files_with_column_stats( &mut self, column_stats: &[FileScanTaskColumnStats], + manifest_schema: Option, ) { let current_snapshot = self.table.metadata().current_snapshot().unwrap(); - let current_schema = current_snapshot.schema(self.table.metadata()).unwrap(); + let current_schema = match manifest_schema { + Some(schema) => schema, + None => current_snapshot.schema(self.table.metadata()).unwrap(), + }; let current_partition_spec = self.table.metadata().default_partition_spec(); let parquet_file_size = self.write_parquet_data_files();