feat: support Spark encode expression via codegen dispatch

docs/source/contributor-guide/sql-file-tests.md

@@ -125,6 +125,16 @@ Skips the file when running on a Spark version older than the specified version.
 -- MinSparkVersion: 3.5
 ```
 
+#### `MaxSparkVersion`
+
+Skips the file when running on a Spark version newer than the specified version (inclusive of
+that version). Use it together with `MinSparkVersion` in a paired fixture when a behavior
+changed between versions and each range needs its own expected output or error class.
+
+```sql
+-- MaxSparkVersion: 3.5
+```
+
 ### Statements
 
 A `statement` block executes DDL or DML and does not check results. Use this for `CREATE TABLE`

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

@@ -565,7 +565,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 | `contains` | ✅ | — |  |
 | `decode` | ✅ | — |  |
 | `elt` | ✅ | Codegen dispatch |  |
-| `encode` | ✅ | — | Spark 4.0+ lowers to `StaticInvoke(Encode.encode)`, which runs via codegen dispatch; on Spark 3.x it falls back |
+| `encode` | ✅ | — |  |
 | `endswith` | ✅ | — |  |
 | `find_in_set` | ✅ | Codegen dispatch |  |
 | `format_number` | ✅ | Codegen dispatch |  |

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

@@ -23,7 +23,7 @@ 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.{CometEncode, CometExpressionSerde, CometStringDecode}
 import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr}
 
 /**
@@ -37,7 +37,7 @@ trait CometExprShim {
 
   def sparkVersionSpecificStringExpressions
       : Map[Class[_ <: Expression], CometExpressionSerde[_]] =
-    Map(classOf[StringDecode] -> CometStringDecode)
+    Map(classOf[StringDecode] -> CometStringDecode, classOf[Encode] -> CometEncode)
   def sparkVersionSpecificMathExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] =
     Map.empty
   def sparkVersionSpecificMiscExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] =

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

@@ -23,7 +23,7 @@ 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.{CometEncode, CometExpressionSerde, CometStringDecode, CometToPrettyString, CometWidthBucket}
 import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr}
 
 /**
@@ -37,7 +37,7 @@ trait CometExprShim {
 
   def sparkVersionSpecificStringExpressions
       : Map[Class[_ <: Expression], CometExpressionSerde[_]] =
-    Map(classOf[StringDecode] -> CometStringDecode)
+    Map(classOf[StringDecode] -> CometStringDecode, classOf[Encode] -> CometEncode)
   def sparkVersionSpecificMathExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] =
     Map(classOf[WidthBucket] -> CometWidthBucket)
   def sparkVersionSpecificMiscExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] =

spark/src/main/spark-3.x/org/apache/comet/serde/CometEncode.scala

@@ -0,0 +1,28 @@
+/*
+ * 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.serde
+
+import org.apache.spark.sql.catalyst.expressions.Encode
+
+/**
+ * Spark 3.x `encode(str, charset)` runs through the codegen dispatcher so Spark's own encoder
+ * handles charset selection and malformed-input behavior. Dual of `CometStringDecode`.
+ */
+object CometEncode extends CometCodegenDispatch[Encode]

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

@@ -0,0 +1,94 @@
+-- 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 (StringType, StringType) -> BinaryType.
+--
+-- `encode` runs through the codegen dispatcher (Spark's own doGenCode inside the Comet
+-- pipeline) so behavior matches Spark exactly across all supported charsets and across the
+-- Spark 4.0 `legacyCharsets` / `legacyErrorAction` modes. This is the dual of the `decode`
+-- codegen-dispatch path (#4465).
+--
+-- Each `query` block runs checkSparkAnswerAndOperator, which fails if the expression fell back
+-- to Spark instead of executing natively, so these are non-vacuous.
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
+
+statement
+CREATE TABLE test_encode(s string) USING parquet
+
+statement
+INSERT INTO test_encode VALUES ('hello'), ('world'), (''), ('café'), (NULL)
+
+-- Charset form over multiple charsets
+
+query
+SELECT encode(s, 'utf-8') FROM test_encode
+
+query
+SELECT encode(s, 'UTF-8') FROM test_encode
+
+query
+SELECT encode(s, 'UTF-16') FROM test_encode
+
+query
+SELECT encode(s, 'UTF-16BE') FROM test_encode
+
+-- UTF-16 emits a byte-order mark and UTF-16LE does not, so the two differ in output, and
+-- UTF-16LE differs from UTF-16BE in byte order.
+
+query
+SELECT encode(s, 'UTF-16LE') FROM test_encode
+
+query
+SELECT encode(s, 'UTF-32') FROM test_encode
+
+query
+SELECT encode(s, 'ISO-8859-1') FROM test_encode
+
+-- US-ASCII: use ASCII-only input so this file stays version-agnostic. Unmappable input under
+-- US-ASCII substitutes on Spark 3.4/3.5 and throws on Spark 4.0+, so it is covered by the paired
+-- fixtures encode_unmappable.sql and encode_unmappable_strict.sql instead.
+
+statement
+CREATE TABLE test_encode_ascii(s string) USING parquet
+
+statement
+INSERT INTO test_encode_ascii VALUES ('hello'), ('world'), (''), (NULL)
+
+query
+SELECT encode(s, 'US-ASCII') FROM test_encode_ascii
+
+-- Literal inputs including NULL and empty string
+
+query
+SELECT encode('hello', 'utf-8'), encode('', 'utf-8'), encode(CAST(NULL AS STRING), 'utf-8')
+
+-- Non-literal charset column: the charset is an ordinary child expression, not required to be
+-- foldable. The dispatcher handles it because it runs Spark's own code.
+
+statement
+CREATE TABLE test_encode_charset(s string, cs string) USING parquet
+
+statement
+INSERT INTO test_encode_charset VALUES ('hello', 'utf-8'), ('world', 'UTF-16BE'), ('café', 'ISO-8859-1'), (NULL, 'utf-8')
+
+query
+SELECT encode(s, cs) FROM test_encode_charset
+
+-- Round-trip: decode(encode(x)) recovers the original string
+
+query
+SELECT decode(encode(s, 'UTF-8'), 'UTF-8') FROM test_encode

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

@@ -0,0 +1,44 @@
+-- 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.
+
+-- encode() charset validation on Spark 3.4 and 3.5.
+--
+-- On those versions `Encode` calls `String.getBytes(charset)` directly, so any charset the JVM
+-- knows is accepted (including ones outside Spark 4.0's allowlist, e.g. windows-1252) and an
+-- unknown name surfaces as `UnsupportedEncodingException` through `Platform.throwException`.
+-- Spark 4.0+ instead routes through `CharsetProvider.forName`, which rejects anything outside
+-- VALID_CHARSETS unless `spark.sql.legacy.javaCharsets` is set; those two branches are covered by
+-- encode_invalid_charset_strict.sql and encode_legacy_charsets.sql.
+-- MaxSparkVersion: 3.5
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
+
+statement
+CREATE TABLE test_encode_charset_legacy(s string) USING parquet
+
+statement
+INSERT INTO test_encode_charset_legacy VALUES ('hello'), ('café'), (''), (NULL)
+
+-- windows-1252 is not in Spark 4.0's VALID_CHARSETS but is a JVM charset, so 3.x accepts it.
+-- This is also the sentinel query: it fails if `encode` did not execute natively, so the
+-- expect_error query below cannot pass vacuously through an operator-level fallback.
+
+query
+SELECT encode(s, 'windows-1252') FROM test_encode_charset_legacy
+
+-- A charset name the JVM does not know at all.
+query expect_error(UnsupportedEncodingException)
+SELECT encode('hello', 'NO-SUCH-CHARSET')

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

@@ -0,0 +1,50 @@
+-- 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.
+
+-- encode() charset validation in Spark 4.0's default mode.
+--
+-- `CharsetProvider.forName` permits only us-ascii, iso-8859-1, utf-8, utf-16be, utf-16le, utf-16
+-- and utf-32 unless `spark.sql.legacy.javaCharsets` is enabled, and otherwise raises
+-- `INVALID_PARAMETER_VALUE.CHARSET`. On Spark 3.4/3.5 a JVM charset outside that list is simply
+-- accepted; see encode_invalid_charset.sql. Enabling the legacy flag restores that behavior on
+-- 4.0+; see encode_legacy_charsets.sql.
+-- MinSparkVersion: 4.0
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
+
+statement
+CREATE TABLE test_encode_charset_strict(s string) USING parquet
+
+statement
+INSERT INTO test_encode_charset_strict VALUES ('hello'), ('café'), (''), (NULL)
+
+-- Sentinel: an allowlisted charset, asserting `encode` executes natively so the expect_error
+-- queries below trip the kernel rather than being satisfied by an operator-level Spark fallback.
+
+query
+SELECT encode(s, 'utf-8') FROM test_encode_charset_strict
+
+-- windows-1252 is a valid JVM charset but is outside VALID_CHARSETS.
+query expect_error(INVALID_PARAMETER_VALUE)
+SELECT encode('hello', 'windows-1252')
+
+-- Column input takes the same path.
+query expect_error(INVALID_PARAMETER_VALUE)
+SELECT encode(s, 'windows-1252') FROM test_encode_charset_strict
+
+-- A charset name the JVM does not know at all is rejected by the same check.
+query expect_error(INVALID_PARAMETER_VALUE)
+SELECT encode('hello', 'NO-SUCH-CHARSET')

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

@@ -0,0 +1,45 @@
+-- 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.
+
+-- encode() under `spark.sql.legacy.javaCharsets=true` on Spark 4.0+.
+--
+-- The flag is baked into the `Encode` node at analysis time and carried through to
+-- `StaticInvoke(classOf[Encode], "encode", ...)` as a literal, so it reaches the codegen
+-- dispatcher and `CharsetProvider.forName` accepts any JVM charset again. Without the flag the
+-- same queries raise `INVALID_PARAMETER_VALUE.CHARSET`; see encode_invalid_charset_strict.sql.
+-- MinSparkVersion: 4.0
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
+-- Config: spark.sql.legacy.javaCharsets=true
+
+statement
+CREATE TABLE test_encode_legacy_charsets(s string) USING parquet
+
+-- All rows are representable in windows-1252 ('é' is 0xE9), so the strict coding-error action
+-- that stays in effect here does not fire.
+statement
+INSERT INTO test_encode_legacy_charsets VALUES ('hello'), ('café'), (''), (NULL)
+
+query
+SELECT encode(s, 'windows-1252') FROM test_encode_legacy_charsets
+
+query
+SELECT encode('café', 'windows-1252'), encode('hello', 'Shift_JIS')
+
+-- The allowlisted charsets keep working with the flag on.
+
+query
+SELECT encode(s, 'UTF-16LE') FROM test_encode_legacy_charsets

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

@@ -0,0 +1,50 @@
+-- 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.
+
+-- encode() over characters the target charset cannot represent, on Spark 3.4 and 3.5.
+--
+-- On those versions `Encode` is a plain BinaryExpression whose eval is
+-- `input.toString.getBytes(charset)`, and `String.getBytes` uses an encoder configured with
+-- `CodingErrorAction.REPLACE`. Unmappable characters therefore become the charset's replacement
+-- byte (0x3F, '?') rather than raising. Spark 4.0+ builds the encoder with
+-- `CodingErrorAction.REPORT` and throws instead; that branch is covered by the paired fixture
+-- encode_unmappable_strict.sql.
+-- MaxSparkVersion: 3.5
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
+
+statement
+CREATE TABLE test_encode_unmappable(s string) USING parquet
+
+statement
+INSERT INTO test_encode_unmappable VALUES ('café'), ('中文'), ('hello'), (''), (NULL)
+
+-- Column input: 'é' and the CJK characters are unmappable in US-ASCII and are substituted.
+
+query
+SELECT encode(s, 'US-ASCII') FROM test_encode_unmappable
+
+-- 'é' is representable in ISO-8859-1 (0xE9) but the CJK characters are not.
+
+query
+SELECT encode(s, 'ISO-8859-1') FROM test_encode_unmappable
+
+-- Literal input, pinning the substituted bytes rather than only asserting Spark/Comet agreement.
+
+query
+SELECT encode('café', 'US-ASCII') = CAST('caf?' AS BINARY),
+       encode('中文', 'US-ASCII') = CAST('??' AS BINARY),
+       encode('中文', 'ISO-8859-1') = CAST('??' AS BINARY)