feat(wip): add ManifestEvalVisitor and InclusiveProjection

Author: sdd

crates/iceberg/src/expr/predicate.rs

@@ -46,7 +46,7 @@ impl<T: Debug, const N: usize> Debug for LogicalExpression<T, N> {
 }
 
 impl<T, const N: usize> LogicalExpression<T, N> {
-    fn new(inputs: [Box<T>; N]) -> Self {
+    pub(crate) fn new(inputs: [Box<T>; N]) -> Self {
         Self { inputs }
     }
 
@@ -82,9 +82,9 @@ where
 #[derive(PartialEq)]
 pub struct UnaryExpression<T> {
     /// Operator of this predicate, must be single operand operator.
-    op: PredicateOperator,
+    pub(crate) op: PredicateOperator,
     /// Term of this predicate, for example, `a` in `a IS NULL`.
-    term: T,
+    pub(crate) term: T,
 }
 
 impl<T: Debug> Debug for UnaryExpression<T> {
@@ -116,13 +116,17 @@ impl<T> UnaryExpression<T> {
         debug_assert!(op.is_unary());
         Self { op, term }
     }
+
+    pub(crate) fn field_id(&self) -> i32 {
+        self.term.field().id
+    }
 }
 
 /// Binary predicate, for example, `a > 10`.
 #[derive(PartialEq)]
 pub struct BinaryExpression<T> {
     /// Operator of this predicate, must be binary operator, such as `=`, `>`, `<`, etc.
-    op: PredicateOperator,
+    pub(crate) op: PredicateOperator,
     /// Term of this predicate, for example, `a` in `a > 10`.
     term: T,
     /// Literal of this predicate, for example, `10` in `a > 10`.
@@ -144,6 +148,10 @@ impl<T> BinaryExpression<T> {
         debug_assert!(op.is_binary());
         Self { op, term, literal }
     }
+
+    pub(crate) fn field_id(&self) -> i32 {
+        self.term.field().id
+    }
 }
 
 impl<T: Display> Display for BinaryExpression<T> {
@@ -165,7 +173,7 @@ impl<T: Bind> Bind for BinaryExpression<T> {
 #[derive(PartialEq)]
 pub struct SetExpression<T> {
     /// Operator of this predicate, must be set operator, such as `IN`, `NOT IN`, etc.
-    op: PredicateOperator,
+    pub(crate) op: PredicateOperator,
     /// Term of this predicate, for example, `a` in `a in (1, 2, 3)`.
     term: T,
     /// Literals of this predicate, for example, `(1, 2, 3)` in `a in (1, 2, 3)`.
@@ -187,6 +195,10 @@ impl<T> SetExpression<T> {
         debug_assert!(op.is_set());
         Self { op, term, literals }
     }
+
+    pub(crate) fn field_id(&self) -> i32 {
+        self.term.field().id
+    }
 }
 
 impl<T: Bind> Bind for SetExpression<T> {
@@ -209,6 +221,9 @@ impl<T: Display + Debug> Display for SetExpression<T> {
 /// Unbound predicate expression before binding to a schema.
 #[derive(Debug, PartialEq)]
 pub enum Predicate {
+    AlwaysTrue,
+    AlwaysFalse,
+
     /// And predicate, for example, `a > 10 AND b < 20`.
     And(LogicalExpression<Predicate, 2>),
     /// Or predicate, for example, `a > 10 OR b < 20`.
@@ -366,6 +381,12 @@ impl Bind for Predicate {
 impl Display for Predicate {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match self {
+            Predicate::AlwaysTrue => {
+                write!(f, "TRUE")
+            }
+            Predicate::AlwaysFalse => {
+                write!(f, "FALSE")
+            }
             Predicate::And(expr) => {
                 write!(f, "({}) AND ({})", expr.inputs()[0], expr.inputs()[1])
             }
@@ -453,6 +474,8 @@ impl Predicate {
     /// ```
     pub fn negate(self) -> Predicate {
         match self {
+            Predicate::AlwaysTrue => Predicate::AlwaysFalse,
+            Predicate::AlwaysFalse => Predicate::AlwaysTrue,
             Predicate::And(expr) => Predicate::Or(LogicalExpression::new(
                 expr.inputs.map(|expr| Box::new(expr.negate())),
             )),
@@ -599,6 +622,14 @@ impl Display for BoundPredicate {
     }
 }
 
+pub(crate) trait PredicateVisitor<T> {
+    fn visit(predicate: Predicate) -> T;
+}
+
+pub(crate) trait BoundPredicateVisitor<T> {
+    fn visit(predicate: BoundPredicate) -> T;
+}
+
 #[cfg(test)]
 mod tests {
     use std::ops::Not;

crates/iceberg/src/scan.rs

@@ -19,22 +19,21 @@
 
 use crate::arrow::ArrowReaderBuilder;
 use crate::expr::BoundPredicate::AlwaysTrue;
-use crate::expr::{Bind, BoundPredicate, BoundReference, LogicalExpression, Predicate, PredicateOperator};
+use crate::expr::{Bind, BoundPredicate, LogicalExpression, Predicate, PredicateOperator};
 use crate::io::FileIO;
 use crate::spec::{
-    DataContentType, FieldSummary, Manifest, ManifestEntry, ManifestEntryRef, ManifestFile,
-    PartitionField, PartitionSpec, PartitionSpecRef, Schema, SchemaRef, SnapshotRef,
-    TableMetadataRef, Transform,
+    DataContentType, FieldSummary, ManifestEntryRef, ManifestFile, PartitionField,
+    PartitionSpecRef, Schema, SchemaRef, SnapshotRef, TableMetadataRef,
 };
 use crate::table::Table;
 use crate::{Error, ErrorKind};
 use arrow_array::RecordBatch;
+use async_stream::try_stream;
 use futures::stream::{iter, BoxStream};
 use futures::StreamExt;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::Arc;
-use async_stream::try_stream;
 
 /// Builder to create table scan.
 pub struct TableScanBuilder<'a> {
@@ -173,7 +172,6 @@ pub type FileScanTaskStream = BoxStream<'static, crate::Result<FileScanTask>>;
 impl TableScan {
     /// Returns a stream of file scan tasks.
 
-
     pub async fn plan_files(&self) -> crate::Result<FileScanTaskStream> {
         // Cache `PartitionEvaluator`s created as part of this scan
         let mut partition_evaluator_cache: HashMap<i32, PartitionEvaluator> = HashMap::new();
@@ -201,7 +199,7 @@ impl TableScan {
                     Some(
                         partition_evaluator_cache
                             .entry(manifest.partition_spec_id())
-                            .or_insert_with_key(self.create_partition_evaluator(filter))
+                            .or_insert_with_key(|key| self.create_partition_evaluator(key, filter))
                             .deref(),
                     )
                 } else {
@@ -236,17 +234,19 @@ impl TableScan {
                     }
                 }
             }
-        }.boxed())
+        }
+        .boxed())
     }
 
-    fn create_partition_evaluator(&self, filter: &Predicate) -> fn(&i32) -> crate::Result<PartitionEvaluator> {
-        |&id| {
-            // TODO: predicate binding not yet merged to main
-            let bound_predicate = filter.bind(self.schema.clone(), self.case_sensitive)?;
+    fn create_partition_evaluator(&self, id: &i32, filter: &Predicate) -> PartitionEvaluator {
+        // TODO: predicate binding not yet merged to main
+        let bound_predicate = filter
+            .bind(self.schema.clone(), self.case_sensitive)
+            .unwrap();
 
-            let partition_spec = self.table_metadata.partition_spec_by_id(id).unwrap();
-            PartitionEvaluator::new(partition_spec.clone(), bound_predicate, self.schema.clone())
-        }
+        let partition_spec = self.table_metadata.partition_spec_by_id(*id).unwrap();
+        PartitionEvaluator::new(partition_spec.clone(), bound_predicate, self.schema.clone())
+            .unwrap()
     }
 
     pub async fn to_arrow(&self) -> crate::Result<ArrowRecordBatchStream> {
@@ -315,7 +315,11 @@ struct ManifestEvalVisitor {
 }
 
 impl ManifestEvalVisitor {
-    fn new(partition_schema: SchemaRef, partition_filter: Predicate, case_sensitive: bool) -> crate::Result<Self> {
+    fn new(
+        partition_schema: SchemaRef,
+        partition_filter: Predicate,
+        case_sensitive: bool,
+    ) -> crate::Result<Self> {
         let partition_filter = partition_filter.bind(partition_schema.clone(), case_sensitive)?;
 
         Ok(Self {
@@ -485,7 +489,11 @@ impl InclusiveProjection {
         }
 
         parts.iter().fold(Predicate::AlwaysTrue, |res, &part| {
-            res.and(part.transform.project(&part.name, &predicate))
+            if let Some(pred_for_part) = part.transform.project(&part.name, &predicate)? {
+                res.and(pred_for_part)
+            } else {
+                res
+            }
         })
     }
 }

crates/iceberg/src/spec/transform.rs

@@ -18,6 +18,7 @@
 //! Transforms in iceberg.
 
 use crate::error::{Error, Result};
+use crate::expr::{BoundPredicate, Predicate};
 use crate::spec::datatypes::{PrimitiveType, Type};
 use crate::ErrorKind;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
@@ -261,6 +262,12 @@ impl Transform {
             _ => self == other,
         }
     }
+
+    pub fn project(&self, name: &str, predicate: &BoundPredicate) -> Result<Option<Predicate>> {
+        // Waiting on https://github.com/apache/iceberg-rust/pull/269
+        // to deliver https://github.com/apache/iceberg-rust/issues/264
+        todo!()
+    }
 }
 
 impl Display for Transform {