feat: add rescale

Author: zhjwpku

src/iceberg/expression/decimal.cc

@@ -1234,6 +1234,18 @@ static inline uint64_t UInt64FromBigEndian(const uint8_t* bytes, int32_t length)
 #endif
 }
 
+static bool RescaleWouldCauseDataLoss(const Decimal& value, int32_t delta_scale,
+                                      const Decimal& multiplier, Decimal* result) {
+  if (delta_scale < 0) {
+    auto result = value.Divide(multiplier);
+    assert(result);
+    return result->second != 0;
+  }
+
+  *result = value * multiplier;
+  return (value < 0) ? *result > value : *result < value;
+}
+
 }  // namespace
 
 Result<Decimal> Decimal::FromReal(float x, int32_t precision, int32_t scale) {
@@ -1322,7 +1334,23 @@ Result<Decimal> Decimal::Rescale(int32_t orig_scale, int32_t new_scale) const {
     return *this;
   }
 
-  return NotImplemented("Decimal::Rescale is not implemented yet");
+  const int32_t delta_scale = new_scale - orig_scale;
+  const int32_t abs_delta_scale = std::abs(delta_scale);
+  Decimal out;
+
+  assert(abs_delta_scale <= kMaxScale);
+
+  auto& multiplier = kDecimal128PowersOfTen[abs_delta_scale];
+
+  const bool rescale_would_cause_data_loss =
+      RescaleWouldCauseDataLoss(*this, delta_scale, multiplier, &out);
+
+  if (rescale_would_cause_data_loss) {
+    return Invalid("Rescale would cause data loss: {} -> {}", ToIntegerString(),
+                   out.ToIntegerString());
+  }
+
+  return out;
 }
 
 bool Decimal::FitsInPrecision(int32_t precision) const {

test/decimal_test.cc

@@ -558,7 +558,9 @@ TEST(TestDecimalFromReal, FromFloat) {
       // 2**14 - 2**-10
       FromFloatTestParam{
           .real = 16383.999f, .precision = 8, .scale = 3, .expected_string = "16383.999"},
-      FromFloatTestParam{16383.999f, .precision = 19, .scale = 3,
+      FromFloatTestParam{.real = 16383.999f,
+                         .precision = 19,
+                         .scale = 3,
                          .expected_string = "16383.999"},
       // 1 - 2**-24
       FromFloatTestParam{.real = 0.99999994f,
@@ -1291,4 +1293,13 @@ TEST(DecimalTest, Negate) {
   check(Decimal(12, 0xFFFFFFFFFFFFFFFFULL), Decimal(-13, 1));
 }
 
+TEST(DecimalTest, Rescale) {
+  ASSERT_EQ(Decimal(11100), Decimal(111).Rescale(0, 2).value());
+  ASSERT_EQ(Decimal(111), Decimal(11100).Rescale(2, 0).value());
+  ASSERT_EQ(Decimal(5), Decimal(500000).Rescale(6, 1).value());
+  ASSERT_EQ(Decimal(500000), Decimal(5).Rescale(1, 6).value());
+
+  ASSERT_THAT(Decimal(5).Rescale(1, 6), IsError(ErrorKind::kInvalid));
+}
+
 }  // namespace iceberg