[SPARK-27174][SQL] Add support for casting integer types to binary

Co-authored-by: Philip Stutz <philip.stutzgmail.com>

## What changes were proposed in this pull request?

This PR adds support for casting

* `ByteType`
* `ShortType`
* `IntegerType`
* `LongType`

to `BinaryType`.

## How was this patch tested?

We added unit tests for casting instances of the above types. For validation, we used Javas `DataOutputStream` to compare the resulting byte array with the result of `Cast`.

We state that the contribution is our original work and that we license the work to the project under the project’s open source license.

cloud-fan we'd appreciate a review if you find the time, thx

Closes apache#24107 from s1ck/cast_to_binary.

Authored-by: Martin Junghanns <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: s1ck

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala

@@ -44,6 +44,7 @@ object Cast {
     case (_, StringType) => true
 
     case (StringType, BinaryType) => true
+    case (_: IntegralType, BinaryType) => true
 
     case (StringType, BooleanType) => true
     case (DateType, BooleanType) => true
@@ -326,6 +327,10 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
   // BinaryConverter
   private[this] def castToBinary(from: DataType): Any => Any = from match {
     case StringType => buildCast[UTF8String](_, _.getBytes)
+    case ByteType => buildCast[Byte](_, NumberConverter.toBinary)
+    case ShortType => buildCast[Short](_, NumberConverter.toBinary)
+    case IntegerType => buildCast[Int](_, NumberConverter.toBinary)
+    case LongType => buildCast[Long](_, NumberConverter.toBinary)
   }
 
   // UDFToBoolean
@@ -908,7 +913,11 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
 
   private[this] def castToBinaryCode(from: DataType): CastFunction = from match {
     case StringType =>
-      (c, evPrim, evNull) => code"$evPrim = $c.getBytes();"
+      (c, evPrim, evNull) =>
+        code"$evPrim = $c.getBytes();"
+    case _: IntegralType =>
+      (c, evPrim, evNull) =>
+        code"$evPrim = ${NumberConverter.getClass.getName.stripSuffix("$")}.toBinary($c);"
   }
 
   private[this] def castToDateCode(

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

@@ -169,4 +169,39 @@ object NumberConverter {
     }
     UTF8String.fromBytes(java.util.Arrays.copyOfRange(temp, resultStartPos, temp.length))
   }
+
+  def toBinary(l: Long): Array[Byte] = {
+    val result = new Array[Byte](8)
+    result(0) = (l >>> 56 & 0xFF).toByte
+    result(1) = (l >>> 48 & 0xFF).toByte
+    result(2) = (l >>> 40 & 0xFF).toByte
+    result(3) = (l >>> 32 & 0xFF).toByte
+    result(4) = (l >>> 24 & 0xFF).toByte
+    result(5) = (l >>> 16 & 0xFF).toByte
+    result(6) = (l >>> 8 & 0xFF).toByte
+    result(7) = (l & 0xFF).toByte
+    result
+  }
+
+  def toBinary(i: Int): Array[Byte] = {
+    val result = new Array[Byte](4)
+    result(0) = (i >>> 24 & 0xFF).toByte
+    result(1) = (i >>> 16 & 0xFF).toByte
+    result(2) = (i >>> 8 & 0xFF).toByte
+    result(3) = (i & 0xFF).toByte
+    result
+  }
+
+  def toBinary(s: Short): Array[Byte] = {
+    val result = new Array[Byte](2)
+    result(0) = (s >>> 8 & 0xFF).toByte
+    result(1) = (s & 0xFF).toByte
+    result
+  }
+
+  def toBinary(s: Byte): Array[Byte] = {
+    val result = new Array[Byte](1)
+    result(0) = s
+    result
+  }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala

@@ -215,11 +215,6 @@ class AnalysisErrorSuite extends AnalysisTest {
     testRelation2.groupBy('a)(sum(UnresolvedStar(None))),
     "Invalid usage of '*'" :: "in expression 'sum'" :: Nil)
 
-  errorTest(
-    "bad casts",
-    testRelation.select(Literal(1).cast(BinaryType).as('badCast)),
-  "cannot cast" :: Literal(1).dataType.simpleString :: BinaryType.simpleString :: Nil)
-
   errorTest(
     "sorting by unsupported column types",
     mapRelation.orderBy('map.asc),

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

@@ -17,8 +17,11 @@
 
 package org.apache.spark.sql.catalyst.util
 
+import java.nio.ByteBuffer
+import java.nio.ByteOrder.BIG_ENDIAN
+
 import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.catalyst.util.NumberConverter.convert
+import org.apache.spark.sql.catalyst.util.NumberConverter.{convert, toBinary}
 import org.apache.spark.unsafe.types.UTF8String
 
 class NumberConverterSuite extends SparkFunSuite {
@@ -37,4 +40,47 @@ class NumberConverterSuite extends SparkFunSuite {
     checkConv("11abc", 10, 16, "B")
   }
 
+  test("byte to binary") {
+    checkToBinary(0.toByte)
+    checkToBinary(1.toByte)
+    checkToBinary(-1.toByte)
+    checkToBinary(Byte.MaxValue)
+    checkToBinary(Byte.MinValue)
+  }
+
+  test("short to binary") {
+    checkToBinary(0.toShort)
+    checkToBinary(1.toShort)
+    checkToBinary(-1.toShort)
+    checkToBinary(Short.MaxValue)
+    checkToBinary(Short.MinValue)
+  }
+
+  test("integer to binary") {
+    checkToBinary(0)
+    checkToBinary(1)
+    checkToBinary(-1)
+    checkToBinary(Int.MaxValue)
+    checkToBinary(Int.MinValue)
+  }
+
+  test("long to binary") {
+    checkToBinary(0L)
+    checkToBinary(1L)
+    checkToBinary(-1L)
+    checkToBinary(Long.MaxValue)
+    checkToBinary(Long.MinValue)
+  }
+
+  def checkToBinary[T](in: T): Unit = in match {
+    case b: Byte =>
+      assert(toBinary(b) === ByteBuffer.allocate(1).order(BIG_ENDIAN).put(b).array())
+    case s: Short =>
+      assert(toBinary(s) === ByteBuffer.allocate(2).order(BIG_ENDIAN).putShort(s).array())
+    case i: Int =>
+      assert(toBinary(i) === ByteBuffer.allocate(4).order(BIG_ENDIAN).putInt(i).array())
+    case l: Long =>
+      assert(toBinary(l) === ByteBuffer.allocate(8).order(BIG_ENDIAN).putLong(l).array())
+  }
+
 }

sql/core/src/test/resources/sql-tests/inputs/cast.sql

@@ -40,6 +40,19 @@ SELECT CAST('-9223372036854775809' AS long);
 SELECT CAST('9223372036854775807' AS long);
 SELECT CAST('9223372036854775808' AS long);
 
+-- cast string to its binary representation
+SELECT HEX(CAST('abc' AS binary));
+
+-- cast integral values to their corresponding binary representation
+SELECT HEX(CAST(CAST(123 AS byte) AS binary));
+SELECT HEX(CAST(CAST(-123 AS byte) AS binary));
+SELECT HEX(CAST(123S AS binary));
+SELECT HEX(CAST(-123S AS binary));
+SELECT HEX(CAST(123 AS binary));
+SELECT HEX(CAST(-123 AS binary));
+SELECT HEX(CAST(123L AS binary));
+SELECT HEX(CAST(-123L AS binary));
+
 DESC FUNCTION boolean;
 DESC FUNCTION EXTENDED boolean;
 -- TODO: migrate all cast tests here.

sql/core/src/test/resources/sql-tests/results/cast.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 24
+-- Number of queries: 33
 
 
 -- !query 0
@@ -179,20 +179,92 @@ NULL
 
 
 -- !query 22
-DESC FUNCTION boolean
+SELECT HEX(CAST('abc' AS binary))
 -- !query 22 schema
-struct<function_desc:string>
+struct<hex(CAST(abc AS BINARY)):string>
 -- !query 22 output
+616263
+
+
+-- !query 23
+SELECT HEX(CAST(CAST(123 AS byte) AS binary))
+-- !query 23 schema
+struct<hex(CAST(CAST(123 AS TINYINT) AS BINARY)):string>
+-- !query 23 output
+7B
+
+
+-- !query 24
+SELECT HEX(CAST(CAST(-123 AS byte) AS binary))
+-- !query 24 schema
+struct<hex(CAST(CAST(-123 AS TINYINT) AS BINARY)):string>
+-- !query 24 output
+85
+
+
+-- !query 25
+SELECT HEX(CAST(123S AS binary))
+-- !query 25 schema
+struct<hex(CAST(123 AS BINARY)):string>
+-- !query 25 output
+007B
+
+
+-- !query 26
+SELECT HEX(CAST(-123S AS binary))
+-- !query 26 schema
+struct<hex(CAST(-123 AS BINARY)):string>
+-- !query 26 output
+FF85
+
+
+-- !query 27
+SELECT HEX(CAST(123 AS binary))
+-- !query 27 schema
+struct<hex(CAST(123 AS BINARY)):string>
+-- !query 27 output
+0000007B
+
+
+-- !query 28
+SELECT HEX(CAST(-123 AS binary))
+-- !query 28 schema
+struct<hex(CAST(-123 AS BINARY)):string>
+-- !query 28 output
+FFFFFF85
+
+
+-- !query 29
+SELECT HEX(CAST(123L AS binary))
+-- !query 29 schema
+struct<hex(CAST(123 AS BINARY)):string>
+-- !query 29 output
+000000000000007B
+
+
+-- !query 30
+SELECT HEX(CAST(-123L AS binary))
+-- !query 30 schema
+struct<hex(CAST(-123 AS BINARY)):string>
+-- !query 30 output
+FFFFFFFFFFFFFF85
+
+
+-- !query 31
+DESC FUNCTION boolean
+-- !query 31 schema
+struct<function_desc:string>
+-- !query 31 output
 Class: org.apache.spark.sql.catalyst.expressions.Cast
 Function: boolean
 Usage: boolean(expr) - Casts the value `expr` to the target data type `boolean`.
 
 
--- !query 23
+-- !query 32
 DESC FUNCTION EXTENDED boolean
--- !query 23 schema
+-- !query 32 schema
 struct<function_desc:string>
--- !query 23 output
+-- !query 32 output
 Class: org.apache.spark.sql.catalyst.expressions.Cast
 Extended Usage:
     No example/argument for boolean.