feat(expression): cast numeric literals to decimal type

src/iceberg/expression/literal.cc

@@ -22,6 +22,7 @@
 #include <cmath>
 #include <concepts>
 #include <cstdint>
+#include <format>
 #include <string>
 #include <vector>
 
@@ -85,6 +86,14 @@ class LiteralCaster {
   /// Cast from Fixed type to target type.
   static Result<Literal> CastFromFixed(const Literal& literal,
                                        const std::shared_ptr<PrimitiveType>& target_type);
+
+  /// Cast an integer value (from Int or Long) to a decimal target type.
+  static Result<Literal> CastIntegerToDecimal(
+      int64_t value, const std::shared_ptr<PrimitiveType>& target_type);
+
+  /// Cast a floating-point value (from Float or Double) to a decimal target type.
+  static Result<Literal> CastRealToDecimal(
+      double value, const std::shared_ptr<PrimitiveType>& target_type);
 };
 
 Literal LiteralCaster::BelowMinLiteral(std::shared_ptr<PrimitiveType> type) {
@@ -95,6 +104,71 @@ Literal LiteralCaster::AboveMaxLiteral(std::shared_ptr<PrimitiveType> type) {
   return Literal(Literal::AboveMax{}, std::move(type));
 }
 
+Result<Literal> LiteralCaster::CastIntegerToDecimal(
+    int64_t value, const std::shared_ptr<PrimitiveType>& target_type) {
+  const auto& decimal_type = internal::checked_cast<const DecimalType&>(*target_type);
+  const int32_t scale = decimal_type.scale();
+  // DecimalType does not bound its scale, but Rescale indexes a powers-of-ten table sized
+  // for [0, kMaxScale]; reject an out-of-range scale here rather than reading past it.
+  if (scale < 0 || scale > Decimal::kMaxScale) {
+    return InvalidArgument("Cannot cast {} as a {} value", value,
+                           target_type->ToString());
+  }
+  // An integer has scale 0, so scaling it to the target scale multiplies the unscaled
+  // value by 10^scale; Rescale rejects a target scale that would overflow the value.
+  ICEBERG_ASSIGN_OR_RAISE(auto decimal, Decimal(value).Rescale(0, scale));
+  if (!decimal.FitsInPrecision(decimal_type.precision())) {
+    return InvalidArgument("Cannot cast {} as a {} value", value,
+                           target_type->ToString());
+  }
+  return Literal::Decimal(decimal.value(), decimal_type.precision(),
+                          decimal_type.scale());
+}
+
+Result<Literal> LiteralCaster::CastRealToDecimal(
+    double value, const std::shared_ptr<PrimitiveType>& target_type) {
+  const auto& decimal_type = internal::checked_cast<const DecimalType&>(*target_type);
+  const int32_t target_scale = decimal_type.scale();
+  if (target_scale < 0 || target_scale > Decimal::kMaxScale) {
+    return InvalidArgument("Cannot cast {} as a {} value", value,
+                           target_type->ToString());
+  }
+  if (!std::isfinite(value)) {
+    return InvalidArgument("Cannot cast {} as a {} value", value,
+                           target_type->ToString());
+  }
+
+  // Parse the shortest round-tripping decimal representation of the value (matching
+  // Java's BigDecimal.valueOf(double), which goes through Double.toString) into a
+  // full-precision decimal, then round to the target scale below.
+  int32_t parsed_scale = 0;
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto parsed, Decimal::FromString(std::format("{}", value), nullptr, &parsed_scale));
+
+  Decimal unscaled = parsed;
+  if (parsed_scale > target_scale) {
+    // Drop excess fractional digits with HALF_UP rounding (round half away from zero, as
+    // Java does), since Rescale itself only truncates and rejects any dropped remainder.
+    const int32_t drop = parsed_scale - target_scale;
+    ICEBERG_ASSIGN_OR_RAISE(auto divisor, Decimal(1).Rescale(0, drop));
+    ICEBERG_ASSIGN_OR_RAISE(auto divmod, parsed.Divide(divisor));
+    Decimal quotient = divmod.first;
+    Decimal remainder = Decimal::Abs(divmod.second);
+    if (remainder * Decimal(2) >= divisor) {
+      quotient += (value < 0) ? Decimal(-1) : Decimal(1);
+    }
+    unscaled = quotient;
+  } else if (parsed_scale < target_scale) {
+    ICEBERG_ASSIGN_OR_RAISE(unscaled, parsed.Rescale(parsed_scale, target_scale));
+  }
+
+  if (!unscaled.FitsInPrecision(decimal_type.precision())) {
+    return InvalidArgument("Cannot cast {} as a {} value", value,
+                           target_type->ToString());
+  }
+  return Literal::Decimal(unscaled.value(), decimal_type.precision(), target_scale);
+}
+
 Result<Literal> LiteralCaster::CastFromInt(
     const Literal& literal, const std::shared_ptr<PrimitiveType>& target_type) {
   auto int_val = std::get<int32_t>(literal.value_);
@@ -109,6 +183,8 @@ Result<Literal> LiteralCaster::CastFromInt(
       return Literal::Double(static_cast<double>(int_val));
     case TypeId::kDate:
       return Literal::Date(int_val);
+    case TypeId::kDecimal:
+      return CastIntegerToDecimal(static_cast<int64_t>(int_val), target_type);
     default:
       return NotSupported("Cast from Int to {} is not implemented",
                           target_type->ToString());
@@ -153,6 +229,8 @@ Result<Literal> LiteralCaster::CastFromLong(
       return Literal::TimestampNs(long_val);
     case TypeId::kTimestampTzNs:
       return Literal::TimestampTzNs(long_val);
+    case TypeId::kDecimal:
+      return CastIntegerToDecimal(long_val, target_type);
     default:
       return NotSupported("Cast from Long to {} is not supported",
                           target_type->ToString());
@@ -166,6 +244,8 @@ Result<Literal> LiteralCaster::CastFromFloat(
   switch (target_type->type_id()) {
     case TypeId::kDouble:
       return Literal::Double(static_cast<double>(float_val));
+    case TypeId::kDecimal:
+      return CastRealToDecimal(static_cast<double>(float_val), target_type);
     default:
       return NotSupported("Cast from Float to {} is not supported",
                           target_type->ToString());
@@ -186,6 +266,8 @@ Result<Literal> LiteralCaster::CastFromDouble(
       }
       return Literal::Float(static_cast<float>(double_val));
     }
+    case TypeId::kDecimal:
+      return CastRealToDecimal(double_val, target_type);
     default:
       return NotSupported("Cast from Double to {} is not supported",
                           target_type->ToString());

src/iceberg/test/literal_test.cc

@@ -150,6 +150,64 @@ TEST(LiteralTest, DoubleCastToOverflow) {
   EXPECT_TRUE(min_result->IsBelowMin());
 }
 
+TEST(LiteralTest, IntegerCastToDecimal) {
+  // An integer default is scaled up to the target decimal scale: 12 -> 12.00 keeps the
+  // unscaled value 1200 at precision/scale (9, 2).
+  auto int_result = Literal::Int(12).CastTo(decimal(9, 2));
+  ASSERT_THAT(int_result, IsOk());
+  EXPECT_EQ(*int_result, Literal::Decimal(1200, 9, 2));
+
+  auto long_result = Literal::Long(int64_t{12}).CastTo(decimal(18, 3));
+  ASSERT_THAT(long_result, IsOk());
+  EXPECT_EQ(*long_result, Literal::Decimal(12000, 18, 3));
+
+  // A value whose scaled form needs more digits than the target precision is rejected.
+  EXPECT_THAT(Literal::Int(12345).CastTo(decimal(4, 0)),
+              IsError(ErrorKind::kInvalidArgument));
+  EXPECT_THAT(Literal::Long(int64_t{100}).CastTo(decimal(4, 2)),
+              IsError(ErrorKind::kInvalidArgument));
+
+  // A scale beyond the decimal powers-of-ten table is rejected rather than read past it
+  // (DecimalType does not bound its scale on construction).
+  EXPECT_THAT(Literal::Int(1).CastTo(std::make_shared<DecimalType>(9, 40)),
+              IsError(ErrorKind::kInvalidArgument));
+}
+
+TEST(LiteralTest, RealCastToDecimal) {
+  // A double is scaled to the target scale keeping its unscaled value: 9.99 ->
+  // 999@(10,2).
+  auto d = Literal::Double(9.99).CastTo(decimal(10, 2));
+  ASSERT_THAT(d, IsOk());
+  EXPECT_EQ(*d, Literal::Decimal(999, 10, 2));
+
+  auto f = Literal::Float(1.5f).CastTo(decimal(4, 3));
+  ASSERT_THAT(f, IsOk());
+  EXPECT_EQ(*f, Literal::Decimal(1500, 4, 3));
+
+  // Rounding is HALF_UP (round half away from zero), matching Java: 2.5 -> 3 (not the
+  // banker's-rounding 2), and -2.5 -> -3.
+  auto half_up = Literal::Double(2.5).CastTo(decimal(2, 0));
+  ASSERT_THAT(half_up, IsOk());
+  EXPECT_EQ(*half_up, Literal::Decimal(3, 2, 0));
+
+  auto neg_half_up = Literal::Double(-2.5).CastTo(decimal(2, 0));
+  ASSERT_THAT(neg_half_up, IsOk());
+  EXPECT_EQ(*neg_half_up, Literal::Decimal(-3, 2, 0));
+
+  // Below the halfway point rounds down.
+  auto round_down = Literal::Double(2.4).CastTo(decimal(2, 0));
+  ASSERT_THAT(round_down, IsOk());
+  EXPECT_EQ(*round_down, Literal::Decimal(2, 2, 0));
+
+  // A value whose rounded form exceeds the target precision is rejected.
+  EXPECT_THAT(Literal::Double(123.4).CastTo(decimal(2, 0)),
+              IsError(ErrorKind::kInvalidArgument));
+  // Non-finite values cannot be represented as a decimal.
+  EXPECT_THAT(
+      Literal::Double(std::numeric_limits<double>::quiet_NaN()).CastTo(decimal(4, 2)),
+      IsError(ErrorKind::kInvalidArgument));
+}
+
 // Error cases for casts
 TEST(LiteralTest, CastToError) {
   std::vector<uint8_t> data = {0x01, 0x02, 0x03, 0x04};