[SPARK-45887][SQL] Align codegen and non-codegen implementation of Encode

### What changes were proposed in this pull request?
In the PR, I propose to change the implementation of interpretation mode, and make it consistent to codegen. Both implementation raise the same error with new error class `INVALID_PARAMETER_VALUE.CHARSET`.

### Why are the changes needed?
To make codegen and non-codegen of the `Encode` expression consistent. So, users will observe the same behaviour in both modes.

### Does this PR introduce _any_ user-facing change?
Yes, if user code depends on error from `encode()`.

### How was this patch tested?
By running the following test suites:
```
$ PYSPARK_PYTHON=python3 build/sbt "sql/testOnly org.apache.spark.sql.SQLQueryTestSuite -- -z string-functions.sql"
$ build/sbt "core/testOnly *SparkThrowableSuite"
$ build/sbt "test:testOnly *.StringFunctionsSuite"
```

### Was this patch authored or co-authored using generative AI tooling?
No.

Closes #43759 from MaxGekk/restrict-charsets-in-encode.

Authored-by: Max Gekk <[email protected]>
Signed-off-by: Max Gekk <[email protected]>

Author: MaxGekk

common/utils/src/main/resources/error/error-classes.json

@@ -2042,6 +2042,11 @@
           "expects an integer value in [0, <upper>), but got <invalidValue>."
         ]
       },
+      "CHARSET" : {
+        "message" : [
+          "expects one of the charsets 'US-ASCII', 'ISO-8859-1', 'UTF-8', 'UTF-16BE', 'UTF-16LE', 'UTF-16', but got <charset>."
+        ]
+      },
       "DATETIME_UNIT" : {
         "message" : [
           "expects one of the units without quotes YEAR, QUARTER, MONTH, WEEK, DAY, DAYOFYEAR, HOUR, MINUTE, SECOND, MILLISECOND, MICROSECOND, but got the string literal <invalidValue>."

docs/sql-error-conditions-invalid-parameter-value-error-class.md

@@ -45,6 +45,10 @@ expects one of binary formats 'base64', 'hex', 'utf-8', but got `<invalidFormat>
 
 expects an integer value in [0, `<upper>`), but got `<invalidValue>`.
 
+## CHARSET
+
+expects one of the charsets 'US-ASCII', 'ISO-8859-1', 'UTF-8', 'UTF-16BE', 'UTF-16LE', 'UTF-16', but got `<charset>`.
+
 ## DATETIME_UNIT
 
 expects one of the units without quotes YEAR, QUARTER, MONTH, WEEK, DAY, DAYOFYEAR, HOUR, MINUTE, SECOND, MILLISECOND, MICROSECOND, but got the string literal `<invalidValue>`.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.catalyst.expressions
 
+import java.io.UnsupportedEncodingException
 import java.text.{BreakIterator, DecimalFormat, DecimalFormatSymbols}
 import java.util.{Base64 => JBase64}
 import java.util.{HashMap, Locale, Map => JMap}
@@ -2694,17 +2695,25 @@ case class Encode(value: Expression, charset: Expression)
 
   protected override def nullSafeEval(input1: Any, input2: Any): Any = {
     val toCharset = input2.asInstanceOf[UTF8String].toString
-    input1.asInstanceOf[UTF8String].toString.getBytes(toCharset)
+    try {
+      input1.asInstanceOf[UTF8String].toString.getBytes(toCharset)
+    } catch {
+      case _: UnsupportedEncodingException =>
+        throw QueryExecutionErrors.invalidCharsetError(prettyName, toCharset)
+    }
   }
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
-    nullSafeCodeGen(ctx, ev, (string, charset) =>
+    nullSafeCodeGen(ctx, ev, (string, charset) => {
+      val toCharset = ctx.freshName("toCharset")
       s"""
+        String $toCharset = $charset.toString();
         try {
-          ${ev.value} = $string.toString().getBytes($charset.toString());
+          ${ev.value} = $string.toString().getBytes($toCharset);
         } catch (java.io.UnsupportedEncodingException e) {
-          org.apache.spark.unsafe.Platform.throwException(e);
-        }""")
+          throw QueryExecutionErrors.invalidCharsetError("$prettyName", $toCharset);
+        }"""
+    })
   }
 
   override protected def withNewChildrenInternal(

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -2758,4 +2758,13 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
         "upper" -> size.toString,
         "invalidValue" -> pos.toString))
   }
+
+  def invalidCharsetError(functionName: String, charset: String): RuntimeException = {
+    new SparkIllegalArgumentException(
+      errorClass = "INVALID_PARAMETER_VALUE.CHARSET",
+      messageParameters = Map(
+        "functionName" -> toSQLId(functionName),
+        "parameter" -> toSQLId("charset"),
+        "charset" -> charset))
+  }
 }

sql/core/src/test/resources/sql-tests/analyzer-results/ansi/string-functions.sql.out

@@ -640,6 +640,21 @@ Project [rpad(cast(0x57 as string), 5, abc) AS rpad(X'57', 5, abc)#x]
 +- OneRowRelation
 
 
+-- !query
+select encode('hello', 'Windows-xxx')
+-- !query analysis
+Project [encode(hello, Windows-xxx) AS encode(hello, Windows-xxx)#x]
++- OneRowRelation
+
+
+-- !query
+select encode(scol, ecol) from values('hello', 'Windows-xxx') as t(scol, ecol)
+-- !query analysis
+Project [encode(scol#x, ecol#x) AS encode(scol, ecol)#x]
++- SubqueryAlias t
+   +- LocalRelation [scol#x, ecol#x]
+
+
 -- !query
 select decode()
 -- !query analysis

sql/core/src/test/resources/sql-tests/analyzer-results/string-functions.sql.out

@@ -640,6 +640,21 @@ Project [rpad(cast(0x57 as string), 5, abc) AS rpad(X'57', 5, abc)#x]
 +- OneRowRelation
 
 
+-- !query
+select encode('hello', 'Windows-xxx')
+-- !query analysis
+Project [encode(hello, Windows-xxx) AS encode(hello, Windows-xxx)#x]
++- OneRowRelation
+
+
+-- !query
+select encode(scol, ecol) from values('hello', 'Windows-xxx') as t(scol, ecol)
+-- !query analysis
+Project [encode(scol#x, ecol#x) AS encode(scol, ecol)#x]
++- SubqueryAlias t
+   +- LocalRelation [scol#x, ecol#x]
+
+
 -- !query
 select decode()
 -- !query analysis

sql/core/src/test/resources/sql-tests/inputs/string-functions.sql

@@ -117,6 +117,10 @@ SELECT lpad(x'57', 5, 'abc');
 SELECT rpad('abc', 5, x'57');
 SELECT rpad(x'57', 5, 'abc');
 
+-- encode
+select encode('hello', 'Windows-xxx');
+select encode(scol, ecol) from values('hello', 'Windows-xxx') as t(scol, ecol);
+
 -- decode
 select decode();
 select decode(encode('abc', 'utf-8'));

sql/core/src/test/resources/sql-tests/results/ansi/string-functions.sql.out

@@ -803,6 +803,40 @@ struct<rpad(X'57', 5, abc):string>
 Wabca
 
 
+-- !query
+select encode('hello', 'Windows-xxx')
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.SparkIllegalArgumentException
+{
+  "errorClass" : "INVALID_PARAMETER_VALUE.CHARSET",
+  "sqlState" : "22023",
+  "messageParameters" : {
+    "charset" : "Windows-xxx",
+    "functionName" : "`encode`",
+    "parameter" : "`charset`"
+  }
+}
+
+
+-- !query
+select encode(scol, ecol) from values('hello', 'Windows-xxx') as t(scol, ecol)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.SparkIllegalArgumentException
+{
+  "errorClass" : "INVALID_PARAMETER_VALUE.CHARSET",
+  "sqlState" : "22023",
+  "messageParameters" : {
+    "charset" : "Windows-xxx",
+    "functionName" : "`encode`",
+    "parameter" : "`charset`"
+  }
+}
+
+
 -- !query
 select decode()
 -- !query schema

sql/core/src/test/resources/sql-tests/results/string-functions.sql.out

@@ -735,6 +735,40 @@ struct<rpad(X'57', 5, abc):string>
 Wabca
 
 
+-- !query
+select encode('hello', 'Windows-xxx')
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.SparkIllegalArgumentException
+{
+  "errorClass" : "INVALID_PARAMETER_VALUE.CHARSET",
+  "sqlState" : "22023",
+  "messageParameters" : {
+    "charset" : "Windows-xxx",
+    "functionName" : "`encode`",
+    "parameter" : "`charset`"
+  }
+}
+
+
+-- !query
+select encode(scol, ecol) from values('hello', 'Windows-xxx') as t(scol, ecol)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.SparkIllegalArgumentException
+{
+  "errorClass" : "INVALID_PARAMETER_VALUE.CHARSET",
+  "sqlState" : "22023",
+  "messageParameters" : {
+    "charset" : "Windows-xxx",
+    "functionName" : "`encode`",
+    "parameter" : "`charset`"
+  }
+}
+
+
 -- !query
 select decode()
 -- !query schema