GH-55340: [C++] Check out-of-range decimal scales before indexing powers-of-ten tables

cpp/src/arrow/acero/hash_aggregate_test.cc

@@ -956,6 +956,28 @@ TEST_P(GroupBy, SumMeanProductDecimal) {
   }
 }
 
+TEST_P(GroupBy, ProductDecimalScaleOutOfRange) {
+  // GH-55340: the product is rescaled by the decimal type's scale, which must be
+  // within [0, kMaxScale]
+  auto key = ArrayFromJSON(int64(), "[1, 2]");
+  for (const auto& [unscaled_type, max_scale] :
+       std::vector<std::pair<std::shared_ptr<DataType>, int32_t>>{
+           {decimal128(38, 0), 38}, {decimal256(76, 0), 76}}) {
+    auto unscaled = ArrayFromJSON(unscaled_type, R"(["2", "3"])");
+    for (int32_t scale : {-1, max_scale + 1}) {
+      ASSERT_OK_AND_ASSIGN(auto type,
+                           DecimalType::Make(unscaled_type->id(), max_scale, scale));
+      ASSERT_OK_AND_ASSIGN(auto values, unscaled->View(type));
+      for (bool use_threads : {true, false}) {
+        SCOPED_TRACE(use_threads ? "parallel/merged" : "serial");
+        EXPECT_RAISES_WITH_MESSAGE_THAT(
+            NotImplemented, HasSubstr("scale outside"),
+            GroupByTest({values}, {key}, {{"hash_product", nullptr}}, use_threads));
+      }
+    }
+  }
+}
+
 TEST_P(GroupBy, MeanOnly) {
   for (bool use_threads : {true, false}) {
     SCOPED_TRACE(use_threads ? "parallel/merged" : "serial");

cpp/src/arrow/compute/kernels/aggregate_basic.cc

@@ -425,6 +425,8 @@ struct ProductInit {
 
   template <typename Type>
   enable_if_decimal<Type, Status> Visit(const Type&) {
+    using AccType = typename ProductImpl<Type>::AccType;
+    RETURN_NOT_OK(MultiplyTraits<AccType>::CheckType(*type));
     state.reset(new ProductImpl<Type>(type, options));
     return Status::OK();
   }

cpp/src/arrow/compute/kernels/aggregate_internal.h

@@ -79,6 +79,8 @@ template <typename Type, typename Enable = void>
 struct MultiplyTraits {
   using CType = typename TypeTraits<Type>::CType;
 
+  static Status CheckType(const DataType&) { return Status::OK(); }
+
   constexpr static CType one(const DataType&) { return static_cast<CType>(1); }
 
   constexpr static CType Multiply(const DataType&, CType lhs, CType rhs) {
@@ -90,6 +92,17 @@ template <typename Type>
 struct MultiplyTraits<Type, enable_if_decimal<Type>> {
   using CType = typename TypeTraits<Type>::CType;
 
+  // one() and Multiply() rescale by the type's scale, which IncreaseScaleBy()
+  // and ReduceScaleBy() only accept in [0, kMaxScale]
+  static Status CheckType(const DataType& ty) {
+    const int32_t scale = static_cast<const Type&>(ty).scale();
+    if (scale < 0 || scale > CType::kMaxScale) {
+      return Status::NotImplemented("Product of decimal type with scale outside [0, ",
+                                    CType::kMaxScale, "]: ", ty);
+    }
+    return Status::OK();
+  }
+
   constexpr static CType one(const DataType& ty) {
     // Return 1 scaled to output type scale
     return CType(1).IncreaseScaleBy(static_cast<const Type&>(ty).scale());

cpp/src/arrow/compute/kernels/aggregate_test.cc

@@ -846,6 +846,23 @@ TEST(TestDecimalProductKernel, ScalarAggregateOptions) {
   }
 }
 
+TEST(TestDecimalProductKernel, ScaleOutOfRange) {
+  // GH-55340: the product is rescaled by the decimal type's scale, which must be
+  // within [0, kMaxScale]
+  for (const auto& [unscaled_type, max_scale] :
+       std::vector<std::pair<std::shared_ptr<DataType>, int32_t>>{
+           {decimal128(38, 0), 38}, {decimal256(76, 0), 76}}) {
+    auto unscaled = ArrayFromJSON(unscaled_type, R"(["2", "3"])");
+    for (int32_t scale : {-1, max_scale + 1}) {
+      ASSERT_OK_AND_ASSIGN(auto type,
+                           DecimalType::Make(unscaled_type->id(), max_scale, scale));
+      ASSERT_OK_AND_ASSIGN(auto values, unscaled->View(type));
+      EXPECT_RAISES_WITH_MESSAGE_THAT(
+          NotImplemented, ::testing::HasSubstr("scale outside"), Product(values));
+    }
+  }
+}
+
 TEST(TestProductKernel, Overflow) {
   EXPECT_THAT(Product(ArrayFromJSON(int64(), "[8589934592, 8589934593]")),
               ResultWith(Datum(static_cast<int64_t>(8589934592))));

cpp/src/arrow/compute/kernels/hash_aggregate_numeric.cc

@@ -315,6 +315,11 @@ struct GroupedProductImpl final
   using CType = typename Base::CType;
   using InputCType = typename Base::InputCType;
 
+  Status Init(ExecContext* ctx, const KernelInitArgs& args) override {
+    RETURN_NOT_OK(MultiplyTraits<AccType>::CheckType(*args.inputs[0]));
+    return Base::Init(ctx, args);
+  }
+
   static CType NullValue(const DataType& out_type) {
     return MultiplyTraits<AccType>::one(out_type);
   }

cpp/src/arrow/compute/kernels/scalar_cast_numeric.cc

@@ -335,6 +335,38 @@ struct CastFunctor<HalfFloatType, StringType, enable_if_t<true>> {
   }
 };
 
+// ----------------------------------------------------------------------
+// Decimal rescaling for truncating casts
+
+// IncreaseScaleBy() and ReduceScaleBy() look up a table of powers of ten and only
+// accept a scale change in [0, kMaxScale].  A scale change computed from decimal
+// types or from a parsed decimal string is not bounded this way (a decimal type's
+// scale can be any int32_t value), so it must be checked first.
+
+template <typename Decimal>
+Decimal UnsafeIncreaseScaleBy(const Decimal& val, int64_t increase_by, Status* st) {
+  DCHECK_GE(increase_by, 0);
+  if (ARROW_PREDICT_FALSE(increase_by > Decimal::kMaxScale)) {
+    // Any non-zero value overflows
+    if (val != Decimal{}) {
+      *st = Status::Invalid("Rescaling decimal value by ", increase_by,
+                            " digits would overflow");
+    }
+    return Decimal{};
+  }
+  return val.IncreaseScaleBy(static_cast<int32_t>(increase_by));
+}
+
+template <typename Decimal>
+Decimal UnsafeReduceScaleBy(const Decimal& val, int64_t reduce_by) {
+  DCHECK_GE(reduce_by, 0);
+  if (ARROW_PREDICT_FALSE(reduce_by > Decimal::kMaxScale)) {
+    // Any value is less than 10^(kMaxScale + 1) in magnitude, so it truncates to zero
+    return Decimal{};
+  }
+  return val.ReduceScaleBy(static_cast<int32_t>(reduce_by), /*round=*/false);
+}
+
 // ----------------------------------------------------------------------
 // Decimal to integer
 
@@ -364,7 +396,8 @@ struct UnsafeUpscaleDecimalToInteger : public DecimalToIntegerMixin {
 
   template <typename OutValue, typename Arg0Value>
   OutValue Call(KernelContext* ctx, Arg0Value val, Status* st) const {
-    return ToInteger<OutValue>(ctx, val.IncreaseScaleBy(-in_scale_), st);
+    return ToInteger<OutValue>(
+        ctx, UnsafeIncreaseScaleBy(val, -static_cast<int64_t>(in_scale_), st), st);
   }
 };
 
@@ -373,7 +406,7 @@ struct UnsafeDownscaleDecimalToInteger : public DecimalToIntegerMixin {
 
   template <typename OutValue, typename Arg0Value>
   OutValue Call(KernelContext* ctx, Arg0Value val, Status* st) const {
-    return ToInteger<OutValue>(ctx, val.ReduceScaleBy(in_scale_, false), st);
+    return ToInteger<OutValue>(ctx, UnsafeReduceScaleBy(val, in_scale_), st);
   }
 };
 
@@ -553,21 +586,22 @@ struct DecimalConversions<Decimal128, Decimal32> {
 
 struct UnsafeUpscaleDecimal {
   template <typename OutValue, typename Arg0Value>
-  OutValue Call(KernelContext*, Arg0Value val, Status*) const {
+  OutValue Call(KernelContext*, Arg0Value val, Status* st) const {
     using Conv = DecimalConversions<OutValue, Arg0Value>;
-    return Conv::ConvertOutput(Conv::ConvertInput(std::move(val)).IncreaseScaleBy(by_));
+    return Conv::ConvertOutput(
+        UnsafeIncreaseScaleBy(Conv::ConvertInput(std::move(val)), by_, st));
   }
-  int32_t by_;
+  int64_t by_;
 };
 
 struct UnsafeDownscaleDecimal {
   template <typename OutValue, typename Arg0Value>
   OutValue Call(KernelContext*, Arg0Value val, Status*) const {
     using Conv = DecimalConversions<OutValue, Arg0Value>;
     return Conv::ConvertOutput(
-        Conv::ConvertInput(std::move(val)).ReduceScaleBy(by_, false));
+        UnsafeReduceScaleBy(Conv::ConvertInput(std::move(val)), by_));
   }
-  int32_t by_;
+  int64_t by_;
 };
 
 struct SafeRescaleDecimal {
@@ -604,15 +638,17 @@ struct CastFunctor<O, I,
     const auto out_scale = out_type.scale();
 
     if (options.allow_decimal_truncate) {
-      if (in_scale < out_scale) {
+      // Computed in 64 bits as the scales can be any int32_t values
+      const int64_t delta_scale = static_cast<int64_t>(out_scale) - in_scale;
+      if (delta_scale > 0) {
         // Unsafe upscale
         applicator::ScalarUnaryNotNullStateful<O, I, UnsafeUpscaleDecimal> kernel(
-            UnsafeUpscaleDecimal{out_scale - in_scale});
+            UnsafeUpscaleDecimal{delta_scale});
         return kernel.Exec(ctx, batch, out);
       } else {
         // Unsafe downscale
         applicator::ScalarUnaryNotNullStateful<O, I, UnsafeDownscaleDecimal> kernel(
-            UnsafeDownscaleDecimal{in_scale - out_scale});
+            UnsafeDownscaleDecimal{-delta_scale});
         return kernel.Exec(ctx, batch, out);
       }
     }
@@ -675,9 +711,10 @@ struct StringToDecimal {
 
     if (ARROW_PREDICT_TRUE(r_parse.ok())) {
       if (allow_truncate_) {
-        return (parsed_scale < out_scale_)
-                   ? parsed_out.IncreaseScaleBy(out_scale_ - parsed_scale)
-                   : parsed_out.ReduceScaleBy(parsed_scale - out_scale_, false);
+        // The parsed scale comes from the string's exponent and can be arbitrarily large
+        const int64_t delta_scale = static_cast<int64_t>(out_scale_) - parsed_scale;
+        return (delta_scale > 0) ? UnsafeIncreaseScaleBy(parsed_out, delta_scale, st)
+                                 : UnsafeReduceScaleBy(parsed_out, -delta_scale);
       }
 
       auto maybe_rescaled = parsed_out.Rescale(parsed_scale, out_scale_);

cpp/src/arrow/compute/kernels/scalar_cast_test.cc

@@ -18,6 +18,7 @@
 #include <cstdint>
 #include <cstdio>
 #include <functional>
+#include <limits>
 #include <memory>
 #include <string>
 #include <vector>
@@ -83,6 +84,20 @@ static std::vector<std::shared_ptr<DataType>> kBaseBinaryTypes = {
 static std::vector<std::shared_ptr<DataType>> kBaseBinaryAndViewTypes = {
     binary(), utf8(), large_binary(), large_utf8(), utf8_view(), binary_view()};
 
+// The decimal type of the same width and precision with another scale, which may be
+// out of the range that DecimalXXX::FromString() can produce
+static std::shared_ptr<DataType> DecimalTypeWithScale(
+    const std::shared_ptr<DataType>& type, int32_t scale) {
+  const auto& decimal_type = checked_cast<const DecimalType&>(*type);
+  return DecimalType::Make(type->id(), decimal_type.precision(), scale).ValueOrDie();
+}
+
+// The same unscaled decimal values with another scale
+static std::shared_ptr<Array> DecimalViewWithScale(const std::shared_ptr<Array>& array,
+                                                   int32_t scale) {
+  return array->View(DecimalTypeWithScale(array->type(), scale)).ValueOrDie();
+}
+
 static void AssertBufferSame(const Array& left, const Array& right, size_t buffer_index) {
   ASSERT_EQ(left.data()->buffers[buffer_index].get(),
             right.data()->buffers[buffer_index].get());
@@ -764,6 +779,33 @@ TEST(Cast, Decimal256ToInt) {
   CheckCast(negative_scale, ArrayFromJSON(int64(), "[1234567890000, -120000]"), options);
 }
 
+TEST(Cast, DecimalToIntTruncateScaleOutOfRange) {
+  // GH-55340: a decimal type's scale is not bounded by kMaxScale, so neither is the
+  // scale change of a truncating cast
+  auto options = CastOptions::Unsafe(int64());
+  for (const auto& [unscaled_type, max_scale] :
+       std::vector<std::pair<std::shared_ptr<DataType>, int32_t>>{
+           {decimal32(9, 0), 9},
+           {decimal64(18, 0), 18},
+           {decimal128(38, 0), 38},
+           {decimal256(76, 0), 76}}) {
+    auto unscaled = ArrayFromJSON(unscaled_type, R"(["12", "-3", "0", null])");
+    auto zero = ArrayFromJSON(unscaled_type, R"(["0", null])");
+    // Downscaling truncates to zero
+    for (int32_t scale :
+         {max_scale, max_scale + 1, std::numeric_limits<int32_t>::max()}) {
+      CheckCast(DecimalViewWithScale(unscaled, scale),
+                ArrayFromJSON(int64(), "[0, 0, 0, null]"), options);
+    }
+    // Upscaling overflows, except for zero
+    for (int32_t scale : {-max_scale - 1, std::numeric_limits<int32_t>::min()}) {
+      CheckCastFails(DecimalViewWithScale(unscaled, scale), options);
+      CheckCast(DecimalViewWithScale(zero, scale), ArrayFromJSON(int64(), "[0, null]"),
+                options);
+    }
+  }
+}
+
 TEST(Cast, IntegerToDecimal) {
   for (auto decimal_type :
        {decimal32(9, 2), decimal64(18, 2), decimal128(22, 2), decimal256(22, 2)}) {
@@ -2145,6 +2187,31 @@ TEST(Cast, Decimal256ToDecimal128) {
   }
 }
 
+TEST(Cast, DecimalToDecimalTruncateScaleOutOfRange) {
+  // GH-55340: a decimal type's scale is not bounded by kMaxScale, so neither is the
+  // scale change of a truncating cast
+  auto options = CastOptions::Unsafe();
+  const std::vector<std::shared_ptr<DataType>> unscaled_types = {
+      decimal32(9, 0), decimal64(18, 0), decimal128(38, 0), decimal256(76, 0)};
+  for (const auto& in_type : unscaled_types) {
+    auto unscaled = ArrayFromJSON(in_type, R"(["12", "-3", "0", null])");
+    for (const auto& out_type : unscaled_types) {
+      // Downscaling by more than kMaxScale truncates to zero
+      for (int32_t scale : {77, std::numeric_limits<int32_t>::max()}) {
+        CheckCast(DecimalViewWithScale(unscaled, scale),
+                  ArrayFromJSON(out_type, R"(["0", "0", "0", null])"), options);
+      }
+      // Upscaling by more than kMaxScale overflows
+      options.to_type = DecimalTypeWithScale(out_type, 77);
+      CheckCastFails(unscaled, options);
+      options.to_type =
+          DecimalTypeWithScale(out_type, std::numeric_limits<int32_t>::max());
+      CheckCastFails(DecimalViewWithScale(unscaled, std::numeric_limits<int32_t>::min()),
+                     options);
+    }
+  }
+}
+
 TEST(Cast, FloatingToDecimal) {
   for (auto float_type : {float32(), float64()}) {
     for (auto decimal_type :
@@ -3133,6 +3200,24 @@ TEST(Cast, StringToDecimal) {
   }
 }
 
+TEST(Cast, StringToDecimalTruncateScaleOutOfRange) {
+  // GH-55340: the scale of a decimal string comes from its exponent and is not
+  // bounded by kMaxScale
+  auto options = CastOptions::Unsafe();
+  for (auto string_type : {utf8(), large_utf8()}) {
+    for (auto decimal_type :
+         {decimal32(9, 0), decimal64(18, 0), decimal128(38, 0), decimal256(76, 0)}) {
+      // Downscaling by more than kMaxScale truncates to zero
+      CheckCast(ArrayFromJSON(string_type,
+                              R"(["1E-77", "-12E-1000000000", "1E-2147483647", null])"),
+                ArrayFromJSON(decimal_type, R"(["0", "0", "0", null])"), options);
+      // Upscaling by more than kMaxScale overflows
+      options.to_type = DecimalTypeWithScale(decimal_type, 77);
+      CheckCastFails(ArrayFromJSON(string_type, R"(["1"])"), options);
+    }
+  }
+}
+
 TEST(Cast, StringToTimestamp) {
   for (auto string_type : {utf8(), large_utf8()}) {
     auto strings = ArrayFromJSON(string_type, R"(["1970-01-01", null, "2000-02-29"])");