diff --git a/datafusion_iceberg/src/table/mod.rs b/datafusion_iceberg/src/table/mod.rs index 98677e09..aac472fd 100644 --- a/datafusion_iceberg/src/table/mod.rs +++ b/datafusion_iceberg/src/table/mod.rs @@ -1932,20 +1932,38 @@ fn apply_position_deletes( ), ]; - let delete_sequence_index = delete_plan - .schema() - .index_of(DELETE_FILE_SEQUENCE_NUMBER_COLUMN)?; - let data_sequence_index = data_plan - .schema() - .index_of(DATA_FILE_SEQUENCE_NUMBER_COLUMN)?; + let sequence_filter = + position_delete_sequence_filter(&delete_plan.schema(), &data_plan.schema())?; + + Ok(Arc::new(HashJoinExec::try_new( + delete_plan, + data_plan, + join_on, + Some(sequence_filter), + &JoinType::RightAnti, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + false, + )?)) +} + +fn position_delete_sequence_filter( + delete_schema: &SchemaRef, + data_schema: &SchemaRef, +) -> 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)?; let filter_schema = Arc::new(ArrowSchema::new(vec![ Field::new("delete_sequence_number", DataType::Int64, false), Field::new("data_sequence_number", DataType::Int64, false), ])); - let sequence_filter = JoinFilter::new( + Ok(JoinFilter::new( Arc::new(BinaryExpr::new( Arc::new(Column::new("delete_sequence_number", 0)), - Operator::Gt, + // Iceberg v2 position deletes apply when their sequence number is + // greater than or equal to the data file sequence number. + Operator::GtEq, Arc::new(Column::new("data_sequence_number", 1)), )), vec![ @@ -1959,19 +1977,7 @@ fn apply_position_deletes( }, ], filter_schema, - ); - - Ok(Arc::new(HashJoinExec::try_new( - delete_plan, - data_plan, - join_on, - Some(sequence_filter), - &JoinType::RightAnti, - None, - PartitionMode::CollectLeft, - NullEquality::NullEqualsNothing, - false, - )?)) + )) } fn value_to_scalarvalue(value: &Value) -> Result { @@ -2335,7 +2341,10 @@ struct PartitionDeleteFileIndex { mod tests { use datafusion::{ - arrow::array::{Int64Array, StringArray}, + arrow::{ + array::{BooleanArray, Int64Array, StringArray}, + record_batch::RecordBatch, + }, execution::object_store::ObjectStoreUrl, prelude::SessionContext, scalar::ScalarValue, @@ -2366,7 +2375,10 @@ mod tests { use crate::{catalog::catalog::IcebergCatalog, table::fake_object_store_url, DataFusionTable}; - use super::{DataSourceExec, ExecutionPlan, ParquetSource, TableProvider}; + use super::{ + position_delete_sequence_filter, DataSourceExec, ExecutionPlan, ParquetSource, + TableProvider, + }; use datafusion::datasource::physical_plan::FileScanConfig; /// Walk `plan` and assert every parquet scan node carries the @@ -2397,6 +2409,46 @@ mod tests { found } + #[test] + fn position_deletes_include_equal_data_sequence_numbers() { + let delete_schema = Arc::new(datafusion::arrow::datatypes::Schema::new(vec![ + datafusion::arrow::datatypes::Field::new( + super::DELETE_FILE_SEQUENCE_NUMBER_COLUMN, + datafusion::arrow::datatypes::DataType::Int64, + false, + ), + ])); + let data_schema = Arc::new(datafusion::arrow::datatypes::Schema::new(vec![ + datafusion::arrow::datatypes::Field::new( + super::DATA_FILE_SEQUENCE_NUMBER_COLUMN, + datafusion::arrow::datatypes::DataType::Int64, + false, + ), + ])); + let filter = position_delete_sequence_filter(&delete_schema, &data_schema).unwrap(); + let batch = RecordBatch::try_new( + filter.schema().clone(), + vec![ + Arc::new(Int64Array::from(vec![4, 5, 6])), + Arc::new(Int64Array::from(vec![5, 5, 5])), + ], + ) + .unwrap(); + + let result = filter + .expression() + .evaluate(&batch) + .unwrap() + .into_array(batch.num_rows()) + .unwrap(); + let result = result.as_any().downcast_ref::().unwrap(); + + assert_eq!( + result.values().iter().collect::>(), + [false, true, true] + ); + } + #[tokio::test] async fn test_scan_installs_caching_parquet_reader_factory() { let object_store = ObjectStoreBuilder::memory();