(datafusion-comet) branch main updated: feat: decode() expression when using 'utf-8' encoding (#1697)

[ag...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

agrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git


The following commit(s) were added to refs/heads/main by this push:
     new dae7aad1f feat: decode() expression when using 'utf-8' encoding (#1697)
dae7aad1f is described below

commit dae7aad1fddf77e4345fd5ff0b2104fcaf936a9d
Author: Matt Butrovich <mb...@users.noreply.github.com>
AuthorDate: Fri May 2 11:30:35 2025 -0400

    feat: decode() expression when using 'utf-8' encoding (#1697)
---
 native/spark-expr/src/conversion_funcs/cast.rs      |  5 ++++-
 .../org/apache/comet/serde/QueryPlanSerde.scala     | 21 +++++++++++++++++++++
 .../scala/org/apache/comet/CometFuzzTestSuite.scala | 17 ++++++++++++++++-
 3 files changed, 41 insertions(+), 2 deletions(-)

diff --git a/native/spark-expr/src/conversion_funcs/cast.rs b/native/spark-expr/src/conversion_funcs/cast.rs
index 59647f2f0..6142abf2b 100644
--- a/native/spark-expr/src/conversion_funcs/cast.rs
+++ b/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()
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index 689e9b9bc..7cadedec3 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/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)
diff --git a/spark/src/test/scala/org/apache/comet/CometFuzzTestSuite.scala b/spark/src/test/scala/org/apache/comet/CometFuzzTestSuite.scala
index 2930ac87a..235c03e7d 100644
--- a/spark/src/test/scala/org/apache/comet/CometFuzzTestSuite.scala
+++ b/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


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org