[SPARK-54206][CONNECT] Support BINARY type data in SparkConnectResultSet

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

Add BINARY type support to the Spark Connect JDBC client

### Why are the changes needed?

BINARY is a fundamental SQL type required for JDBC compliance and interoperability.

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

Yes, it's part of a new feature under Spark connect JDBC support.

### How was this patch tested?

Added new UTs in  `SparkConnectJdbcDataTypeSuite`

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

No

Closes #53018 from vinodkc/br_SPARK-54206.

Authored-by: vinodkc <vinod.kc.in@gmail.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
(cherry picked from commit ecaec3d)
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: vinodkc

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

@@ -137,8 +137,11 @@ class SparkConnectResultSet(
   override def getBigDecimal(columnIndex: Int, scale: Int): java.math.BigDecimal =
     throw new SQLFeatureNotSupportedException
 
-  override def getBytes(columnIndex: Int): Array[Byte] =
-    throw new SQLFeatureNotSupportedException
+  override def getBytes(columnIndex: Int): Array[Byte] = {
+    getColumnValue(columnIndex, null: Array[Byte]) { idx =>
+      currentRow.get(idx).asInstanceOf[Array[Byte]]
+    }
+  }
 
   override def getDate(columnIndex: Int): Date = {
     getColumnValue(columnIndex, null: Date) { idx => currentRow.getDate(idx) }
@@ -187,7 +190,7 @@ class SparkConnectResultSet(
     throw new SQLFeatureNotSupportedException
 
   override def getBytes(columnLabel: String): Array[Byte] =
-    throw new SQLFeatureNotSupportedException
+    getBytes(findColumn(columnLabel))
 
   override def getDate(columnLabel: String): Date =
     getDate(findColumn(columnLabel))

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

@@ -37,6 +37,7 @@ private[jdbc] object JdbcTypeUtils {
     case StringType => Types.VARCHAR
     case _: DecimalType => Types.DECIMAL
     case DateType => Types.DATE
+    case BinaryType => Types.BINARY
     case other =>
       throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
   }
@@ -53,14 +54,15 @@ private[jdbc] object JdbcTypeUtils {
     case StringType => classOf[String].getName
     case _: DecimalType => classOf[JBigDecimal].getName
     case DateType => classOf[Date].getName
+    case BinaryType => classOf[Array[Byte]].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 |
          _: DecimalType => true
-    case NullType | BooleanType | StringType | DateType => false
+    case NullType | BooleanType | StringType | DateType | BinaryType => false
     case other =>
       throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
   }
@@ -77,6 +79,7 @@ private[jdbc] object JdbcTypeUtils {
     case StringType => 255
     case DecimalType.Fixed(p, _) => p
     case DateType => 10
+    case BinaryType => Int.MaxValue
     case other =>
       throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
   }
@@ -85,7 +88,7 @@ private[jdbc] object JdbcTypeUtils {
     case FloatType => 7
     case DoubleType => 15
     case NullType | BooleanType | ByteType | ShortType | IntegerType | LongType | StringType |
-         DateType => 0
+         DateType | BinaryType => 0
     case DecimalType.Fixed(_, s) => s
     case other =>
       throw new SQLFeatureNotSupportedException(s"DataType $other is not supported yet.")
@@ -101,6 +104,7 @@ private[jdbc] object JdbcTypeUtils {
     case StringType =>
       getPrecision(field)
     case DateType => 10 // length of `YYYY-MM-DD`
+    case BinaryType => Int.MaxValue
     // 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

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

@@ -259,7 +259,8 @@ class SparkConnectJdbcDataTypeSuite extends ConnectFunSuite with RemoteSparkSess
       ("cast(1 AS BIGINT)", (rs: ResultSet) => rs.getLong(999)),
       ("cast(1 AS FLOAT)", (rs: ResultSet) => rs.getFloat(999)),
       ("cast(1 AS DOUBLE)", (rs: ResultSet) => rs.getDouble(999)),
-      ("cast(1 AS DECIMAL(10,5))", (rs: ResultSet) => rs.getBigDecimal(999))
+      ("cast(1 AS DECIMAL(10,5))", (rs: ResultSet) => rs.getBigDecimal(999)),
+      ("CAST(X'0A0B0C' AS BINARY)", (rs: ResultSet) => rs.getBytes(999))
     ).foreach {
       case (query, getter) =>
         withExecuteQuery(s"SELECT $query") { rs =>
@@ -284,13 +285,18 @@ class SparkConnectJdbcDataTypeSuite extends ConnectFunSuite with RemoteSparkSess
       ("cast(1 AS FLOAT)", (rs: ResultSet) => rs.getFloat(1), 1.toFloat),
       ("cast(1 AS DOUBLE)", (rs: ResultSet) => rs.getDouble(1), 1.toDouble),
       ("cast(1 AS DECIMAL(10,5))", (rs: ResultSet) => rs.getBigDecimal(1),
-        new java.math.BigDecimal("1.00000"))
+        new java.math.BigDecimal("1.00000")),
+      ("CAST(X'0A0B0C' AS BINARY)", (rs: ResultSet) => rs.getBytes(1),
+        Array[Byte](0x0A, 0x0B, 0x0C))
     ).foreach {
       case (query, getter, expectedValue) =>
         var resultSet: Option[ResultSet] = None
         withExecuteQuery(s"SELECT $query") { rs =>
           assert(rs.next())
-          assert(getter(rs) === expectedValue)
+          expectedValue match {
+            case arr: Array[Byte] => assert(getter(rs).asInstanceOf[Array[Byte]].sameElements(arr))
+            case other => assert(getter(rs) === other)
+          }
           assert(!rs.wasNull)
           resultSet = Some(rs)
         }
@@ -410,4 +416,70 @@ class SparkConnectJdbcDataTypeSuite extends ConnectFunSuite with RemoteSparkSess
       assert(!rs.next())
     }
   }
+
+  test("get binary type") {
+    val testBytes = Array[Byte](0x01, 0x02, 0x03, 0x04, 0x05)
+    val hexString = testBytes.map(b => "%02X".format(b)).mkString
+    withExecuteQuery(s"SELECT CAST(X'$hexString' AS BINARY)") { rs =>
+      assert(rs.next())
+      val bytes = rs.getBytes(1)
+      assert(bytes !== null)
+      assert(bytes.length === testBytes.length)
+      assert(bytes.sameElements(testBytes))
+      assert(!rs.wasNull)
+      assert(!rs.next())
+
+      val metaData = rs.getMetaData
+      assert(metaData.getColumnCount === 1)
+      assert(metaData.getColumnType(1) === Types.BINARY)
+      assert(metaData.getColumnTypeName(1) === "BINARY")
+      assert(metaData.getColumnClassName(1) === "[B")
+      assert(metaData.isSigned(1) === false)
+    }
+  }
+
+  test("get binary type with null") {
+    withExecuteQuery("SELECT cast(null as binary)") { rs =>
+      assert(rs.next())
+      assert(rs.getBytes(1) === null)
+      assert(rs.wasNull)
+      assert(!rs.next())
+
+      val metaData = rs.getMetaData
+      assert(metaData.getColumnCount === 1)
+      assert(metaData.getColumnType(1) === Types.BINARY)
+      assert(metaData.getColumnTypeName(1) === "BINARY")
+      assert(metaData.getColumnClassName(1) === "[B")
+    }
+  }
+
+  test("get binary type by column label") {
+    val testBytes = Array[Byte](0x0A, 0x0B, 0x0C)
+    val hexString = testBytes.map(b => "%02X".format(b)).mkString
+    withExecuteQuery(s"SELECT CAST(X'$hexString' AS BINARY) as test_binary") { rs =>
+      assert(rs.next())
+      val bytes = rs.getBytes("test_binary")
+      assert(bytes !== null)
+      assert(bytes.length === testBytes.length)
+      assert(bytes.sameElements(testBytes))
+      assert(!rs.wasNull)
+      assert(!rs.next())
+
+      val metaData = rs.getMetaData
+      assert(metaData.getColumnCount === 1)
+      assert(metaData.getColumnName(1) === "test_binary")
+      assert(metaData.getColumnLabel(1) === "test_binary")
+    }
+  }
+
+  test("get empty binary") {
+    withExecuteQuery("SELECT CAST(X'' AS BINARY)") { rs =>
+      assert(rs.next())
+      val bytes = rs.getBytes(1)
+      assert(bytes !== null)
+      assert(bytes.length === 0)
+      assert(!rs.wasNull)
+      assert(!rs.next())
+    }
+  }
 }