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

Author: sdd

crates/iceberg/src/arrow/delete_file_manager.rs

@@ -328,11 +328,44 @@ impl DeleteFileManager {
     ///
     /// 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, RoaringTreemap>> {
-        // TODO
+        let mut result: HashMap<String, RoaringTreemap> = 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",
+                    ));
+                };
+
+                result
+                    .entry(file_path.to_string())
+                    .or_default()
+                    .insert(pos as u64);
+            }
+        }
 
-        Ok(HashMap::default())
+        Ok(result)
     }
 
     /// Parses record batch streams from individual equality delete files
@@ -452,27 +485,67 @@ mod tests {
             .load_deletes(&file_scan_tasks[0].deletes, file_io, 5)
             .await
             .unwrap();
+
+        let result = delete_file_manager
+            .get_delete_vector_for_task(&file_scan_tasks[0])
+            .unwrap();
+        assert_eq!(result.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.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();
 
@@ -520,7 +593,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(),
@@ -532,13 +605,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/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);
+}