From 1335780ab457b6720e06c348d8828cc3a98949f1 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 23 Sep 2026 12:02:34 +0300 Subject: [PATCH] Avoid position delete metadata column collisions --- datafusion_iceberg/src/table/mod.rs | 275 +++++++++++++++----- datafusion_iceberg/tests/position_delete.rs | 173 +++++++++++- 2 files changed, 373 insertions(+), 75 deletions(-) diff --git a/datafusion_iceberg/src/table/mod.rs b/datafusion_iceberg/src/table/mod.rs index e46c7144..1cf5bd79 100644 --- a/datafusion_iceberg/src/table/mod.rs +++ b/datafusion_iceberg/src/table/mod.rs @@ -154,6 +154,59 @@ fn row_lineage_field(name: &str, field_id: i32) -> Field { )])) } +#[derive(Clone, Copy, Debug)] +struct PositionDeleteDataColumns { + file_path: usize, + row_position: usize, + sequence_number: usize, +} + +impl PositionDeleteDataColumns { + fn project(self, projection: &[usize]) -> Result { + let projected_index = |scan_index| { + projection + .iter() + .position(|index| *index == scan_index) + .ok_or_else(|| { + DataFusionError::Plan(format!( + "Position delete scan column {scan_index} is missing from the projection" + )) + }) + }; + + Ok(Self { + file_path: projected_index(self.file_path)?, + row_position: projected_index(self.row_position)?, + sequence_number: projected_index(self.sequence_number)?, + }) + } +} + +fn unique_internal_column_name( + file_schema: &ArrowSchema, + partition_columns: &[Field], + base: &str, +) -> String { + let is_available = |candidate: &str| { + file_schema + .fields() + .iter() + .all(|field| field.name() != candidate) + && partition_columns + .iter() + .all(|field| field.name() != candidate) + }; + + if is_available(base) { + return base.to_owned(); + } + + (1..) + .map(|suffix| format!("{base}_{suffix}")) + .find(|candidate| is_available(candidate)) + .expect("an internal column name must be available") +} + /// When the view tracks source arrow ids (the `overrides` map is non-empty), /// reshape each top-level field's `PARQUET:field_id` metadata so it matches /// what the underlying SELECT actually produces at run time: @@ -691,13 +744,17 @@ async fn table_scan( .cloned() .unwrap_or((0..arrow_schema.fields().len()).collect_vec()); + let mut data_file_path_scan_index = None; if enable_data_file_path_column { + data_file_path_scan_index = Some(file_schema.fields().len() + table_partition_cols.len()); table_partition_cols.push(Field::new(DATA_FILE_PATH_COLUMN, DataType::Utf8, false)); } - if enable_manifest_file_path_column { + let manifest_file_path_partition_index = enable_manifest_file_path_column.then(|| { + let index = table_partition_cols.len(); table_partition_cols.push(Field::new(MANIFEST_FILE_PATH_COLUMN, DataType::Utf8, false)); - } + index + }); // All files have to be grouped according to their partition values. This is done by using a HashMap with the partition values as the key. // This way data files with the same partition value are mapped to the same vector. @@ -908,30 +965,53 @@ async fn table_scan( .any(|files| !files.position_deletes.is_empty()); let include_data_file_path_column = enable_data_file_path_column || has_position_deletes; - // Position deletes identify rows by data-file path and absolute row position. Keep both - // columns internal unless the caller explicitly requested the data-file path column. - if has_position_deletes && !enable_data_file_path_column { - table_partition_cols.push(Field::new(DATA_FILE_PATH_COLUMN, DataType::Utf8, false)); - statistics - .column_statistics - .push(ColumnStatistics::new_unknown()); - } + // Carry physical indices into delete reconciliation so valid user columns cannot shadow the + // internal path, row-position, or sequence-number columns. + let position_delete_file_path_index = if has_position_deletes { + Some(if let Some(index) = data_file_path_scan_index { + index + } else { + let name = unique_internal_column_name( + file_schema.as_ref(), + &table_partition_cols, + DATA_FILE_PATH_COLUMN, + ); + // PartitionedFile values always place the data-file path before the manifest path. + // Preserve that physical order when the data-file path is internal to delete handling. + let partition_index = + manifest_file_path_partition_index.unwrap_or(table_partition_cols.len()); + let index = file_schema.fields().len() + partition_index; + table_partition_cols.insert(partition_index, Field::new(name, DataType::Utf8, false)); + statistics + .column_statistics + .push(ColumnStatistics::new_unknown()); + index + }) + } else { + None + }; if enable_row_lineage { table_partition_cols.push(Field::new(FIRST_ROW_ID_COLUMN, DataType::Int64, true)); statistics .column_statistics .push(ColumnStatistics::new_unknown()); } - if has_position_deletes || enable_last_updated_sequence_number_column { - table_partition_cols.push(Field::new( - DATA_FILE_SEQUENCE_NUMBER_COLUMN, - DataType::Int64, - false, - )); - statistics - .column_statistics - .push(ColumnStatistics::new_unknown()); - } + let sequence_number_column = + if has_position_deletes || enable_last_updated_sequence_number_column { + let name = unique_internal_column_name( + file_schema.as_ref(), + &table_partition_cols, + DATA_FILE_SEQUENCE_NUMBER_COLUMN, + ); + let index = file_schema.fields().len() + table_partition_cols.len(); + table_partition_cols.push(Field::new(name.clone(), DataType::Int64, false)); + statistics + .column_statistics + .push(ColumnStatistics::new_unknown()); + Some((name, index)) + } else { + None + }; if scan_changed_data_files { table_partition_cols.push(Field::new(CHANGE_FILE_STATUS_COLUMN, DataType::Utf8, false)); statistics @@ -955,10 +1035,20 @@ async fn table_scan( .map(Arc::new) .collect::>(), ); - if has_position_deletes || enable_data_file_row_position_column || enable_row_id_column { + let needs_row_position = + has_position_deletes || enable_data_file_row_position_column || enable_row_id_column; + let row_position_column = needs_row_position.then(|| { + let name = unique_internal_column_name( + file_schema.as_ref(), + &table_partition_cols, + DATA_FILE_ROW_POSITION_COLUMN, + ); + let index = file_schema.fields().len() + table_partition_cols.len(); + (name, index) + }); + if let Some((name, _)) = &row_position_column { table_schema_builder = table_schema_builder.with_virtual_columns(vec![Arc::new( - Field::new(DATA_FILE_ROW_POSITION_COLUMN, DataType::Int64, false) - .with_extension_type(RowNumber), + Field::new(name, DataType::Int64, false).with_extension_type(RowNumber), )]); statistics .column_statistics @@ -973,7 +1063,30 @@ async fn table_scan( &requested_projection, enable_row_id_column, enable_last_updated_sequence_number_column, + row_position_column + .as_ref() + .map(|(name, _)| name.as_str()) + .unwrap_or(DATA_FILE_ROW_POSITION_COLUMN), + sequence_number_column + .as_ref() + .map(|(name, _)| name.as_str()) + .unwrap_or(DATA_FILE_SEQUENCE_NUMBER_COLUMN), )?; + let position_delete_data_columns = if has_position_deletes { + Some(PositionDeleteDataColumns { + file_path: position_delete_file_path_index.expect("position deletes require a path"), + row_position: row_position_column + .as_ref() + .expect("position deletes require a row position") + .1, + sequence_number: sequence_number_column + .as_ref() + .expect("position deletes require a sequence number") + .1, + }) + } else { + None + }; // See `use_parquet_row_filter_pushdown` for the wide-scan / narrow-predicate rationale. let filter_columns: std::collections::HashSet<_> = parquet_filters .flat_map(|f| f.column_refs().into_iter().cloned()) @@ -996,7 +1109,6 @@ async fn table_scan( let parquet_reader_factory = parquet_reader_factory.clone(); let projection_expr = projection_expr.clone(); let scan_projection = scan_projection.clone(); - let scan_schema = scan_schema.clone(); let change_manifest_sequence_numbers = change_manifest_sequence_numbers.clone(); let mut data_files = data_file_groups .remove(&partition_value) @@ -1048,18 +1160,25 @@ async fn table_scan( } }); - if !delete_files.position_deletes.is_empty() { - for column_name in [ - DATA_FILE_PATH_COLUMN, - DATA_FILE_ROW_POSITION_COLUMN, - DATA_FILE_SEQUENCE_NUMBER_COLUMN, + let position_delete_plan_columns = if !delete_files.position_deletes.is_empty() { + let scan_columns = position_delete_data_columns.ok_or_else(|| { + DataFusionError::Plan( + "Position delete files require internal scan columns".to_owned(), + ) + })?; + for index in [ + scan_columns.file_path, + scan_columns.row_position, + scan_columns.sequence_number, ] { - let index = scan_schema.index_of(column_name)?; if !equality_projection.contains(&index) { equality_projection.push(index); } } - } + Some(scan_columns.project(&equality_projection)?) + } else { + None + }; let mut plan = stream::iter(delete_files.equality_deletes.iter()) .map(Ok::<_, DataFusionError>) @@ -1299,6 +1418,12 @@ async fn table_scan( parquet_reader_factory, table.object_store(), &active_data_file_paths, + position_delete_plan_columns.ok_or_else(|| { + DataFusionError::Plan( + "Position delete files require projected internal columns" + .to_owned(), + ) + })?, ) .await?; } @@ -1500,6 +1625,8 @@ fn row_lineage_projection( requested_projection: &[usize], enable_row_id: bool, enable_last_updated_sequence_number: bool, + row_position_column_name: &str, + sequence_number_column_name: &str, ) -> Result<(Vec, PhysicalProjection), DataFusionError> { fn projected_column( scan_projection: &mut Vec, @@ -1530,11 +1657,7 @@ fn row_lineage_projection( RowLineageKind::RowId, projected_column(&mut scan_projection, scan_schema, PHYSICAL_ROW_ID_COLUMN)?, projected_column(&mut scan_projection, scan_schema, FIRST_ROW_ID_COLUMN)?, - projected_column( - &mut scan_projection, - scan_schema, - DATA_FILE_ROW_POSITION_COLUMN, - )?, + projected_column(&mut scan_projection, scan_schema, row_position_column_name)?, )) } else if enable_last_updated_sequence_number && name == LAST_UPDATED_SEQUENCE_NUMBER_COLUMN { @@ -1549,7 +1672,7 @@ fn row_lineage_projection( projected_column( &mut scan_projection, scan_schema, - DATA_FILE_SEQUENCE_NUMBER_COLUMN, + sequence_number_column_name, )?, )) } else { @@ -1852,6 +1975,7 @@ async fn apply_position_deletes( parquet_reader_factory: Arc, object_store: Arc, active_data_file_paths: &HashSet, + data_columns: PositionDeleteDataColumns, ) -> Result, DataFusionError> { let mut parquet_delete_files = Vec::new(); let mut deletion_vector_files = Vec::new(); @@ -1887,14 +2011,14 @@ async fn apply_position_deletes( if !deletion_vector_files.is_empty() { let vectors = load_deletion_vectors(deletion_vector_files, object_store).await?; let predicate = Arc::new(DeletionVectorPredicate::new( - Arc::new(Column::new_with_schema( - DATA_FILE_PATH_COLUMN, - &data_plan.schema(), - )?), - Arc::new(Column::new_with_schema( - DATA_FILE_ROW_POSITION_COLUMN, - &data_plan.schema(), - )?), + Arc::new(Column::new( + data_plan.schema().field(data_columns.file_path).name(), + data_columns.file_path, + )), + Arc::new(Column::new( + data_plan.schema().field(data_columns.row_position).name(), + data_columns.row_position, + )), vectors, )); data_plan = Arc::new(FilterExec::try_new(predicate, data_plan)?); @@ -1964,31 +2088,33 @@ async fn apply_position_deletes( .build(); let delete_plan: Arc = DataSourceExec::from_data_source(delete_scan_config); + let column_at = + |schema: &SchemaRef, index: usize| -> Result, DataFusionError> { + let field = schema.fields().get(index).ok_or_else(|| { + DataFusionError::Plan(format!( + "Position delete column index {index} is outside a {}-column schema", + schema.fields().len() + )) + })?; + Ok(Arc::new(Column::new(field.name(), index))) + }; let join_on = vec![ ( - Arc::new(Column::new_with_schema( - POSITION_DELETE_FILE_PATH_COLUMN, - &delete_plan.schema(), - )?) as Arc, - Arc::new(Column::new_with_schema( - DATA_FILE_PATH_COLUMN, - &data_plan.schema(), - )?) as Arc, + column_at(&delete_plan.schema(), 0)?, + column_at(&data_plan.schema(), data_columns.file_path)?, ), ( - Arc::new(Column::new_with_schema( - POSITION_DELETE_POS_COLUMN, - &delete_plan.schema(), - )?) as Arc, - Arc::new(Column::new_with_schema( - DATA_FILE_ROW_POSITION_COLUMN, - &data_plan.schema(), - )?) as Arc, + column_at(&delete_plan.schema(), 1)?, + column_at(&data_plan.schema(), data_columns.row_position)?, ), ]; - let sequence_filter = - position_delete_sequence_filter(&delete_plan.schema(), &data_plan.schema())?; + let sequence_filter = position_delete_sequence_filter( + &delete_plan.schema(), + 2, + &data_plan.schema(), + data_columns.sequence_number, + )?; Ok(Arc::new(HashJoinExec::try_new( delete_plan, @@ -2088,10 +2214,27 @@ async fn load_deletion_vectors( fn position_delete_sequence_filter( delete_schema: &SchemaRef, + delete_sequence_index: usize, data_schema: &SchemaRef, + data_sequence_index: usize, ) -> Result { - let delete_sequence_index = delete_schema.index_of(DELETE_FILE_SEQUENCE_NUMBER_COLUMN)?; - let data_sequence_index = data_schema.index_of(DATA_FILE_SEQUENCE_NUMBER_COLUMN)?; + for (schema, index, side) in [ + (delete_schema, delete_sequence_index, "delete"), + (data_schema, data_sequence_index, "data"), + ] { + let field = schema.fields().get(index).ok_or_else(|| { + DataFusionError::Plan(format!( + "Position delete {side} sequence index {index} is outside a {}-column schema", + schema.fields().len() + )) + })?; + if field.data_type() != &DataType::Int64 { + return plan_err!( + "Position delete {side} sequence column at index {index} must be Int64, got {}", + field.data_type() + ); + } + } let filter_schema = Arc::new(ArrowSchema::new(vec![ Field::new("delete_sequence_number", DataType::Int64, false), Field::new("data_sequence_number", DataType::Int64, false), @@ -2610,7 +2753,7 @@ mod tests { false, ), ])); - let filter = position_delete_sequence_filter(&delete_schema, &data_schema).unwrap(); + let filter = position_delete_sequence_filter(&delete_schema, 0, &data_schema, 0).unwrap(); let batch = RecordBatch::try_new( filter.schema().clone(), vec![ diff --git a/datafusion_iceberg/tests/position_delete.rs b/datafusion_iceberg/tests/position_delete.rs index 59274db7..35ea4ab2 100644 --- a/datafusion_iceberg/tests/position_delete.rs +++ b/datafusion_iceberg/tests/position_delete.rs @@ -14,7 +14,10 @@ use datafusion::{ }, prelude::SessionContext, }; -use datafusion_iceberg::catalog::catalog::IcebergCatalog; +use datafusion_iceberg::{ + catalog::catalog::IcebergCatalog, + table::{DataFusionTable, DataFusionTableConfigBuilder}, +}; use futures::{stream, TryStreamExt}; use iceberg_rust::{ arrow::write::write_equality_deletes_parquet_partitioned, @@ -23,6 +26,7 @@ use iceberg_rust::{ spec::{ manifest::{Content, DataFile, FileFormat, Status}, namespace::Namespace, + partition::{PartitionField, PartitionSpec, Transform}, puffin::{Blob, PuffinWriter, STANDARD_BLOB_TYPE_DELETION_VECTOR_V1}, schema::Schema, types::{PrimitiveType, StructField, Type}, @@ -47,7 +51,12 @@ async fn run_query(query: &str, ctx: &SessionContext) -> Vec { .expect("query execution failed") } -fn write_position_delete_file(path: &str, data_file_path: &str, positions: &[i64]) -> DataFile { +fn write_position_delete_file( + path: &str, + data_file_path: &str, + partition: Struct, + positions: &[i64], +) -> DataFile { let schema = Arc::new(ArrowSchema::new(vec![ Field::new("file_path", DataType::Utf8, false).with_metadata(HashMap::from([( PARQUET_FIELD_ID_META_KEY.to_string(), @@ -78,7 +87,7 @@ fn write_position_delete_file(path: &str, data_file_path: &str, positions: &[i64 .with_content(Content::PositionDeletes) .with_file_path(path.to_string()) .with_file_format(FileFormat::Parquet) - .with_partition(Struct::from_iter(Vec::<(String, Option)>::new())) + .with_partition(partition) .with_record_count(metadata.file_metadata().num_rows()) .with_file_size_in_bytes(i64::try_from(file_size).unwrap()) .with_column_sizes(None) @@ -180,6 +189,52 @@ async fn applies_v2_position_deletes() { initial_default: None, write_default: None, }) + .with_struct_field(StructField { + id: 3, + name: "__data_file_path".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::Long), + doc: None, + initial_default: None, + write_default: None, + }) + .with_struct_field(StructField { + id: 4, + name: "__iceberg_file_row_position".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::String), + doc: None, + initial_default: None, + write_default: None, + }) + .with_struct_field(StructField { + id: 5, + name: "__iceberg_data_sequence_number".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::String), + doc: None, + initial_default: None, + write_default: None, + }) + .with_struct_field(StructField { + id: 6, + name: "category".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::String), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap(); + + let partition_spec = PartitionSpec::builder() + .with_partition_field(PartitionField::new( + 6, + 1000, + "category", + Transform::Identity, + )) .build() .unwrap(); @@ -187,6 +242,7 @@ async fn applies_v2_position_deletes() { .with_name("orders") .with_location(&table_dir) .with_schema(schema) + .with_partition_spec(partition_spec) .build(&["test".to_owned()], catalog.clone()) .await .unwrap(); @@ -199,8 +255,12 @@ async fn applies_v2_position_deletes() { run_query( "INSERT INTO warehouse.test.orders VALUES - (1, 'one'), (2, 'two'), (3, 'three'), - (4, 'four'), (5, 'five'), (6, 'six')", + (1, 'one', 101, 'row-one', 'seq-one', 'a'), + (2, 'two', 102, 'row-two', 'seq-two', 'a'), + (3, 'three', 103, 'row-three', 'seq-three', 'a'), + (4, 'four', 104, 'row-four', 'seq-four', 'a'), + (5, 'five', 105, 'row-five', 'seq-five', 'a'), + (6, 'six', 106, 'row-six', 'seq-six', 'a')", &ctx, ) .await; @@ -217,13 +277,14 @@ async fn applies_v2_position_deletes() { .try_collect::>() .await .unwrap(); - let (_, data_manifest_entry) = data_files + let (data_manifest_path, data_manifest_entry) = data_files .iter() .find(|(_, entry)| { entry.status() != &Status::Deleted && entry.data_file().content() == &Content::Data }) .unwrap(); let data_file_path = data_manifest_entry.data_file().file_path().clone(); + let partition = data_manifest_entry.data_file().partition().clone(); let delete_dir = format!("{table_dir}/data"); std::fs::create_dir_all(&delete_dir).unwrap(); @@ -231,11 +292,13 @@ async fn applies_v2_position_deletes() { write_position_delete_file( &format!("{delete_dir}/position-delete-1.parquet"), &data_file_path, + partition.clone(), &[1, 4], ), write_position_delete_file( &format!("{delete_dir}/position-delete-2.parquet"), &data_file_path, + partition, &[4, 5], ), ]; @@ -247,6 +310,61 @@ async fn applies_v2_position_deletes() { .await .unwrap(); + let Tabular::Table(table_with_manifest_metadata) = + catalog.clone().load_tabular(&identifier).await.unwrap() + else { + panic!("orders should be an Iceberg table"); + }; + let metadata_config = DataFusionTableConfigBuilder::default() + .enable_data_file_path_column(false) + .enable_data_file_row_position_column(false) + .enable_manifest_file_path_column(true) + .build() + .unwrap(); + let metadata_ctx = SessionContext::new(); + metadata_ctx + .register_table( + "orders_with_manifest_metadata", + Arc::new(DataFusionTable::new_with_config( + Tabular::Table(table_with_manifest_metadata), + None, + None, + None, + Some(metadata_config), + )), + ) + .unwrap(); + let metadata_batches = run_query( + "SELECT __manifest_file_path FROM orders_with_manifest_metadata LIMIT 1", + &metadata_ctx, + ) + .await; + let manifest_paths = metadata_batches[0] + .column_by_name("__manifest_file_path") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(manifest_paths.value(0), data_manifest_path); + + let metadata_batches = run_query( + "SELECT id, payload FROM orders_with_manifest_metadata ORDER BY id", + &metadata_ctx, + ) + .await; + assert_batches_eq!( + [ + "+----+---------+", + "| id | payload |", + "+----+---------+", + "| 1 | one |", + "| 3 | three |", + "| 4 | four |", + "+----+---------+", + ], + &metadata_batches + ); + let batches = run_query( "SELECT id, payload FROM warehouse.test.orders ORDER BY id", &ctx, @@ -265,6 +383,26 @@ async fn applies_v2_position_deletes() { &batches ); + let batches = run_query( + "SELECT __iceberg_data_sequence_number, id, __data_file_path, + __iceberg_file_row_position + FROM warehouse.test.orders ORDER BY id", + &ctx, + ) + .await; + assert_batches_eq!( + [ + "+--------------------------------+----+------------------+-----------------------------+", + "| __iceberg_data_sequence_number | id | __data_file_path | __iceberg_file_row_position |", + "+--------------------------------+----+------------------+-----------------------------+", + "| seq-one | 1 | 101 | row-one |", + "| seq-three | 3 | 103 | row-three |", + "| seq-four | 4 | 104 | row-four |", + "+--------------------------------+----+------------------+-----------------------------+", + ], + &batches + ); + let batches = run_query( "SELECT id FROM warehouse.test.orders WHERE id IN (2, 3, 5) ORDER BY id", &ctx, @@ -276,7 +414,9 @@ async fn applies_v2_position_deletes() { ); run_query( - "INSERT INTO warehouse.test.orders VALUES (7, 'seven'), (8, 'eight')", + "INSERT INTO warehouse.test.orders VALUES + (7, 'seven', 107, 'row-seven', 'seq-seven', 'a'), + (8, 'eight', 108, 'row-eight', 'seq-eight', 'a')", &ctx, ) .await; @@ -291,10 +431,25 @@ async fn applies_v2_position_deletes() { ); let equality_rows = run_query( - "SELECT id FROM warehouse.test.orders WHERE id IN (3, 7)", + "SELECT id, category FROM warehouse.test.orders WHERE id IN (3, 7)", &ctx, ) .await; + let equality_schema = Arc::new(ArrowSchema::new(vec![ + Field::new("id", DataType::Int64, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "1".to_string(), + )])), + Field::new("category", DataType::Utf8, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "6".to_string(), + )])), + ])); + let equality_rows = equality_rows + .into_iter() + .map(|batch| RecordBatch::try_new(equality_schema.clone(), batch.columns().to_vec())) + .collect::, _>>() + .unwrap(); let Tabular::Table(mut table) = catalog.clone().load_tabular(&identifier).await.unwrap() else { panic!("orders should be an Iceberg table"); }; @@ -302,7 +457,7 @@ async fn applies_v2_position_deletes() { &table, stream::iter(equality_rows.into_iter().map(Ok::<_, ArrowError>)), None, - &[1], + &[1, 6], ) .await .unwrap();