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

Author: sdd

crates/iceberg/src/arrow/delete_file_manager.rs

@@ -21,6 +21,7 @@ use std::pin::Pin;
 use std::sync::{Arc, OnceLock, RwLock};
 use std::task::{Context, Poll};
 
+use arrow_array::{Int64Array, StringArray};
 use futures::channel::oneshot;
 use futures::future::join_all;
 use futures::{StreamExt, TryStreamExt};
@@ -169,7 +170,7 @@ impl CachingDeleteFileManager {
     ///      vector maps that resulted from any positional delete or delete vector files into a
     ///      single map and persist it in the state.
     ///
-    ///  
+    ///
     ///  Conceptually, the data flow is like this:
     /// ```none
     ///                                          FileScanTaskDeleteFile
@@ -393,14 +394,44 @@ impl CachingDeleteFileManager {
     ///
     /// 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",
+                ));
+            };
 
-        Err(Error::new(
-            ErrorKind::FeatureUnsupported,
-            "parsing of positional deletes is not yet supported",
-        ))
+            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",
+                    ));
+                };
+
+                result
+                    .entry(file_path.to_string())
+                    .or_default()
+                    .insert(pos as u64);
+            }
+        }
+
+        Ok(result)
     }
 
     /// Parses record batch streams from individual equality delete files
@@ -514,38 +545,74 @@ mod tests {
             .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 = CachingDeleteFileManager::new(file_io.clone(), 10);
 
         let file_scan_tasks = setup(table_location);
 
-        let result = delete_file_manager
+        let _ = delete_file_manager
             .load_deletes(&file_scan_tasks[0].deletes, file_scan_tasks[0].schema_ref())
             .await;
 
-        assert!(result.is_err_and(|e| e.kind() == ErrorKind::FeatureUnsupported));
+        let result = delete_file_manager
+            .get_delete_vector_for_task(&file_scan_tasks[0])
+            .unwrap();
+        assert_eq!(result.read().unwrap().len(), 3); // pos dels from pos del file 1 and 2
+
+        let result = delete_file_manager
+            .get_delete_vector_for_task(&file_scan_tasks[1])
+            .unwrap();
+        assert_eq!(result.read().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();
 
@@ -596,7 +663,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(),
@@ -608,13 +675,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],
             },
         ];
 

crates/iceberg/src/delete_vector.rs

@@ -42,6 +42,15 @@ impl DeleteVector {
     pub(crate) fn intersect_assign(&mut self, other: &DeleteVector) {
         self.inner.bitor_assign(&other.inner);
     }
+
+    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.

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);
+}