feat: remove flag to selectively enable delete file processing in vavour of it being always on

Author: sdd

crates/iceberg/src/arrow/delete_filter.rs

@@ -73,21 +73,21 @@ struct DeleteFileFilterState {
 }
 
 #[derive(Clone, Debug, Default)]
-pub struct DeleteFilter {
+pub(crate) struct DeleteFilter {
     state: Arc<RwLock<DeleteFileFilterState>>,
 }
 
 impl DeleteFilter {
     /// Retrieve a delete vector for the data file associated with a given file scan task
-    pub fn get_delete_vector(
+    pub(crate) fn get_delete_vector(
         &self,
         file_scan_task: &FileScanTask,
     ) -> Option<Arc<Mutex<DeleteVector>>> {
         self.get_delete_vector_for_path(file_scan_task.data_file_path())
     }
 
     /// Retrieve a delete vector for a data file
-    pub fn get_delete_vector_for_path(
+    pub(crate) fn get_delete_vector_for_path(
         &self,
         delete_file_path: &str,
     ) -> Option<Arc<Mutex<DeleteVector>>> {
@@ -111,7 +111,7 @@ impl DeleteFilter {
     }
 
     /// Builds eq delete predicate for the provided task.
-    pub async fn build_equality_delete_predicate(
+    pub(crate) async fn build_equality_delete_predicate(
         &self,
         file_scan_task: &FileScanTask,
     ) -> Result<Option<BoundPredicate>> {

crates/iceberg/src/arrow/reader.rs

@@ -64,7 +64,6 @@ pub struct ArrowReaderBuilder {
     concurrency_limit_data_files: usize,
     row_group_filtering_enabled: bool,
     row_selection_enabled: bool,
-    delete_file_support_enabled: bool,
 }
 
 impl ArrowReaderBuilder {
@@ -78,7 +77,6 @@ impl ArrowReaderBuilder {
             concurrency_limit_data_files: num_cpus,
             row_group_filtering_enabled: true,
             row_selection_enabled: false,
-            delete_file_support_enabled: false,
         }
     }
 
@@ -107,12 +105,6 @@ impl ArrowReaderBuilder {
         self
     }
 
-    /// Determines whether to enable delete file support.
-    pub fn with_delete_file_support_enabled(mut self, delete_file_support_enabled: bool) -> Self {
-        self.delete_file_support_enabled = delete_file_support_enabled;
-        self
-    }
-
     /// Build the ArrowReader.
     pub fn build(self) -> ArrowReader {
         ArrowReader {
@@ -125,7 +117,6 @@ impl ArrowReaderBuilder {
             concurrency_limit_data_files: self.concurrency_limit_data_files,
             row_group_filtering_enabled: self.row_group_filtering_enabled,
             row_selection_enabled: self.row_selection_enabled,
-            delete_file_support_enabled: self.delete_file_support_enabled,
         }
     }
 }
@@ -142,7 +133,6 @@ pub struct ArrowReader {
 
     row_group_filtering_enabled: bool,
     row_selection_enabled: bool,
-    delete_file_support_enabled: bool,
 }
 
 impl ArrowReader {
@@ -154,7 +144,6 @@ impl ArrowReader {
         let concurrency_limit_data_files = self.concurrency_limit_data_files;
         let row_group_filtering_enabled = self.row_group_filtering_enabled;
         let row_selection_enabled = self.row_selection_enabled;
-        let delete_file_support_enabled = self.delete_file_support_enabled;
 
         let stream = tasks
             .map_ok(move |task| {
@@ -167,7 +156,6 @@ impl ArrowReader {
                     self.delete_file_loader.clone(),
                     row_group_filtering_enabled,
                     row_selection_enabled,
-                    delete_file_support_enabled,
                 )
             })
             .map_err(|err| {
@@ -187,26 +175,12 @@ impl ArrowReader {
         delete_file_loader: CachingDeleteFileLoader,
         row_group_filtering_enabled: bool,
         row_selection_enabled: bool,
-        delete_file_support_enabled: bool,
     ) -> Result<ArrowRecordBatchStream> {
-        if !delete_file_support_enabled && !task.deletes.is_empty() {
-            return Err(Error::new(
-                ErrorKind::FeatureUnsupported,
-                "Delete file support is not enabled",
-            ));
-        }
-
         let should_load_page_index =
             (row_selection_enabled && task.predicate.is_some()) || !task.deletes.is_empty();
 
-        let delete_filter_rx = delete_file_loader.load_deletes(
-            if delete_file_support_enabled {
-                &task.deletes
-            } else {
-                &[]
-            },
-            task.schema.clone(),
-        );
+        let delete_filter_rx = delete_file_loader.load_deletes(&task.deletes, task.schema.clone());
+
         let mut record_batch_stream_builder = Self::create_parquet_record_batch_stream_builder(
             &task.data_file_path,
             file_io.clone(),

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

@@ -59,11 +59,6 @@ pub struct TableScanBuilder<'a> {
     concurrency_limit_manifest_files: usize,
     row_group_filtering_enabled: bool,
     row_selection_enabled: bool,
-
-    // TODO: defaults to false for now whilst delete file processing
-    // is still being worked on but will switch to a default of true
-    // once this work is complete
-    delete_file_processing_enabled: bool,
 }
 
 impl<'a> TableScanBuilder<'a> {
@@ -82,7 +77,6 @@ impl<'a> TableScanBuilder<'a> {
             concurrency_limit_manifest_files: num_cpus,
             row_group_filtering_enabled: true,
             row_selection_enabled: false,
-            delete_file_processing_enabled: false,
         }
     }
 
@@ -189,17 +183,6 @@ impl<'a> TableScanBuilder<'a> {
         self
     }
 
-    /// Determines whether to enable delete file processing (currently disabled by default)
-    ///
-    /// When disabled, delete files are ignored.
-    pub fn with_delete_file_processing_enabled(
-        mut self,
-        delete_file_processing_enabled: bool,
-    ) -> Self {
-        self.delete_file_processing_enabled = delete_file_processing_enabled;
-        self
-    }
-
     /// Build the table scan.
     pub fn build(self) -> Result<TableScan> {
         let snapshot = match self.snapshot_id {
@@ -226,7 +209,6 @@ impl<'a> TableScanBuilder<'a> {
                         concurrency_limit_manifest_files: self.concurrency_limit_manifest_files,
                         row_group_filtering_enabled: self.row_group_filtering_enabled,
                         row_selection_enabled: self.row_selection_enabled,
-                        delete_file_processing_enabled: self.delete_file_processing_enabled,
                     });
                 };
                 current_snapshot_id.clone()
@@ -317,7 +299,6 @@ impl<'a> TableScanBuilder<'a> {
             concurrency_limit_manifest_files: self.concurrency_limit_manifest_files,
             row_group_filtering_enabled: self.row_group_filtering_enabled,
             row_selection_enabled: self.row_selection_enabled,
-            delete_file_processing_enabled: self.delete_file_processing_enabled,
         })
     }
 }
@@ -346,7 +327,6 @@ pub struct TableScan {
 
     row_group_filtering_enabled: bool,
     row_selection_enabled: bool,
-    delete_file_processing_enabled: bool,
 }
 
 impl TableScan {
@@ -368,12 +348,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 +358,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,
         )?;
 
         let mut channel_for_manifest_error = file_scan_task_tx.clone();
@@ -404,34 +378,30 @@ impl TableScan {
         });
 
         let mut channel_for_data_manifest_entry_error = file_scan_task_tx.clone();
+        let mut channel_for_delete_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;
+        // 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;
-                }
-            })
-            .await;
-        }
+            if let Err(error) = result {
+                let _ = channel_for_delete_manifest_entry_error
+                    .send(Err(error))
+                    .await;
+            }
+        })
+        .await;
 
         // Process the data file [`ManifestEntry`] stream in parallel
         spawn(async move {

crates/integration_tests/tests/shared_tests/read_positional_deletes.rs

@@ -37,11 +37,7 @@ async fn test_read_table_with_positional_deletes() {
         .await
         .unwrap();
 
-    let scan = table
-        .scan()
-        .with_delete_file_processing_enabled(true)
-        .build()
-        .unwrap();
+    let scan = table.scan().build().unwrap();
     println!("{:?}", scan);
 
     let plan: Vec<_> = scan