You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "JeelRajodiya (via GitHub)" <gi...@apache.org> on 2026/04/03 06:46:30 UTC
[PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
JeelRajodiya opened a new pull request, #21331:
URL: https://github.com/apache/datafusion/pull/21331
**Rationale**
The `datafusion-spark` crate is missing the `encode` function. Spark's [`encode(expr, charset)`](https://spark.apache.org/docs/latest/api/sql/index.html#encode) converts a string or binary value into binary using a specified character encoding — this is commonly used in Spark SQL workloads and needed by engines built on DataFusion that target Spark compatibility.
**What changes are included in this PR?**
Adds `SparkEncode` to `datafusion-spark`'s string functions. It supports **US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, and UTF-16LE** charsets. Binary input is handled via lossy UTF-8 conversion (invalid bytes → U+FFFD), matching Spark/Databricks behavior.
**Are these changes tested?**
Yes — 15 unit tests covering all charsets, case-insensitive charset matching, null handling, binary input with lossy UTF-8, Utf8View columns, unsupported charset errors, and return field nullability.
**Are there any user-facing changes?**
New `encode` scalar function available when using `datafusion-spark`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4186858416
Re: [`extract_charset` visibility](https://github.com/apache/datafusion/pull/21331/files#r3033108629) — Yes, a `decode` function will need the same charset extraction logic, so keeping it `pub(crate)` avoids duplicating that code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4275897689
Hey @andygrove, I realized that I shouldn't be using `enable_ansi_mode` flag inside encode function. In the spark definition they are not binding the ansi mode to encode function.
Moreover we should target Spark 3.5 which is more permissive and doesn't return errors when null inputs are passed. it simply replaces it with `?`. But I've added a TODO in the doc comment pointing at the two real Spark 4.1 configs so a follow-up PR can wire them properly.
**Below are the references to the spark definitions**
Spark 3.5's [`Encode.scala`](https://github.com/apache/spark/blob/2a56312aeb1665b72c608e14926f5d69fd3a17bc/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala#L2698-L2741):
``` scala
protected override def nullSafeEval(input1: Any, input2: Any): Any = {
input1.asInstanceOf[UTF8String].toString.getBytes(toCharset)
}
```
Just calls Java's `String.getBytes`, which replaces unmappable chars with the charset's default byte (?). No `legacyErrorAction`, no config, no exception.
Spark 4.1's [`Encode.scala`](https://github.com/apache/spark/blob/acfae3372874631728243ba13728f6abbf7ee07b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala#L3170-L3228) added two new configs for the strict behavior:
``` scala
case class Encode(str, charset, legacyCharsets: Boolean, legacyErrorAction: Boolean)
def this(value, charset) =
this(value, charset, SQLConf.get.legacyJavaCharsets, SQLConf.get.legacyCodingErrorAction)
```
> Setting legacyErrorAction=true restores the Spark 3.5 `?` behavior.
These `spark.sql.legacy.javaCharsets` and `spark.sql.legacy.codingErrorAction` are supported in 4.1 version. which can be left for future PR. Currently the PR targets Spark 3.5. I've added mentioned in the doc comment as well.
Let me know if we need to iterate on this further.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3035408870
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,500 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {:?}", dt),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+pub(crate) fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
Review Comment:
Should we make it private anyways?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4201235790
Thanks @xanderbailey and @JeelRajodiya -- the PR load is pretty intense! I started the CI for this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-5467871076
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/21331?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `85.92965%` with `28 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 81.53%. Comparing base ([`4ada0dc`](https://app.codecov.io/gh/apache/datafusion/commit/4ada0dcabbf502f751a7c21cf6cb642154637af9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`a88725e`](https://app.codecov.io/gh/apache/datafusion/commit/a88725e65578059644ce2be51e6a887aa74e6bb3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/21331?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [datafusion/spark/src/function/string/encode.rs](https://app.codecov.io/gh/apache/datafusion/pull/21331?src=pr&el=tree&filepath=datafusion%2Fspark%2Fsrc%2Ffunction%2Fstring%2Fencode.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9zcGFyay9zcmMvZnVuY3Rpb24vc3RyaW5nL2VuY29kZS5ycw==) | 85.85% | [24 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/21331?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #21331 +/- ##
========================================
Coverage 81.53% 81.53%
========================================
Files 1123 1124 +1
Lines 406041 406240 +199
Branches 406041 406240 +199
========================================
+ Hits 331049 331214 +165
- Misses 55631 55657 +26
- Partials 19361 19369 +8
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/21331?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3491227491
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,557 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
+ match charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Utf8(Some(s))
+ | ScalarValue::LargeUtf8(Some(s))
+ | ScalarValue::Utf8View(Some(s)) => Ok(s.to_uppercase()),
+ _ => exec_err!("encode charset argument must be a non-null string"),
+ },
+ ColumnarValue::Array(arr) => {
+ fn validate_constant_charset<'a, S: StringArrayType<'a>>(
+ arr: &'a S,
+ ) -> Result<String> {
+ if arr.is_null(0) {
+ return exec_err!(
+ "encode charset argument must be a non-null string"
Review Comment:
done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3084237290
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,554 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
Review Comment:
Thanks! I've updated error handling with `enable_ansi_mode` flag. which I followed from `round.rs`
https://github.com/apache/datafusion/blob/244f89120b254f824b6fed0f8a88c23a74bf09d7/datafusion/spark/src/function/math/round.rs#L260-L271
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-5468875580
<!-- semver-check-comment -->
Thank you for opening this pull request!
Reviewer note: [cargo-semver-checks](https://github.com/obi1kenobi/cargo-semver-checks) reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).
<details>
<summary>Details</summary>
```
Cloning apache/main
Building datafusion-spark v55.0.0 (current)
Built [ 38.849s] (current)
Parsing datafusion-spark v55.0.0 (current)
Parsed [ 0.046s] (current)
Building datafusion-spark v55.0.0 (baseline)
Built [ 31.737s] (baseline)
Parsing datafusion-spark v55.0.0 (baseline)
Parsed [ 0.045s] (baseline)
Checking datafusion-spark v55.0.0 -> v55.0.0 (no change; assume patch)
Checked [ 0.272s] 223 checks: 223 pass, 31 skip
Summary no semver update required
Finished [ 73.255s] datafusion-spark
Building datafusion-sqllogictest v55.0.0 (current)
error: running cargo-doc on crate 'datafusion-sqllogictest' failed with output:
-----
Compiling proc-macro2 v1.0.107
Compiling quote v1.0.47
Compiling unicode-ident v1.0.24
Compiling libc v0.2.189
Checking cfg-if v1.0.4
Checking bytes v1.12.1
Checking memchr v2.8.3
Compiling serde_core v1.0.229
Compiling autocfg v1.5.1
Compiling syn v2.0.119
Compiling syn v3.0.4
Compiling jobserver v0.1.35
Compiling find-msvc-tools v0.1.11
Compiling shlex v2.0.1
Compiling cc v1.4.4
Checking itoa v1.0.18
Checking equivalent v1.0.2
Compiling libm v0.2.16
Checking once_cell v1.21.4
Checking allocator-api2 v0.2.21
Checking foldhash v0.2.0
Compiling num-traits v0.2.19
Checking hashbrown v0.17.1
Checking indexmap v2.14.1
Checking pin-project-lite v0.2.17
Compiling zmij v1.0.23
Checking futures-core v0.3.34
Compiling zerocopy v0.8.56
Checking futures-sink v0.3.34
Checking errno v0.3.14
Compiling tokio-macros v2.7.2
Checking signal-hook-registry v1.4.8
Checking mio v1.2.2
Compiling zerocopy-derive v0.8.56
Checking socket2 v0.6.5
Compiling version_check v0.9.5
Checking num-integer v0.1.47
Checking tokio v1.53.1
Compiling serde v1.0.229
Compiling serde_derive v1.0.229
Compiling serde_json v1.0.151
Checking slab v0.4.12
Checking futures-channel v0.3.34
Compiling getrandom v0.3.4
Checking smallvec v1.15.2
Compiling futures-macro v0.3.34
Compiling synstructure v0.13.2
Checking http v1.5.0
Checking futures-io v0.3.34
Checking futures-task v0.3.34
Checking futures-util v0.3.34
Checking iana-time-zone v0.1.65
Checking chrono v0.4.45
Compiling zerofrom-derive v0.1.7
Compiling tracing-attributes v0.1.31
Checking tracing-core v0.1.36
Checking siphasher v1.0.3
Checking zerofrom v0.1.8
Compiling yoke-derive v0.8.2
Checking rand_core v0.10.1
Compiling getrandom v0.4.3
Checking stable_deref_trait v1.2.1
Checking tracing v0.1.44
Checking num-complex v0.4.6
Compiling zerovec-derive v0.11.6
Checking getrandom v0.2.17
Checking cpufeatures v0.3.1
Compiling pkg-config v0.3.34
Checking phf_shared v0.12.1
Checking half v2.7.1
Checking num-bigint v0.5.1
Compiling ahash v0.8.12
Checking yoke v0.8.3
Compiling displaydoc v0.2.7
Compiling chrono-tz v0.10.4
Checking percent-encoding v2.3.2
Compiling thiserror v2.0.20
Checking zerovec v0.11.8
Checking phf v0.12.1
Checking arrow-schema v59.2.0
Compiling thiserror-impl v2.0.20
Compiling ring v0.17.14
Checking arrow-buffer v59.2.0
Checking arrow-data v59.2.0
Compiling semver v1.0.28
Checking log v0.4.34
Checking tinystr v0.8.4
Checking chacha20 v0.10.2
Checking litemap v0.8.3
Checking untrusted v0.9.0
Checking bitflags v2.13.1
Checking writeable v0.6.4
Checking base64 v0.23.1
Checking icu_locale_core v2.3.0
Checking rand v0.10.2
Checking potential_utf v0.1.6
Checking zerotrie v0.2.5
Compiling zstd-sys v2.0.16+zstd.1.5.7
Compiling async-trait v0.1.92
Compiling icu_normalizer_data v2.3.0
Checking utf8_iter v1.0.4
Compiling icu_properties_data v2.3.0
Checking icu_collections v2.3.0
Checking icu_provider v2.3.1
Checking tokio-util v0.7.19
Checking aho-corasick v1.1.5
Checking arrow-array v59.2.0
Checking ryu v1.0.23
Compiling object v0.39.1
Compiling zstd-safe v7.2.4
Checking regex-syntax v0.8.11
Checking lexical-util v1.0.7
Checking regex-automata v0.4.18
Checking arrow-select v59.2.0
Checking icu_properties v2.3.0
Checking icu_normalizer v2.3.0
Checking either v1.18.0
Compiling rustix v1.1.4
Checking idna_adapter v1.2.2
Checking regex v1.13.1
Checking form_urlencoded v1.2.2
Compiling parking_lot_core v0.9.12
Checking typenum v1.20.1
Compiling crc32fast v1.5.1
Checking unicode-width v0.2.2
Checking idna v1.1.0
Checking lexical-write-integer v1.0.6
Checking lexical-parse-integer v1.0.6
Compiling rustc_version v0.4.1
Checking futures-executor v0.3.34
Compiling pin-project-internal v1.1.13
Checking adler2 v2.0.1
Checking simd-adler32 v0.3.10
Checking scopeguard v1.2.0
Checking lock_api v0.4.14
Checking miniz_oxide v0.9.1
Checking futures v0.3.34
Compiling flatbuffers v25.12.19
Compiling ar_archive_writer v0.5.3
Checking pin-project v1.1.13
Checking lexical-parse-float v1.0.6
Checking lexical-write-float v1.0.6
Checking url v2.5.8
Checking zlib-rs v0.6.7
Compiling psm v0.1.32
Checking unicode-segmentation v1.13.3
Checking comfy-table v7.2.2
Checking lexical-core v1.0.6
Checking itertools v0.14.0
Checking arrow-ord v59.2.0
Checking flate2 v1.1.10
Checking twox-hash v2.1.4
Checking atoi v2.0.0
Compiling stacker v0.1.25
Checking hex v0.4.3
Checking alloc-no-stdlib v2.0.4
Compiling snap v1.1.2
Checking alloc-stdlib v0.2.4
Checking arrow-cast v59.2.0
Checking lz4_flex v0.14.0
Checking parking_lot v0.12.5
Checking csv-core v0.1.13
Checking same-file v1.0.6
Checking humantime v2.4.0
Checking simdutf8 v0.1.5
Checking csv v1.4.0
Checking walkdir v2.5.0
Checking brotli-decompressor v5.0.3
Compiling recursive-proc-macro-impl v0.1.1
Checking subtle v2.6.1
Checking arrow-csv v59.2.0
Checking recursive v0.1.1
Checking arrow-json v59.2.0
Checking object_store v0.13.2
Checking brotli v8.0.4
Checking arrow-string v59.2.0
Checking zstd v0.13.3
Checking arrow-ipc v59.2.0
Checking arrow-arith v59.2.0
Checking arrow-row v59.2.0
Checking uuid v1.26.0
Compiling sqlparser_derive v0.5.0
Compiling seq-macro v0.3.6
Checking arrow v59.2.0
Checking itertools v0.15.0
Checking ppv-lite86 v0.2.21
Checking sqlparser v0.62.0
Checking hybrid-array v0.4.14
Checking parquet v59.2.0
Checking cmov v0.5.4
Checking linux-raw-sys v0.12.1
Checking ctutils v0.4.2
Checking crypto-common v0.2.2
Checking block-buffer v0.12.1
Checking rand_core v0.9.5
Compiling generic-array v0.14.7
Checking const-oid v0.10.2
Checking digest v0.11.3
Checking rand_chacha v0.9.0
Checking rand v0.9.5
Checking datafusion-doc v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/doc)
Checking foldhash v0.1.5
Compiling crossbeam-utils v0.8.22
Checking hashbrown v0.15.5
Checking block-buffer v0.10.4
Checking crypto-common v0.1.7
Compiling heck v0.5.0
Checking fastrand v2.5.0
Checking fixedbitset v0.5.7
Checking tempfile v3.27.0
Checking petgraph v0.8.3
Checking digest v0.10.7
Checking sha2 v0.11.0
Checking md-5 v0.11.0
Checking hashbrown v0.14.5
Compiling datafusion-macros v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/macros)
Checking dashmap v6.2.1
Compiling blake3 v1.8.7
Compiling anyhow v1.0.104
Checking arrayvec v0.7.8
Checking constant_time_eq v0.4.2
Checking blake2 v0.10.6
Compiling liblzma-sys v0.4.8
Checking libbz2-rs-sys v0.2.5
Compiling prost-derive v0.14.4
Checking bzip2 v0.6.1
Checking datafusion-common-runtime v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/common-runtime)
Compiling httparse v1.10.1
Checking compression-core v0.4.32
Checking http-body v1.1.0
Checking glob v0.3.4
Checking base64 v0.22.1
Checking tower-service v0.3.3
Checking atomic-waker v1.1.2
Checking try-lock v0.2.5
Checking fnv v1.0.7
Checking h2 v0.4.19
Checking want v0.3.1
Checking tokio-stream v0.1.19
Checking zeroize v1.9.0
Checking httpdate v1.0.3
Checking hyper v1.11.1
Checking datafusion-common v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/common)
Checking rustls-pki-types v1.15.1
Compiling prost v0.14.4
Checking hyper-util v0.1.20
Checking http-body-util v0.1.5
Checking num-bigint v0.4.8
Checking datafusion-expr-common v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/expr-common)
Checking sync_wrapper v1.0.2
Compiling rustls v0.23.43
Compiling prettyplease v0.2.37
Checking tower-layer v0.3.3
Checking datafusion-physical-expr-common v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/physical-expr-common)
Compiling prost-types v0.14.4
Checking datafusion-functions-window-common v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/functions-window-common)
Checking datafusion-functions-aggregate-common v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/functions-aggregate-common)
Checking datafusion-expr v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/expr)
Checking rustls-webpki v0.103.15
Compiling serde_derive_internals v0.29.1
Compiling schemars v0.8.22
Checking mime v0.3.17
Compiling hashbrown v0.16.1
Compiling schemars_derive v0.8.22
Checking liblzma v0.4.8
Checking compression-codecs v0.4.38
Checking async-compression v0.4.43
Checking axum-core v0.5.6
Checking datafusion-physical-expr v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/physical-expr)
Checking datafusion-execution v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/execution)
Checking datafusion-functions v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/functions)
Checking tower v0.5.3
Compiling multimap v0.10.1
Compiling dyn-clone v1.0.20
Checking matchit v0.8.4
Checking axum v0.8.9
Compiling prost-build v0.14.4
Compiling regress v0.10.5
Compiling pbjson-build v0.8.0
Checking hyper-timeout v0.5.2
Compiling ident_case v1.0.1
Compiling portable-atomic v1.15.0
Compiling strsim v0.11.1
Compiling darling_core v0.23.0
Checking datafusion-physical-plan v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/physical-plan)
Checking datafusion-physical-expr-adapter v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/physical-expr-adapter)
Compiling typify-impl v0.5.0
Checking tonic v0.14.6
Compiling serde_tokenstream v0.2.3
Checking ureq-proto v0.6.1
Compiling bigdecimal v0.4.10
Checking crc-catalog v2.5.0
Checking utf8-zero v0.8.1
Checking utf8parse v0.2.2
Checking tinyvec_macros v0.1.1
Compiling bollard-buildkit-proto v0.7.0
Checking tinyvec v1.12.0
Checking anstyle-parse v1.0.0
Checking ureq v3.4.0
Checking crc v3.4.0
Compiling darling_macro v0.23.0
Checking tonic-prost v0.14.6
Checking datafusion-functions-aggregate v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/functions-aggregate)
Compiling strum_macros v0.28.0
Compiling typify-macro v0.5.0
Compiling structmeta-derive v0.3.0
Checking deranged v0.5.8
Checking time-core v0.1.9
Checking powerfmt v0.2.0
Checking anstyle-query v1.1.5
Checking num-conv v0.2.2
Compiling unsafe-libyaml v0.2.11
Checking colorchoice v1.0.5
Checking is_terminal_polyfill v1.70.2
Checking anstyle v1.0.14
Checking time v0.3.55
Checking anstream v1.0.0
Compiling serde_yaml v0.9.34+deprecated
Compiling structmeta v0.3.0
Checking datafusion-functions-nested v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/functions-nested)
Checking arrow-avro v59.2.0
Checking datafusion-session v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/session)
Compiling typify v0.5.0
Checking datafusion-datasource v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/datasource)
Compiling darling v0.23.0
Checking unicode-normalization v0.1.25
Compiling pbjson-types v0.8.0
Checking datafusion-pruning v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/pruning)
Checking datafusion-catalog v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/catalog)
Checking tokio-rustls v0.26.4
Checking num-rational v0.4.2
Checking num-iter v0.1.46
Compiling async-stream-impl v0.3.6
Compiling serde_repr v0.1.21
Checking openssl-probe v0.2.1
Checking unicode-bidi v0.3.18
Checking unicode-properties v0.1.4
Checking clap_lex v1.1.0
Checking bollard-stubs v1.52.1-rc.29.1.3
Checking stringprep v0.1.5
Checking clap_builder v4.6.6
Checking datafusion-catalog-listing v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/catalog-listing)
Checking datafusion-functions-table v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/functions-table)
Checking async-stream v0.3.6
Checking rustls-native-certs v0.8.4
Checking num v0.4.3
Checking datafusion-physical-optimizer v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/physical-optimizer)
Checking datafusion-datasource-parquet v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/datasource-parquet)
Checking hyper-rustls v0.27.9
Checking datafusion-sql v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/sql)
Checking datafusion-datasource-json v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/datasource-json)
Checking datafusion-datasource-csv v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/datasource-csv)
Checking datafusion-datasource-avro v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/datasource-avro)
Checking datafusion-datasource-arrow v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/datasource-arrow)
Compiling serde_with_macros v3.22.0
Compiling substrait v0.63.0
Compiling parse-display-derive v0.9.1
Checking datafusion-optimizer v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/optimizer)
Checking datafusion-functions-window v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/functions-window)
Checking hyperlocal v0.9.1
Checking pbjson v0.8.0
Compiling clap_derive v4.6.4
Checking xattr v1.6.1
Checking hmac v0.13.0
Checking serde_urlencoded v0.7.1
Checking rand_core v0.6.4
Compiling enum-ordinalize-derive v4.4.2
Compiling fs-err v3.3.1
Checking byteorder v1.5.0
Checking home v0.5.12
Checking fallible-iterator v0.2.0
Checking rustc-hash v2.1.3
Checking clap v4.6.6
Checking astral-tokio-tar v0.6.4
Checking postgres-protocol v0.6.12
Checking datafusion v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/core)
Compiling enum-ordinalize v4.4.2
Checking rand_chacha v0.3.1
Checking parse-display v0.9.1
Checking bollard v0.20.2
Checking serde_with v3.22.0
Checking ferroid v2.0.0
Checking docker_credential v1.4.0
Compiling postgres-derive v0.4.9
Checking phf_shared v0.13.1
Checking etcetera v0.11.0
Checking unicode-width v0.1.14
Checking escape8259 v0.5.3
Checking subst v0.3.8
Checking libtest-mimic v0.8.2
Checking postgres-types v0.2.14
Checking phf v0.13.1
Checking rand v0.8.8
Compiling educe v0.6.0
Checking testcontainers v0.27.3
Checking md-5 v0.10.6
Checking sha1 v0.11.0
Checking console v0.16.4
Checking itertools v0.13.0
Compiling async-recursion v1.1.1
Checking whoami v2.1.3
Checking owo-colors v4.4.0
Checking similar v2.7.0
Checking unit-prefix v0.5.2
Checking indicatif v0.18.6
Checking tokio-postgres v0.7.18
Checking testcontainers-modules v0.15.0
Checking sqllogictest v0.29.1
Checking datafusion-spark v55.0.0 (/home/runner/work/datafusion/datafusion/datafusion/spark)
error[E0433]: cannot find type `BinaryArray` in this scope
--> /home/runner/work/datafusion/datafusion/datafusion/spark/src/function/string/encode.rs:217:39
|
217 | DataType::Null => Ok(Arc::new(BinaryArray::new_null(arr.len()))),
| ^^^^^^^^^^^ use of undeclared type `BinaryArray`
|
help: consider importing this type alias
|
18 + use arrow::array::BinaryArray;
|
For more information about this error, try `rustc --explain E0433`.
error: could not compile `datafusion-spark` (lib) due to 1 previous error
warning: build failed, waiting for other jobs to finish...
-----
error: failed to build rustdoc for crate datafusion-sqllogictest v55.0.0
note: this is usually due to a compilation error in the crate,
and is unlikely to be a bug in cargo-semver-checks
note: the following command can be used to reproduce the error:
cargo new --lib example &&
cd example &&
echo '[workspace]' >> Cargo.toml &&
cargo add --path /home/runner/work/datafusion/datafusion/datafusion/sqllogictest --features avro,backtrace,bytes,chrono,datafusion-substrait,parquet_encryption,postgres,postgres-types,substrait,testcontainers-modules,tokio-postgres &&
cargo check &&
cargo doc
error: aborting due to failure to build rustdoc for crate datafusion-sqllogictest v55.0.0
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "Jefffrey (via GitHub)" <gi...@apache.org>.
Jefffrey commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-5563478261
thanks @JeelRajodiya & co
sorry it hung for so long
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3069645705
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,554 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
Review Comment:
Is returning `?` here rather than throwing an exception the correct behavior to match Spark's default behavior (legacyErrorAction=false)? My understanding is that legacyErrorAction=true would replace with `?` and legacyErrorAction=false should throw an exception
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3035405731
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,500 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {:?}", dt),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+pub(crate) fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
Review Comment:
@xanderbailey Not now, but in future when we add `decode` function this can be reused.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3889011488
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,360 @@
+// 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.
+
+use std::borrow::Cow;
+use std::sync::Arc;
+
+use arrow::array::{
+ Array, ArrayRef, AsArray, BinaryBuilder, StringArray, StringArrayType,
+};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::utils::take_function_args;
+use datafusion_common::{Result, ScalarValue, exec_err, plan_err};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// The charset for each row. `Constant` is a single uppercased charset; `PerRow`
+/// uppercases each row on demand, yielding `None` for null entries.
+enum Charsets<'a> {
+ Constant(&'a str),
+ PerRow(&'a StringArray),
+}
+
+impl Charsets<'_> {
+ fn get(&self, i: usize) -> Option<Cow<'_, str>> {
+ match self {
+ Charsets::Constant(charset) => Some(Cow::Borrowed(charset)),
+ Charsets::PerRow(array) => {
+ (!array.is_null(i)).then(|| Cow::Owned(array.value(i).to_uppercase()))
+ }
+ }
+ }
+}
+
+fn null_binary_array(len: usize) -> ArrayRef {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..len {
+ builder.append_null();
+ }
+ Arc::new(builder.finish())
+}
+
+/// Encodes each row to binary. A row is null if its value or charset is null;
+/// otherwise it is `encode_string(decode(i), charset)`. `decode` yields the row's
+/// value as text — a borrow for string input, or lossy UTF-8 for binary input
+/// (invalid bytes become U+FFFD, matching Spark).
+fn encode_rows<'a>(
+ len: usize,
+ is_null: impl Fn(usize) -> bool,
+ decode: impl Fn(usize) -> Cow<'a, str>,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ let mut builder = BinaryBuilder::with_capacity(len, len * 4);
+ for i in 0..len {
+ match (is_null(i), charsets.get(i)) {
+ (false, Some(charset)) => {
+ builder.append_value(&encode_string(&decode(i), &charset)?)
+ }
+ _ => builder.append_null(),
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+fn encode_array<'a, S: StringArrayType<'a>>(
+ array: &S,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| Cow::Borrowed(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ array: &'a B,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| String::from_utf8_lossy(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_dispatch(arr: &ArrayRef, charsets: &Charsets) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charsets),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charsets),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charsets),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charsets),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charsets),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charsets),
+ DataType::Null => Ok(null_binary_array(arr.len())),
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// The uppercased charset from a scalar, or `None` if it is null.
+fn scalar_charset(scalar: &ScalarValue) -> Result<Option<String>> {
+ match scalar {
+ ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) | ScalarValue::Utf8View(s) => {
+ Ok(s.as_ref().map(|s| s.to_uppercase()))
+ }
+ ScalarValue::Null => Ok(None),
+ other => exec_err!("encode charset argument must be a string, got {other:?}"),
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn coerce_types(&self, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ let [value_type, charset_type] = take_function_args(self.name(), arg_types)?;
+
+ let value_type = match value_type {
+ DataType::Utf8
+ | DataType::LargeUtf8
+ | DataType::Utf8View
+ | DataType::Binary
Review Comment:
Not directly, but Spark Catalyst’s Analyzer inserts the cast specifically the [`ImplicitTypeCasts`](https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala#L188-L238) coercion rule.
Thus:
```sql
encode(binary_value, 'UTF-8')
```
is analyzed as:
```sql
encode(CAST(binary_value AS STRING), 'UTF-8')
```
We preserve binary input in `coerce_types` instead of letting DataFusion cast it to a string because:
- Spark’s binary-to-string cast preserves malformed bytes, and `encode` later converts them to `U+FFFD`.
- DataFusion’s normal Arrow binary-to-UTF8 cast rejects malformed UTF-8 before `encode` runs.
**Example:** `X'FF'` is malformed UTF-8. Spark’s `encode` path converts it to `U+FFFD`, while DataFusion’s normal binary-to-UTF8 cast returns an invalid UTF-8 error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3491184535
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,557 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
+ match charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Utf8(Some(s))
+ | ScalarValue::LargeUtf8(Some(s))
+ | ScalarValue::Utf8View(Some(s)) => Ok(s.to_uppercase()),
+ _ => exec_err!("encode charset argument must be a non-null string"),
+ },
+ ColumnarValue::Array(arr) => {
+ fn validate_constant_charset<'a, S: StringArrayType<'a>>(
+ arr: &'a S,
+ ) -> Result<String> {
+ if arr.is_null(0) {
+ return exec_err!(
+ "encode charset argument must be a non-null string"
+ );
+ }
+ let charset = arr.value(0).to_uppercase();
+ for i in 1..arr.len() {
+ if arr.is_null(i) || arr.value(i).to_uppercase() != charset {
+ return exec_err!(
+ "encode charset argument must be constant across all rows"
+ );
+ }
+ }
+ Ok(charset)
+ }
+ match arr.data_type() {
+ DataType::Utf8 => validate_constant_charset(&arr.as_string::<i32>()),
+ DataType::LargeUtf8 => validate_constant_charset(&arr.as_string::<i64>()),
+ DataType::Utf8View => validate_constant_charset(&arr.as_string_view()),
+ dt => exec_err!("encode charset argument must be a string, got {dt:?}"),
+ }
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ datafusion_common::internal_err!(
+ "return_type should not be called, use return_field_from_args instead"
+ )
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<FieldRef> {
+ let nullable = args.arg_fields.iter().any(|f| f.is_nullable());
+ Ok(Arc::new(Field::new(
+ self.name(),
+ DataType::Binary,
+ nullable,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ if args.args.len() != 2 {
+ return exec_err!(
+ "encode requires exactly 2 arguments, got {}",
+ args.args.len()
+ );
+ }
+
+ let charset = extract_charset(&args.args[1])?;
+
+ // Determine if the result should be scalar or array
+ let len = args.args.iter().find_map(|arg| match arg {
+ ColumnarValue::Array(a) => Some(a.len()),
+ _ => None,
+ });
+ let inferred_length = len.unwrap_or(1);
+ let is_scalar = len.is_none();
+
+ let string_arr = match &args.args[0] {
+ ColumnarValue::Scalar(scalar) => scalar.to_array_of_size(inferred_length)?,
+ ColumnarValue::Array(array) => Arc::clone(array),
+ };
+
+ let result = encode_dispatch(&string_arr, &charset)?;
+
+ if is_scalar {
+ ScalarValue::try_from_array(&result, 0).map(ColumnarValue::Scalar)
+ } else {
+ Ok(ColumnarValue::Array(result))
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{BinaryArray, StringViewArray};
+ use datafusion_common::config::ConfigOptions;
+
+ /// Helper to invoke encode as a scalar with two literal string arguments.
+ fn eval_encode_scalar(input: ScalarValue, charset: &str) -> Result<ColumnarValue> {
Review Comment:
Done!
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,557 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
+ match charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Utf8(Some(s))
+ | ScalarValue::LargeUtf8(Some(s))
+ | ScalarValue::Utf8View(Some(s)) => Ok(s.to_uppercase()),
+ _ => exec_err!("encode charset argument must be a non-null string"),
+ },
+ ColumnarValue::Array(arr) => {
+ fn validate_constant_charset<'a, S: StringArrayType<'a>>(
+ arr: &'a S,
+ ) -> Result<String> {
+ if arr.is_null(0) {
+ return exec_err!(
+ "encode charset argument must be a non-null string"
+ );
+ }
+ let charset = arr.value(0).to_uppercase();
+ for i in 1..arr.len() {
+ if arr.is_null(i) || arr.value(i).to_uppercase() != charset {
+ return exec_err!(
+ "encode charset argument must be constant across all rows"
+ );
+ }
+ }
+ Ok(charset)
+ }
+ match arr.data_type() {
+ DataType::Utf8 => validate_constant_charset(&arr.as_string::<i32>()),
+ DataType::LargeUtf8 => validate_constant_charset(&arr.as_string::<i64>()),
+ DataType::Utf8View => validate_constant_charset(&arr.as_string_view()),
+ dt => exec_err!("encode charset argument must be a string, got {dt:?}"),
+ }
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ datafusion_common::internal_err!(
+ "return_type should not be called, use return_field_from_args instead"
+ )
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<FieldRef> {
+ let nullable = args.arg_fields.iter().any(|f| f.is_nullable());
+ Ok(Arc::new(Field::new(
+ self.name(),
+ DataType::Binary,
+ nullable,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ if args.args.len() != 2 {
+ return exec_err!(
+ "encode requires exactly 2 arguments, got {}",
+ args.args.len()
+ );
+ }
Review Comment:
Done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4268946419
Thanks for iterating on this @JeelRajodiya. One issue I noticed:
```
spark-sql> SELECT hex(encode('A', 'UTF-32'));
00000041
```
This PR returns `0000FEFF00000041`, with a BOM. Both Spark 3.5 and Spark 4.1 return the four-byte form.
Once challenge for this PR is that there is different behavior across Spark versions for the `encode` expression. Which Spark version is this PR targeting? It would be good to document that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4416114820
@andygrove Can you reivew the PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3084153683
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,554 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
Review Comment:
I missed adding support for UTF-32, I've added it now with respective tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "Jefffrey (via GitHub)" <gi...@apache.org>.
Jefffrey commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3433258274
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,557 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
+ match charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Utf8(Some(s))
+ | ScalarValue::LargeUtf8(Some(s))
+ | ScalarValue::Utf8View(Some(s)) => Ok(s.to_uppercase()),
+ _ => exec_err!("encode charset argument must be a non-null string"),
+ },
+ ColumnarValue::Array(arr) => {
+ fn validate_constant_charset<'a, S: StringArrayType<'a>>(
+ arr: &'a S,
+ ) -> Result<String> {
+ if arr.is_null(0) {
+ return exec_err!(
+ "encode charset argument must be a non-null string"
+ );
+ }
+ let charset = arr.value(0).to_uppercase();
+ for i in 1..arr.len() {
+ if arr.is_null(i) || arr.value(i).to_uppercase() != charset {
+ return exec_err!(
+ "encode charset argument must be constant across all rows"
+ );
+ }
Review Comment:
Spark seems to support charsets per row
```python
>>> spark.sql("select encode(a, b) from values (X'a', 'US-ASCII'), (X'deadbeef', 'UTF-16be') t(a, b)").show()
+-------------------+
| encode(a, b)|
+-------------------+
| [0A]|
|[07 AD FF FD FF FD]|
+-------------------+
```
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,557 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
+ match charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Utf8(Some(s))
+ | ScalarValue::LargeUtf8(Some(s))
+ | ScalarValue::Utf8View(Some(s)) => Ok(s.to_uppercase()),
+ _ => exec_err!("encode charset argument must be a non-null string"),
+ },
+ ColumnarValue::Array(arr) => {
+ fn validate_constant_charset<'a, S: StringArrayType<'a>>(
+ arr: &'a S,
+ ) -> Result<String> {
+ if arr.is_null(0) {
+ return exec_err!(
+ "encode charset argument must be a non-null string"
+ );
+ }
+ let charset = arr.value(0).to_uppercase();
+ for i in 1..arr.len() {
+ if arr.is_null(i) || arr.value(i).to_uppercase() != charset {
+ return exec_err!(
+ "encode charset argument must be constant across all rows"
+ );
+ }
+ }
+ Ok(charset)
+ }
+ match arr.data_type() {
+ DataType::Utf8 => validate_constant_charset(&arr.as_string::<i32>()),
+ DataType::LargeUtf8 => validate_constant_charset(&arr.as_string::<i64>()),
+ DataType::Utf8View => validate_constant_charset(&arr.as_string_view()),
+ dt => exec_err!("encode charset argument must be a string, got {dt:?}"),
+ }
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ datafusion_common::internal_err!(
+ "return_type should not be called, use return_field_from_args instead"
+ )
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<FieldRef> {
+ let nullable = args.arg_fields.iter().any(|f| f.is_nullable());
+ Ok(Arc::new(Field::new(
+ self.name(),
+ DataType::Binary,
+ nullable,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ if args.args.len() != 2 {
+ return exec_err!(
+ "encode requires exactly 2 arguments, got {}",
+ args.args.len()
+ );
+ }
+
+ let charset = extract_charset(&args.args[1])?;
+
+ // Determine if the result should be scalar or array
+ let len = args.args.iter().find_map(|arg| match arg {
+ ColumnarValue::Array(a) => Some(a.len()),
+ _ => None,
+ });
+ let inferred_length = len.unwrap_or(1);
+ let is_scalar = len.is_none();
+
+ let string_arr = match &args.args[0] {
+ ColumnarValue::Scalar(scalar) => scalar.to_array_of_size(inferred_length)?,
+ ColumnarValue::Array(array) => Arc::clone(array),
+ };
+
+ let result = encode_dispatch(&string_arr, &charset)?;
+
+ if is_scalar {
+ ScalarValue::try_from_array(&result, 0).map(ColumnarValue::Scalar)
+ } else {
+ Ok(ColumnarValue::Array(result))
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{BinaryArray, StringViewArray};
+ use datafusion_common::config::ConfigOptions;
+
+ /// Helper to invoke encode as a scalar with two literal string arguments.
+ fn eval_encode_scalar(input: ScalarValue, charset: &str) -> Result<ColumnarValue> {
Review Comment:
Could we move all these tests into SLTs?
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,557 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
Review Comment:
Are the binary paths actually in use? The signature of the function seems to imply they are cast to string as part of coercion
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,557 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
+ match charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Utf8(Some(s))
+ | ScalarValue::LargeUtf8(Some(s))
+ | ScalarValue::Utf8View(Some(s)) => Ok(s.to_uppercase()),
+ _ => exec_err!("encode charset argument must be a non-null string"),
+ },
+ ColumnarValue::Array(arr) => {
+ fn validate_constant_charset<'a, S: StringArrayType<'a>>(
+ arr: &'a S,
+ ) -> Result<String> {
+ if arr.is_null(0) {
+ return exec_err!(
+ "encode charset argument must be a non-null string"
+ );
+ }
+ let charset = arr.value(0).to_uppercase();
+ for i in 1..arr.len() {
+ if arr.is_null(i) || arr.value(i).to_uppercase() != charset {
+ return exec_err!(
+ "encode charset argument must be constant across all rows"
+ );
+ }
+ }
+ Ok(charset)
+ }
+ match arr.data_type() {
+ DataType::Utf8 => validate_constant_charset(&arr.as_string::<i32>()),
+ DataType::LargeUtf8 => validate_constant_charset(&arr.as_string::<i64>()),
+ DataType::Utf8View => validate_constant_charset(&arr.as_string_view()),
+ dt => exec_err!("encode charset argument must be a string, got {dt:?}"),
+ }
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ datafusion_common::internal_err!(
+ "return_type should not be called, use return_field_from_args instead"
+ )
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<FieldRef> {
+ let nullable = args.arg_fields.iter().any(|f| f.is_nullable());
+ Ok(Arc::new(Field::new(
+ self.name(),
+ DataType::Binary,
+ nullable,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ if args.args.len() != 2 {
+ return exec_err!(
+ "encode requires exactly 2 arguments, got {}",
+ args.args.len()
+ );
+ }
Review Comment:
```suggestion
let [arg, charset] = take_function_args("encode", args.args)?;
```
More ergonomic way of destructuring expected args
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,557 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
+ match charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Utf8(Some(s))
+ | ScalarValue::LargeUtf8(Some(s))
+ | ScalarValue::Utf8View(Some(s)) => Ok(s.to_uppercase()),
+ _ => exec_err!("encode charset argument must be a non-null string"),
+ },
+ ColumnarValue::Array(arr) => {
+ fn validate_constant_charset<'a, S: StringArrayType<'a>>(
+ arr: &'a S,
+ ) -> Result<String> {
+ if arr.is_null(0) {
+ return exec_err!(
+ "encode charset argument must be a non-null string"
Review Comment:
Spark allows null charset
```python
>>> spark.sql("select encode(a, b) from values (X'a', 'US-ASCII'), (X'deadbeef', null) t(a, b)").show()
+------------+
|encode(a, b)|
+------------+
| [0A]|
| NULL|
+------------+
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4208499064
Please rerun the tasks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3035515940
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,500 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {:?}", dt),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+pub(crate) fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
Review Comment:
sure!
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,500 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {:?}", dt),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+pub(crate) fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
Review Comment:
sure! will make it private
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4182977077
Hi @Zeel-e6x, thanks for the request (https://github.com/apache/datafusion/pull/21331#issuecomment-4182976963). Only whitelisted users can trigger benchmarks. Allowed users: [Dandandan](https://github.com/Dandandan), [Fokko](https://github.com/Fokko), [Jefffrey](https://github.com/Jefffrey), [Omega359](https://github.com/Omega359), [adriangb](https://github.com/adriangb), [alamb](https://github.com/alamb), [asubiotto](https://github.com/asubiotto), [brunal](https://github.com/brunal), [buraksenn](https://github.com/buraksenn), [cetra3](https://github.com/cetra3), [codephage2020](https://github.com/codephage2020), [comphead](https://github.com/comphead), [erenavsarogullari](https://github.com/erenavsarogullari), [etseidl](https://github.com/etseidl), [friendlymatthew](https://github.com/friendlymatthew), [gabotechs](https://github.com/gabotechs), [geoffreyclaude](https://github.com/geoffreyclaude), [grtlr](https://github.com/grtlr), [haohuaijin](https://github.com/haohuaijin), [jon
athanc-n](https://github.com/jonathanc-n), [kevinjqliu](https://github.com/kevinjqliu), [klion26](https://github.com/klion26), [kosiew](https://github.com/kosiew), [kumarUjjawal](https://github.com/kumarUjjawal), [kunalsinghdadhwal](https://github.com/kunalsinghdadhwal), [liamzwbao](https://github.com/liamzwbao), [mbutrovich](https://github.com/mbutrovich), [mzabaluev](https://github.com/mzabaluev), [neilconway](https://github.com/neilconway), [rluvaton](https://github.com/rluvaton), [sdf-jkl](https://github.com/sdf-jkl), [timsaucer](https://github.com/timsaucer), [xudong963](https://github.com/xudong963), [zhuqi-lucas](https://github.com/zhuqi-lucas).
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3069639830
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,554 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
Review Comment:
Spark also supports `UTF-32`. It would be worth adding a comment here explaining why this isn't or can't be supported.
```
arguments = """
Arguments:
* str - a string expression
* charset - one of the charsets 'US-ASCII', 'ISO-8859-1', 'UTF-8', 'UTF-16BE', 'UTF-16LE', 'UTF-16', 'UTF-32' to encode `str` into a BINARY. It is case insensitive.
""",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3035409055
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,500 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {:?}", dt),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+pub(crate) fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
+ match charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Utf8(Some(s))
+ | ScalarValue::LargeUtf8(Some(s))
+ | ScalarValue::Utf8View(Some(s)) => Ok(s.to_uppercase()),
+ _ => exec_err!("encode charset argument must be a non-null string"),
+ },
+ ColumnarValue::Array(arr) => {
+ fn validate_constant_charset<'a, S: StringArrayType<'a>>(
+ arr: &'a S,
+ ) -> Result<String> {
+ if arr.is_null(0) {
+ return exec_err!(
+ "encode charset argument must be a non-null string"
+ );
+ }
+ let charset = arr.value(0).to_uppercase();
+ for i in 1..arr.len() {
+ if arr.is_null(i) || arr.value(i).to_uppercase() != charset {
+ return exec_err!(
+ "encode charset argument must be constant across all rows"
+ );
+ }
+ }
+ Ok(charset)
+ }
+ match arr.data_type() {
+ DataType::Utf8 => validate_constant_charset(&arr.as_string::<i32>()),
+ DataType::LargeUtf8 => validate_constant_charset(&arr.as_string::<i64>()),
+ DataType::Utf8View => validate_constant_charset(&arr.as_string_view()),
+ dt => exec_err!("encode charset argument must be a string, got {:?}", dt),
+ }
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ datafusion_common::internal_err!(
+ "return_type should not be called, use return_field_from_args instead"
+ )
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<FieldRef> {
+ let nullable = args.arg_fields.iter().any(|f| f.is_nullable());
+ Ok(Arc::new(Field::new(
+ self.name(),
+ DataType::Binary,
+ nullable,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ if args.args.len() != 2 {
+ return exec_err!(
+ "encode requires exactly 2 arguments, got {}",
+ args.args.len()
+ );
+ }
+
+ let charset = extract_charset(&args.args[1])?;
+
+ // Determine if the result should be scalar or array
+ let len = args.args.iter().find_map(|arg| match arg {
+ ColumnarValue::Array(a) => Some(a.len()),
+ _ => None,
+ });
+ let inferred_length = len.unwrap_or(1);
+ let is_scalar = len.is_none();
+
+ let string_arr = match &args.args[0] {
+ ColumnarValue::Scalar(scalar) => scalar.to_array_of_size(inferred_length)?,
+ ColumnarValue::Array(array) => Arc::clone(array),
+ };
+
+ let result = encode_dispatch(&string_arr, &charset)?;
+
+ if is_scalar {
+ ScalarValue::try_from_array(&result, 0).map(ColumnarValue::Scalar)
+ } else {
+ Ok(ColumnarValue::Array(result))
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
Review Comment:
Sure!
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,500 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
Review Comment:
Sure!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "xanderbailey (via GitHub)" <gi...@apache.org>.
xanderbailey commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4200571371
They will normally pick it up within a week or so. If not we can ping them here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4201474785
I pushed the fixes for clippy errors. @alamb Can you rerun the checks please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3491266696
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,557 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
Review Comment:
no they were not in use. I was coercing binary to string. I've fixed it now by preserving binary and string values in the `coerce_types`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "Zeel-e6x (via GitHub)" <gi...@apache.org>.
Zeel-e6x commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4182976963
run benchmarks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "xanderbailey (via GitHub)" <gi...@apache.org>.
xanderbailey commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3033120928
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,500 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
Review Comment:
Pretty sure this will handle surrogate pairs but it's worth adding some emoji tests to check?
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,500 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {:?}", dt),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+pub(crate) fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
Review Comment:
Nit: does this need to be crate public?
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,500 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {:?}", dt),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+pub(crate) fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
+ match charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Utf8(Some(s))
+ | ScalarValue::LargeUtf8(Some(s))
+ | ScalarValue::Utf8View(Some(s)) => Ok(s.to_uppercase()),
+ _ => exec_err!("encode charset argument must be a non-null string"),
+ },
+ ColumnarValue::Array(arr) => {
+ fn validate_constant_charset<'a, S: StringArrayType<'a>>(
+ arr: &'a S,
+ ) -> Result<String> {
+ if arr.is_null(0) {
+ return exec_err!(
+ "encode charset argument must be a non-null string"
+ );
+ }
+ let charset = arr.value(0).to_uppercase();
+ for i in 1..arr.len() {
+ if arr.is_null(i) || arr.value(i).to_uppercase() != charset {
+ return exec_err!(
+ "encode charset argument must be constant across all rows"
+ );
+ }
+ }
+ Ok(charset)
+ }
+ match arr.data_type() {
+ DataType::Utf8 => validate_constant_charset(&arr.as_string::<i32>()),
+ DataType::LargeUtf8 => validate_constant_charset(&arr.as_string::<i64>()),
+ DataType::Utf8View => validate_constant_charset(&arr.as_string_view()),
+ dt => exec_err!("encode charset argument must be a string, got {:?}", dt),
+ }
+ }
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ datafusion_common::internal_err!(
+ "return_type should not be called, use return_field_from_args instead"
+ )
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<FieldRef> {
+ let nullable = args.arg_fields.iter().any(|f| f.is_nullable());
+ Ok(Arc::new(Field::new(
+ self.name(),
+ DataType::Binary,
+ nullable,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ if args.args.len() != 2 {
+ return exec_err!(
+ "encode requires exactly 2 arguments, got {}",
+ args.args.len()
+ );
+ }
+
+ let charset = extract_charset(&args.args[1])?;
+
+ // Determine if the result should be scalar or array
+ let len = args.args.iter().find_map(|arg| match arg {
+ ColumnarValue::Array(a) => Some(a.len()),
+ _ => None,
+ });
+ let inferred_length = len.unwrap_or(1);
+ let is_scalar = len.is_none();
+
+ let string_arr = match &args.args[0] {
+ ColumnarValue::Scalar(scalar) => scalar.to_array_of_size(inferred_length)?,
+ ColumnarValue::Array(array) => Arc::clone(array),
+ };
+
+ let result = encode_dispatch(&string_arr, &charset)?;
+
+ if is_scalar {
+ ScalarValue::try_from_array(&result, 0).map(ColumnarValue::Scalar)
+ } else {
+ Ok(ColumnarValue::Array(result))
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
Review Comment:
LargeBinary and BinaryView tests might be nice
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3084153683
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,554 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
Review Comment:
I missed adding support for UTF-32, I've added it now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-4200339441
Hey @xanderbailey, Do I need to mention the maintainers for review?
I'm planning to open more PRs for implementing other functions but I'm waiting for this to get merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-5466046200
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-5467944167
I'm still working on the PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3889454401
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,360 @@
+// 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.
+
+use std::borrow::Cow;
+use std::sync::Arc;
+
+use arrow::array::{
+ Array, ArrayRef, AsArray, BinaryBuilder, StringArray, StringArrayType,
+};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::utils::take_function_args;
+use datafusion_common::{Result, ScalarValue, exec_err, plan_err};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// The charset for each row. `Constant` is a single uppercased charset; `PerRow`
+/// uppercases each row on demand, yielding `None` for null entries.
+enum Charsets<'a> {
+ Constant(&'a str),
+ PerRow(&'a StringArray),
+}
+
+impl Charsets<'_> {
+ fn get(&self, i: usize) -> Option<Cow<'_, str>> {
+ match self {
+ Charsets::Constant(charset) => Some(Cow::Borrowed(charset)),
+ Charsets::PerRow(array) => {
+ (!array.is_null(i)).then(|| Cow::Owned(array.value(i).to_uppercase()))
+ }
+ }
+ }
+}
+
+fn null_binary_array(len: usize) -> ArrayRef {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..len {
+ builder.append_null();
+ }
+ Arc::new(builder.finish())
+}
+
+/// Encodes each row to binary. A row is null if its value or charset is null;
+/// otherwise it is `encode_string(decode(i), charset)`. `decode` yields the row's
+/// value as text — a borrow for string input, or lossy UTF-8 for binary input
+/// (invalid bytes become U+FFFD, matching Spark).
+fn encode_rows<'a>(
+ len: usize,
+ is_null: impl Fn(usize) -> bool,
+ decode: impl Fn(usize) -> Cow<'a, str>,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ let mut builder = BinaryBuilder::with_capacity(len, len * 4);
+ for i in 0..len {
+ match (is_null(i), charsets.get(i)) {
+ (false, Some(charset)) => {
+ builder.append_value(&encode_string(&decode(i), &charset)?)
+ }
+ _ => builder.append_null(),
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+fn encode_array<'a, S: StringArrayType<'a>>(
+ array: &S,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| Cow::Borrowed(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ array: &'a B,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| String::from_utf8_lossy(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_dispatch(arr: &ArrayRef, charsets: &Charsets) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charsets),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charsets),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charsets),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charsets),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charsets),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charsets),
+ DataType::Null => Ok(null_binary_array(arr.len())),
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// The uppercased charset from a scalar, or `None` if it is null.
+fn scalar_charset(scalar: &ScalarValue) -> Result<Option<String>> {
+ match scalar {
+ ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) | ScalarValue::Utf8View(s) => {
+ Ok(s.as_ref().map(|s| s.to_uppercase()))
+ }
+ ScalarValue::Null => Ok(None),
+ other => exec_err!("encode charset argument must be a string, got {other:?}"),
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn coerce_types(&self, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ let [value_type, charset_type] = take_function_args(self.name(), arg_types)?;
+
+ let value_type = match value_type {
+ DataType::Utf8
+ | DataType::LargeUtf8
+ | DataType::Utf8View
+ | DataType::Binary
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "xanderbailey (via GitHub)" <gi...@apache.org>.
xanderbailey commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3035424596
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,500 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp > 255 { b'?' } else { cp as u8 }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {:?}", dt),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+pub(crate) fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
Review Comment:
Maybe until you need it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3491226097
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,557 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BinaryBuilder, StringArrayType};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::types::{
+ NativeType, logical_binary, logical_null, logical_string,
+};
+use datafusion_common::{Result, ScalarValue, exec_err};
+use datafusion_expr::{
+ Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+ Signature, TypeSignatureClass, Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::coercible(
+ vec![
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![
+ TypeSignatureClass::Native(logical_null()),
+ TypeSignatureClass::Native(logical_binary()),
+ ],
+ NativeType::String,
+ ),
+ Coercion::new_implicit(
+ TypeSignatureClass::Native(logical_string()),
+ vec![TypeSignatureClass::Native(logical_null())],
+ NativeType::String,
+ ),
+ ],
+ Volatility::Immutable,
+ ),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// Encodes a string array using the given charset, producing a BinaryArray.
+fn encode_array<'a, S: StringArrayType<'a>>(
+ string_array: &S,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(string_array.len(), string_array.len() * 4);
+ for i in 0..string_array.len() {
+ if string_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = string_array.value(i);
+ let encoded = encode_string(s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Encodes a binary array using lossy UTF-8 conversion, then re-encodes with the given charset.
+/// Invalid UTF-8 bytes become U+FFFD (replacement character), matching Spark behavior.
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ binary_array: &'a B,
+ charset: &str,
+) -> Result<ArrayRef> {
+ let mut builder =
+ BinaryBuilder::with_capacity(binary_array.len(), binary_array.len() * 4);
+ for i in 0..binary_array.len() {
+ if binary_array.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = String::from_utf8_lossy(binary_array.value(i));
+ let encoded = encode_string(&s, charset)?;
+ builder.append_value(&encoded);
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+/// Dispatches to the correct typed array encoder based on the DataType.
+fn encode_dispatch(arr: &ArrayRef, charset: &str) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charset),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charset),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charset),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charset),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charset),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charset),
+ DataType::Null => {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..arr.len() {
+ builder.append_null();
+ }
+ Ok(Arc::new(builder.finish()))
+ }
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// Extracts a charset string from a ColumnarValue, normalizing to uppercase.
+fn extract_charset(charset_arg: &ColumnarValue) -> Result<String> {
+ match charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Utf8(Some(s))
+ | ScalarValue::LargeUtf8(Some(s))
+ | ScalarValue::Utf8View(Some(s)) => Ok(s.to_uppercase()),
+ _ => exec_err!("encode charset argument must be a non-null string"),
+ },
+ ColumnarValue::Array(arr) => {
+ fn validate_constant_charset<'a, S: StringArrayType<'a>>(
+ arr: &'a S,
+ ) -> Result<String> {
+ if arr.is_null(0) {
+ return exec_err!(
+ "encode charset argument must be a non-null string"
+ );
+ }
+ let charset = arr.value(0).to_uppercase();
+ for i in 1..arr.len() {
+ if arr.is_null(i) || arr.value(i).to_uppercase() != charset {
+ return exec_err!(
+ "encode charset argument must be constant across all rows"
+ );
+ }
Review Comment:
I've added support for per row charset and also added slt tests for that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "Jefffrey (via GitHub)" <gi...@apache.org>.
Jefffrey commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3499475395
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,360 @@
+// 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.
+
+use std::borrow::Cow;
+use std::sync::Arc;
+
+use arrow::array::{
+ Array, ArrayRef, AsArray, BinaryBuilder, StringArray, StringArrayType,
+};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::utils::take_function_args;
+use datafusion_common::{Result, ScalarValue, exec_err, plan_err};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// The charset for each row. `Constant` is a single uppercased charset; `PerRow`
+/// uppercases each row on demand, yielding `None` for null entries.
+enum Charsets<'a> {
+ Constant(&'a str),
+ PerRow(&'a StringArray),
+}
+
+impl Charsets<'_> {
+ fn get(&self, i: usize) -> Option<Cow<'_, str>> {
+ match self {
+ Charsets::Constant(charset) => Some(Cow::Borrowed(charset)),
+ Charsets::PerRow(array) => {
+ (!array.is_null(i)).then(|| Cow::Owned(array.value(i).to_uppercase()))
+ }
+ }
+ }
+}
+
+fn null_binary_array(len: usize) -> ArrayRef {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..len {
+ builder.append_null();
+ }
+ Arc::new(builder.finish())
+}
+
+/// Encodes each row to binary. A row is null if its value or charset is null;
+/// otherwise it is `encode_string(decode(i), charset)`. `decode` yields the row's
+/// value as text — a borrow for string input, or lossy UTF-8 for binary input
+/// (invalid bytes become U+FFFD, matching Spark).
+fn encode_rows<'a>(
+ len: usize,
+ is_null: impl Fn(usize) -> bool,
+ decode: impl Fn(usize) -> Cow<'a, str>,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ let mut builder = BinaryBuilder::with_capacity(len, len * 4);
+ for i in 0..len {
+ match (is_null(i), charsets.get(i)) {
+ (false, Some(charset)) => {
+ builder.append_value(&encode_string(&decode(i), &charset)?)
+ }
+ _ => builder.append_null(),
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+fn encode_array<'a, S: StringArrayType<'a>>(
+ array: &S,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| Cow::Borrowed(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ array: &'a B,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| String::from_utf8_lossy(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_dispatch(arr: &ArrayRef, charsets: &Charsets) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charsets),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charsets),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charsets),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charsets),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charsets),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charsets),
+ DataType::Null => Ok(null_binary_array(arr.len())),
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// The uppercased charset from a scalar, or `None` if it is null.
+fn scalar_charset(scalar: &ScalarValue) -> Result<Option<String>> {
+ match scalar {
+ ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) | ScalarValue::Utf8View(s) => {
+ Ok(s.as_ref().map(|s| s.to_uppercase()))
+ }
+ ScalarValue::Null => Ok(None),
+ other => exec_err!("encode charset argument must be a string, got {other:?}"),
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn coerce_types(&self, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ let [value_type, charset_type] = take_function_args(self.name(), arg_types)?;
+
+ let value_type = match value_type {
+ DataType::Utf8
+ | DataType::LargeUtf8
+ | DataType::Utf8View
+ | DataType::Binary
+ | DataType::LargeBinary
+ | DataType::BinaryView
+ | DataType::Null => value_type.clone(),
+ other => {
+ return plan_err!(
+ "encode expects a string or binary first argument, got {other:?}"
+ );
+ }
+ };
+
+ match charset_type {
+ DataType::Utf8
+ | DataType::LargeUtf8
+ | DataType::Utf8View
+ | DataType::Null => {}
+ other => {
+ return plan_err!(
+ "encode expects a string charset second argument, got {other:?}"
+ );
+ }
+ }
+
+ Ok(vec![value_type, DataType::Utf8])
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ datafusion_common::internal_err!(
+ "return_type should not be called, use return_field_from_args instead"
+ )
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<FieldRef> {
+ let nullable = args.arg_fields.iter().any(|f| f.is_nullable());
+ Ok(Arc::new(Field::new(
+ self.name(),
+ DataType::Binary,
+ nullable,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ let [value, charset_arg] = take_function_args(self.name(), args.args)?;
+
+ let len = [&value, &charset_arg]
+ .into_iter()
+ .find_map(|arg| match arg {
+ ColumnarValue::Array(a) => Some(a.len()),
+ _ => None,
+ });
+ let inferred_length = len.unwrap_or(1);
+ let is_scalar = len.is_none();
+
+ let value_arr = match value {
+ ColumnarValue::Scalar(scalar) => scalar.to_array_of_size(inferred_length)?,
+ ColumnarValue::Array(array) => array,
+ };
+
+ // A null charset yields a null result for that row (Spark NullIntolerant).
+ let result = match &charset_arg {
+ ColumnarValue::Scalar(scalar) => match scalar_charset(scalar)? {
+ Some(charset) => {
+ encode_dispatch(&value_arr, &Charsets::Constant(&charset))?
+ }
+ None => null_binary_array(value_arr.len()),
Review Comment:
```suggestion
None => {
return Ok(ColumnarValue::Scalar(ScalarValue::Binary(None)));
}
```
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,360 @@
+// 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.
+
+use std::borrow::Cow;
+use std::sync::Arc;
+
+use arrow::array::{
+ Array, ArrayRef, AsArray, BinaryBuilder, StringArray, StringArrayType,
+};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::utils::take_function_args;
+use datafusion_common::{Result, ScalarValue, exec_err, plan_err};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// The charset for each row. `Constant` is a single uppercased charset; `PerRow`
+/// uppercases each row on demand, yielding `None` for null entries.
+enum Charsets<'a> {
+ Constant(&'a str),
+ PerRow(&'a StringArray),
+}
+
+impl Charsets<'_> {
+ fn get(&self, i: usize) -> Option<Cow<'_, str>> {
+ match self {
+ Charsets::Constant(charset) => Some(Cow::Borrowed(charset)),
+ Charsets::PerRow(array) => {
+ (!array.is_null(i)).then(|| Cow::Owned(array.value(i).to_uppercase()))
+ }
+ }
+ }
+}
+
+fn null_binary_array(len: usize) -> ArrayRef {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..len {
+ builder.append_null();
+ }
+ Arc::new(builder.finish())
+}
+
+/// Encodes each row to binary. A row is null if its value or charset is null;
+/// otherwise it is `encode_string(decode(i), charset)`. `decode` yields the row's
+/// value as text — a borrow for string input, or lossy UTF-8 for binary input
+/// (invalid bytes become U+FFFD, matching Spark).
+fn encode_rows<'a>(
+ len: usize,
+ is_null: impl Fn(usize) -> bool,
+ decode: impl Fn(usize) -> Cow<'a, str>,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ let mut builder = BinaryBuilder::with_capacity(len, len * 4);
+ for i in 0..len {
+ match (is_null(i), charsets.get(i)) {
+ (false, Some(charset)) => {
+ builder.append_value(&encode_string(&decode(i), &charset)?)
+ }
+ _ => builder.append_null(),
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+fn encode_array<'a, S: StringArrayType<'a>>(
+ array: &S,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| Cow::Borrowed(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ array: &'a B,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| String::from_utf8_lossy(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_dispatch(arr: &ArrayRef, charsets: &Charsets) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charsets),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charsets),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charsets),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charsets),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charsets),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charsets),
+ DataType::Null => Ok(null_binary_array(arr.len())),
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// The uppercased charset from a scalar, or `None` if it is null.
+fn scalar_charset(scalar: &ScalarValue) -> Result<Option<String>> {
+ match scalar {
+ ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) | ScalarValue::Utf8View(s) => {
+ Ok(s.as_ref().map(|s| s.to_uppercase()))
+ }
+ ScalarValue::Null => Ok(None),
+ other => exec_err!("encode charset argument must be a string, got {other:?}"),
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn coerce_types(&self, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ let [value_type, charset_type] = take_function_args(self.name(), arg_types)?;
+
+ let value_type = match value_type {
+ DataType::Utf8
+ | DataType::LargeUtf8
+ | DataType::Utf8View
+ | DataType::Binary
Review Comment:
does Spark also operate on binary types for this function? source code here seems to suggest only strings
https://github.com/apache/spark/blob/dd4b74184e36625aff54d3695a134839a9721bd3/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala#L3276-L3280
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3889011488
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,360 @@
+// 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.
+
+use std::borrow::Cow;
+use std::sync::Arc;
+
+use arrow::array::{
+ Array, ArrayRef, AsArray, BinaryBuilder, StringArray, StringArrayType,
+};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::utils::take_function_args;
+use datafusion_common::{Result, ScalarValue, exec_err, plan_err};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// The charset for each row. `Constant` is a single uppercased charset; `PerRow`
+/// uppercases each row on demand, yielding `None` for null entries.
+enum Charsets<'a> {
+ Constant(&'a str),
+ PerRow(&'a StringArray),
+}
+
+impl Charsets<'_> {
+ fn get(&self, i: usize) -> Option<Cow<'_, str>> {
+ match self {
+ Charsets::Constant(charset) => Some(Cow::Borrowed(charset)),
+ Charsets::PerRow(array) => {
+ (!array.is_null(i)).then(|| Cow::Owned(array.value(i).to_uppercase()))
+ }
+ }
+ }
+}
+
+fn null_binary_array(len: usize) -> ArrayRef {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..len {
+ builder.append_null();
+ }
+ Arc::new(builder.finish())
+}
+
+/// Encodes each row to binary. A row is null if its value or charset is null;
+/// otherwise it is `encode_string(decode(i), charset)`. `decode` yields the row's
+/// value as text — a borrow for string input, or lossy UTF-8 for binary input
+/// (invalid bytes become U+FFFD, matching Spark).
+fn encode_rows<'a>(
+ len: usize,
+ is_null: impl Fn(usize) -> bool,
+ decode: impl Fn(usize) -> Cow<'a, str>,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ let mut builder = BinaryBuilder::with_capacity(len, len * 4);
+ for i in 0..len {
+ match (is_null(i), charsets.get(i)) {
+ (false, Some(charset)) => {
+ builder.append_value(&encode_string(&decode(i), &charset)?)
+ }
+ _ => builder.append_null(),
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+fn encode_array<'a, S: StringArrayType<'a>>(
+ array: &S,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| Cow::Borrowed(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ array: &'a B,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| String::from_utf8_lossy(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_dispatch(arr: &ArrayRef, charsets: &Charsets) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charsets),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charsets),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charsets),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charsets),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charsets),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charsets),
+ DataType::Null => Ok(null_binary_array(arr.len())),
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// The uppercased charset from a scalar, or `None` if it is null.
+fn scalar_charset(scalar: &ScalarValue) -> Result<Option<String>> {
+ match scalar {
+ ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) | ScalarValue::Utf8View(s) => {
+ Ok(s.as_ref().map(|s| s.to_uppercase()))
+ }
+ ScalarValue::Null => Ok(None),
+ other => exec_err!("encode charset argument must be a string, got {other:?}"),
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn coerce_types(&self, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ let [value_type, charset_type] = take_function_args(self.name(), arg_types)?;
+
+ let value_type = match value_type {
+ DataType::Utf8
+ | DataType::LargeUtf8
+ | DataType::Utf8View
+ | DataType::Binary
Review Comment:
Not directly, but Spark Catalyst’s Analyzer inserts the cast specifically through the [`ImplicitTypeCasts`](https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala#L188-L238) coercion rule.
Thus:
```sql
encode(binary_value, 'UTF-8')
```
is analyzed as:
```sql
encode(CAST(binary_value AS STRING), 'UTF-8')
```
We preserve binary input in `coerce_types` instead of letting DataFusion cast it to a string because:
- Spark’s binary-to-string cast preserves malformed bytes, and `encode` later converts them to `U+FFFD`.
- DataFusion’s normal Arrow binary-to-UTF8 cast rejects malformed UTF-8 before `encode` runs.
**Example:** `X'FF'` is malformed UTF-8. Spark’s `encode` path converts it to `U+FFFD`, while DataFusion’s normal binary-to-UTF8 cast returns an invalid UTF-8 error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "Jefffrey (via GitHub)" <gi...@apache.org>.
Jefffrey commented on code in PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#discussion_r3889423310
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,362 @@
+// 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.
+
+use std::borrow::Cow;
+use std::sync::Arc;
+
+use arrow::array::{
+ Array, ArrayRef, AsArray, BinaryBuilder, StringArray, StringArrayType,
+};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::utils::take_function_args;
+use datafusion_common::{Result, ScalarValue, exec_err, plan_err};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// The charset for each row. `Constant` is a single uppercased charset; `PerRow`
+/// uppercases each row on demand, yielding `None` for null entries.
+enum Charsets<'a> {
+ Constant(&'a str),
+ PerRow(&'a StringArray),
+}
+
+impl Charsets<'_> {
+ fn get(&self, i: usize) -> Option<Cow<'_, str>> {
+ match self {
+ Charsets::Constant(charset) => Some(Cow::Borrowed(charset)),
+ Charsets::PerRow(array) => {
+ (!array.is_null(i)).then(|| Cow::Owned(array.value(i).to_uppercase()))
+ }
+ }
+ }
+}
+
+fn null_binary_array(len: usize) -> ArrayRef {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..len {
+ builder.append_null();
+ }
+ Arc::new(builder.finish())
+}
+
+/// Encodes each row to binary. A row is null if its value or charset is null;
+/// otherwise it is `encode_string(decode(i), charset)`. `decode` yields the row's
+/// value as text — a borrow for string input, or lossy UTF-8 for binary input
+/// (invalid bytes become U+FFFD, matching Spark).
+fn encode_rows<'a>(
+ len: usize,
+ is_null: impl Fn(usize) -> bool,
+ decode: impl Fn(usize) -> Cow<'a, str>,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ let mut builder = BinaryBuilder::with_capacity(len, len * 4);
+ for i in 0..len {
+ match (is_null(i), charsets.get(i)) {
+ (false, Some(charset)) => {
+ builder.append_value(&encode_string(&decode(i), &charset)?)
+ }
+ _ => builder.append_null(),
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+fn encode_array<'a, S: StringArrayType<'a>>(
+ array: &S,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| Cow::Borrowed(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ array: &'a B,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| String::from_utf8_lossy(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_dispatch(arr: &ArrayRef, charsets: &Charsets) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charsets),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charsets),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charsets),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charsets),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charsets),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charsets),
+ DataType::Null => Ok(null_binary_array(arr.len())),
Review Comment:
```suggestion
DataType::Null => Ok(Arc::new(BinaryArray::new_null(arr.len()))),
```
and can get rid of `null_binary_array()` function
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,362 @@
+// 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.
+
+use std::borrow::Cow;
+use std::sync::Arc;
+
+use arrow::array::{
+ Array, ArrayRef, AsArray, BinaryBuilder, StringArray, StringArrayType,
+};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::utils::take_function_args;
+use datafusion_common::{Result, ScalarValue, exec_err, plan_err};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
Review Comment:
```suggestion
///
/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
```
TODO shouldnt be part of the doc comment
##########
datafusion/spark/src/function/string/encode.rs:
##########
@@ -0,0 +1,360 @@
+// 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.
+
+use std::borrow::Cow;
+use std::sync::Arc;
+
+use arrow::array::{
+ Array, ArrayRef, AsArray, BinaryBuilder, StringArray, StringArrayType,
+};
+use arrow::datatypes::{DataType, Field, FieldRef};
+use datafusion_common::utils::take_function_args;
+use datafusion_common::{Result, ScalarValue, exec_err, plan_err};
+use datafusion_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+ Volatility,
+};
+
+/// Spark-compatible `encode` expression.
+/// Encodes a string or binary value into binary using the specified character encoding.
+/// Binary input is interpreted as UTF-8 with lossy conversion (invalid bytes become U+FFFD).
+///
+/// # Target Spark version
+/// Emulates Spark 3.5 semantics:
+/// - Accepts canonical charset names and common aliases (`UTF8`, `LATIN1`,
+/// `ISO88591`, `ASCII`, `UTF-32BE`, etc.).
+/// - Unmappable characters (non-ASCII in `US-ASCII`, code points above
+/// `U+00FF` in `ISO-8859-1`) are silently replaced with `?`.
+/// - `UTF-32` is an alias for `UTF-32BE` (no BOM), matching both Spark 3.5
+/// and Spark 4.1.
+///
+/// # Spark 4.0 differences (not implemented)
+/// Spark 4.0 tightened `encode` in two ways, each gated by a `spark.sql.legacy.*`
+/// config that can restore the 3.5 behavior:
+///
+/// - Charset whitelist — rejects aliases with `INVALID_PARAMETER_VALUE.CHARSET`.
+/// Controlled by `spark.sql.legacy.javaCharsets`.
+/// - Unmappable characters — raises `MALFORMED_CHARACTER_CODING`.
+/// Controlled by `spark.sql.legacy.codingErrorAction`.
+///
+/// TODO: wire both configs so Spark 4.0 behavior can be selected at runtime.
+/// See: <https://spark.apache.org/docs/4.0.0/sql-migration-guide.html>
+///
+/// <https://spark.apache.org/docs/latest/api/sql/index.html#encode>
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEncode {
+ signature: Signature,
+}
+
+impl Default for SparkEncode {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SparkEncode {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::user_defined(Volatility::Immutable),
+ }
+ }
+}
+
+/// Encodes a single string value using the specified charset.
+/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
+fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+ match charset {
+ "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
+ "US-ASCII" | "ASCII" => Ok(s
+ .chars()
+ .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
+ .collect()),
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
+ .chars()
+ .map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })
+ .collect()),
+ "UTF-16BE" | "UTF16BE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16LE" | "UTF16LE" => {
+ let mut bytes = Vec::new();
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-16" | "UTF16" => {
+ // BOM (big-endian marker) followed by UTF-16BE encoded bytes
+ let mut bytes = vec![0xFE, 0xFF];
+ for code_unit in s.encode_utf16() {
+ bytes.extend_from_slice(&code_unit.to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ // Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
+ "UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_be_bytes());
+ }
+ Ok(bytes)
+ }
+ "UTF-32LE" | "UTF32LE" => {
+ let mut bytes = Vec::with_capacity(s.len() * 4);
+ for c in s.chars() {
+ bytes.extend_from_slice(&(c as u32).to_le_bytes());
+ }
+ Ok(bytes)
+ }
+ _ => exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ ),
+ }
+}
+
+/// The charset for each row. `Constant` is a single uppercased charset; `PerRow`
+/// uppercases each row on demand, yielding `None` for null entries.
+enum Charsets<'a> {
+ Constant(&'a str),
+ PerRow(&'a StringArray),
+}
+
+impl Charsets<'_> {
+ fn get(&self, i: usize) -> Option<Cow<'_, str>> {
+ match self {
+ Charsets::Constant(charset) => Some(Cow::Borrowed(charset)),
+ Charsets::PerRow(array) => {
+ (!array.is_null(i)).then(|| Cow::Owned(array.value(i).to_uppercase()))
+ }
+ }
+ }
+}
+
+fn null_binary_array(len: usize) -> ArrayRef {
+ let mut builder = BinaryBuilder::new();
+ for _ in 0..len {
+ builder.append_null();
+ }
+ Arc::new(builder.finish())
+}
+
+/// Encodes each row to binary. A row is null if its value or charset is null;
+/// otherwise it is `encode_string(decode(i), charset)`. `decode` yields the row's
+/// value as text — a borrow for string input, or lossy UTF-8 for binary input
+/// (invalid bytes become U+FFFD, matching Spark).
+fn encode_rows<'a>(
+ len: usize,
+ is_null: impl Fn(usize) -> bool,
+ decode: impl Fn(usize) -> Cow<'a, str>,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ let mut builder = BinaryBuilder::with_capacity(len, len * 4);
+ for i in 0..len {
+ match (is_null(i), charsets.get(i)) {
+ (false, Some(charset)) => {
+ builder.append_value(&encode_string(&decode(i), &charset)?)
+ }
+ _ => builder.append_null(),
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+fn encode_array<'a, S: StringArrayType<'a>>(
+ array: &S,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| Cow::Borrowed(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_binary_array<'a, B: arrow::array::BinaryArrayType<'a>>(
+ array: &'a B,
+ charsets: &Charsets,
+) -> Result<ArrayRef> {
+ encode_rows(
+ array.len(),
+ |i| array.is_null(i),
+ |i| String::from_utf8_lossy(array.value(i)),
+ charsets,
+ )
+}
+
+fn encode_dispatch(arr: &ArrayRef, charsets: &Charsets) -> Result<ArrayRef> {
+ match arr.data_type() {
+ DataType::Utf8 => encode_array(&arr.as_string::<i32>(), charsets),
+ DataType::LargeUtf8 => encode_array(&arr.as_string::<i64>(), charsets),
+ DataType::Utf8View => encode_array(&arr.as_string_view(), charsets),
+ DataType::Binary => encode_binary_array(&arr.as_binary::<i32>(), charsets),
+ DataType::LargeBinary => encode_binary_array(&arr.as_binary::<i64>(), charsets),
+ DataType::BinaryView => encode_binary_array(&arr.as_binary_view(), charsets),
+ DataType::Null => Ok(null_binary_array(arr.len())),
+ dt => exec_err!("encode expects a string or binary argument, got {dt:?}"),
+ }
+}
+
+/// The uppercased charset from a scalar, or `None` if it is null.
+fn scalar_charset(scalar: &ScalarValue) -> Result<Option<String>> {
+ match scalar {
+ ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) | ScalarValue::Utf8View(s) => {
+ Ok(s.as_ref().map(|s| s.to_uppercase()))
+ }
+ ScalarValue::Null => Ok(None),
+ other => exec_err!("encode charset argument must be a string, got {other:?}"),
+ }
+}
+
+impl ScalarUDFImpl for SparkEncode {
+ fn name(&self) -> &str {
+ "encode"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn coerce_types(&self, arg_types: &[DataType]) -> Result<Vec<DataType>> {
+ let [value_type, charset_type] = take_function_args(self.name(), arg_types)?;
+
+ let value_type = match value_type {
+ DataType::Utf8
+ | DataType::LargeUtf8
+ | DataType::Utf8View
+ | DataType::Binary
Review Comment:
this is worth leaving a comment here about (keep it succinct and to the point)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "JeelRajodiya (via GitHub)" <gi...@apache.org>.
JeelRajodiya commented on PR #21331:
URL: https://github.com/apache/datafusion/pull/21331#issuecomment-5560895671
if it looks good, Can we get this merged @Jefffrey ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add Spark-compatible `encode` function to datafusion-spark [datafusion]
Posted by "Jefffrey (via GitHub)" <gi...@apache.org>.
Jefffrey merged PR #21331:
URL: https://github.com/apache/datafusion/pull/21331
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org