Add type checking

Author: chenghao-intel

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

@@ -16,6 +16,7 @@
  */
 package org.apache.spark.sql.catalyst.expressions
 
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenFallback, CodeGenContext, GeneratedExpressionCode}
 import org.apache.spark.sql.types._
 
@@ -50,13 +51,16 @@ case class SortArray(child: Expression)
   override def dataType: DataType = child.dataType
   override def inputTypes: Seq[AbstractDataType] = Seq(ArrayType)
 
+  override def checkInputDataTypes(): TypeCheckResult = child.dataType match {
+    case _ @ ArrayType(n: AtomicType, _) => TypeCheckResult.TypeCheckSuccess
+    case other => TypeCheckResult.TypeCheckFailure(
+                    s"Type $other is not supported for ordering operations")
+  }
+
   @transient
   private lazy val lt: (Any, Any) => Boolean = {
     val ordering = child.dataType match {
-      case ArrayType(elementType, _) => elementType match {
-        case n: AtomicType => n.ordering.asInstanceOf[Ordering[Any]]
-        case other => sys.error(s"Type $other does not support ordered operations")
-      }
+      case _ @ ArrayType(n: AtomicType, _) => n.ordering.asInstanceOf[Ordering[Any]]
     }
 
     (left, right) => {

sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala

@@ -287,6 +287,12 @@ class DataFrameFunctionsSuite extends QueryTest {
         Row(Seq[Int](), Seq[String]()),
         Row(null, null))
     )
+
+    val df2 = Seq((Array[Array[Int]](Array(2)), "x")).toDF("a", "b")
+    assert(intercept[AnalysisException] {
+      df2.selectExpr("sort_array(a)").collect()
+    }.getMessage().contains("Type ArrayType(ArrayType(IntegerType,false),true) " +
+      "is not supported for ordering operations"))
   }
 
   test("array size function") {