Revert "[SPARK-33428][SQL] Conv UDF use BigInt to avoid Long value overflow"

This reverts commit 5f9a7fe.

Author: cloud-fan

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/NumberConverter.scala

@@ -21,37 +21,64 @@ import org.apache.spark.unsafe.types.UTF8String
 
 object NumberConverter {
 
+  /**
+   * Divide x by m as if x is an unsigned 64-bit integer. Examples:
+   * unsignedLongDiv(-1, 2) == Long.MAX_VALUE unsignedLongDiv(6, 3) == 2
+   * unsignedLongDiv(0, 5) == 0
+   *
+   * @param x is treated as unsigned
+   * @param m is treated as signed
+   */
+  private def unsignedLongDiv(x: Long, m: Int): Long = {
+    if (x >= 0) {
+      x / m
+    } else {
+      // Let uval be the value of the unsigned long with the same bits as x
+      // Two's complement => x = uval - 2*MAX - 2
+      // => uval = x + 2*MAX + 2
+      // Now, use the fact: (a+b)/c = a/c + b/c + (a%c+b%c)/c
+      x / m + 2 * (Long.MaxValue / m) + 2 / m + (x % m + 2 * (Long.MaxValue % m) + 2 % m) / m
+    }
+  }
+
   /**
    * Decode v into value[].
    *
-   * @param v is treated as an BigInt
+   * @param v is treated as an unsigned 64-bit integer
    * @param radix must be between MIN_RADIX and MAX_RADIX
    */
-  private def decode(v: BigInt, radix: Int, value: Array[Byte]): Unit = {
+  private def decode(v: Long, radix: Int, value: Array[Byte]): Unit = {
     var tmpV = v
     java.util.Arrays.fill(value, 0.asInstanceOf[Byte])
     var i = value.length - 1
     while (tmpV != 0) {
-      val q = tmpV / radix
-      value(i) = (tmpV - q * radix).byteValue
+      val q = unsignedLongDiv(tmpV, radix)
+      value(i) = (tmpV - q * radix).asInstanceOf[Byte]
       tmpV = q
       i -= 1
     }
   }
 
   /**
-   * Convert value[] into a BigInt. If a negative digit is found,
-   * ignore the suffix starting there.
+   * Convert value[] into a long. On overflow, return -1 (as mySQL does). If a
+   * negative digit is found, ignore the suffix starting there.
    *
    * @param radix  must be between MIN_RADIX and MAX_RADIX
    * @param fromPos is the first element that should be considered
    * @return the result should be treated as an unsigned 64-bit integer.
    */
-  private def encode(radix: Int, fromPos: Int, value: Array[Byte]): BigInt = {
-    var v: BigInt = BigInt(0)
+  private def encode(radix: Int, fromPos: Int, value: Array[Byte]): Long = {
+    var v: Long = 0L
+    val bound = unsignedLongDiv(-1 - radix, radix) // Possible overflow once
     var i = fromPos
     while (i < value.length && value(i) >= 0) {
-      v = (v * radix) + BigInt(value(i))
+      if (v >= bound) {
+        // Check for overflow
+        if (unsignedLongDiv(-1 - value(i), radix) < v) {
+          return -1
+        }
+      }
+      v = v * radix + value(i)
       i += 1
     }
     v
@@ -102,7 +129,7 @@ object NumberConverter {
       return null
     }
 
-    val (negative, first) = if (n(0) == '-') (true, 1) else (false, 0)
+    var (negative, first) = if (n(0) == '-') (true, 1) else (false, 0)
 
     // Copy the digits in the right side of the array
     val temp = new Array[Byte](64)
@@ -113,8 +140,19 @@ object NumberConverter {
     }
     char2byte(fromBase, temp.length - n.length + first, temp)
 
-    // Do the conversion by going through a BigInt
-    val v: BigInt = encode(fromBase, temp.length - n.length + first, temp)
+    // Do the conversion by going through a 64 bit integer
+    var v = encode(fromBase, temp.length - n.length + first, temp)
+    if (negative && toBase > 0) {
+      if (v < 0) {
+        v = -1
+      } else {
+        v = -v
+      }
+    }
+    if (toBase < 0 && v < 0) {
+      v = -v
+      negative = true
+    }
     decode(v, Math.abs(toBase), temp)
 
     // Find the first non-zero digit or the last digits if all are zero.
@@ -125,7 +163,7 @@ object NumberConverter {
     byte2char(Math.abs(toBase), firstNonZeroPos, temp)
 
     var resultStartPos = firstNonZeroPos
-    if (negative) {
+    if (negative && toBase < 0) {
       resultStartPos = firstNonZeroPos - 1
       temp(resultStartPos) = '-'
     }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/MathExpressionsSuite.scala

@@ -158,7 +158,7 @@ class MathExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
   test("conv") {
     checkEvaluation(Conv(Literal("3"), Literal(10), Literal(2)), "11")
     checkEvaluation(Conv(Literal("-15"), Literal(10), Literal(-16)), "-F")
-    checkEvaluation(Conv(Literal("-15"), Literal(10), Literal(16)), "-F")
+    checkEvaluation(Conv(Literal("-15"), Literal(10), Literal(16)), "FFFFFFFFFFFFFFF1")
     checkEvaluation(Conv(Literal("big"), Literal(36), Literal(16)), "3A48")
     checkEvaluation(Conv(Literal.create(null, StringType), Literal(36), Literal(16)), null)
     checkEvaluation(Conv(Literal("3"), Literal.create(null, IntegerType), Literal(16)), null)
@@ -168,12 +168,10 @@ class MathExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
     checkEvaluation(
       Conv(Literal(""), Literal(10), Literal(16)), null)
     checkEvaluation(
-      Conv(Literal("9223372036854775807"), Literal(36), Literal(16)), "12DDAC15F246BAF8C0D551AC7")
+      Conv(Literal("9223372036854775807"), Literal(36), Literal(16)), "FFFFFFFFFFFFFFFF")
     // If there is an invalid digit in the number, the longest valid prefix should be converted.
     checkEvaluation(
       Conv(Literal("11abc"), Literal(10), Literal(16)), "B")
-    checkEvaluation(Conv(Literal("c8dcdfb41711fc9a1f17928001d7fd61"), Literal(16), Literal(10)),
-      "266992441711411603393340504520074460513")
   }
 
   test("e") {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/NumberConverterSuite.scala

@@ -34,9 +34,9 @@ class NumberConverterSuite extends SparkFunSuite {
   test("convert") {
     checkConv("3", 10, 2, "11")
     checkConv("-15", 10, -16, "-F")
-    checkConv("-15", 10, 16, "-F")
+    checkConv("-15", 10, 16, "FFFFFFFFFFFFFFF1")
     checkConv("big", 36, 16, "3A48")
-    checkConv("9223372036854775807", 36, 16, "12DDAC15F246BAF8C0D551AC7")
+    checkConv("9223372036854775807", 36, 16, "FFFFFFFFFFFFFFFF")
     checkConv("11abc", 10, 16, "B")
   }
 

sql/core/src/test/scala/org/apache/spark/sql/MathFunctionsSuite.scala

@@ -200,7 +200,7 @@ class MathFunctionsSuite extends QueryTest with SharedSparkSession {
     checkAnswer(df.selectExpr("""conv("100", 2, 10)"""), Row("4"))
     checkAnswer(df.selectExpr("""conv("-10", 16, -10)"""), Row("-16"))
     checkAnswer(
-      df.selectExpr("""conv("9223372036854775807", 36, -16)"""), Row("12DDAC15F246BAF8C0D551AC7"))
+      df.selectExpr("""conv("9223372036854775807", 36, -16)"""), Row("-1")) // for overflow
   }
 
   test("floor") {

sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala

@@ -525,9 +525,6 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "udf_xpath_short",
     "udf_xpath_string",
 
-    // [SPARK-33428][SQL] CONV UDF use BigInt to avoid Long value overflow
-    "udf_conv",
-
     // These tests DROP TABLE that don't exist (but do not specify IF EXISTS)
     "alter_rename_partition1",
     "date_1",
@@ -1006,6 +1003,7 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "udf_concat_insert1",
     "udf_concat_insert2",
     "udf_concat_ws",
+    "udf_conv",
     "udf_cos",
     "udf_count",
     "udf_date_add",