Repository navigation
fix: [branch-1.1] decode invalid UTF-8 in nested Parquet BINARY values read as strings (#7106) - #7340
Conversation
…ings (apache#7106) * fix: decode invalid UTF-8 in nested Parquet BINARY values read as strings When a Parquet BINARY column without the UTF8 annotation is read as a string, through a read schema or spark.sql.parquet.binaryAsString, the native scan converts nested values (struct fields, list elements, map keys and values) in parquet_convert_array. That fell through to Arrow's safe cast, which turns invalid UTF-8 into NULL, so those values silently became NULL where Spark keeps the bytes. Convert Binary to Utf8 with Spark's cast instead. It replaces each ill-formed sequence with U+FFFD, the decode CAST(binary AS string) and native shuffle already share (apache#4764). Top-level columns are unchanged: DataFusion's file-schema coercion still decodes them straight into Utf8 (apache#7032). To keep the zero-copy conversion Arrow's cast gave valid nested values, cast_binary_to_string checks the bytes the rows span once for the default and UTF8 styles. It doesn't check the whole value buffer, which a slice shares with its parent. When the bytes are valid, it returns the input's value buffer narrowed to them, with a slice's offsets rebased. Otherwise rows inside the valid prefix are copied without being checked again, and only the rest go through the U+FFFD decoder. This also speeds up plain CAST(binary AS string). Split out of apache#6246. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: skip the UTF-8 fast path when nulls hide most of the payload cast_binary_to_string validated every byte the rows span, including the payload of null rows, which the row-by-row decoder skipped. After nullif or a CASE with column branches, a mostly-null batch can keep most of its bytes under nulls, so the cast scanned all of them and the reused buffer kept them alive downstream. Count the bytes of the non-null rows first. When null rows span more than half of the bytes, skip the check and decode the live rows one at a time as before, into a value buffer sized by the live bytes instead of the whole span. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: decode invalid UTF-8 in nested LargeBinary and BinaryView values The Parquet reader keeps the types a file's ARROW:schema stores, and DataFusion's file-schema coercion only changes top-level columns. So when an Arrow writer stores a list element, struct field or map value as LargeBinary or BinaryView, the native scan hands it to parquet_convert_array as that type. Only the Binary arm used Spark's cast, so these values still went through Arrow's safe cast and a malformed value became NULL. Convert LargeBinary and BinaryView to Binary first and use the same Spark cast. For LargeBinary that rewrites the offsets and keeps the value buffer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: copy a valid slice that fills little of its value buffer A slice shares its parent's value buffer, and Buffer::slice_with_length narrows the view but keeps the whole allocation. When cast_binary_to_string reused a valid slice's bytes, every output batch cut from one large array, such as aggregate output, held that array's whole allocation. DataFusion's sort charges each batch the capacity of its buffers, so it counted the allocation once per batch and spilled where a compact copy didn't. Reuse the buffer only when the rows' bytes fill at least half of its allocation. Otherwise copy those already validated bytes into a compact buffer with fresh offsets, which still skips the row-by-row decoder. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor: share offsets when casting valid binary to string cast_binary_to_string rebuilt the offsets whenever it copied the values of valid input, even for an unsliced array whose offsets needed no rebasing. Decide each buffer on its own: reuse the offsets when they start at zero and fill at least half of their allocation, and the values when the rows' bytes fill at least half of theirs. An unsliced array keeps its offsets when only its value buffer is oversized, while the first slice of a large array still gets offsets that don't hold the parent's allocation. Also update the function's doc comment, which still said the bytes are taken as is, and pass parquet_convert_array_impl's Binary to Utf8 cast only the options it reads. Add rows to the nested ParquetReadSuite fixture: a null list element, struct field and map value ahead of the invalid row, and a row whose list, struct and map are null. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: decode invalid UTF-8 in nested dictionary-encoded BINARY values A file's ARROW:schema can dictionary-encode a nested BINARY leaf, which the reader then returns as Dictionary(_, Binary | LargeBinary). Those fell through to Arrow's safe cast and turned malformed values into NULL. Decode the dictionary values through the Binary arm and expand the keys. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: copy the validity bitmap a small slice shares with its parent The fast path for valid input compacted the offsets and values of a slice that fills little of their allocations, but cloned the input's validity bitmap, which still held the parent's whole allocation. A sort that charges every batch for what it retains then spilled where the base did not, for example on nullable binary group keys emitted in small slices. Apply the same rule to the bitmap: reuse it when the slice's bits fill at least half of its allocation, and copy them otherwise. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: keep the bytes of a map key read from unannotated BINARY Decoding turned the keys FF and EF BF BD into the same U+FFFD, so a lookup of one returned the other's value and map_from_entries failed with DUPLICATED_MAP_KEY where Spark, which tells keys apart by their bytes, succeeds. A key now keeps its bytes at any depth under a map's key field, unchecked, as a top-level value does. Map values still decode. Decoding a dictionary now touches only the entries the batch references, so a wide malformed entry that one batch uses is not decoded again for every other batch of the row group. A batch whose decoded values would not fit a string column is reported with the byte count instead of an offset overflow. In the binary-to-string fallback, count the live bytes over runs of non-null rows instead of row by row, so sparse nulls don't cost a pass over the batch. In the row that holds the first ill-formed sequence, copy its validated prefix and decode only the rest. Size the builder for the validated prefix plus three bytes per remaining byte, and copy the result compact when a replacement still left it filling under half of its allocation. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix: copy the bytes of a valid binary cast outside a scan A plain CAST(binary AS string) can run on a batch imported over the Arrow C stream, where every buffer keeps the whole batch alive, wide sibling columns included, for as long as a sort or a join holds the result. Share the input's buffers only when a scan adapts its schema, where the reader owns them, and copy the validated bytes once otherwise, as the cast did before. Also cover a Dictionary(Int32, Binary) map key, link the tracking issue for the per-batch dictionary overflow from the code, and list that limit under the known issues with the batch size and the growth of ill-formed bytes. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> (cherry picked from commit 621b9cc)
github-actions
Bot
added
bug
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks @manuzhang. I compared the changes with 621b9cc and the only differences are the ones listed in the description. I also read the in_map_key threading through the struct, list, map and dictionary arms, and the dropped (ArrayRef, bool) pair loses nothing. Approving. Please merge this after #7339 and once the pending jobs finish.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
The newer backport (#7339) has merged, and the Rust and Spark scan jobs passed on this commit. I found one safety issue in the newly added nested map-key path. The fixture checks that two keys remain distinct, but its byte-level assertions do not detect an invalid Arrow string. The inline comment includes a reproducer using the branch’s pinned Arrow version.
| // and `map_from_entries` fails with DUPLICATED_MAP_KEY. A key keeps its bytes | ||
| // instead, unchecked, as a top-level value does (#7032). | ||
| let (offsets, values, nulls) = array.as_binary::<i32>().clone().into_parts(); | ||
| let keys = unsafe { StringArray::new_unchecked(offsets, values, nulls) }; |
There was a problem hiding this comment.
The malformed key in this PR’s own fixture violates the precondition of StringArray::new_unchecked. With Arrow 59.3.0, I passed the fixture’s two keys (41 FF 42 and 41 EF BF BD 42) to StringArray::try_new using the same offsets, bytes and validity. It rejects the first key at byte index 1. The unchecked constructor requires that the checked one would succeed, and a later string operation can obtain an invalid Rust &str. Preserving distinct map keys matters, so replacing both with U+FFFD would also lose information. Could we keep the keys as binary until a byte-preserving boundary, fall back to Spark for the affected scan, or return a checked error before constructing the string array? Pinned Arrow constructor contract
There was a problem hiding this comment.
@sunchao We already merged the same backport to 1.2 so we probably need to merge this one too. Is this issue here specific to this backport only?
There was a problem hiding this comment.
Thanks @sunchao, you're right that the map-key arm breaks the new_unchecked contract when a key is not valid UTF-8. To answer my own question: this isn't specific to the backport. The same code came from #7106 and is on main and branch-1.2 (and in 1.2.0-rc1).
It is the same class of problem as the top-level path, though. A top-level unannotated BINARY column read as a string already reaches a Utf8 array unchecked in every release, because arrow-rs validates UTF-8 only for UTF8-annotated columns (#7032, #4764). Before #7106 these map keys went through Arrow's safe cast and became NULL. Now they keep Spark's bytes and carry the same risk as a top-level value. So I'd like to merge this so branch-1.1 matches branch-1.2, and fix the map-key path with your suggestions in a follow-up: #7360.
Which issue does this PR close?
Backport of #7106 to
branch-1.1. Part of #7032 (the nested values) and #4764.Rationale for this change
When a Parquet BINARY column without the UTF8 annotation is read as a string, through a read schema or
spark.sql.parquet.binaryAsString=true, the native scan converts nested values (struct fields, list elements, map keys and values) inparquet_convert_array. That fell through to Arrow's safe cast, which turns invalid UTF-8 into NULL, so those values silently became NULL where Spark keeps the bytes.branch-1.1has this bug. See #7106 for the full rationale.The fix also goes to
branch-1.2in #7339, opened together with this one. Per the backporting guide, this one should not merge before that one is ready.What changes are included in this PR?
The cherry-pick of 621b9cc from
main.branch-1.1differs frommainin several places the source commit touches, so these parts were applied by hand:native/core/src/parquet/parquet_support.rs: on this branchparquet_convert_array_impl,parquet_convert_struct_to_structandparquet_convert_map_to_mapreturn a bareArrayRef, not the(ArrayRef, bool)field-overlap pair that fix: match Spark nullness for missing nested Parquet fields #7085 introduced onmain. The newin_map_keyparameter is threaded through the branch's signatures, and the newBinary/LargeBinary/BinaryViewandDictionaryarms return a bare array. The helper functionsreferenced_dictionary_valuesanddecoded_bytesand the four new unit tests are taken verbatim from the source commit. TheSparkErrorimport conflict was resolved the same way as onbranch-1.2.native/core/src/parquet/parquet_exec.rs: the new testnested_arrow_binary_types_read_as_string_decode_invalid_utf8is taken from the source commit, with the three trailing datetime-rebase arguments dropped from itsinit_datasource_execcall, because that function doesn't take them on this branch.native/spark-expr/src/conversion_funcs/cast.rs: only the import line conflicted.mainimportsformat_decimal_strthere and this branch doesn't; the backport adds only thearrow::bufferimports the fix needs. The rest of the file's changes applied cleanly.SparkCastOptions::new,SparkCastOptions::new_without_timezoneandSparkParquetOptions::newstill take theallow_incompatflag thatmainhas since removed, so the backported code and tests passfalseto them.docs/source/user-guide/latest/compatibility/index.md: this branch's "Strings with non-UTF-8 bytes" section predates the per-source list thatmainhas, so the source commit's new bullet is folded into the existing paragraph as a sentence: nested BINARY values read as strings are now decoded, top-level values and map keys are not checked, with a link to Native Parquet scan returns NULL for nested BINARY values that are not valid UTF-8 when they are read as STRING #7032.docs/source/user-guide/latest/parquet.mddoes not exist on this branch, so its changes (the Native Parquet scan returns NULL for nested BINARY values that are not valid UTF-8 when they are read as STRING #7032 known-issue entry and the Nested dictionary-encoded BINARY read as string overflows Utf8 offsets when flattened per batch #7254 dictionary limit) are dropped.How are these changes tested?
The tests from #7106: the Rust unit tests in
parquet_support.rs,parquet_exec.rsandcast.rs, and the newParquetReadSuitetest "native scan decodes invalid UTF-8 in nested BINARY read as string".Locally on this branch:
cargo check --all-targets,cargo clippy --all-targets --workspace -- -D warningsandcargo fmt --checkinnative/pass.parquet_supportandparquet_exectests indatafusion-comet(22 tests) and thecasttests indatafusion-comet-spark-expr(95 tests) pass../mvnw test -Dtest=none -Dsuites="org.apache.comet.parquet.ParquetReadV1Suite native scan decodes"passes aftermake core.🤖 Generated with Claude Code