feat: always process delete files in plan phase. Opt-out flag only applies to reader

Author: sdd

crates/iceberg/src/scan/context.rs

@@ -45,7 +45,7 @@ pub(crate) struct ManifestFileContext {
     object_cache: Arc<ObjectCache>,
     snapshot_schema: SchemaRef,
     expression_evaluator_cache: Arc<ExpressionEvaluatorCache>,
-    delete_file_index: Option<DeleteFileIndex>,
+    delete_file_index: DeleteFileIndex,
 }
 
 /// Wraps a [`ManifestEntryRef`] alongside the objects that are needed
@@ -58,7 +58,7 @@ pub(crate) struct ManifestEntryContext {
     pub bound_predicates: Option<Arc<BoundPredicates>>,
     pub partition_spec_id: i32,
     pub snapshot_schema: SchemaRef,
-    pub delete_file_index: Option<DeleteFileIndex>,
+    pub delete_file_index: DeleteFileIndex,
 }
 
 impl ManifestFileContext {
@@ -105,16 +105,13 @@ impl ManifestEntryContext {
     /// consume this `ManifestEntryContext`, returning a `FileScanTask`
     /// created from it
     pub(crate) async fn into_file_scan_task(self) -> Result<FileScanTask> {
-        let deletes = if let Some(delete_file_index) = self.delete_file_index {
-            delete_file_index
-                .get_deletes_for_data_file(
-                    self.manifest_entry.data_file(),
-                    self.manifest_entry.sequence_number(),
-                )
-                .await?
-        } else {
-            vec![]
-        };
+        let deletes = self
+            .delete_file_index
+            .get_deletes_for_data_file(
+                self.manifest_entry.data_file(),
+                self.manifest_entry.sequence_number(),
+            )
+            .await?;
 
         Ok(FileScanTask {
             start: 0,
@@ -188,24 +185,19 @@ impl PlanContext {
         &self,
         manifest_list: Arc<ManifestList>,
         tx_data: Sender<ManifestEntryContext>,
-        delete_file_idx_and_tx: Option<(DeleteFileIndex, Sender<ManifestEntryContext>)>,
+        delete_file_idx: DeleteFileIndex,
+        delete_file_tx: Sender<ManifestEntryContext>,
     ) -> Result<Box<impl Iterator<Item = Result<ManifestFileContext>> + 'static>> {
         let manifest_files = manifest_list.entries().iter();
 
         // TODO: Ideally we could ditch this intermediate Vec as we return an iterator.
         let mut filtered_mfcs = vec![];
 
         for manifest_file in manifest_files {
-            let (delete_file_idx, tx) = if manifest_file.content == ManifestContentType::Deletes {
-                let Some((delete_file_idx, tx)) = delete_file_idx_and_tx.as_ref() else {
-                    continue;
-                };
-                (Some(delete_file_idx.clone()), tx.clone())
+            let tx = if manifest_file.content == ManifestContentType::Deletes {
+                delete_file_tx.clone()
             } else {
-                (
-                    delete_file_idx_and_tx.as_ref().map(|x| x.0.clone()),
-                    tx_data.clone(),
-                )
+                tx_data.clone()
             };
 
             let partition_bound_predicate = if self.predicate.is_some() {
@@ -233,7 +225,7 @@ impl PlanContext {
                 manifest_file,
                 partition_bound_predicate,
                 tx,
-                delete_file_idx,
+                delete_file_idx.clone(),
             );
 
             filtered_mfcs.push(Ok(mfc));
@@ -247,7 +239,7 @@ impl PlanContext {
         manifest_file: &ManifestFile,
         partition_filter: Option<Arc<BoundPredicate>>,
         sender: Sender<ManifestEntryContext>,
-        delete_file_index: Option<DeleteFileIndex>,
+        delete_file_index: DeleteFileIndex,
     ) -> ManifestFileContext {
         let bound_predicates =
             if let (Some(ref partition_bound_predicate), Some(snapshot_bound_predicate)) =

crates/iceberg/src/scan/mod.rs

@@ -368,12 +368,7 @@ impl TableScan {
         // used to stream the results back to the caller
         let (file_scan_task_tx, file_scan_task_rx) = channel(concurrency_limit_manifest_entries);
 
-        let delete_file_idx_and_tx: Option<(DeleteFileIndex, Sender<DeleteFileContext>)> =
-            if self.delete_file_processing_enabled {
-                Some(DeleteFileIndex::new())
-            } else {
-                None
-            };
+        let (delete_file_idx, delete_file_tx) = DeleteFileIndex::new();
 
         let manifest_list = plan_context.get_manifest_list().await?;
 
@@ -383,9 +378,8 @@ impl TableScan {
         let manifest_file_contexts = plan_context.build_manifest_file_contexts(
             manifest_list,
             manifest_entry_data_ctx_tx,
-            delete_file_idx_and_tx.as_ref().map(|(delete_file_idx, _)| {
-                (delete_file_idx.clone(), manifest_entry_delete_ctx_tx)
-            }),
+            delete_file_idx.clone(),
+            manifest_entry_delete_ctx_tx.clone(),
         )?;
 
         let mut channel_for_manifest_error = file_scan_task_tx.clone();
@@ -405,33 +399,29 @@ impl TableScan {
 
         let mut channel_for_data_manifest_entry_error = file_scan_task_tx.clone();
 
-        if let Some((_, delete_file_tx)) = delete_file_idx_and_tx {
-            let mut channel_for_delete_manifest_entry_error = file_scan_task_tx.clone();
-
-            // Process the delete file [`ManifestEntry`] stream in parallel
-            spawn(async move {
-                let result = manifest_entry_delete_ctx_rx
-                    .map(|me_ctx| Ok((me_ctx, delete_file_tx.clone())))
-                    .try_for_each_concurrent(
-                        concurrency_limit_manifest_entries,
-                        |(manifest_entry_context, tx)| async move {
-                            spawn(async move {
-                                Self::process_delete_manifest_entry(manifest_entry_context, tx)
-                                    .await
-                            })
-                            .await
-                        },
-                    )
-                    .await;
+        let mut channel_for_delete_manifest_entry_error = file_scan_task_tx.clone();
 
-                if let Err(error) = result {
-                    let _ = channel_for_delete_manifest_entry_error
-                        .send(Err(error))
-                        .await;
-                }
-            })
-            .await;
-        }
+        // Process the delete file [`ManifestEntry`] stream in parallel
+        spawn(async move {
+            let result = manifest_entry_delete_ctx_rx
+                .map(|me_ctx| Ok((me_ctx, delete_file_tx.clone())))
+                .try_for_each_concurrent(
+                    concurrency_limit_manifest_entries,
+                    |(manifest_entry_context, tx)| async move {
+                        spawn(async move {
+                            Self::process_delete_manifest_entry(manifest_entry_context, tx).await
+                        })
+                        .await
+                    },
+                )
+                .await;
+
+            if let Err(error) = result {
+                let _ = channel_for_delete_manifest_entry_error
+                    .send(Err(error))
+                    .await;
+            }
+        });
 
         // Process the data file [`ManifestEntry`] stream in parallel
         spawn(async move {
@@ -461,7 +451,8 @@ impl TableScan {
         let mut arrow_reader_builder = ArrowReaderBuilder::new(self.file_io.clone())
             .with_data_file_concurrency_limit(self.concurrency_limit_data_files)
             .with_row_group_filtering_enabled(self.row_group_filtering_enabled)
-            .with_row_selection_enabled(self.row_selection_enabled);
+            .with_row_selection_enabled(self.row_selection_enabled)
+            .with_delete_file_support_enabled(self.delete_file_processing_enabled);
 
         if let Some(batch_size) = self.batch_size {
             arrow_reader_builder = arrow_reader_builder.with_batch_size(batch_size);

crates/integration_tests/tests/shared_tests/read_positional_deletes.rs

@@ -39,7 +39,7 @@ async fn test_read_table_with_positional_deletes() {
 
     let scan = table
         .scan()
-        .with_delete_file_processing_enabled(true)
+        .with_delete_file_processing_enabled(false)
         .build()
         .unwrap();
     println!("{:?}", scan);
@@ -54,20 +54,14 @@ async fn test_read_table_with_positional_deletes() {
     println!("{:?}", plan);
 
     // Scan plan phase should include delete files in file plan
-    // when with_delete_file_processing_enabled == true
+    // even when with_delete_file_processing_enabled == false
     assert_eq!(plan[0].deletes.len(), 2);
 
-    // 😱 If we don't support positional deletes, we should fail when we try to read a table that
-    // has positional deletes. The table has 12 rows, and 2 are deleted, see provision.py
     let result = scan.to_arrow().await.unwrap().try_collect::<Vec<_>>().await;
 
+    // `delete_file_processing_enabled == false` is propagated to the reader. Since
+    // we have some delete files, this causes it to return an Err.
     assert!(result.is_err_and(|e| e.kind() == FeatureUnsupported));
-
-    // When we get support for it:
-    // let batch_stream = scan.to_arrow().await.unwrap();
-    // let batches: Vec<_> = batch_stream.try_collect().await.is_err();
-    // let num_rows: usize = batches.iter().map(|v| v.num_rows()).sum();
-    // assert_eq!(num_rows, 10);
 }
 
 #[tokio::test]
@@ -99,8 +93,6 @@ async fn test_read_table_with_positional_deletes_with_delete_support_enabled() {
         .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