[EPIC] Consistent handling of invalid UTF-8 in native StringType (ingress policy)

[andygrove]

Background

Spark's StringType can hold arbitrary bytes, including sequences that are not valid UTF-8 (Spark stores them verbatim via UTF8String). Arrow's Utf8/LargeUtf8 types require valid UTF-8, and arrow-rs StringArray::value hands out a &str via an unchecked decode that trusts the array's UTF-8 invariant. As a result, any native string kernel that reads an Arrow string array containing invalid UTF-8 (via value()) materializes an invalid &str, which is undefined behaviour (iterating chars, slicing on char boundaries, etc. can misbehave, panic, or be miscompiled).

Comet currently handles invalid-UTF-8 strings inconsistently across sites. This EPIC tracks giving Comet a single, coherent policy.

Current state

String-producing sites (resolved -- decode JVM-compatibly to valid UTF-8):

Site	Behavior
Native shuffle get_string	Decodes via decode_utf8_spark_lossy (matches the JVM's new String(bytes, UTF_8) byte-for-byte, including surrogate-range cases). #4521
CAST(binary AS string)	Same decode_utf8_spark_lossy (shared with shuffle). #4488 / PR #4763

Both produce valid UTF-8 that matches Spark's rendered output, so neither feeds the downstream-kernel UB.

String-ingress sites (the gaps):

Site	Behavior	Problem
Native (Rust) Parquet reader	Rejects invalid UTF-8 (encountered non UTF-8 data)	Comet errors where Spark succeeds (#4121)
JVM -> native Arrow FFI import (jni_api.rs from_ffi)	Passes bytes through unchecked	Latent UB: invalid UTF-8 enters the native pipeline and downstream value() is unsound

The two ingress paths are simultaneously too strict (reject) and too loose (unchecked) for the same input.

Gap A -- native scan rejects invalid UTF-8 (#4121)

The arrow-rs Parquet reader validates UTF-8 and errors on non-UTF-8 STRING columns. Spark reads them fine. Surfaces in Spark 4.1.1's hll.sql; currently worked around by disabling Comet for that file.

Gap B -- FFI import does not validate (confirmed)

arrow::ffi::from_ffi builds the imported array with ArrayData::new_unchecked and does not validate UTF-8 (arrow-array src/ffi.rs even carries a // Should FFI be checking validity? comment). Comet adds no validation on the import side (prepare_output's validate_full() runs only on the output/export path under a debug flag). So invalid-UTF-8 string bytes arriving from a JVM-side source (a JVM-side scan, or any Spark -> Comet columnar handoff) enter the native pipeline as an Arrow Utf8 array that lies about its validity.

Severity: a latent default-config UB hazard. It requires (1) genuinely non-UTF-8 string bytes, (2) flowing through the unchecked FFI path rather than the validating native reader, and (3) consumption by a native string kernel that reads &str. That conjunction is uncommon, which is why it has not surfaced as crashes -- but it is unsound by default.

Proposed direction

Establish a single ingress policy: native string data must be valid UTF-8. Decode (not reinterpret) at the boundary using the existing decode_utf8_spark_lossy (in spark-expr/src/utils.rs), which is zero-copy for the valid-UTF-8 common case:

• Gap A: read the Parquet STRING column as Binary, then convert to Utf8 via decode_utf8_spark_lossy instead of erroring -- consistent with cast/shuffle and with Spark.
• Gap B: decode imported Utf8/LargeUtf8 columns through decode_utf8_spark_lossy at the FFI import boundary.

This unifies all four sites on one decoder and removes the UB. It touches the scan/FFI hot path, so it needs a perf evaluation (the from_utf8 validation pass is SIMD-fast and the native Rust reader already pays it, but string-heavy workloads should be benchmarked) and belongs in its own PR, not bundled with the cast fix.

Alternatives considered

• Document + accept as a rare known limitation (leaves the UB in default config).
• Validate-and-error at import like the native scan (trades rare UB for rare query failures / Spark-incompatibility).
• Harden every string kernel to be byte-safe (impractically large surface).

Observable caveat (already documented)

Decoding (rather than preserving raw bytes) diverges from Spark only under byte-level round-trips: CAST(CAST(X'FF' AS STRING) AS BINARY) returns X'FF' in Spark but X'EFBFBD' (UTF-8 of U+FFFD) in Comet; likewise octet_length/hashing on the raw bytes. This is noted in the compatibility guide and the contributor guide.

Tasks

• Decide the ingress policy (decode-at-boundary vs alternatives)
• Gap A: fix the native scan to read+decode invalid-UTF-8 STRING columns (Comet native scan rejects invalid UTF-8 byte sequences in STRING column (hll.sql on Spark 4.1) #4121); re-enable hll.sql
• Gap B: decode imported string columns at the FFI boundary
• Benchmark the validation/decode cost on string-heavy scans and FFI handoffs
• Confirm there are no other ingress points (e.g. row -> columnar conversion)

Related

• [Bug] CAST(BinaryType AS StringType) uses unsafe from_utf8_unchecked (undefined behaviour) #4488 / PR fix: decode CAST(binary AS string) JVM-compatibly instead of unsafe reinterpret #4763 -- CAST(binary AS string) decode (done)
• native shuffle: get_string should not panic on non-UTF-8 bytes (use lossy decode) #4521 -- native shuffle get_string decode (done)
• Comet native scan rejects invalid UTF-8 byte sequences in STRING column (hll.sql on Spark 4.1) #4121 -- native scan rejects invalid UTF-8 (Gap A)

[andygrove]

Investigation notes: Gap A (#4121) implementation constraint

Digging into Gap A (native scan rejects invalid UTF-8), the "read the STRING column as Binary, then decode via decode_utf8_spark_lossy" direction runs into a constraint worth recording before anyone picks it up.

The crux

arrow-rs validates UTF-8 during Parquet decode, gated purely on whether the target Arrow field for the column is Utf8/LargeUtf8 (the parquet crate's arrow/buffer/offset_buffer.rs::try_push raises encountered non UTF-8 data). If the target field is Binary, no validation runs. So the fix does require the reader to emit Binary for STRING columns, then decode Binary -> Utf8 on the Comet side.

The problem is choosing that read type. It is decided inside DataFusion's ParquetOpener via options.with_schema(physical_file_schema), where physical_file_schema is computed by apply_file_schema_type_coercions(logical, physical). That function only coerces Binary -> Utf8 (plus view types), never Utf8 -> Binary. DataFusion 54's ParquetSource exposes no hook for us to override the read schema (no with_schema_adapter, and the opener does not consult a SchemaAdapterFactory). DataFusion has a binary_as_string option but no string_as_binary.

So there is no clean Comet-only insertion point in stock DataFusion 54 to force STRING columns to be read as Binary. This applies to both the native_datafusion scan (parquet_exec.rs) and native_iceberg_compat (iceberg_scan.rs), which share the expr adapter. native_comet (the JVM reader) is unaffected.

The Comet-side decode half is straightforward

Independent of how we source the read-as-Binary capability:

• Expose decode_utf8_spark_lossy (currently pub(crate) in native/shuffle/src/spark_unsafe/unsafe_object.rs), ideally by relocating it to datafusion-comet-spark-expr so both scan and shuffle depend on it and the cast path can reuse it later. Note cast_binary_formatter (spark-expr/.../cast.rs) still uses from_utf8_unchecked today, so any Binary -> Utf8 adapter cast must not reuse it.
• Route the (Binary physical, Utf8 logical) pair in schema_adapter.rs (currently a default Spark Cast) through the lossy decoder.

Options for read-as-Binary

1. Upstream DataFusion first. Add a symmetric string_as_binary read coercion (mirror of binary_as_string). Comet then keeps the required schema as Utf8, sets the flag, and the existing SparkPhysicalExprAdapter inserts the Binary -> Utf8 cast at the scan boundary that we route through decode_utf8_spark_lossy. Cleanest layering; depends on a DF release or pinning DF to a git rev.
2. Comet-local reader-factory hook. Inject an ARROW:schema forcing Binary and carry Binary through the logical schema so the coercion does not undo it. Single Comet PR, but fights the coercion and is fragile around nested types.
3. Comet-local custom FileOpener/FileSource. Fork the ParquetOpener to control ArrowReaderOptions::with_schema directly. Full control, but duplicates predicate pushdown, page-index, encryption, and metadata-cache logic.

Option 1 seems like the right home for the mechanism. Flagging here so we can decide the direction before the Gap A PR.

[andygrove]

Investigation notes: Gap B (FFI import UTF-8 validation)

Mapping the JVM -> native Arrow FFI import surface for Gap B (imported Utf8/LargeUtf8 arrays are built with arrow's unchecked from_ffi / from_ffi_and_data_type via ArrayData::new_unchecked, so invalid UTF-8 enters native as an Arrow string array that lies about its validity).

Import sites (the complete set)

Three distinct JVM -> native import points, none of which validate UTF-8 today:

Site	Location	Hotness	Carries
A. ScanExec stream	batch_from_ffi -> from_ffi_and_data_type, aligned_stream_reader.rs:95-98	Hottest: per batch, per column, for all query input	native_comet JVM Parquet reader, Spark columnar handoff, shuffle reads, mapInArrow (everything ScanExec pulls)
B. columnarToRow	from_ffi, jni_api.rs:1296	Per batch/column, only for columnar->row queries	JVM columnar batch -> Spark UnsafeRow
C. JVM UDF result	from_ffi, jvm_udf/mod.rs:239	Per UDF eval	Result of a JVM-side Spark UDF

The existing safeguard does not cover this

prepare_output's validate_full() (jni_api.rs:685) is gated on COMET_DEBUG_ENABLED, runs only on the export (native -> JVM) path, and checks structural offsets rather than UTF-8 well-formedness of imported data. It does not touch any import site.

What a fix needs

1. Expose the decoder (shared with Gap A): decode_utf8_spark_lossy is pub(crate) in the shuffle crate (native/shuffle/src/spark_unsafe/unsafe_object.rs:40) and is not reachable from native/core. Promote to pub or relocate to datafusion-comet-spark-expr.
2. A nested-aware string-decode walker (new code): nothing in scan.rs/copy.rs recurses into nested types today (only top-level dictionaries are unpacked). It must handle Utf8, LargeUtf8, Dictionary(_, Utf8), List/LargeList/FixedSizeList, Struct, and Map. The fast path is a single str::from_utf8 SIMD pass over the values buffer with a zero-copy borrow; it rebuilds/allocates only on genuinely invalid bytes.
3. Apply it at all three boundaries. For the dominant path A, the natural spot is copy_or_unpack_array / copy_array (native/core/src/execution/operators/copy.rs), already the per-column choke point invoked at scan.rs:159. Plain Utf8/LargeUtf8 currently falls through to a bare Arc::clone there, so a decode branch has to be inserted (dictionary columns are already rebuilt at that point).

Two subtleties for the design

• Fast-path soundness: a whole-buffer from_utf8 being valid does not by itself guarantee each element's [offset[i], offset[i+1]) slice is on a char boundary (a malformed producer could split a codepoint), and value() slices per offset with an unchecked decode. To stay both fast and sound, the valid path should also confirm offsets do not land on UTF-8 continuation bytes (cheap O(number of strings) check), otherwise fall back to per-element decode. Legitimate Spark producers always satisfy this; the guard is for the adversarial-bytes case that motivates this gap.
• Performance: this adds a validation pass per string column per batch on the hottest path (A). Consistent with the EPIC's benchmarking task, the change should include a string-heavy scan benchmark.

Scope

Separate PR from Gap A, but shares the decoder-exposure change and the same decode-via-decode_utf8_spark_lossy policy. If Gap A lands first, exposing the decoder is already done.