Repository navigation
[EPIC] Consistent handling of invalid UTF-8 in native StringType (ingress policy) #4764
Description
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
STRINGcolumn asBinary, then convert toUtf8viadecode_utf8_spark_lossyinstead of erroring -- consistent with cast/shuffle and with Spark. - Gap B: decode imported
Utf8/LargeUtf8columns throughdecode_utf8_spark_lossyat 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
STRINGcolumns (Comet native scan rejects invalid UTF-8 byte sequences in STRING column (hll.sql on Spark 4.1) #4121); re-enablehll.sql - Gap B: decode imported string columns at the FFI boundary
- Gap C: BINARY columns read as STRING in the native scan. Nested values that are not valid UTF-8 become NULL, and top-level values reach a
Utf8array unvalidated (Native Parquet scan returns NULL for nested BINARY values that are not valid UTF-8 when they are read as STRING #7032) - 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_stringdecode (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)
Activity
andygrove commentedon Jul 7, 2026
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(theparquetcrate'sarrow/buffer/offset_buffer.rs::try_pushraisesencountered non UTF-8 data). If the target field isBinary, no validation runs. So the fix does require the reader to emitBinaryfor STRING columns, then decodeBinary->Utf8on the Comet side.The problem is choosing that read type. It is decided inside DataFusion's
ParquetOpenerviaoptions.with_schema(physical_file_schema), wherephysical_file_schemais computed byapply_file_schema_type_coercions(logical, physical). That function only coerces Binary -> Utf8 (plus view types), never Utf8 -> Binary. DataFusion 54'sParquetSourceexposes no hook for us to override the read schema (nowith_schema_adapter, and the opener does not consult aSchemaAdapterFactory). DataFusion has abinary_as_stringoption but nostring_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 thenative_datafusionscan (parquet_exec.rs) andnative_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:
decode_utf8_spark_lossy(currentlypub(crate)innative/shuffle/src/spark_unsafe/unsafe_object.rs), ideally by relocating it todatafusion-comet-spark-exprso both scan and shuffle depend on it and the cast path can reuse it later. Notecast_binary_formatter(spark-expr/.../cast.rs) still usesfrom_utf8_uncheckedtoday, so any Binary -> Utf8 adapter cast must not reuse it.(Binary physical, Utf8 logical)pair inschema_adapter.rs(currently a default SparkCast) through the lossy decoder.Options for read-as-Binary
string_as_binaryread coercion (mirror ofbinary_as_string). Comet then keeps the required schema asUtf8, sets the flag, and the existingSparkPhysicalExprAdapterinserts theBinary -> Utf8cast at the scan boundary that we route throughdecode_utf8_spark_lossy. Cleanest layering; depends on a DF release or pinning DF to a git rev.ARROW:schemaforcingBinaryand carryBinarythrough the logical schema so the coercion does not undo it. Single Comet PR, but fights the coercion and is fragile around nested types.FileOpener/FileSource. Fork theParquetOpenerto controlArrowReaderOptions::with_schemadirectly. 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 commentedon Jul 7, 2026
Investigation notes: Gap B (FFI import UTF-8 validation)
Mapping the JVM -> native Arrow FFI import surface for Gap B (imported
Utf8/LargeUtf8arrays are built with arrow's uncheckedfrom_ffi/from_ffi_and_data_typeviaArrayData::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:
batch_from_ffi->from_ffi_and_data_type,aligned_stream_reader.rs:95-98mapInArrow(everythingScanExecpulls)from_ffi,jni_api.rs:1296UnsafeRowfrom_ffi,jvm_udf/mod.rs:239The existing safeguard does not cover this
prepare_output'svalidate_full()(jni_api.rs:685) is gated onCOMET_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
decode_utf8_spark_lossyispub(crate)in the shuffle crate (native/shuffle/src/spark_unsafe/unsafe_object.rs:40) and is not reachable fromnative/core. Promote topubor relocate todatafusion-comet-spark-expr.scan.rs/copy.rsrecurses into nested types today (only top-level dictionaries are unpacked). It must handleUtf8,LargeUtf8,Dictionary(_, Utf8),List/LargeList/FixedSizeList,Struct, andMap. The fast path is a singlestr::from_utf8SIMD pass over the values buffer with a zero-copy borrow; it rebuilds/allocates only on genuinely invalid bytes.copy_or_unpack_array/copy_array(native/core/src/execution/operators/copy.rs), already the per-column choke point invoked atscan.rs:159. PlainUtf8/LargeUtf8currently falls through to a bareArc::clonethere, so a decode branch has to be inserted (dictionary columns are already rebuilt at that point).Two subtleties for the design
from_utf8being 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), andvalue()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.Scope
Separate PR from Gap A, but shares the decoder-exposure change and the same decode-via-
decode_utf8_spark_lossypolicy. If Gap A lands first, exposing the decoder is already done.docs: document known correctness issues in the compatibility guide (#…
andygrove commentedon Aug 5, 2026
Note for Gap B: import site C now has consumers that depend on it not decoding
Two things about import site C (JVM UDF result,
from_ffiatnative/spark-expr/src/jvm_udf/mod.rs:239) have changed since the Gap B investigation notes above were written. Both affect how Gap B should be designed, so recording them here.Site C is no longer just user UDFs
The table above describes site C as "Result of a JVM-side Spark UDF", hotness "per UDF eval". That is now an undercount. Since #4728 routes
Unsupported(not justIncompatible) through the JVM codegen dispatcher, every serde mixing inCodegenDispatchFallbackreturns its results through site C. There are currently 21 such serdes, covering ordinary built-in expressions on default config, not just registered user UDFs.So site C is closer in character to site A than the table suggests: it carries built-in expression results on default-config queries. That matters for the benchmarking task, and it matters for the blast radius of any decode inserted there.
A decode at site C would reverse a fix that is currently in review
PR #5267 fixes
spark.sql.function.concatBinaryAsString=trueproducing wrong bytes, by marking the affectedconcatUnsupportedso Spark's generated code runs it inside the Comet pipeline and the original bytes survive. The mechanism only works because site C passes bytes through unchecked. The JVM writes the rawUTF8Stringbytes into theVarCharVector(CometBatchKernelCodegenOutput.scala:243), and nativehexreads them back out.If Gap B decodes imported
Utf8/LargeUtf8at site C unconditionally,hex(concat(cast(b AS string), x))goes back to returningEFBFBDEFBFBD41and that PR's SQL file test starts failing. The revert would be silent at review time and only show up as a red test.Whether that is a problem depends on which way we go on the underlying policy question, which is really the first task in this EPIC ("decide the ingress policy"). #5267 is effectively arguing for an option not listed under "Alternatives considered": keep byte-exact subtrees out of native execution entirely, so that the bytes never need to be representable in Arrow
Utf8. That option needs site C to be exempt from Gap B's decode, plus a rule for which expressions qualify. The alternative is that we hold the line from #4763 (decode, accept the documented byte-round-trip divergence) and #5267 becomes a docs change instead.Either way, Gap B should not treat site C as a mechanical "add a decode here" the way sites A and B are. Flagging so whoever picks it up makes that call deliberately.
docs: document known correctness issues in the compatibility guide (a…
kazuyukitanimura commentedon Sep 16, 2026
We need a assignee for this one
andygrove commentedon Sep 27, 2026
Status as of
31b38196d: Gap B is done. #5310 decodes invalid UTF-8 at every JVM-to-native import site:ScanExec::import_column, native columnar-to-row and the JVM UDF result. So the "passes bytes through unchecked" row of the current-state table no longer holds, and the policy question has in practice been settled as decode-at-the-boundary. The benchmark item is covered byCometStringFfiImportBenchmarkandbenches/utf8_decode.rs. What remains is Gap A (#4121).fix: decode invalid UTF-8 in nested Parquet BINARY values read as str…
fix: decode invalid UTF-8 in nested Parquet BINARY values read as str…
fix: decode invalid UTF-8 in nested Parquet BINARY values read as str…
fix: decode invalid UTF-8 in nested Parquet BINARY values read as str…
fix: decode invalid UTF-8 in nested Parquet BINARY values read as str…
fix: decode invalid UTF-8 in nested Parquet BINARY values read as str…
fix: decode invalid UTF-8 in nested Parquet BINARY values read as str…
fix: decode invalid UTF-8 in nested Parquet BINARY values read as str…