From 6608e918ce0cb40dd8c17947c308c2c2a09551ad Mon Sep 17 00:00:00 2001
From: pprudhvi <30983614+pprudhvi@users.noreply.github.com>
Date: Tue, 21 Jan 2020 17:30:00 +0530
Subject: [PATCH] =?UTF-8?q?ARROW-7631:=20return=20zero=20if=20there=20is?=
 =?UTF-8?q?=20an=20overflow=20while=20downscaling=20a=20d=E2=80=A6=20(#13)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* ARROW-7631: return zero if there is an overflow while downscaling a decimal

  - also add castDecimalNullOnOverflow function

* inline helper function
---
 .../gandiva/function_registry_arithmetic.cc   |  4 +++
 cpp/src/gandiva/precompiled/decimal_ops.cc    |  5 ++-
 .../gandiva/precompiled/decimal_wrapper.cc    | 35 +++++++++++++++++--
 3 files changed, 40 insertions(+), 4 deletions(-)

diff --git a/cpp/src/gandiva/function_registry_arithmetic.cc b/cpp/src/gandiva/function_registry_arithmetic.cc
index 7f11d9e1bc666..f098533243c9d 100644
--- a/cpp/src/gandiva/function_registry_arithmetic.cc
+++ b/cpp/src/gandiva/function_registry_arithmetic.cc
@@ -56,6 +56,10 @@ std::vector<NativeFunction> GetArithmeticFunctionRegistry() {
       UNARY_SAFE_NULL_IF_NULL(castDECIMAL, {}, decimal128, decimal128),
       UNARY_UNSAFE_NULL_IF_NULL(castDECIMAL, {}, utf8, decimal128),
 
+      NativeFunction("castDECIMALNullOnOverflow", {}, DataTypeVector{decimal128()},
+                     decimal128(), kResultNullInternal,
+                     "castDECIMALNullOnOverflow_decimal128"),
+
       UNARY_SAFE_NULL_IF_NULL(castDATE, {}, int64, date64),
 
       // add/sub/multiply/divide/mod
diff --git a/cpp/src/gandiva/precompiled/decimal_ops.cc b/cpp/src/gandiva/precompiled/decimal_ops.cc
index 902c8d799aebe..494c3a0bc6089 100644
--- a/cpp/src/gandiva/precompiled/decimal_ops.cc
+++ b/cpp/src/gandiva/precompiled/decimal_ops.cc
@@ -544,7 +544,10 @@ static BasicDecimal128 ModifyScaleAndPrecision(const BasicDecimalScalar128& x,
     return x.value().IncreaseScaleBy(delta_scale);
   } else {
     // Do not do any rounding, that is handled by the caller.
-    return x.value().ReduceScaleBy(-delta_scale, false);
+    auto result = x.value().ReduceScaleBy(-delta_scale, false);
+    DECIMAL_OVERFLOW_IF(BasicDecimal128::Abs(result) > GetMaxValue(out_precision),
+                        overflow);
+    return result;
   }
 }
 
diff --git a/cpp/src/gandiva/precompiled/decimal_wrapper.cc b/cpp/src/gandiva/precompiled/decimal_wrapper.cc
index 66ccb3e0ddbd1..e4102257658e0 100644
--- a/cpp/src/gandiva/precompiled/decimal_wrapper.cc
+++ b/cpp/src/gandiva/precompiled/decimal_wrapper.cc
@@ -229,14 +229,43 @@ void castDECIMAL_float32(float in, int32_t x_precision, int32_t x_scale,
 }
 
 FORCE_INLINE
-void castDECIMAL_decimal128(int64_t x_high, uint64_t x_low, int32_t x_precision,
-                            int32_t x_scale, int32_t out_precision, int32_t out_scale,
-                            int64_t* out_high, int64_t* out_low) {
+boolean castDecimal_return_overflow(int64_t x_high, uint64_t x_low, int32_t x_precision,
+                                    int32_t x_scale, int32_t out_precision,
+                                    int32_t out_scale, int64_t* out_high,
+                                    int64_t* out_low) {
   gandiva::BasicDecimalScalar128 x({x_high, x_low}, x_precision, x_scale);
   bool overflow = false;
   auto out = gandiva::decimalops::Convert(x, out_precision, out_scale, &overflow);
   *out_high = out.high_bits();
   *out_low = out.low_bits();
+  return overflow;
+}
+
+FORCE_INLINE
+void castDECIMAL_decimal128(int64_t x_high, uint64_t x_low, int32_t x_precision,
+                            int32_t x_scale, int32_t out_precision, int32_t out_scale,
+                            int64_t* out_high, int64_t* out_low) {
+  castDecimal_return_overflow(x_high, x_low, x_precision, x_scale, out_precision,
+                              out_scale, out_high, out_low);
+}
+
+FORCE_INLINE
+void castDECIMALNullOnOverflow_decimal128(int64_t x_high, uint64_t x_low,
+                                          int32_t x_precision, int32_t x_scale,
+                                          boolean x_isvalid, bool* out_valid,
+                                          int32_t out_precision, int32_t out_scale,
+                                          int64_t* out_high, int64_t* out_low) {
+  *out_valid = true;
+
+  if (!x_isvalid) {
+    *out_valid = false;
+    return;
+  }
+
+  if (castDecimal_return_overflow(x_high, x_low, x_precision, x_scale, out_precision,
+                                  out_scale, out_high, out_low)) {
+    *out_valid = false;
+  }
 }
 
 FORCE_INLINE