feat: add transform project (apache#371)

Add Transform::Project for inclusive predicate projection, which transforms a BoundPredicate to an inclusive predicate on partition values.

Author: zhjwpku

src/iceberg/expression/predicate.h

@@ -27,6 +27,7 @@
 #include "iceberg/expression/expression.h"
 #include "iceberg/expression/literal.h"
 #include "iceberg/expression/term.h"
+#include "iceberg/iceberg_export.h"
 
 namespace iceberg {
 

src/iceberg/test/transform_test.cc

@@ -23,8 +23,14 @@
 #include <regex>
 #include <utility>
 
+#include "iceberg/expression/predicate.h"
+#include "iceberg/expression/term.h"
+#include "iceberg/result.h"
 #include "iceberg/transform_function.h"
 #include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/projection_util_internal.h"
 
 namespace iceberg {
 namespace {
@@ -240,6 +246,66 @@ bool Transform::SatisfiesOrderOf(const Transform& other) const {
   std::unreachable();
 }
 
+Result<std::unique_ptr<UnboundPredicate>> Transform::Project(
+    std::string_view name, const std::shared_ptr<BoundPredicate>& predicate) {
+  switch (transform_type_) {
+    case TransformType::kIdentity:
+      return ProjectionUtil::IdentityProject(name, predicate);
+    case TransformType::kBucket: {
+      // If the predicate has a transformed child that matches the given transform, return
+      // a predicate.
+      if (predicate->term()->kind() == Term::Kind::kTransform) {
+        const auto boundTransform =
+            internal::checked_pointer_cast<BoundTransform>(predicate->term());
+        if (*this == *boundTransform->transform()) {
+          return ProjectionUtil::RemoveTransform(name, predicate);
+        } else {
+          return nullptr;
+        }
+      }
+      ICEBERG_ASSIGN_OR_RAISE(auto func, Bind(predicate->term()->type()));
+      return ProjectionUtil::BucketProject(name, predicate, func);
+    }
+    case TransformType::kTruncate: {
+      // If the predicate has a transformed child that matches the given transform, return
+      // a predicate.
+      if (predicate->term()->kind() == Term::Kind::kTransform) {
+        const auto boundTransform =
+            internal::checked_pointer_cast<BoundTransform>(predicate->term());
+        if (*this == *boundTransform->transform()) {
+          return ProjectionUtil::RemoveTransform(name, predicate);
+        } else {
+          return nullptr;
+        }
+      }
+      ICEBERG_ASSIGN_OR_RAISE(auto func, Bind(predicate->term()->type()));
+      return ProjectionUtil::TruncateProject(name, predicate, func);
+    }
+    case TransformType::kYear:
+    case TransformType::kMonth:
+    case TransformType::kDay:
+    case TransformType::kHour: {
+      // If the predicate has a transformed child that matches the given transform, return
+      // a predicate.
+      if (predicate->term()->kind() == Term::Kind::kTransform) {
+        const auto boundTransform =
+            internal::checked_pointer_cast<BoundTransform>(predicate->term());
+        if (*this == *boundTransform->transform()) {
+          return ProjectionUtil::RemoveTransform(name, predicate);
+        } else {
+          return nullptr;
+        }
+      }
+      ICEBERG_ASSIGN_OR_RAISE(auto func, Bind(predicate->term()->type()));
+      return ProjectionUtil::TemporalProject(name, predicate, func);
+    }
+    case TransformType::kUnknown:
+    case TransformType::kVoid:
+      return nullptr;
+  }
+  std::unreachable();
+}
+
 bool TransformFunction::Equals(const TransformFunction& other) const {
   return transform_type_ == other.transform_type_ && *source_type_ == *other.source_type_;
 }

src/iceberg/transform.cc

@@ -23,6 +23,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <string_view>
 #include <utility>
 #include <variant>
 
@@ -164,11 +165,23 @@ class ICEBERG_EXPORT Transform : public util::Formattable {
   /// For example, sorting by day(ts) will produce an ordering that is also by month(ts)
   /// or year(ts). However, sorting by day(ts) will not satisfy the order of hour(ts) or
   /// identity(ts).
-  ///
+  /// \param other The other transform to compare with.
   /// \return true if ordering by this transform is equivalent to ordering by the other
   /// transform.
   bool SatisfiesOrderOf(const Transform& other) const;
 
+  /// \brief Transforms a BoundPredicate to an inclusive predicate on the partition values
+  /// produced by the transform.
+  ///
+  /// This inclusive transform guarantees that if predicate->Test(value) is true, then
+  /// Projected(transform(value)) is true.
+  /// \param name The name of the partition column.
+  /// \param predicate The predicate to project.
+  /// \return A Result containing either a unique pointer to the projected predicate,
+  /// nullptr if the projection cannot be performed, or an Error if the projection fails.
+  Result<std::unique_ptr<UnboundPredicate>> Project(
+      std::string_view name, const std::shared_ptr<BoundPredicate>& predicate);
+
   /// \brief Returns a string representation of this transform (e.g., "bucket[16]").
   std::string ToString() const override;
 

src/iceberg/transform.h

@@ -83,6 +83,9 @@ class ICEBERG_EXPORT TruncateTransform : public TransformFunction {
   /// \brief Returns the same type as source_type.
   std::shared_ptr<Type> ResultType() const override;
 
+  /// \brief Returns the width to truncate to.
+  int32_t width() const { return width_; }
+
   /// \brief Create a TruncateTransform.
   /// \param source_type Type of the input data.
   /// \param width The width to truncate to.

src/iceberg/transform_function.h

@@ -123,6 +123,9 @@ class Uuid;
 class Expression;
 class Literal;
 
+class BoundPredicate;
+class UnboundPredicate;
+
 class DataTableScan;
 class FileScanTask;
 class ScanTask;