[SPARK-54205][CONNECT] Supports Decimal type data in SparkConnectResultSet

### What changes were proposed in this pull request?

Spark connect has supported JDBC protocol with a few commonly used SQL data types. But currently it's missing the support for Decimal data which is also very commonly used to store money objects. I would like to have it support Decimal data type.

### Why are the changes needed?

Right now, a user is able to read Decimal data from SQL by converting the data to string, and then parse the string into Java BigDecimal object. But since JDBC driver is already able to fetch the data as Java BigDecimal type, we can save the effort converting it back and forth. Instead, we just pass through the data we obtain from the raw JDBC result set.

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

It's part of a new feature under Spark connect JDBC support.

### How was this patch tested?

I have created a test new unit test named **'get decimal type'** and it covers my changes. Also the test case aligns with the tests for fetching other data types.

### Was this patch authored or co-authored using generative AI tooling?

No

Closes #52947 from cty123/cty123/support-spark-connect-decimaltype.

Lead-authored-by: cty123 <[email protected]>
Co-authored-by: cty <[email protected]>
Signed-off-by: yangjie01 <[email protected]>

Author: cty123

sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala

@@ -257,11 +257,17 @@ class SparkConnectResultSet(
   override def getCharacterStream(columnLabel: String): Reader =
     throw new SQLFeatureNotSupportedException
 
-  override def getBigDecimal(columnIndex: Int): java.math.BigDecimal =
-    throw new SQLFeatureNotSupportedException
+  override def getBigDecimal(columnIndex: Int): java.math.BigDecimal = {
+    if (currentRow.isNullAt(columnIndex - 1)) {
+      _wasNull = true
+      return null
+    }
+    _wasNull = false
+    currentRow.getDecimal(columnIndex - 1)
+  }
 
   override def getBigDecimal(columnLabel: String): java.math.BigDecimal =
-    throw new SQLFeatureNotSupportedException
+    getBigDecimal(findColumn(columnLabel))
 
   override def isBeforeFirst: Boolean = {
     checkOpen()

sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcTypeUtils.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.connect.client.jdbc.util
 
 import java.lang.{Boolean => JBoolean, Byte => JByte, Double => JDouble, Float => JFloat, Long => JLong, Short => JShort}
+import java.math.{BigDecimal => JBigDecimal}
 import java.sql.{Array => _, _}
 
 import org.apache.spark.sql.types._
@@ -34,6 +35,7 @@ private[jdbc] object JdbcTypeUtils {
     case FloatType => Types.FLOAT
     case DoubleType => Types.DOUBLE
     case StringType => Types.VARCHAR
+    case _: DecimalType => Types.DECIMAL
     case other =>
       throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
   }
@@ -48,12 +50,14 @@ private[jdbc] object JdbcTypeUtils {
     case FloatType => classOf[JFloat].getName
     case DoubleType => classOf[JDouble].getName
     case StringType => classOf[String].getName
+    case _: DecimalType => classOf[JBigDecimal].getName
     case other =>
       throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
   }
 
   def isSigned(field: StructField): Boolean = field.dataType match {
-    case ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType => true
+    case ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType |
+         _: DecimalType => true
     case NullType | BooleanType | StringType => false
     case other =>
       throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
@@ -69,6 +73,7 @@ private[jdbc] object JdbcTypeUtils {
     case FloatType => 7
     case DoubleType => 15
     case StringType => 255
+    case DecimalType.Fixed(p, _) => p
     case other =>
       throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
   }
@@ -77,6 +82,7 @@ private[jdbc] object JdbcTypeUtils {
     case FloatType => 7
     case DoubleType => 15
     case NullType | BooleanType | ByteType | ShortType | IntegerType | LongType | StringType => 0
+    case DecimalType.Fixed(_, s) => s
     case other =>
       throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
   }
@@ -90,6 +96,12 @@ private[jdbc] object JdbcTypeUtils {
     case DoubleType => 24
     case StringType =>
       getPrecision(field)
+    // precision + negative sign + leading zero + decimal point, like DECIMAL(5,5) = -0.12345
+    case DecimalType.Fixed(p, s) if p == s => p + 3
+    // precision + negative sign, like DECIMAL(5,0) = -12345
+    case DecimalType.Fixed(p, s) if s == 0 => p + 1
+    // precision + negative sign + decimal point, like DECIMAL(5,2) = -123.45
+    case DecimalType.Fixed(p, _) => p + 2
     case other =>
       throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
   }

sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectJdbcDataTypeSuite.scala

@@ -215,4 +215,37 @@ class SparkConnectJdbcDataTypeSuite extends ConnectFunSuite with RemoteSparkSess
       assert(metaData.getColumnDisplaySize(1) === 255)
     }
   }
+
+  test("get decimal type") {
+    Seq(
+      ("123.45", 37, 2, 39),
+      ("-0.12345", 5, 5, 8),
+      ("-0.12345", 6, 5, 8),
+      ("-123.45", 5, 2, 7),
+      ("12345", 5, 0, 6),
+      ("-12345", 5, 0, 6)
+    ).foreach {
+      case (value, precision, scale, expectedColumnDisplaySize) =>
+        val decimalType = s"DECIMAL($precision,$scale)"
+        withExecuteQuery(s"SELECT cast('$value' as $decimalType)") { rs =>
+          assert(rs.next())
+          assert(rs.getBigDecimal(1) === new java.math.BigDecimal(value))
+          assert(!rs.wasNull)
+          assert(!rs.next())
+
+          val metaData = rs.getMetaData
+          assert(metaData.getColumnCount === 1)
+          assert(metaData.getColumnName(1) === s"CAST($value AS $decimalType)")
+          assert(metaData.getColumnLabel(1) === s"CAST($value AS $decimalType)")
+          assert(metaData.getColumnType(1) === Types.DECIMAL)
+          assert(metaData.getColumnTypeName(1) === decimalType)
+          assert(metaData.getColumnClassName(1) === "java.math.BigDecimal")
+          assert(metaData.isSigned(1) === true)
+          assert(metaData.getPrecision(1) === precision)
+          assert(metaData.getScale(1) === scale)
+          assert(metaData.getColumnDisplaySize(1) === expectedColumnDisplaySize)
+          assert(metaData.getColumnDisplaySize(1) >= value.size)
+      }
+    }
+  }
 }