int_to_binary

Author: coderfender

docs/source/user-guide/latest/compatibility.md

@@ -106,6 +106,7 @@ The following cast operations are generally compatible with Spark except for the
 | byte | double |  |
 | byte | decimal |  |
 | byte | string |  |
+| byte | binary |  |
 | short | boolean |  |
 | short | byte |  |
 | short | integer |  |
@@ -114,6 +115,7 @@ The following cast operations are generally compatible with Spark except for the
 | short | double |  |
 | short | decimal |  |
 | short | string |  |
+| short | binary |  |
 | integer | boolean |  |
 | integer | byte |  |
 | integer | short |  |
@@ -122,6 +124,7 @@ The following cast operations are generally compatible with Spark except for the
 | integer | double |  |
 | integer | decimal |  |
 | integer | string |  |
+| integer | binary |  |
 | long | boolean |  |
 | long | byte |  |
 | long | short |  |
@@ -130,6 +133,7 @@ The following cast operations are generally compatible with Spark except for the
 | long | double |  |
 | long | decimal |  |
 | long | string |  |
+| long | binary |  |
 | float | boolean |  |
 | float | byte |  |
 | float | short |  |
@@ -160,7 +164,6 @@ The following cast operations are generally compatible with Spark except for the
 | string | long |  |
 | string | float |  |
 | string | double |  |
-| string | binary |  |
 | string | date | Only supports years between 262143 BC and 262142 AD |
 | binary | string |  |
 | date | string |  |

native/spark-expr/src/conversion_funcs/cast.rs

@@ -20,13 +20,13 @@ use crate::{timezone, BinaryOutputStyle};
 use crate::{EvalMode, SparkError, SparkResult};
 use arrow::array::builder::StringBuilder;
 use arrow::array::{
-    BooleanBuilder, Decimal128Builder, DictionaryArray, GenericByteArray, ListArray,
+    BinaryBuilder, BooleanBuilder, Decimal128Builder, DictionaryArray, GenericByteArray, ListArray,
     PrimitiveBuilder, StringArray, StructArray,
 };
 use arrow::compute::can_cast_types;
 use arrow::datatypes::{
-    i256, ArrowDictionaryKeyType, ArrowNativeType, DataType, Decimal256Type, GenericBinaryType,
-    Schema,
+    i256, ArrowDictionaryKeyType, ArrowNativeType, DataType, Decimal256Type,
+    GenericBinaryType, Schema,
 };
 use arrow::{
     array::{
@@ -311,22 +311,22 @@ fn can_cast_from_byte(to_type: &DataType, _: &SparkCastOptions) -> bool {
     use DataType::*;
     matches!(
         to_type,
-        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Decimal128(_, _)
+        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Decimal128(_, _) | Binary
     )
 }
 
 fn can_cast_from_short(to_type: &DataType, _: &SparkCastOptions) -> bool {
     use DataType::*;
     matches!(
         to_type,
-        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Decimal128(_, _)
+        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Decimal128(_, _) | Binary
     )
 }
 
 fn can_cast_from_int(to_type: &DataType, options: &SparkCastOptions) -> bool {
     use DataType::*;
     match to_type {
-        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Utf8 => true,
+        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Utf8 | Binary => true,
         Decimal128(_, _) => {
             // incompatible: no overflow check
             options.allow_incompat
@@ -338,7 +338,7 @@ fn can_cast_from_int(to_type: &DataType, options: &SparkCastOptions) -> bool {
 fn can_cast_from_long(to_type: &DataType, options: &SparkCastOptions) -> bool {
     use DataType::*;
     match to_type {
-        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 => true,
+        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 | Binary => true,
         Decimal128(_, _) => {
             // incompatible: no overflow check
             options.allow_incompat
@@ -501,6 +501,29 @@ macro_rules! cast_float_to_string {
     }};
 }
 
+// eval mode is not needed since all ints can be implemented in binary format
+macro_rules! cast_whole_num_to_binary {
+    ($array:expr, $primitive_type:ty, $byte_size:expr) => {{
+        let input_arr = $array
+            .as_any()
+            .downcast_ref::<$primitive_type>()
+            .ok_or_else(|| SparkError::Internal("Expected numeric array".to_string()))?;
+
+        let len = input_arr.len();
+        let mut builder = BinaryBuilder::with_capacity(len, len * $byte_size);
+
+        for i in 0..input_arr.len() {
+            if input_arr.is_null(i) {
+                builder.append_null();
+            } else {
+                builder.append_value(input_arr.value(i).to_be_bytes());
+            }
+        }
+
+        Ok(Arc::new(builder.finish()) as ArrayRef)
+    }};
+}
+
 macro_rules! cast_int_to_int_macro {
     (
         $array: expr,
@@ -1096,6 +1119,10 @@ fn cast_array(
             Ok(cast_with_options(&array, to_type, &CAST_OPTIONS)?)
         }
         (Binary, Utf8) => Ok(cast_binary_to_string::<i32>(&array, cast_options)?),
+        (Int8, Binary) => cast_whole_num_to_binary!(&array, Int8Array, 1),
+        (Int16, Binary) => cast_whole_num_to_binary!(&array, Int16Array, 2),
+        (Int32, Binary) => cast_whole_num_to_binary!(&array, Int32Array, 4),
+        (Int64, Binary) => cast_whole_num_to_binary!(&array, Int64Array, 8),
         _ if cast_options.is_adapting_schema
             || is_datafusion_spark_compatible(from_type, to_type) =>
         {

spark/src/main/scala/org/apache/comet/expressions/CometCast.scala

@@ -21,7 +21,7 @@ package org.apache.comet.expressions
 
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression, Literal}
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DecimalType, NullType, StructType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, DataType, DataTypes, DecimalType, NullType, StructType}
 
 import org.apache.comet.CometConf
 import org.apache.comet.CometSparkSessionExtensions.withInfo
@@ -126,6 +126,7 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
         isSupported(dt.elementType, DataTypes.StringType, timeZoneId, evalMode)
       case (dt: ArrayType, dt1: ArrayType) =>
         isSupported(dt.elementType, dt1.elementType, timeZoneId, evalMode)
+      case (from: DataType, _: BinaryType) => canCastToBinary(from)
       case (dt: DataType, _) if dt.typeName == "timestamp_ntz" =>
         // https://github.com/apache/datafusion-comet/issues/378
         toType match {
@@ -344,6 +345,12 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
     case _ => Unsupported(Some(s"Cast from DecimalType to $toType is not supported"))
   }
 
+  private def canCastToBinary(fromType: DataType): SupportLevel = fromType match {
+    case DataTypes.ByteType | DataTypes.ShortType | DataTypes.IntegerType | DataTypes.LongType =>
+      Compatible()
+    case _ => Unsupported(Some(s"Cast from BinaryType to $fromType is not supported"))
+  }
+
   private def unsupported(fromType: DataType, toType: DataType): Unsupported = {
     Unsupported(Some(s"Cast from $fromType to $toType is not supported"))
   }

spark/src/test/scala/org/apache/comet/CometCastSuite.scala

@@ -281,7 +281,7 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
       hasIncompatibleType = usingParquetExecWithIncompatTypes)
   }
 
-  ignore("cast ShortType to BinaryType") {
+  test("cast ShortType to BinaryType") {
     castTest(
       generateShorts(),
       DataTypes.BinaryType,