From f8bc7ec9eed2c2228319ccb0e60138a68a55b27b Mon Sep 17 00:00:00 2001 From: Xander Date: Fri, 28 Aug 2026 10:59:16 +0100 Subject: [PATCH 1/4] refactor(scan): extract scan planning utilities --- crates/iceberg/src/scan/mod.rs | 189 +++++++++++++++++++-------------- 1 file changed, 108 insertions(+), 81 deletions(-) diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 7abbf2dddf..8e205a788d 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -42,7 +42,10 @@ use crate::metadata_columns::{ }; use crate::partitioning::compute_unified_partition_type; use crate::runtime::Runtime; -use crate::spec::{DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, NameMapping, Schema, SnapshotRef}; +use crate::spec::{ + DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, NameMapping, Schema, SchemaRef, SnapshotRef, + StructType, +}; use crate::table::Table; use crate::util::available_parallelism; use crate::{Error, ErrorKind, Result}; @@ -61,6 +64,103 @@ fn resolve_field_id(schema: &Schema, column_name: &str, case_sensitive: bool) -> } } +fn projected_field_ids( + schema: &Schema, + column_names: Option<&[String]>, + case_sensitive: bool, +) -> Result> { + let mut field_ids = vec![]; + let column_names = column_names.map(<[String]>::to_vec).unwrap_or_else(|| { + schema + .as_struct() + .fields() + .iter() + .map(|f| f.name.clone()) + .collect() + }); + + for column_name in column_names.iter() { + if is_metadata_column_name(column_name) { + field_ids.push(get_metadata_field_id(column_name)?); + continue; + } + + let field_id = resolve_field_id(schema, column_name, case_sensitive).ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Column {column_name} not found in table. Schema: {schema}"), + ) + })?; + + schema + .as_struct() + .field_by_id(field_id) + .ok_or_else(|| { + Error::new( + ErrorKind::FeatureUnsupported, + format!( + "Column {column_name} is not a direct child of schema but a nested field, which is not supported now. Schema: {schema}" + ), + ) + })?; + + field_ids.push(field_id); + } + + Ok(field_ids) +} + +fn bind_scan_predicate( + schema: &SchemaRef, + predicate: Option<&Predicate>, + case_sensitive: bool, +) -> Result>> { + predicate + .map(|predicate| predicate.bind(schema.clone(), case_sensitive)) + .transpose() + .map(|predicate| predicate.map(Arc::new)) +} + +fn table_name_mapping(table: &Table) -> Result>> { + Ok(table + .metadata() + .properties() + .get(DEFAULT_SCHEMA_NAME_MAPPING) + .map(|raw| { + serde_json::from_str::(raw).map_err(|error| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Failed to parse table property {DEFAULT_SCHEMA_NAME_MAPPING} as a NameMapping" + ), + ) + .with_source(error) + }) + }) + .transpose()? + .map(Arc::new)) +} + +fn projected_partition_type( + table: &Table, + schema: &Schema, + field_ids: &[i32], +) -> Result>> { + if !field_ids.contains(&RESERVED_FIELD_ID_PARTITION) { + return Ok(None); + } + + compute_unified_partition_type( + table + .metadata() + .partition_specs_iter() + .map(|spec| spec.as_ref()), + schema, + ) + .map(Arc::new) + .map(Some) +} + /// Builder to create table scan. pub struct TableScanBuilder<'a> { table: &'a Table, @@ -233,85 +333,12 @@ impl<'a> TableScanBuilder<'a> { }; let schema = snapshot.schema(self.table.metadata())?; - - let mut field_ids = vec![]; - let column_names = self.column_names.clone().unwrap_or_else(|| { - schema - .as_struct() - .fields() - .iter() - .map(|f| f.name.clone()) - .collect() - }); - - for column_name in column_names.iter() { - // Handle metadata columns (like "_file") - if is_metadata_column_name(column_name) { - field_ids.push(get_metadata_field_id(column_name)?); - continue; - } - - let field_id = - resolve_field_id(&schema, column_name, self.case_sensitive).ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!("Column {column_name} not found in table. Schema: {schema}"), - ) - })?; - - schema - .as_struct() - .field_by_id(field_id) - .ok_or_else(|| { - Error::new( - ErrorKind::FeatureUnsupported, - format!( - "Column {column_name} is not a direct child of schema but a nested field, which is not supported now. Schema: {schema}" - ), - ) - })?; - - field_ids.push(field_id); - } - - let snapshot_bound_predicate = if let Some(ref predicates) = self.filter { - Some(predicates.bind(schema.clone(), self.case_sensitive)?) - } else { - None - }; - - let name_mapping = self - .table - .metadata() - .properties() - .get(DEFAULT_SCHEMA_NAME_MAPPING) - .map(|raw| { - serde_json::from_str::(raw).map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - format!( - "Failed to parse table property {DEFAULT_SCHEMA_NAME_MAPPING} as a NameMapping" - ), - ) - .with_source(e) - }) - }) - .transpose()? - .map(Arc::new); - - // Compute unified partition type if _partition is projected - let unified_partition_type = if field_ids.contains(&RESERVED_FIELD_ID_PARTITION) { - let partition_type = compute_unified_partition_type( - self.table - .metadata() - .partition_specs_iter() - .map(|s| s.as_ref()), - &schema, - )?; - Some(Arc::new(partition_type)) - } else { - None - }; + let field_ids = + projected_field_ids(&schema, self.column_names.as_deref(), self.case_sensitive)?; + let snapshot_bound_predicate = + bind_scan_predicate(&schema, self.filter.as_ref(), self.case_sensitive)?; + let name_mapping = table_name_mapping(self.table)?; + let unified_partition_type = projected_partition_type(self.table, &schema, &field_ids)?; let plan_context = PlanContext { snapshot, @@ -319,7 +346,7 @@ impl<'a> TableScanBuilder<'a> { snapshot_schema: schema, case_sensitive: self.case_sensitive, predicate: self.filter.map(Arc::new), - snapshot_bound_predicate: snapshot_bound_predicate.map(Arc::new), + snapshot_bound_predicate, object_cache: self.table.object_cache(), field_ids: Arc::new(field_ids), name_mapping, From 222ebea06ec345438a6b628a0abb76c41534cb62 Mon Sep 17 00:00:00 2001 From: Xander Date: Fri, 28 Aug 2026 13:19:58 +0100 Subject: [PATCH 2/4] comments --- crates/iceberg/public-api.txt | 4 +- crates/iceberg/src/scan/mod.rs | 70 ++++++++------------ crates/iceberg/src/spec/name_mapping/mod.rs | 3 - crates/iceberg/src/spec/table_metadata.rs | 71 ++++++++++++++++++++- crates/iceberg/src/spec/table_properties.rs | 63 ++++++++++++++++++ 5 files changed, 159 insertions(+), 52 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 19f6c26968..ce787ac5ba 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -2730,6 +2730,7 @@ pub fn iceberg::spec::TableMetadata::sort_orders_iter(&self) -> impl core::iter: pub fn iceberg::spec::TableMetadata::statistics_for_snapshot(&self, snapshot_id: i64) -> core::option::Option<&iceberg::spec::StatisticsFile> pub fn iceberg::spec::TableMetadata::statistics_iter(&self) -> impl core::iter::traits::exact_size::ExactSizeIterator pub fn iceberg::spec::TableMetadata::table_properties(&self) -> iceberg::Result +pub fn iceberg::spec::TableMetadata::unified_partition_type(&self, schema: &iceberg::spec::Schema) -> iceberg::Result pub fn iceberg::spec::TableMetadata::uuid(&self) -> uuid::Uuid pub async fn iceberg::spec::TableMetadata::write_to(&self, file_io: &iceberg::io::FileIO, metadata_location: &iceberg::MetadataLocation) -> iceberg::Result<()> impl core::clone::Clone for iceberg::spec::TableMetadata @@ -2814,6 +2815,7 @@ pub const iceberg::spec::TableProperties::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABL pub const iceberg::spec::TableProperties::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED_DEFAULT: bool pub const iceberg::spec::TableProperties::PROPERTY_DEFAULT_FILE_FORMAT: &str pub const iceberg::spec::TableProperties::PROPERTY_DEFAULT_FILE_FORMAT_DEFAULT: &str +pub const iceberg::spec::TableProperties::PROPERTY_DEFAULT_NAME_MAPPING: &str pub const iceberg::spec::TableProperties::PROPERTY_DEFAULT_PARTITION_SPEC: &str pub const iceberg::spec::TableProperties::PROPERTY_DEFAULT_SORT_ORDER: &str pub const iceberg::spec::TableProperties::PROPERTY_DELETE_DEFAULT_FILE_FORMAT: &str @@ -2875,6 +2877,7 @@ pub fn iceberg::spec::TableProperties::commit_max_retry_wait_ms(&self) -> u64 pub fn iceberg::spec::TableProperties::commit_min_retry_wait_ms(&self) -> u64 pub fn iceberg::spec::TableProperties::commit_num_retries(&self) -> usize pub fn iceberg::spec::TableProperties::commit_total_retry_timeout_ms(&self) -> u64 +pub fn iceberg::spec::TableProperties::default_name_mapping(&self) -> &core::option::Option> pub fn iceberg::spec::TableProperties::encryption_data_key_length(&self) -> usize pub fn iceberg::spec::TableProperties::encryption_key_id(&self) -> &core::option::Option pub fn iceberg::spec::TableProperties::from_properties(properties: &std::collections::hash::map::HashMap) -> iceberg::Result @@ -3089,7 +3092,6 @@ pub fn iceberg::spec::ViewVersionLog::serialize<__S>(&self, __serializer: __S) - impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::ViewVersionLog pub fn iceberg::spec::ViewVersionLog::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub const iceberg::spec::DEFAULT_SCHEMA_ID: iceberg::spec::SchemaId -pub const iceberg::spec::DEFAULT_SCHEMA_NAME_MAPPING: &str pub const iceberg::spec::INITIAL_ROW_ID: u64 pub const iceberg::spec::LIST_FIELD_NAME: &str pub const iceberg::spec::MAIN_BRANCH: &str diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 8e205a788d..2c547eb4f3 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -40,12 +40,8 @@ use crate::io::FileIO; use crate::metadata_columns::{ RESERVED_FIELD_ID_PARTITION, get_metadata_field_id, is_metadata_column_name, }; -use crate::partitioning::compute_unified_partition_type; use crate::runtime::Runtime; -use crate::spec::{ - DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, NameMapping, Schema, SchemaRef, SnapshotRef, - StructType, -}; +use crate::spec::{DataContentType, Schema, SchemaRef, SnapshotRef, StructType}; use crate::table::Table; use crate::util::available_parallelism; use crate::{Error, ErrorKind, Result}; @@ -64,7 +60,7 @@ fn resolve_field_id(schema: &Schema, column_name: &str, case_sensitive: bool) -> } } -fn projected_field_ids( +fn collect_scan_field_ids( schema: &Schema, column_names: Option<&[String]>, case_sensitive: bool, @@ -121,26 +117,6 @@ fn bind_scan_predicate( .map(|predicate| predicate.map(Arc::new)) } -fn table_name_mapping(table: &Table) -> Result>> { - Ok(table - .metadata() - .properties() - .get(DEFAULT_SCHEMA_NAME_MAPPING) - .map(|raw| { - serde_json::from_str::(raw).map_err(|error| { - Error::new( - ErrorKind::DataInvalid, - format!( - "Failed to parse table property {DEFAULT_SCHEMA_NAME_MAPPING} as a NameMapping" - ), - ) - .with_source(error) - }) - }) - .transpose()? - .map(Arc::new)) -} - fn projected_partition_type( table: &Table, schema: &Schema, @@ -150,15 +126,11 @@ fn projected_partition_type( return Ok(None); } - compute_unified_partition_type( - table - .metadata() - .partition_specs_iter() - .map(|spec| spec.as_ref()), - schema, - ) - .map(Arc::new) - .map(Some) + table + .metadata() + .unified_partition_type(schema) + .map(Arc::new) + .map(Some) } /// Builder to create table scan. @@ -334,10 +306,15 @@ impl<'a> TableScanBuilder<'a> { let schema = snapshot.schema(self.table.metadata())?; let field_ids = - projected_field_ids(&schema, self.column_names.as_deref(), self.case_sensitive)?; + collect_scan_field_ids(&schema, self.column_names.as_deref(), self.case_sensitive)?; let snapshot_bound_predicate = bind_scan_predicate(&schema, self.filter.as_ref(), self.case_sensitive)?; - let name_mapping = table_name_mapping(self.table)?; + let name_mapping = self + .table + .metadata() + .table_properties()? + .default_name_mapping() + .clone(); let unified_partition_type = projected_partition_type(self.table, &schema, &field_ids)?; let plan_context = PlanContext { @@ -691,11 +668,10 @@ pub mod tests { }; use crate::scan::FileScanTask; use crate::spec::{ - DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, DataFileBuilder, DataFileFormat, Datum, - FormatVersion, Literal, MAIN_BRANCH, ManifestEntry, ManifestListWriter, ManifestStatus, - ManifestWriterBuilder, NestedField, Operation, PartitionSpec, PrimitiveType, Schema, - Snapshot, Struct, StructType, Summary, TableMetadata, TableMetadataBuilder, Type, - UnboundPartitionSpec, + DataContentType, DataFileBuilder, DataFileFormat, Datum, FormatVersion, Literal, + MAIN_BRANCH, ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, + NestedField, Operation, PartitionSpec, PrimitiveType, Schema, Snapshot, Struct, StructType, + Summary, TableMetadata, TableMetadataBuilder, TableProperties, Type, UnboundPartitionSpec, }; use crate::table::Table; use crate::test_utils::test_runtime; @@ -1923,7 +1899,8 @@ pub mod tests { #[test] fn test_table_scan_with_name_mapping_property() { let mapping_json = r#"[{"field-id":1,"names":["id","record_id"]}]"#; - let table = table_with_property(DEFAULT_SCHEMA_NAME_MAPPING, mapping_json); + let table = + table_with_property(TableProperties::PROPERTY_DEFAULT_NAME_MAPPING, mapping_json); let table_scan = table.scan().build().unwrap(); let mapping = table_scan @@ -1944,7 +1921,10 @@ pub mod tests { #[test] fn test_table_scan_with_malformed_name_mapping_property() { - let table = table_with_property(DEFAULT_SCHEMA_NAME_MAPPING, "{ not valid json"); + let table = table_with_property( + TableProperties::PROPERTY_DEFAULT_NAME_MAPPING, + "{ not valid json", + ); let err = table .scan() @@ -1961,7 +1941,7 @@ pub mod tests { let mapping_json = r#"[{"field-id":1,"names":["id","record_id"]}]"#; let mut metadata = fixture.table.metadata().clone(); metadata.properties.insert( - DEFAULT_SCHEMA_NAME_MAPPING.to_string(), + TableProperties::PROPERTY_DEFAULT_NAME_MAPPING.to_string(), mapping_json.to_string(), ); let table = Table::builder() diff --git a/crates/iceberg/src/spec/name_mapping/mod.rs b/crates/iceberg/src/spec/name_mapping/mod.rs index db9e44c290..404ea49752 100644 --- a/crates/iceberg/src/spec/name_mapping/mod.rs +++ b/crates/iceberg/src/spec/name_mapping/mod.rs @@ -22,9 +22,6 @@ use std::sync::Arc; use serde::{Deserialize, Serialize}; use serde_with::{DefaultOnNull, serde_as}; -/// Property name for name mapping. -pub const DEFAULT_SCHEMA_NAME_MAPPING: &str = "schema.name-mapping.default"; - /// Iceberg fallback field name to ID mapping. #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)] #[serde(transparent)] diff --git a/crates/iceberg/src/spec/table_metadata.rs b/crates/iceberg/src/spec/table_metadata.rs index bb455a4a5c..9f92edb8a4 100644 --- a/crates/iceberg/src/spec/table_metadata.rs +++ b/crates/iceberg/src/spec/table_metadata.rs @@ -33,14 +33,15 @@ use uuid::Uuid; use super::snapshot::SnapshotReference; pub use super::table_metadata_builder::{TableMetadataBuildResult, TableMetadataBuilder}; use super::{ - DEFAULT_PARTITION_SPEC_ID, PartitionSpecRef, PartitionStatisticsFile, SchemaId, SchemaRef, - SnapshotRef, SnapshotRetention, SortOrder, SortOrderRef, StatisticsFile, StructType, + DEFAULT_PARTITION_SPEC_ID, PartitionSpecRef, PartitionStatisticsFile, Schema, SchemaId, + SchemaRef, SnapshotRef, SnapshotRetention, SortOrder, SortOrderRef, StatisticsFile, StructType, TableProperties, parse_metadata_file_compression, }; use crate::catalog::{METADATA_FOLDER_NAME, MetadataLocation}; use crate::compression::CompressionCodec; use crate::error::{Result, timestamp_ms_to_utc}; use crate::io::FileIO; +use crate::partitioning::compute_unified_partition_type; use crate::spec::EncryptedKey; use crate::{Error, ErrorKind}; @@ -277,6 +278,19 @@ impl TableMetadata { &self.default_partition_type } + /// Returns the unified partition type across every partition spec in the table, resolved + /// against `schema`. + /// + /// Unlike [`Self::default_partition_type`], the result contains all partition fields ever + /// used by the table, so partition values stay readable across partition spec evolution. + /// See [`compute_unified_partition_type`] for the exact merge rules. + pub fn unified_partition_type(&self, schema: &Schema) -> Result { + compute_unified_partition_type( + self.partition_specs_iter().map(|spec| spec.as_ref()), + schema, + ) + } + #[inline] /// Returns spec id of the "current" partition spec. pub fn default_partition_spec_id(&self) -> i32 { @@ -1639,7 +1653,7 @@ mod tests { BlobMetadata, EncryptedKey, INITIAL_ROW_ID, Literal, NestedField, NullOrder, Operation, PartitionSpec, PartitionStatisticsFile, PrimitiveLiteral, PrimitiveType, Schema, Snapshot, SnapshotReference, SnapshotRetention, SortDirection, SortField, SortOrder, StatisticsFile, - Summary, TableProperties, Transform, Type, UnboundPartitionField, + Summary, TableProperties, Transform, Type, UnboundPartitionField, UnboundPartitionSpec, }; use crate::{ErrorKind, TableCreation}; @@ -4364,4 +4378,55 @@ mod tests { "s3://other-bucket/custom-meta" ); } + + #[test] + fn test_unified_partition_type_spans_all_specs() { + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(3, "z", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(); + + let metadata = TableMetadataBuilder::new( + schema.clone(), + UnboundPartitionSpec::builder() + .with_spec_id(0) + .add_partition_field(2, "y", Transform::Identity) + .unwrap() + .build(), + SortOrder::unsorted_order(), + "s3://bucket/table".to_string(), + FormatVersion::V2, + HashMap::new(), + ) + .unwrap() + .build() + .unwrap() + .metadata + .into_builder(None) + .add_partition_spec( + UnboundPartitionSpec::builder() + .add_partition_field(3, "z", Transform::Identity) + .unwrap() + .build(), + ) + .unwrap() + .build() + .unwrap() + .metadata; + + // The default spec only knows about `y`, but `_partition` must expose both. + assert_eq!(metadata.default_partition_type().fields().len(), 1); + + let unified = metadata.unified_partition_type(&schema).unwrap(); + let names: Vec<&str> = unified + .fields() + .iter() + .map(|field| field.name.as_str()) + .collect(); + assert_eq!(names, vec!["y", "z"]); + } } diff --git a/crates/iceberg/src/spec/table_properties.rs b/crates/iceberg/src/spec/table_properties.rs index 7ed1c032d5..c98ee0a344 100644 --- a/crates/iceberg/src/spec/table_properties.rs +++ b/crates/iceberg/src/spec/table_properties.rs @@ -16,12 +16,14 @@ // under the License. use std::collections::HashMap; +use std::sync::Arc; use iceberg_property_macro::Properties; use crate::compression::CompressionCodec; use crate::encryption::AesKeySize; use crate::error::{Error, ErrorKind, Result}; +use crate::spec::NameMapping; use crate::util::location::strip_trailing_slash; fn parse_location_property(path: &str) -> Result { @@ -32,6 +34,18 @@ fn parse_location_property(path: &str) -> Result { Ok(strip_trailing_slash(path).to_string()) } +fn parse_name_mapping(value: &str) -> Result> { + serde_json::from_str::(value) + .map(Arc::new) + .map_err(|error| { + Error::new( + ErrorKind::DataInvalid, + "Failed to parse table property as a NameMapping", + ) + .with_source(error) + }) +} + /// Parse compression codec for metadata files from table properties. /// Retrieves the compression codec property, applies defaults, and parses the value. /// Only "none" (or empty string) and "gzip" are supported for metadata compression. @@ -358,6 +372,15 @@ pub struct TableProperties { getter )] write_object_storage_partitioned_paths: bool, + /// The table's default name mapping, used to assign field ids when reading data files + /// that carry no field id metadata. `None` if `schema.name-mapping.default` is not set. + #[property( + key = Self::PROPERTY_DEFAULT_NAME_MAPPING, + default = None, + parse_with = parse_name_mapping, + getter + )] + default_name_mapping: Option>, } impl TableProperties { @@ -452,6 +475,10 @@ impl TableProperties { /// location. pub const PROPERTY_WRITE_METADATA_PATH: &str = "write.metadata.path"; + /// Property key for the table's default name mapping, stored as a JSON + /// [`NameMapping`] document. + pub const PROPERTY_DEFAULT_NAME_MAPPING: &str = "schema.name-mapping.default"; + /// Compression codec for metadata files (JSON) pub const PROPERTY_METADATA_COMPRESSION_CODEC: &str = "write.metadata.compression-codec"; /// Default metadata compression codec - uncompressed @@ -1233,4 +1260,40 @@ mod tests { assert!(tp.write_object_storage_partitioned_paths); } } + + #[test] + fn test_table_properties_default_name_mapping() { + // Test unset. + let table_properties = TableProperties::try_from(&HashMap::new()).unwrap(); + assert!(table_properties.default_name_mapping().is_none()); + + let table_properties = TableProperties::try_from(&HashMap::from([( + TableProperties::PROPERTY_DEFAULT_NAME_MAPPING.to_string(), + r#"[{"field-id":1,"names":["id","record_id"]}]"#.to_string(), + )])) + .unwrap(); + let mapping = table_properties.default_name_mapping().as_ref().unwrap(); + assert_eq!(mapping.fields().len(), 1); + assert_eq!(mapping.fields()[0].field_id(), Some(1)); + assert_eq!(mapping.fields()[0].names(), &[ + "id".to_string(), + "record_id".to_string() + ]); + } + + #[test] + fn test_table_properties_malformed_name_mapping() { + let error = TableProperties::try_from(&HashMap::from([( + TableProperties::PROPERTY_DEFAULT_NAME_MAPPING.to_string(), + "{ not valid json".to_string(), + )])) + .unwrap_err(); + + assert_eq!(error.kind(), ErrorKind::DataInvalid); + // The property key must survive as error context. + assert!( + format!("{error}").contains(TableProperties::PROPERTY_DEFAULT_NAME_MAPPING), + "{error}" + ); + } } From 5853c0e6ad0d121c9271e3f4c1d14d1ba3479e4d Mon Sep 17 00:00:00 2001 From: Xander Date: Mon, 31 Aug 2026 18:15:23 +0100 Subject: [PATCH 3/4] comments --- crates/iceberg/public-api.txt | 5 ++++- crates/iceberg/src/scan/mod.rs | 3 ++- crates/iceberg/src/spec/name_mapping/mod.rs | 18 ++++++++++++++++++ crates/iceberg/src/spec/table_properties.rs | 16 +--------------- 4 files changed, 25 insertions(+), 17 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index c21d6ad6ec..e8a887b3b2 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -2270,6 +2270,9 @@ pub fn iceberg::spec::NameMapping::eq(&self, other: &iceberg::spec::NameMapping) impl core::fmt::Debug for iceberg::spec::NameMapping pub fn iceberg::spec::NameMapping::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl core::marker::StructuralPartialEq for iceberg::spec::NameMapping +impl core::str::traits::FromStr for iceberg::spec::NameMapping +pub type iceberg::spec::NameMapping::Err = iceberg::Error +pub fn iceberg::spec::NameMapping::from_str(value: &str) -> iceberg::Result impl serde_core::ser::Serialize for iceberg::spec::NameMapping pub fn iceberg::spec::NameMapping::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::spec::NameMapping @@ -2877,7 +2880,7 @@ pub fn iceberg::spec::TableProperties<'properties>::commit_max_retry_wait_ms(&se pub fn iceberg::spec::TableProperties<'properties>::commit_min_retry_wait_ms(&self) -> iceberg::Result pub fn iceberg::spec::TableProperties<'properties>::commit_num_retries(&self) -> iceberg::Result pub fn iceberg::spec::TableProperties<'properties>::commit_total_retry_timeout_ms(&self) -> iceberg::Result -pub fn iceberg::spec::TableProperties<'properties>::default_name_mapping(&self) -> iceberg::Result>> +pub fn iceberg::spec::TableProperties<'properties>::default_name_mapping(&self) -> iceberg::Result> pub fn iceberg::spec::TableProperties<'properties>::encryption_data_key_length(&self) -> iceberg::Result pub fn iceberg::spec::TableProperties<'properties>::encryption_key_id(&self) -> iceberg::Result> pub fn iceberg::spec::TableProperties<'properties>::gc_enabled(&self) -> iceberg::Result diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index ded4a4d6b9..8715040c09 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -313,7 +313,8 @@ impl<'a> TableScanBuilder<'a> { .table .metadata() .table_properties() - .default_name_mapping()?; + .default_name_mapping()? + .map(Arc::new); let unified_partition_type = projected_partition_type(self.table, &schema, &field_ids)?; let plan_context = PlanContext { diff --git a/crates/iceberg/src/spec/name_mapping/mod.rs b/crates/iceberg/src/spec/name_mapping/mod.rs index 404ea49752..a87e829ad0 100644 --- a/crates/iceberg/src/spec/name_mapping/mod.rs +++ b/crates/iceberg/src/spec/name_mapping/mod.rs @@ -17,11 +17,14 @@ //! Iceberg name mapping. +use std::str::FromStr; use std::sync::Arc; use serde::{Deserialize, Serialize}; use serde_with::{DefaultOnNull, serde_as}; +use crate::{Error, ErrorKind, Result}; + /// Iceberg fallback field name to ID mapping. #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)] #[serde(transparent)] @@ -41,6 +44,21 @@ impl NameMapping { } } +impl FromStr for NameMapping { + type Err = Error; + + /// Parses a [`NameMapping`] from its JSON representation. + fn from_str(value: &str) -> Result { + serde_json::from_str(value).map_err(|error| { + Error::new( + ErrorKind::DataInvalid, + "Failed to parse value as a NameMapping", + ) + .with_source(error) + }) + } +} + /// Maps field names to IDs. #[serde_as] #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)] diff --git a/crates/iceberg/src/spec/table_properties.rs b/crates/iceberg/src/spec/table_properties.rs index 3486c7d2c6..01a330e28d 100644 --- a/crates/iceberg/src/spec/table_properties.rs +++ b/crates/iceberg/src/spec/table_properties.rs @@ -16,7 +16,6 @@ // under the License. use std::collections::HashMap; -use std::sync::Arc; use iceberg_property_macro::properties_view; @@ -34,18 +33,6 @@ fn parse_location_property(path: &str) -> Result { Ok(strip_trailing_slash(path).to_string()) } -fn parse_name_mapping(value: &str) -> Result> { - serde_json::from_str::(value) - .map(Arc::new) - .map_err(|error| { - Error::new( - ErrorKind::DataInvalid, - "Failed to parse table property as a NameMapping", - ) - .with_source(error) - }) -} - fn parse_metadata_compression(value: &str) -> Result { // Handle empty string as None if value.is_empty() { @@ -355,10 +342,9 @@ pub struct TableProperties { #[property( key = Self::PROPERTY_DEFAULT_NAME_MAPPING, default = None, - parse_with = parse_name_mapping, getter )] - default_name_mapping: Option>, + default_name_mapping: Option, } } From eb33947670914abf36c4efb1494c7c0296843caa Mon Sep 17 00:00:00 2001 From: Xander Date: Mon, 31 Aug 2026 18:24:47 +0100 Subject: [PATCH 4/4] remove allocation --- crates/iceberg/src/scan/mod.rs | 58 +++++++++++++++------------------- 1 file changed, 26 insertions(+), 32 deletions(-) diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 8715040c09..909edf40ba 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -65,45 +65,39 @@ fn collect_scan_field_ids( column_names: Option<&[String]>, case_sensitive: bool, ) -> Result> { - let mut field_ids = vec![]; - let column_names = column_names.map(<[String]>::to_vec).unwrap_or_else(|| { - schema - .as_struct() - .fields() - .iter() - .map(|f| f.name.clone()) - .collect() - }); - - for column_name in column_names.iter() { - if is_metadata_column_name(column_name) { - field_ids.push(get_metadata_field_id(column_name)?); - continue; - } + let Some(column_names) = column_names else { + return Ok(schema.as_struct().fields().iter().map(|f| f.id).collect()); + }; - let field_id = resolve_field_id(schema, column_name, case_sensitive).ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!("Column {column_name} not found in table. Schema: {schema}"), - ) - })?; + column_names + .iter() + .map(|column_name| { + if is_metadata_column_name(column_name) { + return get_metadata_field_id(column_name); + } - schema - .as_struct() - .field_by_id(field_id) - .ok_or_else(|| { + let field_id = resolve_field_id(schema, column_name, case_sensitive).ok_or_else(|| { Error::new( - ErrorKind::FeatureUnsupported, - format!( - "Column {column_name} is not a direct child of schema but a nested field, which is not supported now. Schema: {schema}" - ), + ErrorKind::DataInvalid, + format!("Column {column_name} not found in table. Schema: {schema}"), ) })?; - field_ids.push(field_id); - } + schema + .as_struct() + .field_by_id(field_id) + .ok_or_else(|| { + Error::new( + ErrorKind::FeatureUnsupported, + format!( + "Column {column_name} is not a direct child of schema but a nested field, which is not supported now. Schema: {schema}" + ), + ) + })?; - Ok(field_ids) + Ok(field_id) + }) + .collect() } fn bind_scan_predicate(