Repository navigation
Conversation
github-actions
Bot
added
bug
d4d898c to
282232f
Compare
|
Hi @manuzhang is this ready for review? |
282232f to
6c310a4
Compare
|
Hi @kazuyukitanimura, thanks for the reminder. It's ready now. |
kazuyukitanimura
left a comment
kazuyukitanimura
left a comment
There was a problem hiding this comment.
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.
|
The PR description says Parquet filter pushdown is skipped while |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
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 ParquetBINARYasStringType. 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_castthrough 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.1at89c729a9697467e8f8d89e3eadf1797a75e2328c: 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:
CometNativeScanserializesbinary_as_string; the native planner forwards it toinit_datasource_exec. Helpers construct the binary read schema, identify affected filter columns, and build the output projection.parquet_convert_array_impladds 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
oneandtwo, queried withs = 'one'. With the flag false, one group was pruned andscan_io_data_byteswas 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
binaryAsStringwhen 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]>
…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 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 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 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 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]>
6c310a4 to
1f4fae6
Compare
github-actions
Bot
added
the
area:Delta
1f4fae6 to
f5408c3
Compare
comphead
left a comment
comphead
left a comment
There was a problem hiding this comment.
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.
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
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 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 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]>
|
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). |
manuzhang
added
the
run-benchmark-check
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]>
1f9ba1e to
e8df434
Compare
|
@andygrove @comphead Do you have further comments? I believe the previous ones have been resolved. |
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
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.
Push string filters other than =, IN and IS [NOT] NULL against the decoded value instead of keeping them above the scan. For a file that annotates the column as a string, the expression adapter drops the cast to Binary and back, in the output projection and in pushed filters, so the file reads and prunes as it does with the setting off. Co-Authored-By: Claude Fable 5.2 <[email protected]>
|
Thanks @andygrove I've made changes as suggested and updated PR description with the benchmark results. Please take another look, thanks! |
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
The PR backlog is huge and the maintainers don't have the bandwidth to review them all, so here is an automated review. Feel free to ignore it.
This is a re-review of the change since e8df434b58, which is a single commit (fc6bf4d671). My earlier request for changes was about ranges and LIKE never reaching the reader for a string column that the file annotates, plus the missing benchmark for the current head. Both are addressed.
SparkPhysicalExprAdapter::unwrap_binary_round_trip now reduces CAST(CAST(s AS BINARY) AS STRING) to the column when the file stores it as Utf8, and it runs before unwrap_cast_to_binary_comparison. I read it against column_cast_to_binary, and it only fires for a Spark Cast to Utf8 over the adapter's own cast to Binary of a Utf8 file column. That is safe because arrow-rs has already validated an annotated column and decoding valid UTF-8 returns it unchanged. An unannotated file keeps the Binary column and the decoding cast, which binary_as_string_reads_annotated_strings_without_casts checks on both sides. The filter side in binary_as_string_pushdown_filter now rewrites every other reference to decode(s) instead of withholding the filter, so range, LIKE and NOT IN filters prune on annotated files and still filter rows on unannotated ones. binary_as_string_prunes_annotated_strings covers both the pruning-only and the row-filter modes. The benchmark in the description shows the annotated full scan with the setting on (67 ms) at or below the setting off (74 ms), and the equality and range cases match as well.
Against branch-1.2, an annotated string column now reads as Utf8 with its filters pruning the same way, so the pruning and full-scan regression is gone. The remaining differences from the release are the intended ones. An unannotated BINARY column is decoded with replacement instead of being read unchecked, and FIXED_LEN_BYTE_ARRAY requested as a string is rejected the way Spark rejects it. The thread about an unannotated column with a Utf8 ARROW:schema hint was withdrawn upstream of this PR, since it also panics on main with the setting off.
I found nothing further to raise on this head. My earlier blocking findings no longer hold.
…12117) # Which issue does this PR close? Relates to apache/datafusion-comet#6246. # Rationale for this change Applications that adapt cached Parquet metadata need to replace file-level key/value entries, for example an Arrow schema hint, before constructing a reader. The existing public constructors require rebuilding `FileMetaData` and `ParquetMetaData` to make this change. That loses private encryption metadata and the file decryptor, so this approach cannot safely handle encrypted files. A setter on the existing metadata builder lets callers update those entries while preserving the state required to read the file. A downstream use is Comet's per-file schema adaptation, which currently has to exclude encrypted files from its metadata rewrite. # What changes are included in this PR? Add `ParquetMetaDataBuilder::set_key_value_metadata(Option<Vec<KeyValue>>)`. It replaces the complete list, with `None` removing the entries. `Arc::make_mut` gives shared file metadata copy-on-write behavior and preserves the other file metadata, row groups, page indexes and decryptor. The update affects only in-memory metadata. # Are these changes tested? A unit test covers adding, replacing and removing entries, including an explicitly empty list. It checks that cached metadata remains unchanged, the other file metadata (including encryption fields when enabled) is preserved, and row groups and page indexes remain shared. An encrypted-read regression covers encrypted and plaintext footers. It updates already-loaded metadata, verifies its page indexes, and reads the encrypted data without supplying decryption keys again. This checks that the cached decryptor survives the update. # Are there any significant user-facing changes? This is an additive public API. Existing method signatures and behavior are unchanged. AI assistance: Codex drafted the implementation and regression tests and ran local validation.
Which issue does this PR close?
Part of #4180 (the
spark.sql.parquet.binaryAsStringaudit 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?
binary_as_stringfor Spark's Parquet scan. Contributed scans retain their existing behavior.SparkPhysicalExprAdapterdrops the cast to Binary and back, in the output projection and in pushed filters. The file is read without the extra casts, and every filter, including ranges and LIKE, prunes as it does with the setting off. An unannotated column that an embeddedARROW:schematypes as a string takes this path too and is read unchecked, as it is on main. No annotation stamping or footer metadata rewrite is added for binaryAsString.CometReadBenchmark --binary-as-string.How are these changes tested?
Native regressions cover safe raw-byte pushdown, conservative pruning for unannotated columns, range, LIKE and NOT IN pruning for annotated columns, reading an annotated column without casts, 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.
ParquetReadSuiteadds Spark-parity coverage for fixed-length binary requested as string versus binary, and for ranges, LIKE and NOT IN on annotated and unannotated columns. The existing partitioned-file, malformed-byte, join and top-k coverage remains.Local validation: the native binaryAsString tests and the three binaryAsString tests in
ParquetReadV1Suite(Spark 4.1) passed. Rust formatting, clippy, Scala Spotless checks andgit diff --checkpassed. Spark 4.2 SQL tests are enabled through CI labels.CometReadBenchmark --binary-as-stringon an Apple M2, OpenJDK 21, best time in ms over 10M rows of a Spark-written (annotated) string column:c1 = '500'c1 < '11'