feat: add positional delete parsing. Add tests for end-to-end positional delete functionality

sdd · sdd · commit 7b755fdc7b43 · 2025-05-07T19:13:22.000+01:00
diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs
@@ -18,6 +18,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use arrow_array::{Int64Array, StringArray};
 use futures::channel::oneshot;
 use futures::future::join_all;
 use futures::{StreamExt, TryStreamExt};
@@ -347,14 +348,44 @@ impl CachingDeleteFileLoader {
     ///
     /// Returns a map of data file path to a delete vector
     async fn parse_positional_deletes_record_batch_stream(
-        stream: ArrowRecordBatchStream,
+        mut stream: ArrowRecordBatchStream,
     ) -> Result<HashMap<String, DeleteVector>> {
-        // TODO
+        let mut result: HashMap<String, DeleteVector> = HashMap::default();
+
+        while let Some(batch) = stream.next().await {
+            let batch = batch?;
+            let schema = batch.schema();
+            let columns = batch.columns();
+
+            let Some(file_paths) = columns[0].as_any().downcast_ref::<StringArray>() else {
+                return Err(Error::new(
+                    ErrorKind::DataInvalid,
+                    "Could not downcast file paths array to StringArray",
+                ));
+            };
+            let Some(positions) = columns[1].as_any().downcast_ref::<Int64Array>() else {
+                return Err(Error::new(
+                    ErrorKind::DataInvalid,
+                    "Could not downcast positions array to Int64Array",
+                ));
+            };
+
+            for (file_path, pos) in file_paths.iter().zip(positions.iter()) {
+                let (Some(file_path), Some(pos)) = (file_path, pos) else {
+                    return Err(Error::new(
+                        ErrorKind::DataInvalid,
+                        "null values in delete file",
+                    ));
+                };
 
-        Err(Error::new(
-            ErrorKind::FeatureUnsupported,
-            "parsing of positional deletes is not yet supported",
-        ))
+                result
+                    .entry(file_path.to_string())
+                    .or_default()
+                    .insert(pos as u64);
+            }
+        }
+
+        Ok(result)
     }
 
     /// Parses record batch streams from individual equality delete files
@@ -395,7 +426,7 @@ mod tests {
     const FIELD_ID_POSITIONAL_DELETE_POS: u64 = 2147483545;
 
     #[tokio::test]
-    async fn test_delete_file_manager_load_deletes() {
+    async fn test_delete_file_loader_load_deletes() {
         let tmp_dir = TempDir::new().unwrap();
         let table_location = tmp_dir.path();
         let file_io = FileIO::from_path(table_location.as_os_str().to_str().unwrap())
@@ -405,37 +436,76 @@ mod tests {
 
         // Note that with the delete file parsing not yet in place, all we can test here is that
         // the call to the loader fails with the expected FeatureUnsupportedError.
-        let delete_file_manager = CachingDeleteFileLoader::new(file_io.clone(), 10);
+        let delete_file_loader = CachingDeleteFileLoader::new(file_io.clone(), 10);
 
         let file_scan_tasks = setup(table_location);
 
-        let result = delete_file_manager
+        let delete_filter = delete_file_loader
             .load_deletes(&file_scan_tasks[0].deletes, file_scan_tasks[0].schema_ref())
             .await
+            .unwrap()
+            .unwrap();
+
+        let result = delete_filter
+            .get_delete_vector(&file_scan_tasks[0])
             .unwrap();
+        assert_eq!(result.lock().unwrap().len(), 3); // pos dels from pos del file 1 and 2
 
-        assert!(result.is_err_and(|e| e.kind() == ErrorKind::FeatureUnsupported));
+        let result = delete_filter
+            .get_delete_vector(&file_scan_tasks[1])
+            .unwrap();
+        assert_eq!(result.lock().unwrap().len(), 3); // pos dels from pos del file 3
     }
 
     fn setup(table_location: &Path) -> Vec<FileScanTask> {
         let data_file_schema = Arc::new(Schema::builder().build().unwrap());
         let positional_delete_schema = create_pos_del_schema();
 
-        let file_path_values = vec![format!("{}/1.parquet", table_location.to_str().unwrap()); 8];
-        let pos_values = vec![0, 1, 3, 5, 6, 8, 1022, 1023];
-
-        let file_path_col = Arc::new(StringArray::from_iter_values(file_path_values));
-        let pos_col = Arc::new(Int64Array::from_iter_values(pos_values));
+        let mut file_path_values = vec![];
+        let mut pos_values = vec![];
+
+        file_path_values.push(vec![
+            format!(
+                "{}/1.parquet",
+                table_location.to_str().unwrap()
+            );
+            3
+        ]);
+        pos_values.push(vec![0, 1, 3]);
+
+        file_path_values.push(vec![
+            format!(
+                "{}/1.parquet",
+                table_location.to_str().unwrap()
+            );
+            3
+        ]);
+        pos_values.push(vec![5, 6, 8]);
+
+        file_path_values.push(vec![
+            format!(
+                "{}/2.parquet",
+                table_location.to_str().unwrap()
+            );
+            3
+        ]);
+        pos_values.push(vec![1022, 1023, 1024]);
+        // 9 rows in total pos deleted across 3 files
 
         let props = WriterProperties::builder()
             .set_compression(Compression::SNAPPY)
             .build();
 
         for n in 1..=3 {
+            let file_path_col = Arc::new(StringArray::from_iter_values(
+                file_path_values.pop().unwrap(),
+            ));
+            let pos_col = Arc::new(Int64Array::from_iter_values(pos_values.pop().unwrap()));
+
             let positional_deletes_to_write =
                 RecordBatch::try_new(positional_delete_schema.clone(), vec![
                     file_path_col.clone(),
-                    pos_col.clone(),
+                    pos_col,
                 ])
                 .unwrap();
 
@@ -486,7 +556,7 @@ mod tests {
                 start: 0,
                 length: 0,
                 record_count: None,
-                data_file_path: "".to_string(),
+                data_file_path: format!("{}/1.parquet", table_location.to_str().unwrap()),
                 data_file_content: DataContentType::Data,
                 data_file_format: DataFileFormat::Parquet,
                 schema: data_file_schema.clone(),
@@ -498,13 +568,13 @@ mod tests {
                 start: 0,
                 length: 0,
                 record_count: None,
-                data_file_path: "".to_string(),
+                data_file_path: format!("{}/2.parquet", table_location.to_str().unwrap()),
                 data_file_content: DataContentType::Data,
                 data_file_format: DataFileFormat::Parquet,
                 schema: data_file_schema.clone(),
                 project_field_ids: vec![],
                 predicate: None,
-                deletes: vec![pos_del_2, pos_del_3],
+                deletes: vec![pos_del_3],
             },
         ];
 
diff --git a/crates/iceberg/src/arrow/delete_filter.rs b/crates/iceberg/src/arrow/delete_filter.rs
@@ -202,43 +202,67 @@ mod tests {
     const FIELD_ID_POSITIONAL_DELETE_POS: u64 = 2147483545;
 
     #[tokio::test]
-    async fn test_delete_file_manager_load_deletes() {
+    async fn test_delete_file_loader_load_deletes() {
         let tmp_dir = TempDir::new().unwrap();
         let table_location = tmp_dir.path();
         let file_io = FileIO::from_path(table_location.as_os_str().to_str().unwrap())
             .unwrap()
             .build()
             .unwrap();
 
-        // Note that with the delete file parsing not yet in place, all we can test here is that
-        // the call to the loader fails with the expected FeatureUnsupportedError.
-        let delete_file_manager = CachingDeleteFileLoader::new(file_io.clone(), 10);
+        let delete_file_loader = CachingDeleteFileLoader::new(file_io.clone(), 10);
 
         let file_scan_tasks = setup(table_location);
 
-        let result = delete_file_manager
+        let delete_filter = delete_file_loader
             .load_deletes(&file_scan_tasks[0].deletes, file_scan_tasks[0].schema_ref())
             .await
+            .unwrap()
+            .unwrap();
+
+        let _ = delete_file_loader
+            .load_deletes(&file_scan_tasks[1].deletes, file_scan_tasks[1].schema_ref())
+            .await
+            .unwrap()
+            .unwrap();
+
+        let result = delete_filter
+            .get_delete_vector(&file_scan_tasks[0])
             .unwrap();
+        assert_eq!(result.lock().unwrap().len(), 12); // pos dels from pos del file 1 and 2
 
-        assert!(result.is_err_and(|e| e.kind() == ErrorKind::FeatureUnsupported));
+        let result = delete_filter
+            .get_delete_vector(&file_scan_tasks[1])
+            .unwrap();
+        assert_eq!(result.lock().unwrap().len(), 8); // pos dels from pos del file 3
     }
 
     fn setup(table_location: &Path) -> Vec<FileScanTask> {
         let data_file_schema = Arc::new(Schema::builder().build().unwrap());
         let positional_delete_schema = create_pos_del_schema();
 
-        let file_path_values = vec![format!("{}/1.parquet", table_location.to_str().unwrap()); 8];
-        let pos_values = vec![0, 1, 3, 5, 6, 8, 1022, 1023];
-
-        let file_path_col = Arc::new(StringArray::from_iter_values(file_path_values));
-        let pos_col = Arc::new(Int64Array::from_iter_values(pos_values));
+        let file_path_values = [
+            vec![format!("{}/1.parquet", table_location.to_str().unwrap()); 8],
+            vec![format!("{}/1.parquet", table_location.to_str().unwrap()); 8],
+            vec![format!("{}/2.parquet", table_location.to_str().unwrap()); 8],
+        ];
+        let pos_values = [
+            vec![0i64, 1, 3, 5, 6, 8, 1022, 1023],
+            vec![0i64, 1, 3, 5, 20, 21, 22, 23],
+            vec![0i64, 1, 3, 5, 6, 8, 1022, 1023],
+        ];
 
         let props = WriterProperties::builder()
             .set_compression(Compression::SNAPPY)
             .build();
 
         for n in 1..=3 {
+            let file_path_vals = file_path_values.get(n - 1).unwrap();
+            let file_path_col = Arc::new(StringArray::from_iter_values(file_path_vals));
+
+            let pos_vals = pos_values.get(n - 1).unwrap();
+            let pos_col = Arc::new(Int64Array::from_iter_values(pos_vals.clone()));
+
             let positional_deletes_to_write =
                 RecordBatch::try_new(positional_delete_schema.clone(), vec![
                     file_path_col.clone(),
@@ -293,7 +317,7 @@ mod tests {
                 start: 0,
                 length: 0,
                 record_count: None,
-                data_file_path: "".to_string(),
+                data_file_path: format!("{}/1.parquet", table_location.to_str().unwrap()),
                 data_file_content: DataContentType::Data,
                 data_file_format: DataFileFormat::Parquet,
                 schema: data_file_schema.clone(),
@@ -305,13 +329,13 @@ mod tests {
                 start: 0,
                 length: 0,
                 record_count: None,
-                data_file_path: "".to_string(),
+                data_file_path: format!("{}/2.parquet", table_location.to_str().unwrap()),
                 data_file_content: DataContentType::Data,
                 data_file_format: DataFileFormat::Parquet,
                 schema: data_file_schema.clone(),
                 project_field_ids: vec![],
                 predicate: None,
-                deletes: vec![pos_del_2, pos_del_3],
+                deletes: vec![pos_del_3],
             },
         ];
 
diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs
@@ -38,6 +38,15 @@ impl DeleteVector {
         let outer = self.inner.bitmaps();
         DeleteVectorIterator { outer, inner: None }
     }
+
+    pub fn insert(&mut self, pos: u64) -> bool {
+        self.inner.insert(pos)
+    }
+
+    #[allow(unused)]
+    pub fn len(&self) -> u64 {
+        self.inner.len()
+    }
 }
 
 // Ideally, we'd just wrap `roaring::RoaringTreemap`'s iterator, `roaring::treemap::Iter` here.
diff --git a/crates/integration_tests/tests/shared_tests/read_positional_deletes.rs b/crates/integration_tests/tests/shared_tests/read_positional_deletes.rs
@@ -69,3 +69,43 @@ async fn test_read_table_with_positional_deletes() {
     // let num_rows: usize = batches.iter().map(|v| v.num_rows()).sum();
     // assert_eq!(num_rows, 10);
 }
+
+#[tokio::test]
+async fn test_read_table_with_positional_deletes_with_delete_support_enabled() {
+    let fixture = get_shared_containers();
+    let rest_catalog = RestCatalog::new(fixture.catalog_config.clone());
+
+    let table = rest_catalog
+        .load_table(
+            &TableIdent::from_strs(["default", "test_positional_merge_on_read_double_deletes"])
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    let scan = table
+        .scan()
+        .with_delete_file_processing_enabled(true)
+        .build()
+        .unwrap();
+    println!("{:?}", scan);
+
+    let plan: Vec<_> = scan
+        .plan_files()
+        .await
+        .unwrap()
+        .try_collect()
+        .await
+        .unwrap();
+    println!("{:?}", plan);
+
+    // Scan plan phase should include delete files in file plan
+    // when with_delete_file_processing_enabled == true
+    assert_eq!(plan[0].deletes.len(), 2);
+
+    // we should see two rows deleted, returning 10 rows instead of 12
+    let batch_stream = scan.to_arrow().await.unwrap();
+    let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
+    let num_rows: usize = batches.iter().map(|v| v.num_rows()).sum();
+    assert_eq!(num_rows, 10);
+}
