fix: decode invalid UTF-8 at the JVM to native FFI import boundary

[manuzhang]

This PR continues the work from #4945.

Which issue does this PR close?

Part of #4764 (EPIC: consistent handling of invalid UTF-8 in native StringType). This PR implements Gap B only, decoding invalid UTF-8 at the JVM to native Arrow FFI import boundary. It does not close the EPIC, which still tracks Gap A (the native scan rejecting invalid UTF-8).

Rationale for this change

Spark's StringType (UTF8String) can hold arbitrary bytes, including sequences that are not valid UTF-8. When a JVM side source hands string columns to native code over the Arrow C Data Interface, arrow-rs imports them with from_ffi / from_ffi_and_data_type, which build the array via ArrayData::new_unchecked and do not validate UTF-8. The imported Arrow Utf8 / LargeUtf8 array then lies about its validity, and any downstream native string kernel that reads &str through arrow-rs's unchecked StringArray::value() (from_utf8_unchecked) exercises undefined behaviour: iterating chars, slicing on char boundaries, and similar operations can misbehave, panic, or be miscompiled. This is a latent, default configuration soundness hazard.

The string producing sites already decode invalid bytes the way Spark renders them: CAST(binary AS string) (#4763) and native shuffle get_string (#4521) both use decode_utf8_spark_lossy. This PR applies the same policy to the string ingress side, so the whole native pipeline agrees on a single invariant: native string data is always valid UTF-8.

What changes are included in this PR?

• A new decode_string_arrays walker in datafusion-comet-common, beside the existing decode_utf8_spark_lossy. It ensures every Utf8 / LargeUtf8 array reachable from an imported column holds valid UTF-8, decoding invalid bytes to Spark's rendered form. It is zero copy for the valid common case (a single from_utf8 validation pass plus an O(number of strings) boundary check, returning the same Arc), and rebuilds element by element only when bytes are genuinely invalid. It recurses through Dictionary, Struct, List, LargeList, FixedSizeList, and Map.
• The walker is called at the three JVM to native FFI import sites that carry string data:
  • ScanExec::pull_next, which handles all native query input (the native_comet JVM reader, Spark columnar handoff, shuffle reads, mapInArrow). Decoding runs before dictionary unpack so compact dictionary values are validated rather than the expanded ones.
  • columnarToRow conversion.
  • JVM UDF result.
• The fast path validates the used byte range and also confirms no element boundary splits a codepoint. This second check is required for soundness: a whole buffer valid "é" (bytes C3 A9) split across two offsets would otherwise hand each element an invalid slice, and value() decodes those unchecked.
• A criterion benchmark for the valid fast path.

No configuration flag gates this. It fixes undefined behaviour in the default configuration and matches the always on behaviour of the sibling cast and shuffle fixes.

The only observable divergence from Spark is the previously documented one: decoding rather than preserving raw bytes differs only under byte level round trips (for example CAST(CAST(X'FF' AS STRING) AS BINARY)), already noted in the compatibility guide.

How are these changes tested?

• Rust unit tests for the walker covering: valid input returned zero copy (pointer identical buffers); invalid bytes decoding to U+FFFD matching the JVM; the split codepoint boundary case; nested Dictionary / Struct / List / FixedSizeList / Map (decode and zero copy paths); null preservation; sliced arrays with a non zero starting offset; and trailing empty strings.
• A Rust test at the ScanExec import site proving an invalid UTF-8 column is decoded through the production per column path.
• A criterion benchmark measuring the valid fast path (about 120 microseconds for an 8192 row batch, a single validation pass with a zero copy return).

[0lai0]

Thanks @manuzhang for picking this up. left some comments

[0lai0]

nit: Could we map this to CometError::Arrow instead of CometError::Internal?

decode_string_arrays is declared as Result<ArrayRef, ArrowError>, and the other two call sites (scan.rs, jvm_udf) already surface that as CometError::Arrow (via ? or an explicit map). Using Internal here misclassifies an Arrow/data failure as a Comet internal error.

arrays.push(decode_string_arrays(&imported)?) should be enough, since CometError implements From<ArrowError>.

[0lai0]

on the Map arm, Arc::new(map.entries().clone()) runs even when nothing needs decoding
Suggestion : Could we inline the struct walk over map.entries() columns (like the Struct arm) so the valid/zero-copy path doesn't allocate that Arc?