feat: add expression visitors

src/iceberg/CMakeLists.txt

@@ -20,10 +20,12 @@ set(ICEBERG_INCLUDES "$<BUILD_INTERFACE:${PROJECT_BINARY_DIR}/src>"
 set(ICEBERG_SOURCES
     arrow_c_data_guard_internal.cc
     catalog/memory/in_memory_catalog.cc
+    expression/binder.cc
     expression/expression.cc
     expression/expressions.cc
     expression/literal.cc
     expression/predicate.cc
+    expression/rewrite_not.cc
     expression/term.cc
     file_reader.cc
     file_writer.cc

src/iceberg/expression/binder.cc

@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/expression/binder.h"
+
+namespace iceberg {
+
+Binder::Binder(const Schema& schema, bool case_sensitive)
+    : schema_(schema), case_sensitive_(case_sensitive) {}
+
+Result<std::shared_ptr<Expression>> Binder::Bind(const Schema& schema,
+                                                 const std::shared_ptr<Expression>& expr,
+                                                 bool case_sensitive) {
+  Binder binder(schema, case_sensitive);
+  return Visit<std::shared_ptr<Expression>, Binder>(expr, binder);
+}
+
+Result<std::shared_ptr<Expression>> Binder::AlwaysTrue() { return True::Instance(); }
+
+Result<std::shared_ptr<Expression>> Binder::AlwaysFalse() { return False::Instance(); }
+
+Result<std::shared_ptr<Expression>> Binder::Not(
+    const std::shared_ptr<Expression>& child_result) {
+  return iceberg::Not::MakeFolded(child_result);
+}
+
+Result<std::shared_ptr<Expression>> Binder::And(
+    const std::shared_ptr<Expression>& left_result,
+    const std::shared_ptr<Expression>& right_result) {
+  return iceberg::And::MakeFolded(left_result, right_result);
+}
+
+Result<std::shared_ptr<Expression>> Binder::Or(
+    const std::shared_ptr<Expression>& left_result,
+    const std::shared_ptr<Expression>& right_result) {
+  return iceberg::Or::MakeFolded(left_result, right_result);
+}
+
+Result<std::shared_ptr<Expression>> Binder::Predicate(
+    const std::shared_ptr<UnboundPredicate>& pred) {
+  ICEBERG_DCHECK(pred != nullptr, "Predicate cannot be null");
+  return pred->Bind(schema_, case_sensitive_);
+}
+
+Result<std::shared_ptr<Expression>> Binder::Predicate(
+    const std::shared_ptr<BoundPredicate>& pred) {
+  ICEBERG_DCHECK(pred != nullptr, "Predicate cannot be null");
+  return InvalidExpression("Found already bound predicate: {}", pred->ToString());
+}
+
+Result<bool> IsBoundVisitor::IsBound(const std::shared_ptr<Expression>& expr) {
+  ICEBERG_DCHECK(expr != nullptr, "Expression cannot be null");
+  IsBoundVisitor visitor;
+  return Visit<bool, IsBoundVisitor>(expr, visitor);
+}
+
+Result<bool> IsBoundVisitor::AlwaysTrue() { return true; }
+
+Result<bool> IsBoundVisitor::AlwaysFalse() { return true; }
+
+Result<bool> IsBoundVisitor::Not(bool child_result) { return child_result; }
+
+Result<bool> IsBoundVisitor::And(bool left_result, bool right_result) {
+  return left_result && right_result;
+}
+
+Result<bool> IsBoundVisitor::Or(bool left_result, bool right_result) {
+  return left_result && right_result;
+}
+
+Result<bool> IsBoundVisitor::Predicate(const std::shared_ptr<BoundPredicate>& pred) {
+  return true;
+}
+
+Result<bool> IsBoundVisitor::Predicate(const std::shared_ptr<UnboundPredicate>& pred) {
+  return false;
+}
+
+}  // namespace iceberg

src/iceberg/expression/binder.h

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+/// \file iceberg/expression/binder.h
+/// Bind an expression to a schema.
+
+#include "iceberg/expression/expression_visitor.h"
+
+namespace iceberg {
+
+class ICEBERG_EXPORT Binder : public ExpressionVisitor<std::shared_ptr<Expression>> {
+ public:
+  Binder(const Schema& schema, bool case_sensitive);
+
+  static Result<std::shared_ptr<Expression>> Bind(const Schema& schema,
+                                                  const std::shared_ptr<Expression>& expr,
+                                                  bool case_sensitive);
+
+  Result<std::shared_ptr<Expression>> AlwaysTrue() override;
+  Result<std::shared_ptr<Expression>> AlwaysFalse() override;
+  Result<std::shared_ptr<Expression>> Not(
+      const std::shared_ptr<Expression>& child_result) override;
+  Result<std::shared_ptr<Expression>> And(
+      const std::shared_ptr<Expression>& left_result,
+      const std::shared_ptr<Expression>& right_result) override;
+  Result<std::shared_ptr<Expression>> Or(
+      const std::shared_ptr<Expression>& left_result,
+      const std::shared_ptr<Expression>& right_result) override;
+  Result<std::shared_ptr<Expression>> Predicate(
+      const std::shared_ptr<BoundPredicate>& pred) override;
+  Result<std::shared_ptr<Expression>> Predicate(
+      const std::shared_ptr<UnboundPredicate>& pred) override;
+
+ private:
+  const Schema& schema_;
+  const bool case_sensitive_;
+};
+
+class ICEBERG_EXPORT IsBoundVisitor : public ExpressionVisitor<bool> {
+ public:
+  static Result<bool> IsBound(const std::shared_ptr<Expression>& expr);
+
+  Result<bool> AlwaysTrue() override;
+  Result<bool> AlwaysFalse() override;
+  Result<bool> Not(bool child_result) override;
+  Result<bool> And(bool left_result, bool right_result) override;
+  Result<bool> Or(bool left_result, bool right_result) override;
+  Result<bool> Predicate(const std::shared_ptr<BoundPredicate>& pred) override;
+  Result<bool> Predicate(const std::shared_ptr<UnboundPredicate>& pred) override;
+};
+
+// TODO(gangwu): add the Java parity `ReferenceVisitor`
+
+}  // namespace iceberg

src/iceberg/expression/expression.cc

@@ -120,6 +120,24 @@ Result<std::unique_ptr<Not>> Not::Make(std::shared_ptr<Expression> child) {
   return std::unique_ptr<Not>(new Not(std::move(child)));
 }
 
+Result<std::shared_ptr<Expression>> Not::MakeFolded(std::shared_ptr<Expression> child) {
+  if (child->op() == Expression::Operation::kTrue) {
+    return False::Instance();
+  }
+
+  if (child->op() == Expression::Operation::kFalse) {
+    return True::Instance();
+  }
+
+  // not(not(x)) = x
+  if (child->op() == Expression::Operation::kNot) {
+    const auto& not_expr = internal::checked_cast<const ::iceberg::Not&>(*child);
+    return not_expr.child();
+  }
+
+  return Not::Make(std::move(child));
+}
+
 Not::Not(std::shared_ptr<Expression> child) : child_(std::move(child)) {
   ICEBERG_DCHECK(child_ != nullptr, "Not expression cannot have null child");
 }

src/iceberg/expression/expression.h

@@ -28,6 +28,7 @@
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
 #include "iceberg/util/formattable.h"
+#include "iceberg/util/macros.h"
 
 namespace iceberg {
 
@@ -80,6 +81,11 @@ class ICEBERG_EXPORT Expression : public util::Formattable {
   }
 
   std::string ToString() const override { return "Expression"; }
+
+  virtual bool is_unbound_predicate() const { return false; }
+  virtual bool is_bound_predicate() const { return false; }
+  virtual bool is_unbound_aggregate() const { return false; }
+  virtual bool is_bound_aggregate() const { return false; }
 };
 
 /// \brief An Expression that is always true.
@@ -137,6 +143,44 @@ class ICEBERG_EXPORT And : public Expression {
   static Result<std::unique_ptr<And>> Make(std::shared_ptr<Expression> left,
                                            std::shared_ptr<Expression> right);
 
+  /// \brief Creates a folded And expression from two sub-expressions.
+  ///
+  /// \param left The left operand of the AND expression
+  /// \param right The right operand of the AND expression
+  /// \param args Additional operands of the AND expression
+  /// \return A Result containing a shared pointer to the folded And expression, or an
+  /// error if left or right is nullptr
+  /// \note A folded And expression is an expression that is equivalent to the original
+  /// expression, but with the And operation removed. For example, (true and x) = x.
+  template <typename... Args>
+  static Result<std::shared_ptr<Expression>> MakeFolded(std::shared_ptr<Expression> left,
+                                                        std::shared_ptr<Expression> right,
+                                                        Args&&... args)
+    requires std::conjunction_v<std::is_same<Args, std::shared_ptr<Expression>>...>
+  {
+    if constexpr (sizeof...(args) == 0) {
+      if (left->op() == Expression::Operation::kFalse ||
+          right->op() == Expression::Operation::kFalse) {
+        return False::Instance();
+      }
+
+      if (left->op() == Expression::Operation::kTrue) {
+        return right;
+      }
+
+      if (right->op() == Expression::Operation::kTrue) {
+        return left;
+      }
+
+      return And::Make(std::move(left), std::move(right));
+    } else {
+      ICEBERG_ASSIGN_OR_THROW(auto and_expr,
+                              And::Make(std::move(left), std::move(right)));
+
+      return And::MakeFolded(std::move(and_expr), std::forward<Args>(args)...);
+    }
+  }
+
   /// \brief Returns the left operand of the AND expression.
   ///
   /// \return The left operand of the AND expression
@@ -175,6 +219,43 @@ class ICEBERG_EXPORT Or : public Expression {
   static Result<std::unique_ptr<Or>> Make(std::shared_ptr<Expression> left,
                                           std::shared_ptr<Expression> right);
 
+  /// \brief Creates a folded Or expression from two sub-expressions.
+  ///
+  /// \param left The left operand of the OR expression
+  /// \param right The right operand of the OR expression
+  /// \param args Additional operands of the OR expression
+  /// \return A Result containing a shared pointer to the folded Or expression, or an
+  /// error if left or right is nullptr
+  /// \note A folded Or expression is an expression that is equivalent to the original
+  /// expression, but with the Or operation removed. For example, (false or x) = x.
+  template <typename... Args>
+  static Result<std::shared_ptr<Expression>> MakeFolded(std::shared_ptr<Expression> left,
+                                                        std::shared_ptr<Expression> right,
+                                                        Args&&... args)
+    requires std::conjunction_v<std::is_same<Args, std::shared_ptr<Expression>>...>
+  {
+    if constexpr (sizeof...(args) == 0) {
+      if (left->op() == Expression::Operation::kTrue ||
+          right->op() == Expression::Operation::kTrue) {
+        return True::Instance();
+      }
+
+      if (left->op() == Expression::Operation::kFalse) {
+        return right;
+      }
+
+      if (right->op() == Expression::Operation::kFalse) {
+        return left;
+      }
+
+      return Or::Make(std::move(left), std::move(right));
+    } else {
+      ICEBERG_ASSIGN_OR_THROW(auto or_expr, Or::Make(std::move(left), std::move(right)));
+
+      return Or::MakeFolded(std::move(or_expr), std::forward<Args>(args)...);
+    }
+  }
+
   /// \brief Returns the left operand of the OR expression.
   ///
   /// \return The left operand of the OR expression
@@ -211,6 +292,16 @@ class ICEBERG_EXPORT Not : public Expression {
   /// \return A Result containing a unique pointer to Not, or an error if child is nullptr
   static Result<std::unique_ptr<Not>> Make(std::shared_ptr<Expression> child);
 
+  /// \brief Creates a folded Not expression from a child expression.
+  ///
+  /// \param child The expression to negate
+  /// \return A Result containing a shared pointer to the folded Not expression, or an
+  /// error if child is nullptr
+  /// \note A folded Not expression is an expression that is equivalent to the original
+  /// expression, but with the Not operation removed. For example, not(not(x)) = x.
+  static Result<std::shared_ptr<Expression>> MakeFolded(
+      std::shared_ptr<Expression> child);
+
   /// \brief Returns the child expression.
   ///
   /// \return The child expression being negated