diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 00040053a6..2bed9fa35f 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1353,6 +1353,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 @@ -1383,11 +1384,34 @@ 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::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 +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 @@ -1445,6 +1469,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 211ebf8f0c..75ac0267a5 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() } @@ -219,6 +226,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 { @@ -350,6 +360,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 6d23e94704..4679dc58fc 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() @@ -103,6 +107,37 @@ fn collect_scan_field_ids( .collect() } +fn collect_column_stats_field_ids( + schema: &Schema, + column_names: Option<&[String]>, + case_sensitive: bool, +) -> Result>> { + column_names + .map(|column_names| { + column_names + .iter() + .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() +} + fn bind_scan_predicate( schema: &SchemaRef, predicate: Option<&Predicate>, @@ -135,6 +170,7 @@ pub struct TableScanBuilder<'a> { table: &'a Table, // Defaults to none which means select all columns column_names: Option>, + column_stats_names: Option>, snapshot_id: Option, batch_size: Option, case_sensitive: bool, @@ -154,6 +190,7 @@ impl<'a> TableScanBuilder<'a> { Self { table, column_names: None, + column_stats_names: None, snapshot_id: None, batch_size: None, case_sensitive: true, @@ -211,6 +248,27 @@ 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 must name + /// primitive fields, which may be nested. By default, tasks carry no statistics. + /// + /// [`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, + ) -> Self { + self.column_stats_names = Some( + column_names + .into_iter() + .map(|item| item.to_string()) + .collect(), + ); + self + } + /// Set the snapshot to scan. The scan projects the schema that snapshot was /// written with. When not set, the scan reads the current snapshot with the /// table's current schema. @@ -323,6 +381,11 @@ impl<'a> TableScanBuilder<'a> { }; 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.as_deref(), + self.case_sensitive, + )?; let snapshot_bound_predicate = bind_scan_predicate(&schema, self.filter.as_ref(), self.case_sensitive)?; let name_mapping = self @@ -358,6 +421,7 @@ impl<'a> TableScanBuilder<'a> { expression_evaluator_cache: Arc::new(ExpressionEvaluatorCache::new()), unified_partition_type, sort_orders, + column_stats_field_ids, }; Ok(TableScan { @@ -702,7 +766,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; @@ -710,11 +774,11 @@ 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, - SortDirection, SortField, SortOrder, Struct, Summary, TableMetadataBuilder, + SortDirection, SortField, SortOrder, Struct, StructType, Summary, TableMetadataBuilder, TableProperties, Transform, Type, UnboundPartitionSpec, }; use crate::table::Table; @@ -772,6 +836,44 @@ mod tests { ); } + /// 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, + "s", + Type::Struct(StructType::new(vec![ + NestedField::optional(2, "f", PrimitiveType::Long.into()).into(), + ])), + ) + .into(), + ]) + .build() + .unwrap() + } + + #[test] + fn test_collect_column_stats_field_ids_resolves_nested_field() { + assert_eq!( + 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() @@ -1146,6 +1248,177 @@ mod tests { ); } + /// 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_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() + } + + 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, 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 + .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; + + 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![Some(column_stats(&[])), Some(column_stats(&[]))] + ); + 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}" + ); + } + + 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] async fn test_plan_files_on_table_without_any_snapshots() { let table = TableTestFixture::new_empty().table; @@ -1990,6 +2263,26 @@ 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_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)])) + .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(), + )) .with_case_sensitive(true) .with_key_metadata(Some(vec![1, 2, 3].into_boxed_slice())) .build() @@ -1997,6 +2290,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..9ba9627309 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() @@ -648,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 { @@ -760,6 +826,106 @@ impl From for Result { } } +/// Column statistics of a [`FileScanTask`]'s data file, keyed by field id and limited to the +/// columns requested for the scan. +/// +/// 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, + + /// 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", + with = "crate::scan::task::_serde::bounds" + )] + lower_bounds: HashMap, + + /// Inclusive upper bound per field id. + #[serde( + default, + skip_serializing_if = "HashMap::is_empty", + with = "crate::scan::task::_serde::bounds" + )] + 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 { + 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), + lower_bounds: retain(data_file.lower_bounds(), field_ids), + upper_bounds: retain(data_file.upper_bounds(), field_ids), + } + } + + /// 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 + } + + /// 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 63f6418efd..aa662bfe0e 100644 --- a/crates/iceberg/src/test_utils/scan.rs +++ b/crates/iceberg/src/test_utils/scan.rs @@ -41,10 +41,12 @@ 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, 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; @@ -1100,6 +1102,83 @@ 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 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 = 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(); + + 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), + ))])) + .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()) + .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)`.