[HADP-52545][HADP-45102] Add config not to pad decimal with trailing zeros (apache#243)

* [HADP-45102] Add config not to pad decimal with trailing zeros for compatibility with 2.3.1 (apache#101)

### What changes were proposed in this pull request?
Add config `spark.sql.legacy.decimal.padTrailingZeros` such that padding decimal with trailing zeros can be disabled for spark-sql interface

### Why are the changes needed?
In 2.3.1 decimals are not padded with trailing zeros which is changed in apache#26697. This is to keep compatibility with 2.3.1.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Existing UT updated.

Co-authored-by: tianlzhang <[email protected]>

Author: 2 people

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -3255,6 +3255,14 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val DECIMAL_PAD_TRAILING_ZEROS =
+    buildConf("spark.sql.legacy.decimal.padTrailingZeros")
+      .internal()
+      .doc("Whether to pad decimal numbers with trailing zeros to the scale of the column for " +
+        "spark-sql interface")
+      .booleanConf
+      .createWithDefault(true)
+
   val SQL_OPTIONS_REDACTION_PATTERN = buildConf("spark.sql.redaction.options.regex")
     .doc("Regex to decide which keys in a Spark SQL command's options map contain sensitive " +
       "information. The values of options whose names that match this regex will be redacted " +
@@ -5740,6 +5748,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
 
   def literalPickMinimumPrecision: Boolean = getConf(LITERAL_PICK_MINIMUM_PRECISION)
 
+  def decimalPadTrailingZeros: Boolean = getConf(DECIMAL_PAD_TRAILING_ZEROS)
+
   def continuousStreamingEpochBacklogQueueSize: Int =
     getConf(CONTINUOUS_STREAMING_EPOCH_BACKLOG_QUEUE_SIZE)
 

sql/core/src/main/scala/org/apache/spark/sql/execution/HiveResult.scala

@@ -81,6 +81,14 @@ object HiveResult {
           .map(_.mkString("\t"))
     }
 
+  private def formatDecimal(d: java.math.BigDecimal): String = {
+    if (d.compareTo(java.math.BigDecimal.ZERO) == 0) {
+      java.math.BigDecimal.ZERO.toPlainString
+    } else {
+      d.stripTrailingZeros().toPlainString // Hive strips trailing zeros
+    }
+  }
+
   private def formatDescribeTableOutput(rows: Array[Row]): Seq[String] = {
     rows.map {
       case Row(name: String, dataType: String, comment) =>
@@ -103,7 +111,13 @@ object HiveResult {
     case (i: Instant, TimestampType) => formatters.timestamp.format(i)
     case (l: LocalDateTime, TimestampNTZType) => formatters.timestamp.format(l)
     case (bin: Array[Byte], BinaryType) => new String(bin, StandardCharsets.UTF_8)
-    case (decimal: java.math.BigDecimal, DecimalType()) => decimal.toPlainString
+    case (decimal: java.math.BigDecimal, DecimalType()) =>
+      // HADP-45102: to optionally keep compatibility with 2.3.1 where trailing zeros are stripped
+      if (SQLConf.get.decimalPadTrailingZeros) {
+        decimal.toPlainString
+      } else {
+        formatDecimal(decimal)
+      }
     case (n, _: NumericType) => n.toString
     case (s: String, StringType) => if (nested) "\"" + s + "\"" else s
     case (interval: CalendarInterval, CalendarIntervalType) => interval.toString

sql/core/src/test/scala/org/apache/spark/sql/execution/HiveResultSuite.scala

@@ -65,18 +65,30 @@ class HiveResultSuite extends SharedSparkSession {
   }
 
   test("decimal formatting in hive result") {
-    val df = Seq(new java.math.BigDecimal("1")).toDS()
-    Seq(2, 6, 18).foreach { scala =>
-      val executedPlan =
-        df.selectExpr(s"CAST(value AS decimal(38, $scala))").queryExecution.executedPlan
-      val result = hiveResultString(executedPlan)
-      assert(result.head.split("\\.").last.length === scala)
-    }
+    Seq("true", "false").foreach { padTrailingZeros =>
+      withSQLConf(SQLConf.DECIMAL_PAD_TRAILING_ZEROS.key -> padTrailingZeros) {
+        val df = Seq(new java.math.BigDecimal("1")).toDS()
+        Seq(2, 6, 18).foreach { scale =>
+          val executedPlan =
+            df.selectExpr(s"CAST(value AS decimal(38, $scale))").queryExecution.executedPlan
+          val result = hiveResultString(executedPlan)
+          if (padTrailingZeros.toBoolean) {
+            assert(result.head.split("\\.").last.length === scale)
+          } else {
+            assert(result.head.split("\\.").length == 1)
+          }
+        }
 
-    val executedPlan = Seq(java.math.BigDecimal.ZERO).toDS()
-      .selectExpr(s"CAST(value AS decimal(38, 8))").queryExecution.executedPlan
-    val result = hiveResultString(executedPlan)
-    assert(result.head === "0.00000000")
+        val executedPlan = Seq(java.math.BigDecimal.ZERO).toDS()
+          .selectExpr(s"CAST(value AS decimal(38, 8))").queryExecution.executedPlan
+        val result = hiveResultString(executedPlan)
+        if (padTrailingZeros.toBoolean) {
+          assert(result.head === "0.00000000")
+        } else {
+          assert(result.head === "0")
+        }
+      }
+    }
   }
 
   test("SHOW TABLES in hive result") {