fix: support binaryAsString in the native Parquet scan by manuzhang · Pull Request #6246 · apache/datafusion-comet

manuzhang · 2026-09-26T16:21:46Z

Which issue does this PR close?

Part of #4180 (the spark.sql.parquet.binaryAsString audit item) and #4764. Nested values are handled separately in #7106.

Rationale for this change

With spark.sql.parquet.binaryAsString=true, Spark infers unannotated Parquet BINARY columns as strings. Reading these directly into Arrow string arrays can admit unchecked invalid UTF-8, causing debug panics and undefined behavior in native string functions.

Read those values as binary and use Comet's existing Spark-compatible replacement decoder before native string evaluation. Malformed sequences become U+FFFD, following the same policy as binary-to-string casts and JVM imports. The documented differences from Spark's retained raw bytes still apply to byte lengths, hashing, round trips and value identity.

The intermediate Binary type must also preserve Spark's conversion checks: requesting a FIXED_LEN_BYTE_ARRAY column as StringType must still fail.

What changes are included in this PR?

Serialize binary_as_string for Spark's Parquet scan. Contributed scans retain their existing behavior.
Use Binary for top-level string fields in the logical read schema, with a replacement-decoding output projection. Scans that project no top-level string keep the existing path. Partition fields retain their types.
Compare equality, positive IN and null checks as raw bytes where safe. Equality and IN literals containing U+FFFD, ranges, LIKE, NOT IN and other string predicates remain above the scan. Spark's residual filter re-evaluates every data filter.
Preserve direct-column pruning for the supported equality, IN and null-check predicates on annotated strings. Broader pruning improvements are deferred: ranges and LIKE still do not prune top-level string columns while this setting is enabled, including UTF8-annotated columns. No annotation stamping or footer metadata rewrite is added for binaryAsString.
Validate Spark's original requested schema before the intermediate Binary type can admit FIXED_LEN_BYTE_ARRAY as a string. Raise the structured conversion error at the first data-page request so row filtering cannot hide it, while empty files and format-pruned files retain their behavior. The check is scoped to Parquet conversion validation and preserves Iceberg UUID-to-string adaptation.
Document the supported behavior and pruning limitation. Include full-scan and equality cases in CometReadBenchmark --binary-as-string.

How are these changes tested?

Native regressions cover safe raw-byte pushdown, conservative pruning for annotated and unannotated columns, decoded output filtering, and fixed-length binary rejection with row filtering off/on, including a predicate that would discard every row and an empty file. An adapter regression preserves Iceberg UUID-to-string conversion.

ParquetReadSuite adds Spark-parity coverage for fixed-length binary requested as string versus binary. The existing partitioned-file, malformed-byte, join and top-k coverage remains.

Local validation: all 344 native Parquet tests passed (one existing test ignored). Rust formatting, Scala Spotless checks for the affected files and git diff --check passed. Spark 4.2 SQL tests and the benchmark compilation/lint check are enabled through CI labels. No new throughput measurements are claimed.

kazuyukitanimura · 2026-10-15T18:56:22Z

Hi @manuzhang is this ready for review?

manuzhang · 2026-10-16T04:32:26Z

Hi @kazuyukitanimura, thanks for the reminder. It's ready now.

kazuyukitanimura

native/core/src/execution/operators/dynamic_filter/parquet_reader.rs.
"The dynamic filter path is built against the Utf8 output, but the scan now reads Binary and casts in a pushed projection. What happens to a join dynamic filter or a TopK dynamic filter on a string column when binaryAsString=true? Could you add a test for a join on a string key and an ORDER BY s LIMIT n? If the scan can't accept the filter safely, it should decline it."

This touches the native scan and a serde. Could you run dev/local-ci.sh spark sql_core-1, or add a run-spark-*-tests label? Spark's parquet suites cover binaryAsString, and the PR tier skipped them.

andygrove · 2026-10-18T17:43:33Z

The PR description says Parquet filter pushdown is skipped while spark.sql.parquet.binaryAsString=true, but that isn't in the user docs. Could you add it to docs/source/user-guide/latest/parquet.md, near the Filter Pushdown section or the binaryAsString entry under known issues, so users who turn the config on know why they lose pushdown?

sunchao

Reviewed the full diff from a45445c6ef770b2d008b63ef42478bfdbc3f1e01 to 6c310a48f79f0e53b39ee5a31ad9189ab7616ce7, including both commits and all 15 changed files. The PR remains open and non-draft. Read the existing reviews, conversation comments, inline comments, and threads.

No additional introduced P1/P2 issues found within this review. The previously reported string-filter pruning regression remains reproducible and unresolved, so existing_blockers=true.

Routed skills: review-comet-pr, review-comet-expression-pr, and review-comet-ffi-pr for array-boundary checks.

Summary

Prior state and problem: With spark.sql.parquet.binaryAsString=true, Spark infers unannotated Parquet BINARY as StringType. The native scan previously passed that string type into decoding. Nested binary-to-string adaptation used Arrow’s safe cast, which could replace malformed values with nulls.
Design approach: Read top-level string fields as binary, then apply Comet’s existing Spark-compatible replacement decoder through a scan projection. Nested conversions reuse spark_cast through the schema adapter.
Correctness: Compared Spark’s schema converter, row converter, vector updater, UTF8String, and relevant Parquet tests across 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0. String rendering follows the existing Comet policy. Local tests passed for malformed nested values, nulls, reordered projections, partition values, and missing-column string defaults. The discussed string-key dynamic-filter trigger is currently excluded by Comet’s integer-only join and TopK reader-filter eligibility checks. No additional P1/P2 correctness issue was identified.
Compatibility analysis: The protobuf field is additive and defaults to false. Scala sets it for the core Parquet scan without enabling it for contrib scans. Compared the relevant paths with branch-1.1 at 89c729a9697467e8f8d89e3eadf1797a75e2328c: replacement decoding is the intended change, while loss of pruning on annotated strings is the existing regression described below. No version-specific semantic split requiring a shim was found.
Key design decisions: Partition fields retain their original types. Filters referencing converted top-level string columns are withheld from the reader, while independent filters remain eligible. The residual filter preserves final filtering of decoded values.
Implementation sketch: CometNativeScan serializes binary_as_string; the native planner forwards it to init_datasource_exec. Helpers construct the binary read schema, identify affected filter columns, and build the output projection. parquet_convert_array_impl adds the nested conversion path. Other native call sites explicitly pass false.
Performance: Confirmed the concern already raised in the pruning thread. A local annotated-string fixture had two 500-row groups containing one and two, queried with s = 'one'. With the flag false, one group was pruned and scan_io_data_bytes was 49. With it true, no group was pruned and data reads increased to 98 bytes. Both returned 500 rows after the residual filter. This substantiates the existing P2 concern without claiming a throughput measurement. Annotated strings also traverse the additional conversion path already discussed by reviewers.
Design: Decoding arbitrary bytes at ingress fits Comet’s established string representation. The session-wide schema rewrite is broader than the affected physical columns, however. The author’s proposed per-file coercion control addresses the confirmed pruning problem more precisely, but it is absent from this head.
Abstraction & complexity: The schema and projection helpers are localized and reuse existing cast machinery. No new ownership mechanism or FFI lifecycle is introduced. Retaining the logical string schema and adapting only affected file columns would remove the need for the current static string-filter exclusion.
Behavioral changes worth calling out: Malformed values become replacement characters rather than retaining Spark’s raw bytes, so the documented byte-length, hashing, round-trip, and value-identity differences apply. The nested adapter change also operates independently of binaryAsString when that conversion path is reached. Ordinary annotated string scans are affected by the flag despite not needing binary inference.
Suggested improvements: Resolve the existing pruning regression before merging this head. The per-file approach described in the author’s follow-up is a concrete direction. No additional improvement meeting the P1/P2 reporting bar was identified.

Exact-head CI: run 38850857342 passed Required Checks, native tests, the Spark 4.1 Comet suites, and TPC-H/TPC-DS checks. Logs explicitly confirm both added Rust tests and the added ParquetReadV1Suite test passed. Spark’s SQL suites, Iceberg SQL suites, macOS, and benchmark jobs were skipped.

Local validation: both added Rust tests and two disposable probes passed. The initial build lacked JNI headers; rerunning with an available JDK 17 resolved that limitation. No local JVM integration suite, full Spark SQL/version matrix, or throughput benchmark was run. Disposable test hooks were removed, and the checkout is clean at the requested SHA. Nothing was published.

…ings 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 <[email protected]>

comphead

This head implements everything in Comet now: apache/datafusion#25802 was closed unmerged on 2026-10-20, and the pinned DataFusion is 55.2.0. The thread's pruning, double-cast and dynamic-filter points are not repeated. Four comments.

comphead · 2026-10-23T00:13:33Z

+    }
+}
+
+fn binary_as_string_projection(


A design question that follows from #25802 being closed, and it relates to the double-cast thread. Could the per-file decision move into SparkPhysicalExprAdapter? DataFusion 55.2's opener runs the adapter over the pushed projection for each file whose physical schema differs from the logical one. For a file that annotates the column, the projection should reach the adapter as Comet's Cast to Utf8 over a DataFusion CastExpr to Binary over a Utf8 column. A rule next to unwrap_cast_to_binary_comparison could return the column itself, since arrow-rs has already validated those bytes (validate_utf8 is set for ConvertedType::UTF8), and unannotated files would keep the decode. I would expect the annotated full scan in CometReadBenchmark --binary-as-string to drop back toward its binaryAsString=false time, and the same rule might let ranges and LIKE be pushed against the decoded value again. I have not tried it, and I have not checked that the expression shape reaches the adapter as described. arrow-rs maps JSON-annotated BINARY to Utf8 without validating it, so such a column would be read unchecked, as it is with the setting off.

Please retain range pruning for annotated strings: s < 'two'.

@rich7420 I'd like to leave it out of this PR since it's a more substantial change.

andygrove

Thanks @manuzhang, and sorry that apache/datafusion#25802 didn't land. Reading unannotated columns as binary and decoding at the boundary fixes a real crash, and the binary-literal pushdown for =, IN and IS NULL is a nice way to keep pruning.

I'm requesting changes for one thing. Spark-written files annotate their strings, and arrow-rs already validates those bytes, so for annotated columns this path costs a 2x full scan and loses range and LIKE pruning without fixing anything. That's a regression from 1.1 for anyone who turns binaryAsString on. comphead's suggestion at parquet_exec.rs:219 looks like the way through: collapse the cast pair back to the column in SparkPhysicalExprAdapter for files that annotate it, and keep the decode for those that don't. Could you try that, and rerun CometReadBenchmark --binary-as-string so we can see the annotated full scan back near its binaryAsString=false time?

Since #7106 is approved and will likely land first, the parquet.md bullet will also need to combine both PRs' parts, as comphead described.

…ings 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 <[email protected]>

…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 <[email protected]> * 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 <[email protected]> * 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 <[email protected]> * 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 <[email protected]> * 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 <[email protected]> * 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 <[email protected]> * 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 <[email protected]> * 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 <[email protected]> * 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 <[email protected]> --------- Co-authored-by: Claude Opus 5.5 <[email protected]>

…ings (#7106) (#7339) * 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 (#4764). Top-level columns are unchanged: DataFusion's file-schema coercion still decodes them straight into Utf8 (#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 #6246. * 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. * 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. * 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. * 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. * 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. * 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. * 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. * 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. --------- (cherry picked from commit 621b9cc) Co-authored-by: Claude Opus 5.5 <[email protected]>

…ings (#7106) (#7340) * 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 (#4764). Top-level columns are unchanged: DataFusion's file-schema coercion still decodes them straight into Utf8 (#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 #6246. * 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. * 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. * 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. * 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. * 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. * 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. * 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. * 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. --------- (cherry picked from commit 621b9cc) Co-authored-by: Claude Opus 5.5 <[email protected]>

manuzhang · 2026-11-04T14:36:54Z

Thanks all for the review. I've rebased on #7106 and reran the benchmark, and the full-scan regression is gone (check the numbers in the PR description).

rich7420 · 2026-11-04T17:54:16Z

+/// Arrow Utf8.
+fn binary_as_string_read_field(field: &FieldRef) -> FieldRef {
+    if is_binary_as_string_read(field.data_type()) {
+        Arc::new(field.as_ref().clone().with_data_type(DataType::Binary))


Please preserve Spark’s rejection of FIXED_LEN_BYTE_ARRAY(3) as STRING.

This is fixed now. Please check again.

rich7420 · 2026-11-07T16:56:09Z

+            .iter()
+            .any(|field| is_binary_as_string_read(field.data_type()));
+    let read_base_schema = if binary_as_string {
+        binary_as_string_read_schema(&base_schema)


@manuzhang thanks for the update.

Dictionary string hints still slip through here and panic on malformed bytes in debug builds. Let's handle both shapes in one small helper and test PLAIN and dictionary-encoded pages.

Could you share how to reproduce the panic?

Sorry, I got the scope wrong. An unannotated BINARY column with a Utf8 or Dictionary ARROW:schema hint panics in debug builds, and main does too with binaryAsString off, so it isn't from this PR (same root cause as apache/arrow-rs#11140). I'm withdrawing this request. Repro on main: rich7420@424ab31

When spark.sql.parquet.binaryAsString is enabled, Spark infers unannotated Parquet BINARY columns as StringType. arrow-rs validates UTF-8 only for columns annotated as strings, so decoding such a column straight into Arrow Utf8 produced an invalid string array in release builds, which is undefined behavior, and panicked in debug builds. Read top-level string columns as Binary, then decode them to Utf8 at the scan boundary with the same Spark-compatible replacement the cast (apache#4763) and the FFI import boundary (apache#5310) use. Nested values are handled separately, in Parquet filter pushdown is skipped while the configuration is enabled, because filter expressions are built against the output Utf8 schema and are not yet rewritten against the Binary read schema. Spark's Filter above the scan re-evaluates every data filter, so results stay correct. Co-Authored-By: Claude Opus 5.5 <[email protected]>

With spark.sql.parquet.binaryAsString enabled, the native scan skipped Parquet filter pushdown for every filter, because filters are typed against the Utf8 output schema while top-level string columns are read as Binary. Only filters on those columns are affected, so drop just those and push the rest, which still prune row groups. Spark's Filter above the scan still evaluates every data filter. Co-Authored-By: Claude Opus 5.5 <[email protected]>

With spark.sql.parquet.binaryAsString enabled, the scan reads top-level string columns as Binary, so filters on them were kept out of the scan and pruned nothing. Push =, IN and IS [NOT] NULL on those columns with a Binary literal instead. A valid value decodes to itself and a malformed one to a string containing U+FFFD, so for a literal without U+FFFD the byte comparison keeps exactly the rows the string comparison keeps. Ranges, LIKE, function calls and literals containing U+FFFD stay above the scan, because decoding can reorder values. An AND keeps whichever side can be pushed, since Spark's Filter above the scan re-evaluates every data filter. For a file that annotates the column as a string, the default adapter compares CAST(s AS BINARY), which pruning can't see through. Comet's adapter now compares the column with the literal in the column's own type, so those files prune too. A join or TopK dynamic filter is pushed onto the scan's output, so it is mapped through the decoding projection and compares decoded strings. A test pins that, though today these filters accept only integer keys. Also read only Utf8 fields as Binary. Spark's StringType is never LargeUtf8, and spark_cast has no LargeBinary to LargeUtf8 arm. Co-Authored-By: Claude Opus 5.5 <[email protected]>

With spark.sql.parquet.binaryAsString enabled, the native scan reads Spark-written strings as Binary too and decodes them back. Add a full scan and a selective equality filter over sorted strings in small row groups, each with the setting off and on. Run it alone with --binary-as-string. Co-Authored-By: Claude Opus 5.5 <[email protected]>

The scan tests now check that NOT IN and an IN list holding U+FFFD stay above the scan, and ParquetReadSuite compares NOT IN with Spark. A schema adapter test checks that NOT IN on an annotated column is compared directly, and that a null or invalid UTF-8 Binary literal leaves the filter unchanged. The compatibility guide no longer says Parquet text is always valid UTF-8, and the configuration audit marks binaryAsString as Partial, a status its legend defines. Co-Authored-By: Claude Opus 5.5 <[email protected]>

…ed name The planner binds data filters by position against the required columns followed by the partition columns, so a column's index says which field it reads. Check the required schema's field type at that index instead of folding column names into a set, and drop the fold_name and fold_names imports. A scan that projects no string column now keeps the projection-indices path, since it has nothing to read as Binary. Co-Authored-By: Claude Fable 5.1 <[email protected]>

The schema converter's binaryAsString inference and the BINARY cases of ParquetFilters are the same in 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Co-Authored-By: Claude Fable 5.1 <[email protected]>

Co-Authored-By: Claude Fable 5.2 <[email protected]>

manuzhang · 2026-11-11T14:59:15Z

@andygrove @comphead Do you have further comments? I believe the previous ones have been resolved.

andygrove

Thanks @manuzhang. Rewriting =, IN and IS [NOT] NULL as raw-byte comparisons, collapsing the cast for annotated files, and keeping Spark's rejection of FIXED_LEN_BYTE_ARRAY as STRING all look right to me. I could not find a correctness problem in the pushdown rewrite.

I am keeping my request for changes, because of one thing that is still open. I compared against branch-1.2. There, binaryAsString=true reads strings as Utf8 and pushes every filter unchanged, so a Spark-written string column prunes on s < 'two' and on startsWith. With this PR, ranges and LIKE never reach the reader for any top-level string column in the scan while the setting is on, including columns the file annotates as strings. The description and parquet.md now state this, but it is still a regression from the latest release for anyone who has the setting on, and a tracking issue would not make up for it.

You said on 11-07 that restoring it is a more substantial change. comphead's suggestion at parquet_exec.rs:211 looks like the way through, and the new unwrap_cast_to_binary_comparison already recognizes the CastExpr(col -> Binary) shape. Could the same per-file check also drop the Binary round trip in the output projection when the file annotates the column, and keep the decode only for files that do not? That should bring back range and LIKE pruning and the annotated full-scan time, and then the limitation in parquet.md could go away.

Could you also post CometReadBenchmark --binary-as-string output for the current head? I would like to see the annotated full scan and the c1 = '500' filter with the setting off and on. Your 11-04 comment says the full-scan numbers are in the description, but the description now says no throughput measurements are claimed. I expect the full-scan cost to be one extra UTF-8 validation pass, since valid input shares buffers in the cast, but I would rather see it measured.

github-actions Bot added bug Something isn't working area:scan Parquet scan / data reading area:joins Join operators and dynamic filter pushdown labels Sep 26, 2026

manuzhang force-pushed the binary-as-string-native branch from d4d898c to 282232f Compare September 27, 2026 14:03

This was referenced Sep 27, 2026

Allow a Parquet scan to read binary columns as binary when the table schema declares a string type apache/datafusion#25801

Closed

feat(parquet): add coerce_binary_to_string option to read binary file columns as binary apache/datafusion#25802

Closed

manuzhang force-pushed the binary-as-string-native branch from 282232f to 6c310a4 Compare October 16, 2026 04:29

manuzhang marked this pull request as ready for review October 16, 2026 04:32

kazuyukitanimura reviewed Oct 16, 2026

View reviewed changes

Comment thread native/core/src/parquet/parquet_exec.rs Resolved

Comment thread native/core/src/parquet/parquet_exec.rs Resolved

Comment thread spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala Resolved

Comment thread native/core/src/parquet/parquet_support.rs Resolved

manuzhang mentioned this pull request Oct 19, 2026

fix: decode invalid UTF-8 in nested Parquet BINARY values read as strings #7106

Merged

sunchao reviewed Oct 19, 2026

View reviewed changes

manuzhang force-pushed the binary-as-string-native branch from 6c310a4 to 1f4fae6 Compare October 21, 2026 00:22

github-actions Bot added the area:Delta Delta Lake contrib (native scan, deletion vectors) label Oct 21, 2026

manuzhang force-pushed the binary-as-string-native branch from 1f4fae6 to f5408c3 Compare October 21, 2026 14:59

manuzhang added the run-spark-4.2-tests Run the Spark 4.2 SQL tests on this PR label Oct 21, 2026

rich7420 reviewed Oct 22, 2026

View reviewed changes

Comment thread docs/source/user-guide/latest/parquet.md Resolved

Comment thread docs/source/contributor-guide/spark_configs_support.md Resolved

Comment thread native/core/src/parquet/schema_adapter.rs Resolved

comphead reviewed Oct 23, 2026

View reviewed changes

andygrove requested changes Oct 23, 2026

View reviewed changes

manuzhang force-pushed the binary-as-string-native branch from 293210a to 9c427cd Compare October 30, 2026 16:04

manuzhang requested review from andygrove, comphead, kazuyukitanimura, rich7420 and sunchao November 4, 2026 14:34

rich7420 reviewed Nov 4, 2026

View reviewed changes

comphead mentioned this pull request Nov 5, 2026

fix: report make_date Date32 overflow in both ANSI modes #7343

Open

manuzhang added the run-benchmark-check Run the benchmark compile and lint check on this pull request instead of waiting for the merge queue label Nov 7, 2026

manuzhang mentioned this pull request Nov 7, 2026

feat(parquet): allow updating cached file key-value metadata apache/arrow-rs#12117

Open

rich7420 reviewed Nov 7, 2026

View reviewed changes

manuzhang and others added 10 commits November 10, 2026 22:09

docs: record the Spark versions checked for the binaryAsString audit

00fbaf6

The schema converter's binaryAsString inference and the BINARY cases of ParquetFilters are the same in 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Co-Authored-By: Claude Fable 5.1 <[email protected]>

fix: preserve binaryAsString type checks and annotated string pruning

1fe3c67

fix: retain only binaryAsString conversion validation

a18390f

fix: use Arc::clone in the binaryAsString adapter test

e8df434

Co-Authored-By: Claude Fable 5.2 <[email protected]>

manuzhang force-pushed the binary-as-string-native branch from 1f9ba1e to e8df434 Compare November 10, 2026 14:12

andygrove requested changes Nov 11, 2026

View reviewed changes

Conversation

manuzhang commented Sep 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

How are these changes tested?

Uh oh!

kazuyukitanimura commented Oct 15, 2026

Uh oh!

manuzhang commented Oct 16, 2026

Uh oh!

kazuyukitanimura left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

andygrove commented Oct 18, 2026

Uh oh!

sunchao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

comphead left a comment

Choose a reason for hiding this comment

Uh oh!

comphead Oct 23, 2026

Choose a reason for hiding this comment

Uh oh!

rich7420 Nov 4, 2026

Choose a reason for hiding this comment

Uh oh!

manuzhang Nov 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

andygrove left a comment

Choose a reason for hiding this comment

Uh oh!

manuzhang commented Nov 4, 2026

Uh oh!

rich7420 Nov 4, 2026

Choose a reason for hiding this comment

Uh oh!

manuzhang Nov 7, 2026

Choose a reason for hiding this comment

Uh oh!

rich7420 Nov 7, 2026

Choose a reason for hiding this comment

Uh oh!

manuzhang Nov 10, 2026

Choose a reason for hiding this comment

Uh oh!

rich7420 Nov 11, 2026

Choose a reason for hiding this comment

Uh oh!

manuzhang commented Nov 11, 2026

Uh oh!

andygrove left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

manuzhang commented Sep 26, 2026 •

edited

Loading

manuzhang Nov 7, 2026 •

edited

Loading