fix: decode CAST(binary AS string) JVM-compatibly instead of unsafe reinterpret

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:

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(<binary> 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(<binary> 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(<map> AS <map>)` falls back to Spark even though native `cast_map_to_map` exists (https://github.com/apache/datafusion-comet/issues/4491).

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)).

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};

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🦀");
+    }
+}