Support reverse function with ArrayType input

Author: cfmcgrady

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -915,6 +915,8 @@ object QueryPlanSerde extends Logging with CometExprShim {
       case l @ Length(child) if child.dataType == BinaryType =>
         withInfo(l, "Length on BinaryType is not supported")
         None
+      case r @ Reverse(child) if child.dataType.isInstanceOf[ArrayType] =>
+        convert(r, CometArrayReverse)
       case expr =>
         QueryPlanSerde.exprSerdeMap.get(expr.getClass) match {
           case Some(handler) =>

spark/src/main/scala/org/apache/comet/serde/arrays.scala

@@ -20,11 +20,9 @@
 package org.apache.comet.serde
 
 import scala.annotation.tailrec
-
-import org.apache.spark.sql.catalyst.expressions.{ArrayAppend, ArrayContains, ArrayDistinct, ArrayExcept, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayRemove, ArrayRepeat, ArraysOverlap, ArrayUnion, Attribute, CreateArray, ElementAt, Expression, Flatten, GetArrayItem, Literal}
+import org.apache.spark.sql.catalyst.expressions.{ArrayAppend, ArrayContains, ArrayDistinct, ArrayExcept, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayRemove, ArrayRepeat, ArrayUnion, ArraysOverlap, Attribute, CreateArray, ElementAt, Expression, Flatten, GetArrayItem, Literal, Reverse}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
-
 import org.apache.comet.CometSparkSessionExtensions.withInfo
 import org.apache.comet.serde.QueryPlanSerde._
 import org.apache.comet.shims.CometExprShim
@@ -432,6 +430,25 @@ object CometGetArrayItem extends CometExpressionSerde[GetArrayItem] {
   }
 }
 
+object CometArrayReverse extends CometExpressionSerde[Reverse] with ArraysBase {
+  override def convert(
+                        expr: Reverse,
+                        inputs: Seq[Attribute],
+                        binding: Boolean): Option[ExprOuterClass.Expr] = {
+    val inputTypes = expr.children.map(_.dataType).toSet
+    for (dt <- inputTypes) {
+      if (!isTypeSupported(dt)) {
+        withInfo(expr, s"data type not supported: $dt")
+        return None
+      }
+    }
+    val reverseExprProto = exprToProto(expr.child, inputs, binding)
+    val reverseScalarExpr = scalarFunctionExprToProto("array_reverse", reverseExprProto)
+    optExprWithInfo(reverseScalarExpr, expr, expr.children: _*)
+  }
+
+}
+
 object CometElementAt extends CometExpressionSerde[ElementAt] {
 
   override def convert(

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

@@ -708,4 +708,11 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp
       }
     }
   }
+
+  test("test reverse function") {
+    withTable("t1") {
+      sql("create table t1 using parquet as select sequence(id, 10) as c1 from range(10)")
+      checkSparkAnswerAndOperator("select reverse(c1) AS x FROM t1 ORDER BY c1")
+    }
+  }
 }