refactor: TableScan's plan_files method now operates on manifest files and manifest data files concurrently rather than sequentially

Author: sdd

crates/iceberg/src/arrow/reader.rs

@@ -52,7 +52,7 @@ pub struct ArrowReaderBuilder {
     field_ids: Vec<usize>,
     file_io: FileIO,
     schema: SchemaRef,
-    predicates: Option<BoundPredicate>,
+    predicate: Option<BoundPredicate>,
 }
 
 impl ArrowReaderBuilder {
@@ -63,7 +63,7 @@ impl ArrowReaderBuilder {
             field_ids: vec![],
             file_io,
             schema,
-            predicates: None,
+            predicate: None,
         }
     }
 
@@ -82,7 +82,7 @@ impl ArrowReaderBuilder {
 
     /// Sets the predicates to apply to the scan.
     pub fn with_predicates(mut self, predicates: BoundPredicate) -> Self {
-        self.predicates = Some(predicates);
+        self.predicate = Some(predicates);
         self
     }
 
@@ -93,7 +93,7 @@ impl ArrowReaderBuilder {
             field_ids: self.field_ids,
             schema: self.schema,
             file_io: self.file_io,
-            predicates: self.predicates,
+            predicate: self.predicate,
         }
     }
 }
@@ -105,7 +105,7 @@ pub struct ArrowReader {
     #[allow(dead_code)]
     schema: SchemaRef,
     file_io: FileIO,
-    predicates: Option<BoundPredicate>,
+    predicate: Option<BoundPredicate>,
 }
 
 impl ArrowReader {
@@ -118,7 +118,7 @@ impl ArrowReader {
         let mut collector = CollectFieldIdVisitor {
             field_ids: HashSet::default(),
         };
-        if let Some(predicates) = &self.predicates {
+        if let Some(predicates) = &self.predicate {
             visit(&mut collector, predicates)?;
         }
 
@@ -162,7 +162,7 @@ impl ArrowReader {
         &self,
         parquet_schema: &SchemaDescriptor,
         arrow_schema: &ArrowSchemaRef,
-    ) -> crate::Result<ProjectionMask> {
+    ) -> Result<ProjectionMask> {
         if self.field_ids.is_empty() {
             Ok(ProjectionMask::all())
         } else {
@@ -232,7 +232,7 @@ impl ArrowReader {
         parquet_schema: &SchemaDescriptor,
         collector: &CollectFieldIdVisitor,
     ) -> Result<Option<RowFilter>> {
-        if let Some(predicates) = &self.predicates {
+        if let Some(predicate) = &self.predicate {
             let field_id_map = build_field_id_map(parquet_schema)?;
 
             // Collect Parquet column indices from field ids.
@@ -255,7 +255,7 @@ impl ArrowReader {
             // After collecting required leaf column indices used in the predicate,
             // creates the projection mask for the Arrow predicates.
             let projection_mask = ProjectionMask::leaves(parquet_schema, column_indices.clone());
-            let predicate_func = visit(&mut converter, predicates)?;
+            let predicate_func = visit(&mut converter, predicate)?;
             let arrow_predicate = ArrowPredicateFn::new(projection_mask, predicate_func);
             Ok(Some(RowFilter::new(vec![Box::new(arrow_predicate)])))
         } else {

crates/iceberg/src/error.rs

@@ -333,6 +333,12 @@ define_from_err!(
 
 define_from_err!(std::io::Error, ErrorKind::Unexpected, "IO Operation failed");
 
+define_from_err!(
+    futures::channel::mpsc::SendError,
+    ErrorKind::Unexpected,
+    "Failed to send a message to a channel"
+);
+
 /// Helper macro to check arguments.
 ///
 ///