feat: Support Spark Expression Encode

docs/source/contributor-guide/expression-audits/string_funcs.md

@@ -84,6 +84,14 @@
 - `decode` runs through the codegen dispatcher on all versions (Spark 3.x via `CometStringDecode`, Spark 4.0 via the `StaticInvoke` replacement routed to `CometStaticInvokeCodegenDispatch`), so Spark's own evaluation runs inside the Comet pipeline. This honours the `charset` argument and the Spark 4.0 `legacyCharsets` / `legacyErrorAction` flags, and falls back to Spark when the dispatcher is disabled.
 - Known limitation: because there is no `CometExpressionSerde[StringDecode]` registration, `decode` does not surface in the auto-generated compatibility docs (https://github.com/apache/datafusion-comet/issues/4466).
 
+## encode
+
+- Spark 3.4.3 (audited 2026-07-23): `Encode(value, charset)` converts the input through `UTF8String.toString.getBytes(charset)`, replacing malformed UTF-8 before encoding.
+- Spark 3.5.8 (audited 2026-07-23): unchanged from Spark 3.4.3.
+- Spark 4.0.1 (audited 2026-07-23): refactored to `RuntimeReplaceable` backed by `StaticInvoke(Encode.encode, ...)`. Valid UTF-8 uses a raw-byte fast path; malformed UTF-8 is converted through `UTF8String.toString` before encoding.
+- Spark 4.1.1 (audited 2026-07-23): unchanged from Spark 4.0.1.
+- Comet supports literal UTF-8 charsets by lowering `encode` to `CAST(string AS binary)` and falls back for other charsets. The lowering preserves malformed UTF-8 bytes where Spark replaces them, a known incompatibility tracked by https://github.com/apache/datafusion-comet/issues/4764.
+
 ## endswith
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.

docs/source/user-guide/latest/expressions.md

@@ -549,7 +549,7 @@ expression-level). The `outer` variants are wired but marked `Incompatible`; the
 | `contains` | ✅ |  |
 | `decode` | ✅ |  |
 | `elt` | ✅ |  |
-| `encode` | 🔜 | Lowers to `StaticInvoke(encode)` (not allowlisted); falls back |
+| `encode` | ✅ | UTF-8 only; malformed UTF-8 input can differ from Spark ([#4764](https://github.com/apache/datafusion-comet/issues/4764)) |
 | `endswith` | ✅ |  |
 | `find_in_set` | ✅ |  |
 | `format_number` | ✅ |  |

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

@@ -19,10 +19,14 @@
 
 package org.apache.comet.serde
 
+import java.util.Locale
+
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Base64, BitLength, Cast, Concat, ConcatWs, Contains, Elt, Empty2Null, EndsWith, Expression, FindInSet, FormatNumber, FormatString, GetJsonObject, InitCap, Left, Length, Levenshtein, Like, Literal, Lower, Mask, OctetLength, Overlay, RegExpExtract, RegExpExtractAll, RegExpInStr, RegExpReplace, Right, RLike, SoundEx, StartsWith, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper}
 import org.apache.spark.sql.types.{BinaryType, DataTypes, LongType, StringType}
 
 import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
+import org.apache.comet.expressions.{CometCast, CometEvalMode}
 import org.apache.comet.serde.ExprOuterClass.Expr
 import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType}
 import org.apache.comet.shims.CometTypeShim
@@ -653,6 +657,39 @@ object CometGetJsonObject extends CometCodegenDispatch[GetJsonObject] with Nativ
     }
 }
 
+trait CommonStringExprs {
+
+  def stringEncode(
+      expr: Expression,
+      charset: Expression,
+      value: Expression,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[Expr] = {
+    charset match {
+      case Literal(str, DataTypes.StringType)
+          if str != null && str.toString.toLowerCase(Locale.ROOT) == "utf-8" =>
+        // For valid UTF-8, encode(col, 'utf-8') is byte-equivalent to cast(string AS binary).
+        // Spark StringType can also hold malformed UTF-8 bytes; Spark replaces those bytes during
+        // encode, while the cast preserves them. This limitation is tracked by #4764.
+        val strExpr = exprToProtoInternal(value, inputs, binding)
+        if (strExpr.isDefined) {
+          CometCast.castToProto(
+            expr,
+            None,
+            DataTypes.BinaryType,
+            strExpr.get,
+            CometEvalMode.LEGACY)
+        } else {
+          withFallbackReason(expr, value)
+          None
+        }
+      case _ =>
+        withFallbackReason(expr, "Comet only supports encoding with 'utf-8'.")
+        None
+    }
+  }
+}
+
 // Expressions routed through the JVM codegen dispatcher: no native implementation, so Spark's own
 // doGenCode runs inside the Comet pipeline, matching Spark exactly.
 object CometLevenshtein extends CometCodegenDispatch[Levenshtein]

spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala

@@ -23,13 +23,13 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate.Sum
 
 import org.apache.comet.expressions.CometEvalMode
-import org.apache.comet.serde.{CometExpressionSerde, CometStringDecode}
+import org.apache.comet.serde.{CometExpressionSerde, CometStringDecode, CommonStringExprs}
 import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr}
 
 /**
  * `CometExprShim` acts as a shim for parsing expressions from different Spark versions.
  */
-trait CometExprShim {
+trait CometExprShim extends CommonStringExprs {
   protected def evalMode(c: Cast): CometEvalMode.Value =
     CometEvalModeUtil.fromSparkEvalMode(c.evalMode)
 
@@ -48,7 +48,13 @@ trait CometExprShim {
   def sparkVersionSpecificExprToProtoInternal(
       expr: Expression,
       inputs: Seq[Attribute],
-      binding: Boolean): Option[Expr] = None
+      binding: Boolean): Option[Expr] = {
+    expr match {
+      case e: Encode =>
+        stringEncode(expr, e.charset, e.value, inputs, binding)
+      case _ => None
+    }
+  }
 }
 
 object CometEvalModeUtil {

spark/src/main/spark-3.5/org/apache/comet/shims/CometExprShim.scala

@@ -23,13 +23,13 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate.Sum
 
 import org.apache.comet.expressions.CometEvalMode
-import org.apache.comet.serde.{CometExpressionSerde, CometStringDecode, CometToPrettyString, CometWidthBucket}
+import org.apache.comet.serde.{CometExpressionSerde, CometStringDecode, CometToPrettyString, CometWidthBucket, CommonStringExprs}
 import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr}
 
 /**
  * `CometExprShim` acts as a shim for parsing expressions from different Spark versions.
  */
-trait CometExprShim {
+trait CometExprShim extends CommonStringExprs {
   protected def evalMode(c: Cast): CometEvalMode.Value =
     CometEvalModeUtil.fromSparkEvalMode(c.evalMode)
 
@@ -48,7 +48,13 @@ trait CometExprShim {
   def sparkVersionSpecificExprToProtoInternal(
       expr: Expression,
       inputs: Seq[Attribute],
-      binding: Boolean): Option[Expr] = None
+      binding: Boolean): Option[Expr] = {
+    expr match {
+      case e: Encode =>
+        stringEncode(expr, e.charset, e.value, inputs, binding)
+      case _ => None
+    }
+  }
 }
 
 object CometEvalModeUtil {

spark/src/main/spark-4.x/org/apache/comet/shims/CometExprShimCommon.scala

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.shims
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
+
+import org.apache.comet.serde.CommonStringExprs
+import org.apache.comet.serde.ExprOuterClass.Expr
+
+trait CometExprShimCommon extends CommonStringExprs {
+
+  protected def sparkExprToProto(
+      expr: Expression,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[Expr] = {
+    expr match {
+      // For valid UTF-8, encode(str, 'utf-8') -> cast(string AS binary) is a zero-copy
+      // reinterpret. Malformed UTF-8 differs as described in CommonStringExprs.stringEncode.
+      case s: StaticInvoke
+          if s.staticObject == classOf[Encode] &&
+            s.functionName == "encode" =>
+        s.arguments match {
+          case value +: charset +: _ =>
+            stringEncode(expr, charset, value, inputs, binding)
+          case _ => None
+        }
+
+      case _ => None
+    }
+  }
+}

spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala

@@ -35,7 +35,7 @@ import org.apache.comet.serde.QueryPlanSerde.exprToProtoInternal
  * are identical across minor versions; per-version traits override only `binaryOutputStyle` and
  * supply the matching `CometEvalModeUtil.sumEvalMode`.
  */
-trait Spark4xCometExprShim extends CometExprShim4x {
+trait Spark4xCometExprShim extends CometExprShim4x with CometExprShimCommon {
   protected def evalMode(c: Cast): CometEvalMode.Value =
     CometEvalModeUtil.fromSparkEvalMode(c.evalMode)
 
@@ -105,7 +105,7 @@ trait Spark4xCometExprShim extends CometExprShim4x {
                 .foreach(reasons => s.setTagValue(CometExplainInfo.FALLBACK_REASONS, reasons))
             }
             exprProto
-          case _ => None
+          case _ => sparkExprToProto(s, inputs, binding)
         }
 
       // dayname / monthname (Spark 4.0+) are shared across all 4.x minor versions; see

spark/src/test/resources/sql-tests/expressions/string/encode.sql

@@ -0,0 +1,66 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements.  See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership.  The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License.  You may obtain a copy of the License at
+--
+--   http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied.  See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Tests for the SQL `encode(str, charset)` function.
+--
+-- Spark 3.x: Encode is a BinaryExpression(value, charset).
+-- Spark 4.x+: Encode is RuntimeReplaceable; the analyzer rewrites it to
+--   StaticInvoke(classOf[Encode], BinaryType, "encode", ...)
+
+statement
+CREATE TABLE test_encode_utf8(s string) USING parquet
+
+statement
+INSERT INTO test_encode_utf8 VALUES ('hello'), ('world'), (''), ('café'), (NULL)
+
+query
+SELECT encode(s, 'utf-8') FROM test_encode_utf8
+
+query
+SELECT encode(s, 'UTF-8') FROM test_encode_utf8
+
+-- Mixed-case charset literal exercises toLowerCase normalization
+query
+SELECT encode(s, 'Utf-8') FROM test_encode_utf8
+
+query
+SELECT encode('hello', 'utf-8'), encode('', 'utf-8'), encode(CAST(NULL AS STRING), 'utf-8')
+
+-- Different language(French, Japanese)
+query
+SELECT encode('café', 'utf-8'), encode('日本語', 'utf-8')
+
+-- Spark replaces malformed UTF-8 before encoding, but Comet's cast lowering preserves the raw
+-- bytes. Keep the divergence captured until native string ingress has a consistent UTF-8 policy.
+query ignore(https://github.com/apache/datafusion-comet/issues/4764)
+SELECT encode(CAST(X'FF' AS STRING), 'utf-8')
+
+-- non-UTF-8 falls back to Spark JVM
+statement
+CREATE TABLE test_encode_charset_safe(s string) USING parquet
+
+statement
+INSERT INTO test_encode_charset_safe VALUES ('hello'), ('world'), (''), (NULL)
+
+query expect_fallback(Comet only supports encoding with 'utf-8'.)
+SELECT encode(s, 'UTF-16BE') FROM test_encode_charset_safe
+
+query expect_fallback(Comet only supports encoding with 'utf-8'.)
+SELECT encode(s, 'US-ASCII') FROM test_encode_charset_safe
+
+query expect_fallback(Comet only supports encoding with 'utf-8'.)
+SELECT encode(s, 'ISO-8859-1') FROM test_encode_charset_safe