feat: decode() expression when using 'utf-8' encoding (#1697)

Author: mbutrovich

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

@@ -882,7 +882,10 @@ fn cast_array(
     let array = match &from_type {
         Dictionary(key_type, value_type)
             if key_type.as_ref() == &Int32
-                && (value_type.as_ref() == &Utf8 || value_type.as_ref() == &LargeUtf8) =>
+                && (value_type.as_ref() == &Utf8
+                    || value_type.as_ref() == &LargeUtf8
+                    || value_type.as_ref() == &Binary
+                    || value_type.as_ref() == &LargeBinary) =>
         {
             let dict_array = array
                 .as_any()

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

@@ -19,6 +19,8 @@
 
 package org.apache.comet.serde
 
+import java.util.Locale
+
 import scala.collection.JavaConverters._
 import scala.math.min
 
@@ -1430,6 +1432,25 @@ object QueryPlanSerde extends Logging with CometExprShim {
         val optExpr = scalarFunctionExprToProto("ascii", childExpr)
         optExprWithInfo(optExpr, expr, castExpr)
 
+      case s: StringDecode =>
+        // Right child is the encoding expression.
+        s.right match {
+          case Literal(str, DataTypes.StringType)
+              if str.toString.toLowerCase(Locale.ROOT) == "utf-8" =>
+            // decode(col, 'utf-8') can be treated as a cast with "try" eval mode that puts nulls
+            // for invalid strings.
+            // Left child is the binary expression.
+            castToProto(
+              expr,
+              None,
+              DataTypes.StringType,
+              exprToProtoInternal(s.left, inputs, binding).get,
+              CometEvalMode.TRY)
+          case _ =>
+            withInfo(expr, "Comet only supports decoding with 'utf-8'.")
+            None
+        }
+
       case BitLength(child) =>
         val castExpr = Cast(child, StringType)
         val childExpr = exprToProtoInternal(castExpr, inputs, binding)

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

@@ -35,7 +35,7 @@ import org.apache.spark.sql.execution.SparkPlan
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType
-import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, StructType}
+import org.apache.spark.sql.types._
 
 import org.apache.comet.testing.{DataGenOptions, ParquetGenerator}
 
@@ -188,6 +188,21 @@ class CometFuzzTestSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("decode") {
+    val df = spark.read.parquet(filename)
+    df.createOrReplaceTempView("t1")
+    // We want to make sure that the schema generator wasn't modified to accidentally omit
+    // BinaryType, since then this test would not run any queries and silently pass.
+    var testedBinary = false
+    for (field <- df.schema.fields if field.dataType == BinaryType) {
+      testedBinary = true
+      // Intentionally use odd capitalization of 'utf-8' to test normalization.
+      val sql = s"SELECT decode(${field.name}, 'utF-8') FROM t1"
+      checkSparkAnswerAndOperator(sql)
+    }
+    assert(testedBinary)
+  }
+
   test("Parquet temporal types written as INT96") {
     // int96 coercion in DF does not work with nested types yet
     // https://github.com/apache/datafusion/issues/15763