diff --git a/docs/source/contributor-guide/adding_a_new_expression.md b/docs/source/contributor-guide/adding_a_new_expression.md index 88522b09a8..6cda31f0df 100644 --- a/docs/source/contributor-guide/adding_a_new_expression.md +++ b/docs/source/contributor-guide/adding_a_new_expression.md @@ -478,6 +478,21 @@ pub fn spark_ceil( } ``` +#### Producing strings from arbitrary bytes + +Spark's `StringType` may contain bytes that are not valid UTF-8 (Spark stores them verbatim), but +Arrow's string arrays must be valid UTF-8. Any native code that turns arbitrary bytes into a string +must therefore **decode** them, not reinterpret them. Building a Rust `String`/`&str` from non-UTF-8 +bytes is undefined behaviour, and an Arrow string array that holds invalid UTF-8 is unsound for the +downstream kernels that read it via `StringArray::value` (which assumes the UTF-8 invariant). + +Use `crate::utils::decode_utf8_spark_lossy`. It replaces each ill-formed sequence with `U+FFFD` +exactly as the JVM's `new String(bytes, UTF_8)` does, so Comet's rendered output matches Spark +(including the surrogate-range cases where `str::from_utf8_lossy` would differ). This is the decoder +used by both `CAST(binary AS string)` and the native columnar shuffle. See the +[strings with non-UTF-8 bytes](../user-guide/latest/compatibility/index.md) note in the +compatibility guide for the user-visible behavior. + ### API Differences Between Spark Versions If the expression you're adding has different behavior across different Spark versions, you'll need to account for that in your implementation. There are two tools at your disposal to help with this: diff --git a/docs/source/contributor-guide/expression-audits/conversion_funcs.md b/docs/source/contributor-guide/expression-audits/conversion_funcs.md index 45b6a8f03b..be0f730064 100644 --- a/docs/source/contributor-guide/expression-audits/conversion_funcs.md +++ b/docs/source/contributor-guide/expression-audits/conversion_funcs.md @@ -28,7 +28,7 @@ - Spark 4.0.1 (audited 2026-05-27): `VariantType` added; `StringType` literals replaced with `_: StringType` to accommodate collated strings. `(TimestampType, ByteType|ShortType|IntegerType)` added to `canAnsiCast`. `NullIntolerant` -> `nullIntolerant: Boolean` refactor. New `ToPrettyString.BinaryFormatter` semantics for `Binary -> String` are replicated natively via `spark_binary_formatter`. Numeric-to-numeric matrix unchanged. - Spark 4.1.1 (audited 2026-05-27): `TimeType` added; many `TimeType` arms in `canCast`/`canAnsiCast`. Geospatial `GeographyType` / `GeometryType` types added with their own conversion rules. Numeric-to-numeric matrix unchanged. - Known divergences and gaps: - - `CAST( AS STRING)` uses `unsafe { String::from_utf8_unchecked }` in native code, which is undefined behaviour for non-UTF8 inputs (https://github.com/apache/datafusion-comet/issues/4488). + - `CAST( AS STRING)` decodes the bytes with `decode_utf8_spark_lossy`, replacing invalid UTF-8 with `U+FFFD` exactly as the JVM's `new String(bytes, UTF_8)` does (the same decoder as the native shuffle). It matches Spark's rendered output and diverges only under byte-level round-trips such as `CAST(CAST(x AS STRING) AS BINARY)`. - Spark 4.0 collated `StringType` is not explicitly guarded; pattern equality is expected to keep collated-string casts falling back, but there is no test (https://github.com/apache/datafusion-comet/issues/4489; umbrella #2190). - Spark 4.1 `TimeType` casts have no explicit `Unsupported` arm; they fall back implicitly but do not appear in the auto-generated compatibility doc (https://github.com/apache/datafusion-comet/issues/4490). - `CAST( AS )` falls back to Spark even though native `cast_map_to_map` exists (https://github.com/apache/datafusion-comet/issues/4491). diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index f3b531a18f..fe2361b6a8 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -70,3 +70,23 @@ This is distinct from expressions that have **no** codegen-dispatch path: there, incompatible cases fall back to Spark by default, and `allowIncompatible=true` runs the native (incompatible) path instead. `cast` is the main example; see the [expression reference](../expressions.md) for which expressions have incompatible cases. + +## Strings with non-UTF-8 bytes + +Spark's `StringType` can hold arbitrary bytes, including sequences that are not valid UTF-8 (for +example `CAST(X'FF' AS STRING)`). Arrow's string type requires valid UTF-8, so Comet cannot store +the raw bytes natively. When Comet produces a string from arbitrary bytes (such as +`CAST(binary AS string)` or a columnar shuffle), it decodes them the same way the JVM does +(`new String(bytes, UTF_8)`), replacing each ill-formed sequence with the Unicode replacement +character `U+FFFD`. Spark itself applies the identical replacement whenever such a string is +materialized (collected, printed, or passed to most string functions), so the rendered result +matches Spark. + +The one observable difference is byte-preserving round-trips: Spark keeps the original bytes, so +`CAST(CAST(X'FF' AS STRING) AS BINARY)` returns `X'FF'`, whereas Comet returns the UTF-8 encoding of +`U+FFFD` (`X'EFBFBD'`). Operations that inspect the raw bytes (re-casting to binary, `octet_length`, +hashing) can therefore differ for non-UTF-8 input. + +Separately, Comet's native Parquet scan currently rejects string columns whose stored bytes are not +valid UTF-8 rather than reading them like Spark +([#4121](https://github.com/apache/datafusion-comet/issues/4121)). diff --git a/native/common/src/lib.rs b/native/common/src/lib.rs index a9549badb1..b2d4a57431 100644 --- a/native/common/src/lib.rs +++ b/native/common/src/lib.rs @@ -22,4 +22,4 @@ mod utils; pub use error::{decimal_overflow_error, SparkError, SparkErrorWithContext, SparkResult}; pub use query_context::{create_query_context_map, QueryContext, QueryContextMap}; -pub use utils::bytes_to_i128; +pub use utils::{bytes_to_i128, decode_utf8_spark_lossy}; diff --git a/native/common/src/utils.rs b/native/common/src/utils.rs index 12283db30d..1838d87d40 100644 --- a/native/common/src/utils.rs +++ b/native/common/src/utils.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +use std::borrow::Cow; + /// Converts a slice of bytes to i128. The bytes are serialized in big-endian order by /// `BigInteger.toByteArray()` in Java. pub fn bytes_to_i128(slice: &[u8]) -> i128 { @@ -35,3 +37,175 @@ pub fn bytes_to_i128(slice: &[u8]) -> i128 { i128::from_le_bytes(bytes) } + +/// Decode `bytes` as UTF-8 the way Spark renders `StringType` -- `new String(bytes, UTF_8)` on the +/// JVM -- replacing each ill-formed sequence with a single `U+FFFD` and skipping the same number of +/// bytes the JDK's UTF-8 `CharsetDecoder` (action REPLACE) would. Valid UTF-8 is returned as a +/// zero-cost borrow. +/// +/// This intentionally differs from `str::from_utf8_lossy` for surrogate-range three-byte sequences +/// (`ED A0..BF ..`, e.g. CESU-8 / Java modified-UTF-8 supplementary chars) and for some other +/// ill-formed multi-byte units: `from_utf8_lossy` follows the Unicode "maximal subpart" rule and +/// can emit one `U+FFFD` per byte, whereas the JDK collapses certain ill-formed units into a single +/// `U+FFFD`. Matching the JDK byte-for-byte means Comet renders arbitrary bytes identically to +/// Spark -- whether they arrive via a columnar shuffle or a `CAST(binary AS string)`. The +/// per-class malformed lengths below (E0/ED overlong & surrogate handling, F0/F4 range checks) +/// match the observable replacement behavior of the JDK UTF-8 decoder; they were determined from +/// observed `new String(bytes, UTF_8)` output, not by reviewing the OpenJDK source. +pub fn decode_utf8_spark_lossy(bytes: &[u8]) -> Cow<'_, str> { + // Fast path: well-formed UTF-8 borrows with zero copy (the overwhelmingly common case). + if let Ok(s) = std::str::from_utf8(bytes) { + return Cow::Borrowed(s); + } + + const RC: char = '\u{FFFD}'; + let n = bytes.len(); + let mut out = String::with_capacity(n); + let mut i = 0; + while i < n { + let b1 = bytes[i]; + if b1 < 0x80 { + out.push(b1 as char); + i += 1; + } else if (0xC2..=0xDF).contains(&b1) { + // 2-byte lead. Bad/absent continuation -> single FFFD, skip 1. + if i + 1 < n && (bytes[i + 1] & 0xC0) == 0x80 { + let cp = (((b1 as u32) & 0x1F) << 6) | ((bytes[i + 1] as u32) & 0x3F); + // guards above keep cp a valid scalar (2-byte range, never a surrogate) + out.push(char::from_u32(cp).unwrap()); + i += 2; + } else { + out.push(RC); + i += 1; + } + } else if (0xE0..=0xEF).contains(&b1) { + // 3-byte lead. + if i + 1 >= n { + out.push(RC); // truncated lead at EOF + i = n; + } else { + let b2 = bytes[i + 1]; + if (b1 == 0xE0 && (b2 & 0xE0) == 0x80) || (b2 & 0xC0) != 0x80 { + // overlong (E0 80..9F) or b2 not a continuation -> skip 1 + out.push(RC); + i += 1; + } else if i + 2 >= n { + out.push(RC); // truncated after a valid b2 at EOF + i = n; + } else { + let b3 = bytes[i + 2]; + if (b3 & 0xC0) != 0x80 { + out.push(RC); // b3 not a continuation -> skip 2 + i += 2; + } else { + let cp = (((b1 as u32) & 0x0F) << 12) + | (((b2 as u32) & 0x3F) << 6) + | ((b3 as u32) & 0x3F); + if (0xD800..=0xDFFF).contains(&cp) { + // surrogate (e.g. ED A0 80) -> JDK skips all 3, single FFFD + out.push(RC); + i += 3; + } else { + // surrogate range excluded above -> cp is a valid scalar + out.push(char::from_u32(cp).unwrap()); + i += 3; + } + } + } + } + } else if (0xF0..=0xF4).contains(&b1) { + // 4-byte lead. + if i + 1 >= n { + out.push(RC); + i = n; + } else { + let b2 = bytes[i + 1]; + if (b1 == 0xF0 && !(0x90..=0xBF).contains(&b2)) + || (b1 == 0xF4 && (b2 & 0xF0) != 0x80) + || (b2 & 0xC0) != 0x80 + { + out.push(RC); // bad b2 -> skip 1 + i += 1; + } else if i + 2 >= n { + out.push(RC); + i = n; + } else if (bytes[i + 2] & 0xC0) != 0x80 { + out.push(RC); // b3 not a continuation -> skip 2 + i += 2; + } else if i + 3 >= n { + out.push(RC); + i = n; + } else if (bytes[i + 3] & 0xC0) != 0x80 { + out.push(RC); // b4 not a continuation -> skip 3 + i += 3; + } else { + let cp = (((b1 as u32) & 0x07) << 18) + | (((b2 as u32) & 0x3F) << 12) + | (((bytes[i + 2] as u32) & 0x3F) << 6) + | ((bytes[i + 3] as u32) & 0x3F); + // F0/F4 range guards above keep cp within U+10000..=U+10FFFF -> a valid scalar + out.push(char::from_u32(cp).unwrap()); + i += 4; + } + } + } else { + // Lone continuation (0x80..0xBF), overlong 2-byte leads (0xC0/0xC1), or out-of-range + // 4-byte leads (0xF5..0xFF): each is a single ill-formed byte -> skip 1. + out.push(RC); + i += 1; + } + } + Cow::Owned(out) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Oracle = JDK 17 `new String(bytes, StandardCharsets.UTF_8)` (the renderer Spark uses for + /// StringType). Each row's expected output was verified against the JVM. The decoder must match + /// it byte-for-byte -- including the surrogate-range case where `str::from_utf8_lossy` differs. + #[test] + fn decode_utf8_spark_lossy_matches_jvm_replacement_granularity() { + let cases: &[(&[u8], &str)] = &[ + (&[0xFF, 0xFE, 0x41], "\u{FFFD}\u{FFFD}A"), + (&[0x80, 0x42], "\u{FFFD}B"), + (&[0xE0, 0x80], "\u{FFFD}\u{FFFD}"), + (&[0xF0, 0x80, 0x80, 0x41], "\u{FFFD}\u{FFFD}\u{FFFD}A"), + (&[0xC0, 0xAF], "\u{FFFD}\u{FFFD}"), + // The parity case: Rust's from_utf8_lossy would give three U+FFFD here. + (&[0xED, 0xA0, 0x80], "\u{FFFD}"), + ( + &[0xF4, 0x90, 0x80, 0x80], + "\u{FFFD}\u{FFFD}\u{FFFD}\u{FFFD}", + ), + ]; + for (bytes, expected) in cases { + assert_eq!( + decode_utf8_spark_lossy(bytes), + *expected, + "bytes {bytes:02x?} should render like the JVM" + ); + } + } + + #[test] + fn decode_utf8_spark_lossy_valid_utf8_is_borrowed_zero_copy() { + let s = "café — 日本語 🦀"; + match decode_utf8_spark_lossy(s.as_bytes()) { + Cow::Borrowed(b) => assert_eq!(b, s), + Cow::Owned(_) => panic!("valid UTF-8 must borrow, not allocate"), + } + } + + #[test] + fn decode_utf8_spark_lossy_valid_multibyte_around_invalid_bytes_decodes() { + // 'a' | é (C3 A9) | stray 0xFF | 'b' | 🦀 (F0 9F A6 80) -> valid chars preserved, one FFFD. + let mut bytes = vec![b'a']; + bytes.extend_from_slice("é".as_bytes()); + bytes.push(0xFF); + bytes.push(b'b'); + bytes.extend_from_slice("🦀".as_bytes()); + assert_eq!(decode_utf8_spark_lossy(&bytes), "aé\u{FFFD}b🦀"); + } +} diff --git a/native/shuffle/src/spark_unsafe/unsafe_object.rs b/native/shuffle/src/spark_unsafe/unsafe_object.rs index 55794b144f..65c1b92219 100644 --- a/native/shuffle/src/spark_unsafe/unsafe_object.rs +++ b/native/shuffle/src/spark_unsafe/unsafe_object.rs @@ -18,128 +18,11 @@ use super::list::SparkUnsafeArray; use super::map::SparkUnsafeMap; use super::row::SparkUnsafeRow; -use datafusion_comet_common::bytes_to_i128; +use datafusion_comet_common::{bytes_to_i128, decode_utf8_spark_lossy}; use std::borrow::Cow; const MAX_LONG_DIGITS: u8 = 18; -/// Decode `bytes` as UTF-8 the way Spark renders `StringType` -- `new String(bytes, UTF_8)` on the -/// JVM -- replacing each ill-formed sequence with a single `U+FFFD` and skipping the same number of -/// bytes the JDK's UTF-8 `CharsetDecoder` (action REPLACE) would. Valid UTF-8 is returned as a -/// zero-cost borrow. -/// -/// This intentionally differs from `str::from_utf8_lossy` for surrogate-range three-byte sequences -/// (`ED A0..BF ..`, e.g. CESU-8 / Java modified-UTF-8 supplementary chars) and for some other -/// ill-formed multi-byte units: `from_utf8_lossy` follows the Unicode "maximal subpart" rule and -/// can emit one `U+FFFD` per byte, whereas the JDK collapses certain ill-formed units into a single -/// `U+FFFD`. Matching the JDK byte-for-byte means a Comet columnar shuffle of arbitrary bytes -/// renders identically to a Spark JVM shuffle. The per-class malformed lengths below -/// (E0/ED overlong & surrogate handling, F0/F4 range checks) match the observable replacement -/// behavior of the JDK UTF-8 decoder; they were determined from observed -/// `new String(bytes, UTF_8)` output, not by reviewing the OpenJDK source. -pub(crate) fn decode_utf8_spark_lossy(bytes: &[u8]) -> Cow<'_, str> { - // Fast path: well-formed UTF-8 borrows with zero copy (the overwhelmingly common case). - if let Ok(s) = std::str::from_utf8(bytes) { - return Cow::Borrowed(s); - } - - const RC: char = '\u{FFFD}'; - let n = bytes.len(); - let mut out = String::with_capacity(n); - let mut i = 0; - while i < n { - let b1 = bytes[i]; - if b1 < 0x80 { - out.push(b1 as char); - i += 1; - } else if (0xC2..=0xDF).contains(&b1) { - // 2-byte lead. Bad/absent continuation -> single FFFD, skip 1. - if i + 1 < n && (bytes[i + 1] & 0xC0) == 0x80 { - let cp = (((b1 as u32) & 0x1F) << 6) | ((bytes[i + 1] as u32) & 0x3F); - out.push(char::from_u32(cp).unwrap()); - i += 2; - } else { - out.push(RC); - i += 1; - } - } else if (0xE0..=0xEF).contains(&b1) { - // 3-byte lead. - if i + 1 >= n { - out.push(RC); // truncated lead at EOF - i = n; - } else { - let b2 = bytes[i + 1]; - if (b1 == 0xE0 && (b2 & 0xE0) == 0x80) || (b2 & 0xC0) != 0x80 { - // overlong (E0 80..9F) or b2 not a continuation -> skip 1 - out.push(RC); - i += 1; - } else if i + 2 >= n { - out.push(RC); // truncated after a valid b2 at EOF - i = n; - } else { - let b3 = bytes[i + 2]; - if (b3 & 0xC0) != 0x80 { - out.push(RC); // b3 not a continuation -> skip 2 - i += 2; - } else { - let cp = (((b1 as u32) & 0x0F) << 12) - | (((b2 as u32) & 0x3F) << 6) - | ((b3 as u32) & 0x3F); - if (0xD800..=0xDFFF).contains(&cp) { - // surrogate (e.g. ED A0 80) -> JDK skips all 3, single FFFD - out.push(RC); - i += 3; - } else { - out.push(char::from_u32(cp).unwrap()); - i += 3; - } - } - } - } - } else if (0xF0..=0xF4).contains(&b1) { - // 4-byte lead. - if i + 1 >= n { - out.push(RC); - i = n; - } else { - let b2 = bytes[i + 1]; - if (b1 == 0xF0 && !(0x90..=0xBF).contains(&b2)) - || (b1 == 0xF4 && (b2 & 0xF0) != 0x80) - || (b2 & 0xC0) != 0x80 - { - out.push(RC); // bad b2 -> skip 1 - i += 1; - } else if i + 2 >= n { - out.push(RC); - i = n; - } else if (bytes[i + 2] & 0xC0) != 0x80 { - out.push(RC); // b3 not a continuation -> skip 2 - i += 2; - } else if i + 3 >= n { - out.push(RC); - i = n; - } else if (bytes[i + 3] & 0xC0) != 0x80 { - out.push(RC); // b4 not a continuation -> skip 3 - i += 3; - } else { - let cp = (((b1 as u32) & 0x07) << 18) - | (((b2 as u32) & 0x3F) << 12) - | (((bytes[i + 2] as u32) & 0x3F) << 6) - | ((bytes[i + 3] as u32) & 0x3F); - out.push(char::from_u32(cp).unwrap()); - i += 4; - } - } - } else { - // Lone continuation (0x80..0xBF), overlong 2-byte leads (0xC0/0xC1), or out-of-range - // 4-byte leads (0xF5..0xFF): each is a single ill-formed byte -> skip 1. - out.push(RC); - i += 1; - } - } - Cow::Owned(out) -} - /// A common trait for Spark Unsafe classes that can be used to access the underlying data, /// e.g., `UnsafeRow` and `UnsafeArray`. This defines a set of methods that can be used to /// access the underlying data with index. @@ -351,56 +234,3 @@ macro_rules! impl_primitive_accessors { }; } pub(crate) use impl_primitive_accessors; - -#[cfg(test)] -mod utf8_lossy_tests { - use super::decode_utf8_spark_lossy; - use std::borrow::Cow; - - /// Oracle = JDK 17 `new String(bytes, StandardCharsets.UTF_8)` (the renderer Spark uses for - /// StringType). Each row's expected output was verified against the JVM. The decoder must match - /// it byte-for-byte -- including the surrogate-range case where `str::from_utf8_lossy` differs. - #[test] - fn matches_jvm_replacement_granularity() { - let cases: &[(&[u8], &str)] = &[ - (&[0xFF, 0xFE, 0x41], "\u{FFFD}\u{FFFD}A"), - (&[0x80, 0x42], "\u{FFFD}B"), - (&[0xE0, 0x80], "\u{FFFD}\u{FFFD}"), - (&[0xF0, 0x80, 0x80, 0x41], "\u{FFFD}\u{FFFD}\u{FFFD}A"), - (&[0xC0, 0xAF], "\u{FFFD}\u{FFFD}"), - // The parity case: Rust's from_utf8_lossy would give three U+FFFD here. - (&[0xED, 0xA0, 0x80], "\u{FFFD}"), - ( - &[0xF4, 0x90, 0x80, 0x80], - "\u{FFFD}\u{FFFD}\u{FFFD}\u{FFFD}", - ), - ]; - for (bytes, expected) in cases { - assert_eq!( - decode_utf8_spark_lossy(bytes), - *expected, - "bytes {bytes:02x?} should render like the JVM" - ); - } - } - - #[test] - fn valid_utf8_is_borrowed_zero_copy() { - let s = "café — 日本語 🦀"; - match decode_utf8_spark_lossy(s.as_bytes()) { - Cow::Borrowed(b) => assert_eq!(b, s), - Cow::Owned(_) => panic!("valid UTF-8 must borrow, not allocate"), - } - } - - #[test] - fn valid_multibyte_around_invalid_bytes_decodes() { - // 'a' | é (C3 A9) | stray 0xFF | 'b' | 🦀 (F0 9F A6 80) -> valid chars preserved, one FFFD. - let mut bytes = vec![b'a']; - bytes.extend_from_slice("é".as_bytes()); - bytes.push(0xFF); - bytes.push(b'b'); - bytes.extend_from_slice("🦀".as_bytes()); - assert_eq!(decode_utf8_spark_lossy(&bytes), "aé\u{FFFD}b🦀"); - } -} diff --git a/native/spark-expr/src/conversion_funcs/cast.rs b/native/spark-expr/src/conversion_funcs/cast.rs index 351ccfa777..a60046b03d 100644 --- a/native/spark-expr/src/conversion_funcs/cast.rs +++ b/native/spark-expr/src/conversion_funcs/cast.rs @@ -40,7 +40,7 @@ use crate::utils::{array_with_timezone, cast_timestamp_to_ntz, timestamp_ntz_to_ use crate::EvalMode::Legacy; use crate::{cast_whole_num_to_binary, BinaryOutputStyle}; use crate::{EvalMode, SparkError}; -use arrow::array::builder::StringBuilder; +use arrow::array::builder::{GenericStringBuilder, StringBuilder}; use arrow::array::{ new_null_array, BinaryBuilder, DictionaryArray, GenericByteArray, ListArray, MapArray, StringArray, StructArray, @@ -50,8 +50,8 @@ use arrow::datatypes::{Field, Fields, GenericBinaryType}; use arrow::error::ArrowError; use arrow::{ array::{ - cast::AsArray, types::Int32Type, Array, ArrayRef, GenericStringArray, Int16Array, - Int32Array, Int64Array, Int8Array, OffsetSizeTrait, PrimitiveArray, + cast::AsArray, types::Int32Type, Array, ArrayRef, Int16Array, Int32Array, Int64Array, + Int8Array, OffsetSizeTrait, PrimitiveArray, }, compute::{cast_with_options, take, CastOptions}, record_batch::RecordBatch, @@ -62,6 +62,8 @@ use base64::Engine; use datafusion::common::{internal_err, DataFusionError, Result as DataFusionResult, ScalarValue}; use datafusion::physical_expr::PhysicalExpr; use datafusion::physical_plan::ColumnarValue; +use datafusion_comet_common::decode_utf8_spark_lossy; +use std::borrow::Cow; use std::{ fmt::{Debug, Display, Formatter}, hash::Hash, @@ -801,21 +803,22 @@ fn cast_binary_to_string( .downcast_ref::>>() .unwrap(); - fn binary_formatter(value: &[u8], spark_cast_options: &SparkCastOptions) -> String { - match spark_cast_options.binary_output_style { - Some(s) => spark_binary_formatter(value, s), - None => cast_binary_formatter(value), + // Build with a GenericStringBuilder and append &str straight from the decoder's Cow so the + // common valid-UTF-8 cast path copies the bytes once (into the Arrow value buffer) instead of + // twice (an owned String, then a copy into the buffer). + let mut builder = GenericStringBuilder::::new(); + for value in input.iter() { + match value { + Some(value) => match spark_cast_options.binary_output_style { + // ToPrettyString styles (Spark 4.0+) build owned strings, which is unavoidable. + Some(s) => builder.append_value(spark_binary_formatter(value, s)), + // Default CAST(binary AS string): borrows in the valid path, appends once. + None => builder.append_value(cast_binary_formatter(value)), + }, + None => builder.append_null(), } } - - let output_array = input - .iter() - .map(|value| match value { - Some(value) => Ok(Some(binary_formatter(value, spark_cast_options))), - _ => Ok(None), - }) - .collect::, ArrowError>>()?; - Ok(Arc::new(output_array)) + Ok(Arc::new(builder.finish())) } /// This function mimics the [BinaryFormatter]: https://github.com/apache/spark/blob/v4.0.0/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ToStringBase.scala#L449-L468 @@ -824,7 +827,10 @@ fn cast_binary_to_string( /// Before Spark 4.0.0, the default is SPACE_DELIMITED_UPPERCASE_HEX fn spark_binary_formatter(value: &[u8], binary_output_style: BinaryOutputStyle) -> String { match binary_output_style { - BinaryOutputStyle::Utf8 => String::from_utf8(value.to_vec()).unwrap(), + // Spark's UTF8 BinaryFormatter renders via `UTF8String.fromBytes(bytes).toString`, i.e. + // `new String(bytes, UTF_8)`. Route through the shared JVM-compatible decoder so invalid + // bytes become U+FFFD (matching Spark) instead of panicking on non-UTF-8 input (#4488). + BinaryOutputStyle::Utf8 => decode_utf8_spark_lossy(value).into_owned(), BinaryOutputStyle::Basic => { format!( "{:?}", @@ -853,20 +859,78 @@ fn spark_binary_formatter(value: &[u8], binary_output_style: BinaryOutputStyle) } } -fn cast_binary_formatter(value: &[u8]) -> String { - match String::from_utf8(value.to_vec()) { - Ok(value) => value, - Err(_) => unsafe { String::from_utf8_unchecked(value.to_vec()) }, - } +fn cast_binary_formatter(value: &[u8]) -> Cow<'_, str> { + // CAST(binary AS string) reinterprets the bytes as UTF-8, like Spark's UTF8String.fromBytes. + // Spark keeps the raw bytes, but Arrow's Utf8 type requires valid UTF-8, and building a String + // from non-UTF-8 bytes is undefined behaviour (#4488). Decode JVM-compatibly-lossily instead: + // `decode_utf8_spark_lossy` replaces ill-formed sequences with U+FFFD exactly as Spark's + // `new String(bytes, UTF_8)` does (the same decoder Comet's native shuffle uses, #4521). The + // result is memory-safe valid UTF-8, never feeds invalid bytes into downstream native string + // kernels, and matches Spark's rendered output byte-for-byte. It diverges from Spark only under + // byte-level round-trips such as CAST(CAST(x AS string) AS binary), where Spark still has the + // original bytes and Comet has the U+FFFD replacements. Returning `Cow` lets the valid path + // borrow so the caller appends without an intermediate allocation. + decode_utf8_spark_lossy(value) } #[cfg(test)] mod tests { use super::*; - use arrow::array::{ListArray, NullArray, StringArray}; + use arrow::array::{BinaryArray, ListArray, NullArray, StringArray}; use arrow::buffer::OffsetBuffer; use arrow::datatypes::TimestampMicrosecondType; use arrow::datatypes::{Field, Fields}; + + #[test] + fn test_cast_binary_to_string_replaces_invalid_utf8_jvm_compatibly() { + // Invalid bytes are replaced with U+FFFD instead of reinterpreted as an invalid `str`, + // and the granularity matches the JVM: the surrogate-range sequence [ED A0 80] collapses + // to a single U+FFFD (Rust's `from_utf8_lossy` would emit three). Valid UTF-8 ("abc") is + // preserved exactly and NULL stays NULL. + let input = BinaryArray::from_opt_vec(vec![ + Some(&[0xFFu8, 0xFE][..]), + None, + Some("abc".as_bytes()), + Some(&[0xEDu8, 0xA0, 0x80][..]), + ]); + // binary_output_style defaults to None, i.e. the plain (non-ToPrettyString) cast path. + let cast_options = SparkCastOptions::new(EvalMode::Legacy, "UTC", false); + + let result = cast_binary_to_string::(&input, &cast_options).unwrap(); + + let strings = result.as_string::(); + assert_eq!(strings.len(), 4); + assert_eq!(strings.value(0), "\u{FFFD}\u{FFFD}"); + assert!(strings.is_null(1)); + assert_eq!(strings.value(2), "abc"); + assert_eq!(strings.value(3), "\u{FFFD}"); + } + + #[test] + fn test_cast_binary_to_string_utf8_output_style_replaces_invalid_utf8() { + // Spark's `binaryOutputStyle=UTF8` (Spark 4.0+) ToPrettyString formatter renders binary via + // `new String(bytes, UTF_8)`. Previously this arm called `String::from_utf8(..).unwrap()`, + // which panicked the executor on non-UTF-8 input. It must now decode JVM-compatibly-lossily, + // matching the default cast path's replacement behavior. + let input = BinaryArray::from_opt_vec(vec![ + Some(&[0xFFu8, 0xFE][..]), + None, + Some("abc".as_bytes()), + Some(&[0xEDu8, 0xA0, 0x80][..]), + ]); + let mut cast_options = SparkCastOptions::new(EvalMode::Legacy, "UTC", false); + cast_options.binary_output_style = Some(BinaryOutputStyle::Utf8); + + let result = cast_binary_to_string::(&input, &cast_options).unwrap(); + + let strings = result.as_string::(); + assert_eq!(strings.len(), 4); + assert_eq!(strings.value(0), "\u{FFFD}\u{FFFD}"); + assert!(strings.is_null(1)); + assert_eq!(strings.value(2), "abc"); + assert_eq!(strings.value(3), "\u{FFFD}"); + } + #[test] fn test_cast_unsupported_timestamp_to_date() { // Since datafusion uses chrono::Datetime internally not all dates representable by TimestampMicrosecondType are supported