feat: return Err for unimplemented delete vec parse methods and make DeleteVector::intersect_assign pub(crate)

Author: sdd

crates/iceberg/src/arrow/delete_file_manager.rs

@@ -359,7 +359,10 @@ impl CachingDeleteFileManager {
     ) -> Result<HashMap<String, DeleteVector>> {
         // TODO
 
-        Ok(HashMap::default())
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "parsing of positional deletes is not yet supported",
+        ))
     }
 
     /// Parses record batch streams from individual equality delete files
@@ -370,7 +373,10 @@ impl CachingDeleteFileManager {
     ) -> Result<Predicate> {
         // TODO
 
-        Ok(AlwaysTrue)
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "parsing of equality deletes is not yet supported",
+        ))
     }
 
     /// Builds eq delete predicate for the provided task.
@@ -471,15 +477,16 @@ mod tests {
             .unwrap();
 
         // Note that with the delete file parsing not yet in place, all we can test here is that
-        // the call to the loader does not fail.
+        // 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);
 
-        delete_file_manager
+        let result = delete_file_manager
             .load_deletes(&file_scan_tasks[0].deletes)
-            .await
-            .unwrap();
+            .await;
+
+        assert!(result.is_err_and(|e| e.kind() == ErrorKind::FeatureUnsupported));
     }
 
     fn setup(table_location: &Path) -> Vec<FileScanTask> {

crates/iceberg/src/arrow/reader.rs

@@ -307,7 +307,7 @@ impl ArrowReader {
         if let Some(positional_delete_indexes) = positional_delete_indexes {
             let delete_row_selection = {
                 let positional_delete_indexes = positional_delete_indexes.read().unwrap();
-                
+
                 Self::build_deletes_row_selection(
                     record_batch_stream_builder.metadata().row_groups(),
                     &selected_row_group_indices,

crates/iceberg/src/delete_vector.rs

@@ -39,7 +39,7 @@ impl DeleteVector {
         DeleteVectorIterator { outer, inner: None }
     }
 
-    pub fn intersect_assign(&mut self, other: &DeleteVector) {
+    pub(crate) fn intersect_assign(&mut self, other: &DeleteVector) {
         self.inner.bitor_assign(&other.inner);
     }
 }