feat: implement Literal Transform

Author: zhjwpku

src/iceberg/expression/literal.cc

@@ -130,8 +130,16 @@ Literal Literal::Boolean(bool value) { return {Value{value}, iceberg::boolean()}
 
 Literal Literal::Int(int32_t value) { return {Value{value}, iceberg::int32()}; }
 
+Literal Literal::Date(int32_t value) { return {Value{value}, iceberg::date()}; }
+
 Literal Literal::Long(int64_t value) { return {Value{value}, iceberg::int64()}; }
 
+Literal Literal::Timestamp(int64_t value) { return {Value{value}, iceberg::timestamp()}; }
+
+Literal Literal::TimestampTz(int64_t value) {
+  return {Value{value}, iceberg::timestamp_tz()};
+}
+
 Literal Literal::Float(float value) { return {Value{value}, iceberg::float32()}; }
 
 Literal Literal::Double(double value) { return {Value{value}, iceberg::float64()}; }
@@ -208,12 +216,30 @@ std::partial_ordering Literal::operator<=>(const Literal& other) const {
       return this_val <=> other_val;
     }
 
+    case TypeId::kDate: {
+      auto this_val = std::get<int32_t>(value_);
+      auto other_val = std::get<int32_t>(other.value_);
+      return this_val <=> other_val;
+    }
+
     case TypeId::kLong: {
       auto this_val = std::get<int64_t>(value_);
       auto other_val = std::get<int64_t>(other.value_);
       return this_val <=> other_val;
     }
 
+    case TypeId::kTimestamp: {
+      auto this_val = std::get<int64_t>(value_);
+      auto other_val = std::get<int64_t>(other.value_);
+      return this_val <=> other_val;
+    }
+
+    case TypeId::kTimestampTz: {
+      auto this_val = std::get<int64_t>(value_);
+      auto other_val = std::get<int64_t>(other.value_);
+      return this_val <=> other_val;
+    }
+
     case TypeId::kFloat: {
       auto this_val = std::get<float>(value_);
       auto other_val = std::get<float>(other.value_);

src/iceberg/expression/literal.h

@@ -63,7 +63,10 @@ class ICEBERG_EXPORT Literal {
   /// \brief Factory methods for primitive types
   static Literal Boolean(bool value);
   static Literal Int(int32_t value);
+  static Literal Date(int32_t value);
   static Literal Long(int64_t value);
+  static Literal Timestamp(int64_t value);
+  static Literal TimestampTz(int64_t value);
   static Literal Float(float value);
   static Literal Double(double value);
   static Literal String(std::string value);
@@ -85,6 +88,9 @@ class ICEBERG_EXPORT Literal {
   /// \brief Get the literal type.
   const std::shared_ptr<PrimitiveType>& type() const;
 
+  /// \brief Get the literal value.
+  const Value& value() const { return value_; }
+
   /// \brief Converts this literal to a literal of the given type.
   ///
   /// When a predicate is bound to a concrete data column, literals are converted to match

src/iceberg/manifest_reader_internal.cc

@@ -33,11 +33,6 @@
 
 namespace iceberg {
 
-#define NANOARROW_RETURN_IF_NOT_OK(status, error)                  \
-  if (status != NANOARROW_OK) [[unlikely]] {                       \
-    return InvalidArrowData("Nanoarrow error: {}", error.message); \
-  }
-
 #define PARSE_PRIMITIVE_FIELD(item, array_view, type)                                   \
   for (size_t row_idx = 0; row_idx < array_view->length; row_idx++) {                   \
     if (!ArrowArrayViewIsNull(array_view, row_idx)) {                                   \

src/iceberg/transform.h

@@ -23,9 +23,11 @@
 
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <variant>
 
 #include "iceberg/arrow_c_data.h"
+#include "iceberg/expression/literal.h"
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
@@ -172,6 +174,8 @@ class ICEBERG_EXPORT TransformFunction {
   TransformFunction(TransformType transform_type, std::shared_ptr<Type> source_type);
   /// \brief Transform an input array to a new array
   virtual Result<ArrowArray> Transform(const ArrowArray& data) = 0;
+  /// \brief Transform an input Literal to a new Literal
+  virtual Result<std::optional<Literal>> Transform(const Literal& literal) = 0;
   /// \brief Get the transform type
   TransformType transform_type() const;
   /// \brief Get the source type of transform function

src/iceberg/transform_function.cc

@@ -19,7 +19,11 @@
 
 #include "iceberg/transform_function.h"
 
+#include <cassert>
+#include <chrono>
+
 #include "iceberg/type.h"
+#include "iceberg/util/murmurhash3_internal.h"
 
 namespace iceberg {
 
@@ -30,6 +34,10 @@ Result<ArrowArray> IdentityTransform::Transform(const ArrowArray& input) {
   return NotImplemented("IdentityTransform::Transform");
 }
 
+Result<std::optional<Literal>> IdentityTransform::Transform(const Literal& literal) {
+  return literal;
+}
+
 Result<std::shared_ptr<Type>> IdentityTransform::ResultType() const {
   return source_type();
 }
@@ -51,6 +59,57 @@ Result<ArrowArray> BucketTransform::Transform(const ArrowArray& input) {
   return NotImplemented("BucketTransform::Transform");
 }
 
+Result<std::optional<Literal>> BucketTransform::Transform(const Literal& literal) {
+  assert(literal.type() == source_type());
+  if (literal.IsBelowMin() || literal.IsAboveMax()) {
+    return InvalidArgument(
+        "Cannot apply bucket transform to literal with value {} of type {}",
+        literal.ToString(), source_type()->ToString());
+  }
+  int32_t hash_value = 0;
+  switch (source_type()->type_id()) {
+    case TypeId::kInt:
+    case TypeId::kDate: {
+      auto value = std::get<int32_t>(literal.value());
+      MurmurHash3_x86_32(&value, sizeof(int32_t), 0, &hash_value);
+      break;
+    }
+    case TypeId::kLong:
+    case TypeId::kTime:
+    case TypeId::kTimestamp:
+    case TypeId::kTimestampTz: {
+      auto value = std::get<int64_t>(literal.value());
+      MurmurHash3_x86_32(&value, sizeof(int64_t), 0, &hash_value);
+      break;
+    }
+    case TypeId::kDecimal:
+    case TypeId::kUuid: {
+      auto value = std::get<std::array<uint8_t, 16>>(literal.value());
+      MurmurHash3_x86_32(value.data(), sizeof(uint8_t) * 16, 0, &hash_value);
+      break;
+    }
+    case TypeId::kString: {
+      auto value = std::get<std::string>(literal.value());
+      MurmurHash3_x86_32(value.data(), value.size(), 0, &hash_value);
+      break;
+    }
+    case TypeId::kFixed:
+    case TypeId::kBinary: {
+      auto value = std::get<std::vector<uint8_t>>(literal.value());
+      MurmurHash3_x86_32(value.data(), value.size(), 0, &hash_value);
+      break;
+    }
+    default:
+      std::unreachable();
+  }
+
+  // Calculate the bucket index
+  int32_t bucket_index =
+      (hash_value & std::numeric_limits<int32_t>::max()) % num_buckets_;
+
+  return Literal::Int(bucket_index);
+}
+
 Result<std::shared_ptr<Type>> BucketTransform::ResultType() const {
   return iceberg::int32();
 }
@@ -91,6 +150,46 @@ Result<ArrowArray> TruncateTransform::Transform(const ArrowArray& input) {
   return NotImplemented("TruncateTransform::Transform");
 }
 
+Result<std::optional<Literal>> TruncateTransform::Transform(const Literal& literal) {
+  assert(literal.type() == source_type());
+  if (literal.IsBelowMin() || literal.IsAboveMax()) {
+    return InvalidArgument(
+        "Cannot apply truncate transform to literal with value {} of type {}",
+        literal.ToString(), source_type()->ToString());
+  }
+
+  switch (source_type()->type_id()) {
+    case TypeId::kInt: {
+      auto value = std::get<int32_t>(literal.value());
+      return Literal::Int(value % width_);
+    }
+    case TypeId::kLong: {
+      auto value = std::get<int64_t>(literal.value());
+      return Literal::Long(value % width_);
+    }
+    case TypeId::kDecimal: {
+      // TODO(zhjwpku): Handle decimal truncation logic here
+      return NotImplemented("Truncate for Decimal is not implemented yet");
+    }
+    case TypeId::kString: {
+      auto value = std::get<std::string>(literal.value());
+      if (value.size() > static_cast<size_t>(width_)) {
+        value.resize(width_);
+      }
+      return Literal::String(value);
+    }
+    case TypeId::kBinary: {
+      auto value = std::get<std::vector<uint8_t>>(literal.value());
+      if (value.size() > static_cast<size_t>(width_)) {
+        value.resize(width_);
+      }
+      return Literal::Binary(value);
+    }
+    default:
+      std::unreachable();
+  }
+}
+
 Result<std::shared_ptr<Type>> TruncateTransform::ResultType() const {
   return source_type();
 }
@@ -124,6 +223,34 @@ Result<ArrowArray> YearTransform::Transform(const ArrowArray& input) {
   return NotImplemented("YearTransform::Transform");
 }
 
+Result<std::optional<Literal>> YearTransform::Transform(const Literal& literal) {
+  assert(literal.type() == source_type());
+  if (literal.IsBelowMin() || literal.IsAboveMax()) {
+    return InvalidArgument(
+        "Cannot apply year transform to literal with value {} of type {}",
+        literal.ToString(), source_type()->ToString());
+  }
+
+  using namespace std::chrono;
+  switch (source_type()->type_id()) {
+    case TypeId::kDate: {
+      auto value = std::get<int32_t>(literal.value());
+      auto epoch = sys_days(year{1970} / January / 1);
+      auto ymd = year_month_day(epoch + days{value});
+      return Literal::Int(static_cast<int32_t>(ymd.year()));
+    }
+    case TypeId::kTimestamp:
+    case TypeId::kTimestampTz: {
+      auto value = std::get<int64_t>(literal.value());
+      // Convert milliseconds-since-epoch into a `year_month_day` object
+      auto ymd = year_month_day(floor<days>(sys_time<milliseconds>(milliseconds{value})));
+      return Literal::Int(static_cast<int32_t>(ymd.year()));
+    }
+    default:
+      std::unreachable();
+  }
+}
+
 Result<std::shared_ptr<Type>> YearTransform::ResultType() const {
   return iceberg::int32();
 }
@@ -152,6 +279,46 @@ Result<ArrowArray> MonthTransform::Transform(const ArrowArray& input) {
   return NotImplemented("MonthTransform::Transform");
 }
 
+Result<std::optional<Literal>> MonthTransform::Transform(const Literal& literal) {
+  assert(literal.type() == source_type());
+  if (literal.IsBelowMin() || literal.IsAboveMax()) {
+    return InvalidArgument(
+        "Cannot apply month transform to literal with value {} of type {}",
+        literal.ToString(), source_type()->ToString());
+  }
+
+  using namespace std::chrono;
+  switch (source_type()->type_id()) {
+    case TypeId::kDate: {
+      auto value = std::get<int32_t>(literal.value());
+      auto epoch = sys_days(year{1970} / January / 1);
+      auto ymd = year_month_day(epoch + days{value});
+      auto epoch_ymd = year_month_day(epoch);
+      auto delta = ymd.year() - epoch_ymd.year();
+      // Calculate the month as months from 1970-01
+      // Note: January is month 1, so we subtract 1 to get zero-based
+      // month count.
+      return Literal::Int(static_cast<int32_t>(delta.count() * 12 +
+                                               static_cast<unsigned>(ymd.month()) - 1));
+    }
+    case TypeId::kTimestamp:
+    case TypeId::kTimestampTz: {
+      auto value = std::get<int64_t>(literal.value());
+      // Convert milliseconds-since-epoch into a `year_month_day` object
+      auto ymd = year_month_day(floor<days>(sys_time<milliseconds>(milliseconds{value})));
+      auto epoch_ymd = year_month_day(year{1970} / January / 1);
+      auto delta = ymd.year() - epoch_ymd.year();
+      // Calculate the month as months from 1970-01
+      // Note: January is month 1, so we subtract 1 to get zero-based
+      // month count.
+      return Literal::Int(static_cast<int32_t>(delta.count() * 12 +
+                                               static_cast<unsigned>(ymd.month()) - 1));
+    }
+    default:
+      std::unreachable();
+  }
+}
+
 Result<std::shared_ptr<Type>> MonthTransform::ResultType() const {
   return iceberg::int32();
 }
@@ -180,6 +347,35 @@ Result<ArrowArray> DayTransform::Transform(const ArrowArray& input) {
   return NotImplemented("DayTransform::Transform");
 }
 
+Result<std::optional<Literal>> DayTransform::Transform(const Literal& literal) {
+  assert(literal.type() == source_type());
+  if (literal.IsBelowMin() || literal.IsAboveMax()) {
+    return InvalidArgument(
+        "Cannot apply day transform to literal with value {} of type {}",
+        literal.ToString(), source_type()->ToString());
+  }
+
+  using namespace std::chrono;
+  switch (source_type()->type_id()) {
+    case TypeId::kDate: {
+      // Day is the same as the date value
+      return literal;
+    }
+    case TypeId::kTimestamp:
+    case TypeId::kTimestampTz: {
+      auto value = std::get<int64_t>(literal.value());
+      // Convert milliseconds to `sys_days` (chronological days since epoch)
+      auto timestamp = sys_time<milliseconds>(milliseconds{value});
+      auto days_since_epoch = floor<days>(timestamp);
+
+      return Literal::Date(
+          static_cast<int32_t>(days_since_epoch.time_since_epoch().count()));
+    }
+    default:
+      std::unreachable();
+  }
+}
+
 Result<std::shared_ptr<Type>> DayTransform::ResultType() const { return iceberg::date(); }
 
 Result<std::unique_ptr<TransformFunction>> DayTransform::Make(
@@ -206,6 +402,32 @@ Result<ArrowArray> HourTransform::Transform(const ArrowArray& input) {
   return NotImplemented("HourTransform::Transform");
 }
 
+Result<std::optional<Literal>> HourTransform::Transform(const Literal& literal) {
+  assert(literal.type() == source_type());
+  if (literal.IsBelowMin() || literal.IsAboveMax()) {
+    return InvalidArgument(
+        "Cannot apply hour transform to literal with value {} of type {}",
+        literal.ToString(), source_type()->ToString());
+  }
+
+  using namespace std::chrono;
+  switch (source_type()->type_id()) {
+    case TypeId::kTimestamp:
+    case TypeId::kTimestampTz: {
+      auto value = std::get<int64_t>(literal.value());
+      // Create a `sys_time` object from the milliseconds value
+      auto timestamp = sys_time<milliseconds>(milliseconds{value});
+
+      // Convert the time since epoch directly into hours
+      auto hours_since_epoch = duration_cast<hours>(timestamp.time_since_epoch()).count();
+
+      return Literal::Int(static_cast<int32_t>(hours_since_epoch));
+    }
+    default:
+      std::unreachable();
+  }
+}
+
 Result<std::shared_ptr<Type>> HourTransform::ResultType() const {
   return iceberg::int32();
 }
@@ -233,6 +455,10 @@ Result<ArrowArray> VoidTransform::Transform(const ArrowArray& input) {
   return NotImplemented("VoidTransform::Transform");
 }
 
+Result<std::optional<Literal>> VoidTransform::Transform(const Literal& literal) {
+  return std::nullopt;
+}
+
 Result<std::shared_ptr<Type>> VoidTransform::ResultType() const { return source_type(); }
 
 Result<std::unique_ptr<TransformFunction>> VoidTransform::Make(