[BLAZE-760] Fallback shuffle exchange when range partitioning with unsupported type

spark-extension-shims-spark3/src/test/scala/org/apache/spark/sql/blaze/BlazeQuerySuite.scala

@@ -81,6 +81,17 @@ class BlazeQuerySuite extends org.apache.spark.sql.QueryTest with BaseBlazeSQLSu
       })
   }
 
+  test("binary type in range partitioning") {
+    withTable("t1", "t2") {
+      sql("create table t1(c1 binary, c2 int) using parquet")
+      sql(
+        "insert into t1 values (to_binary('test1', 'utf-8'), 1), (to_binary('test2', 'utf-8'), 2)")
+      spark.table("t1").printSchema()
+      val df = sql("select c2 from t1 order by c1")
+      checkAnswer(df, Seq(Row(1), Row(2)))
+    }
+  }
+
   test("log function with negative input") {
     withTable("t1") {
       sql("create table t1 using parquet as select -1 as c1")

spark-extension/src/main/scala/org/apache/spark/sql/blaze/BlazeConverters.scala

@@ -29,7 +29,7 @@ import org.apache.spark.sql.blaze.BlazeConvertStrategy.convertStrategyTag
 import org.apache.spark.sql.blaze.BlazeConvertStrategy.convertToNonNativeTag
 import org.apache.spark.sql.blaze.BlazeConvertStrategy.isNeverConvert
 import org.apache.spark.sql.blaze.BlazeConvertStrategy.joinSmallerSideTag
-import org.apache.spark.sql.blaze.NativeConverters.StubExpr
+import org.apache.spark.sql.blaze.NativeConverters.{StubExpr, scalarTypeSupported}
 import org.apache.spark.sql.catalyst.expressions.Alias
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.catalyst.expressions.AttributeReference
@@ -272,6 +272,16 @@ object BlazeConverters extends Logging {
         .isInstanceOf[RangePartitioning],
       s"partitioning not supported: ${exec.outputPartitioning}")
 
+    if (exec.outputPartitioning.isInstanceOf[RangePartitioning]) {
+      val unsupportedOrderType = exec.outputPartitioning
+        .asInstanceOf[RangePartitioning]
+        .ordering
+        .find(e => !scalarTypeSupported(e.dataType))
+      assert(
+        unsupportedOrderType.isEmpty,
+        s"Unsupported order type in range partitioning: ${unsupportedOrderType.get}")
+    }
+
     val convertedChild = outputPartitioning match {
       case p
           if p.isInstanceOf[HashPartitioning] || p

spark-extension/src/main/scala/org/apache/spark/sql/blaze/NativeConverters.scala

@@ -118,6 +118,21 @@ object NativeConverters extends Logging {
     scalarTypeBuilder.build()
   }
 
+  def scalarTypeSupported(dataType: DataType): Boolean = {
+    dataType match {
+      case NullType | BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType |
+          DoubleType | StringType | DateType | TimestampType =>
+        true
+      case _: DecimalType => true
+      case at: ArrayType => scalarTypeSupported(at.elementType)
+      case m: MapType =>
+        scalarTypeSupported(m.keyType) && scalarTypeSupported(m.valueType)
+      case s: StructType =>
+        s.fields.forall(e => scalarTypeSupported(e.dataType))
+      case _ => false
+    }
+  }
+
   def convertDataType(sparkDataType: DataType): pb.ArrowType = {
     val arrowTypeBuilder = pb.ArrowType.newBuilder()
     sparkDataType match {