refactor: use StructType instead of Schema for unbound term

Author: HeartLinked

src/iceberg/expression/aggregate.cc

@@ -467,13 +467,13 @@ bool MinAggregate::HasValue(const DataFile& file) const {
 
 template <typename B>
 Result<std::shared_ptr<Expression>> UnboundAggregateImpl<B>::Bind(
-    const Schema& schema, bool case_sensitive) const {
+    const StructType& struct_, bool case_sensitive) const {
   ICEBERG_DCHECK(UnboundAggregateImpl<B>::IsSupportedOp(this->op()),
                  "Unexpected aggregate operation");
 
   std::shared_ptr<B> bound_term;
   if (this->term()) {
-    ICEBERG_ASSIGN_OR_RAISE(bound_term, this->term()->Bind(schema, case_sensitive));
+    ICEBERG_ASSIGN_OR_RAISE(bound_term, this->term()->Bind(struct_, case_sensitive));
   }
 
   switch (this->op()) {

src/iceberg/expression/aggregate.h

@@ -84,7 +84,7 @@ class ICEBERG_EXPORT UnboundAggregateImpl : public UnboundAggregate,
     return BASE::term() ? BASE::term()->reference() : nullptr;
   }
 
-  Result<std::shared_ptr<Expression>> Bind(const Schema& schema,
+  Result<std::shared_ptr<Expression>> Bind(const StructType& struct_,
                                            bool case_sensitive) const override;
 
  private:

src/iceberg/expression/binder.cc

@@ -21,13 +21,13 @@
 
 namespace iceberg {
 
-Binder::Binder(const Schema& schema, bool case_sensitive)
-    : schema_(schema), case_sensitive_(case_sensitive) {}
+Binder::Binder(const StructType& struct_, bool case_sensitive)
+    : struct_(struct_), case_sensitive_(case_sensitive) {}
 
-Result<std::shared_ptr<Expression>> Binder::Bind(const Schema& schema,
+Result<std::shared_ptr<Expression>> Binder::Bind(const StructType& struct_,
                                                  const std::shared_ptr<Expression>& expr,
                                                  bool case_sensitive) {
-  Binder binder(schema, case_sensitive);
+  Binder binder(struct_, case_sensitive);
   return Visit<std::shared_ptr<Expression>, Binder>(expr, binder);
 }
 
@@ -55,7 +55,7 @@ Result<std::shared_ptr<Expression>> Binder::Or(
 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_);
+  return pred->Bind(struct_, case_sensitive_);
 }
 
 Result<std::shared_ptr<Expression>> Binder::Predicate(
@@ -73,7 +73,7 @@ Result<std::shared_ptr<Expression>> Binder::Aggregate(
 Result<std::shared_ptr<Expression>> Binder::Aggregate(
     const std::shared_ptr<UnboundAggregate>& aggregate) {
   ICEBERG_DCHECK(aggregate != nullptr, "Aggregate cannot be null");
-  return aggregate->Bind(schema_, case_sensitive_);
+  return aggregate->Bind(struct_, case_sensitive_);
 }
 
 Result<bool> IsBoundVisitor::IsBound(const std::shared_ptr<Expression>& expr) {

src/iceberg/expression/binder.h

@@ -28,9 +28,9 @@ namespace iceberg {
 
 class ICEBERG_EXPORT Binder : public ExpressionVisitor<std::shared_ptr<Expression>> {
  public:
-  Binder(const Schema& schema, bool case_sensitive);
+  Binder(const StructType& struct_, bool case_sensitive);
 
-  static Result<std::shared_ptr<Expression>> Bind(const Schema& schema,
+  static Result<std::shared_ptr<Expression>> Bind(const StructType& struct_,
                                                   const std::shared_ptr<Expression>& expr,
                                                   bool case_sensitive);
 
@@ -54,7 +54,7 @@ class ICEBERG_EXPORT Binder : public ExpressionVisitor<std::shared_ptr<Expressio
       const std::shared_ptr<UnboundAggregate>& aggregate) override;
 
  private:
-  const Schema& schema_;
+  const StructType& struct_;
   const bool case_sensitive_;
 };
 

src/iceberg/expression/expression.h

@@ -340,14 +340,14 @@ class ICEBERG_EXPORT Unbound {
  public:
   /// \brief Bind this expression to a concrete schema.
   ///
-  /// \param schema The schema to bind against
+  /// \param struct_ The struct type (or schema) to bind against
   /// \param case_sensitive Whether field name matching should be case sensitive
   /// \return A bound expression or an error if binding fails
-  virtual Result<std::shared_ptr<B>> Bind(const Schema& schema,
+  virtual Result<std::shared_ptr<B>> Bind(const StructType& struct_,
                                           bool case_sensitive) const = 0;
 
   /// \brief Overloaded Bind method that uses case-sensitive matching by default.
-  Result<std::shared_ptr<B>> Bind(const Schema& schema) const;
+  Result<std::shared_ptr<B>> Bind(const StructType& struct_) const;
 
   /// \brief Returns the underlying named reference for this unbound term.
   virtual std::shared_ptr<class NamedReference> reference() = 0;

src/iceberg/expression/manifest_evaluator.cc

@@ -363,12 +363,9 @@ Result<std::unique_ptr<ManifestEvaluator>> ManifestEvaluator::MakePartitionFilte
     std::shared_ptr<Expression> expr, const std::shared_ptr<PartitionSpec>& spec,
     const Schema& schema, bool case_sensitive) {
   ICEBERG_ASSIGN_OR_RAISE(auto partition_type, spec->PartitionType(schema));
-  auto field_span = partition_type->fields();
-  std::vector<SchemaField> fields(field_span.begin(), field_span.end());
-  auto partition_schema = std::make_shared<Schema>(fields);
   ICEBERG_ASSIGN_OR_RAISE(auto rewrite_expr, RewriteNot::Visit(std::move(expr)));
   ICEBERG_ASSIGN_OR_RAISE(auto partition_expr,
-                          Binder::Bind(*partition_schema, rewrite_expr, case_sensitive));
+                          Binder::Bind(*partition_type, rewrite_expr, case_sensitive));
   return std::unique_ptr<ManifestEvaluator>(
       new ManifestEvaluator(std::move(partition_expr)));
 }

src/iceberg/expression/predicate.cc

@@ -182,8 +182,8 @@ Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::Negate() const {
 
 template <typename B>
 Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::Bind(
-    const Schema& schema, bool case_sensitive) const {
-  ICEBERG_ASSIGN_OR_RAISE(auto bound_term, BASE::term()->Bind(schema, case_sensitive));
+    const StructType& struct_, bool case_sensitive) const {
+  ICEBERG_ASSIGN_OR_RAISE(auto bound_term, BASE::term()->Bind(struct_, case_sensitive));
 
   if (values_.empty()) {
     return BindUnaryOperation(std::move(bound_term));

src/iceberg/expression/predicate.h

@@ -28,6 +28,7 @@
 #include "iceberg/expression/literal.h"
 #include "iceberg/expression/term.h"
 #include "iceberg/iceberg_export.h"
+#include "iceberg/type.h"
 
 namespace iceberg {
 
@@ -68,7 +69,7 @@ class ICEBERG_EXPORT UnboundPredicate : public virtual Expression,
   std::shared_ptr<NamedReference> reference() override = 0;
 
   /// \brief Bind this UnboundPredicate.
-  Result<std::shared_ptr<Expression>> Bind(const Schema& schema,
+  Result<std::shared_ptr<Expression>> Bind(const StructType& struct_,
                                            bool case_sensitive) const override = 0;
 
   /// \brief Negate this UnboundPredicate.
@@ -125,7 +126,7 @@ class ICEBERG_EXPORT UnboundPredicateImpl : public UnboundPredicate,
 
   std::string ToString() const override;
 
-  Result<std::shared_ptr<Expression>> Bind(const Schema& schema,
+  Result<std::shared_ptr<Expression>> Bind(const StructType& struct_,
                                            bool case_sensitive) const override;
 
   Result<std::shared_ptr<Expression>> Negate() const override;

src/iceberg/expression/projections.cc

@@ -27,6 +27,7 @@
 #include "iceberg/expression/rewrite_not.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
+#include "iceberg/schema.h"
 #include "iceberg/transform.h"
 #include "iceberg/util/macros.h"
 

src/iceberg/expression/residual_evaluator.cc

@@ -27,6 +27,7 @@
 #include "iceberg/schema.h"
 #include "iceberg/schema_internal.h"
 #include "iceberg/transform.h"
+#include "iceberg/type.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg {
@@ -42,8 +43,7 @@ class ResidualVisitor : public BoundVisitor<std::shared_ptr<Expression>> {
                                       const StructLike& partition_data,
                                       bool case_sensitive) {
     ICEBERG_ASSIGN_OR_RAISE(auto partition_type, spec.PartitionType(schema));
-    auto partition_schema = FromStructType(std::move(*partition_type), std::nullopt);
-    return ResidualVisitor(spec, schema, std::move(partition_schema), partition_data,
+    return ResidualVisitor(spec, schema, std::move(partition_type), partition_data,
                            case_sensitive);
   }
 
@@ -202,17 +202,17 @@ class ResidualVisitor : public BoundVisitor<std::shared_ptr<Expression>> {
 
  private:
   ResidualVisitor(const PartitionSpec& spec, const Schema& schema,
-                  std::unique_ptr<Schema> partition_schema,
+                  std::unique_ptr<StructType> partition_type,
                   const StructLike& partition_data, bool case_sensitive)
       : spec_(spec),
         schema_(schema),
-        partition_schema_(std::move(partition_schema)),
+        partition_type_(std::move(partition_type)),
         partition_data_(partition_data),
         case_sensitive_(case_sensitive) {}
 
   const PartitionSpec& spec_;
   const Schema& schema_;
-  std::unique_ptr<Schema> partition_schema_;
+  std::unique_ptr<StructType> partition_type_;
   const StructLike& partition_data_;
   bool case_sensitive_;
 };
@@ -243,9 +243,8 @@ Result<std::shared_ptr<Expression>> ResidualVisitor::Predicate(
     std::shared_ptr<Expression> strict_result = nullptr;
 
     if (strict_projection != nullptr) {
-      ICEBERG_ASSIGN_OR_RAISE(
-          auto bound_strict,
-          strict_projection->Bind(*partition_schema_, case_sensitive_));
+      ICEBERG_ASSIGN_OR_RAISE(auto bound_strict,
+                              strict_projection->Bind(*partition_type_, case_sensitive_));
       if (bound_strict->is_bound_predicate()) {
         ICEBERG_ASSIGN_OR_RAISE(
             strict_result, BoundVisitor::Predicate(
@@ -270,7 +269,7 @@ Result<std::shared_ptr<Expression>> ResidualVisitor::Predicate(
     if (inclusive_projection != nullptr) {
       ICEBERG_ASSIGN_OR_RAISE(
           auto bound_inclusive,
-          inclusive_projection->Bind(*partition_schema_, case_sensitive_));
+          inclusive_projection->Bind(*partition_type_, case_sensitive_));
 
       if (bound_inclusive->is_bound_predicate()) {
         ICEBERG_ASSIGN_OR_RAISE(