Skip to content

fix: support binaryAsString in the native Parquet scan - #6246

Open
manuzhang wants to merge 5 commits into
apache:mainfrom
manuzhang:binary-as-string-native
Open

manuzhang wants to merge 5 commits into
apache:mainfrom
manuzhang:binary-as-string-native

Conversation

@manuzhang

@manuzhang manuzhang commented Sep 26, 2026 •

Copy link
Copy Markdown
Member

Which issue does this PR close?

Part of #4180 (the spark.sql.parquet.binaryAsString audit item) and #4764. Values nested in structs, lists and maps are fixed separately in #7106.

Rationale for this change

This fixes the native Parquet scan with spark.sql.parquet.binaryAsString=true when a column holds bytes that aren't valid UTF-8.

With the setting on, Spark reads a BINARY column without the UTF8 annotation as a string. Comet's native scan asks arrow-rs for such a column as an Arrow string, but arrow-rs checks UTF-8 only for columns the file annotates as strings. So invalid bytes land in an Arrow string array unchecked:

This PR reads these columns as binary and converts them to strings after the scan, replacing each ill-formed sequence with U+FFFD. CAST(binary AS string) (#4763) and data imported from the JVM (#5310) already handle invalid UTF-8 this way (#4764). Collected values still match Spark, which renders these strings the same way. Byte-level results such as hex(value) now show the bytes of U+FFFD where Spark keeps the original bytes, as described under "Strings with non-UTF-8 bytes" in the compatibility guide.

Reading the columns as binary has a cost. A filter on them compares decoded strings, which the scan can't check against the file's statistics, so it stops pruning. Spark's schema doesn't say which columns a file annotates, so this also hits ordinary string columns in files Spark wrote. The filter changes in this PR restore pruning for =, IN and IS [NOT] NULL.

What changes are included in this PR?

  • With the setting on, the native scan reads every top-level string column as Binary and decodes it to Utf8 in a projection at the scan boundary. Spark's schema doesn't say which file columns are annotated, so annotated columns take the same path. NativeScanCommon carries the setting as binary_as_string.
  • Filters on those columns:
    • =, IN and IS [NOT] NULL are pushed with a Binary literal, so value = 'one' becomes value = X'6F6E65'. 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.
    • For a file that annotates the column as a string, Comet's expression adapter turns CAST(value AS BINARY) = X'6F6E65' back into value = 'one', so the file's statistics still prune.
    • Ranges, LIKE, function calls and literals containing U+FFFD stay above the scan, because decoding can reorder values: "a\x80" sorts below "aé" as bytes, but "a�" sorts above it.
    • Filters on other columns are pushed as before. An AND keeps whichever side can be pushed, since Spark's Filter above the scan re-evaluates every data filter.
  • Join and TopK dynamic filters accept only integer keys, so they never reach these columns. If one did, the scan would map it through the decoding projection, and it would compare decoded strings.
  • Docs: Filter Pushdown and the known issue for BINARY values read as strings in parquet.md, and the audit entry in spark_configs_support.md.

Compared with Spark, this prunes more on unannotated columns and less on annotated ones. Spark pushes a filter only when the literal's type matches the column's Parquet type, so it doesn't push = or IN on an unannotated BINARY column, though it does push IS NULL. On an annotated column Spark also pushes ranges, which stay above the scan here while the setting is on.

How are these changes tested?

  • parquet_exec.rs, with three row groups so the returned rows show which ones were pruned:

    • An unannotated column holding a valid value, 0xff and a null: =, IN and IS [NOT] NULL prune, a range and a U+FFFD literal don't, and AND keeps the side it can push. Each runs with and without row-level filtering.
    • An annotated column: =, IN and IS [NOT] NULL prune.
    • A filter pushed onto the scan's output, the way dynamic filters are, compares decoded strings.
  • ParquetReadSuite:

    • Schema inference, Spark answer parity and CometNativeScanExec under both values, as before.
    • New: two partition directories holding a valid value, 0xff and a null, plus a Spark-written file, with row-level filtering off and on. Compares =, IN, IS [NOT] NULL, a range, an OR, a self-join on the string column and ORDER BY value, id LIMIT 3 with Spark, and checks with pushdown_rows_matched that value = 'one' reaches the row filter.
  • CometReadBenchmark --binary-as-string scans Spark-written (annotated) strings with the setting off and on:

    10M rows on an Apple M2, best time of each case. = on annotated strings prunes as it does with the setting off. A full scan takes twice as long, because decoding copies every value.

    Comet case binaryAsString=false true
    Full scan, sum(length(c1)) 72 ms 150 ms
    c1 = '500' 17 ms 16 ms
  • Malformed values nested in arrays, structs and maps are covered in fix: decode invalid UTF-8 in nested Parquet BINARY values read as strings #7106.

🤖 Generated with Claude Code

@kazuyukitanimura

Copy link
Copy Markdown
Contributor

Hi @manuzhang is this ready for review?

@manuzhang
manuzhang force-pushed the binary-as-string-native branch from 282232f to 6c310a4 Compare October 16, 2026 04:29
@manuzhang
manuzhang marked this pull request as ready for review October 16, 2026 04:32
@manuzhang

Copy link
Copy Markdown
Member Author

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

@kazuyukitanimura kazuyukitanimura left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

// In binaryAsString mode the filters are typed against the Utf8 output schema, but the scan
// reads the string columns as Binary, so drop the filters that reference one for the same
// reason. Filters on other columns still prune.
let data_filters = match data_filters {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for fixing this. Reading the string columns as Binary and decoding at the boundary is the right shape. I'm worried about the pruning we lose. Before this PR, a filter like s = 'x' with binaryAsString=true pruned row groups. Now it never reaches the reader. Could we keep that pushdown, for example by rewriting the filter against the Binary read schema with a binary literal for =, IN and IS NULL? If that isn't feasible here, could you show benchmark numbers and explain why losing pruning is acceptable for users who turn this config on?

Ok(false)
}

fn binary_as_string_projection(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For a file column that is already annotated UTF8, I think this reads Utf8, casts to Binary in the adapter, and then casts back to Utf8 in the projection. Could you benchmark a scan of annotated string columns with binaryAsString=false and true against branch-1.1? If there is a regression, deciding per file whether the column is annotated would avoid the double cast.

}
}

test("spark.sql.parquet.binaryAsString uses native scan") {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add these cases to the test?
- A WHERE value = 'one' filter, which exercises the dropped-filter path and compares against Spark.
- An array and a struct built from unannotated binary with a malformed byte.
- A partitioned or multi-file read

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

Comment thread native/core/src/parquet/parquet_support.rs Outdated
@andygrove

Copy link
Copy Markdown
Member

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 sunchao left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

manuzhang added a commit to manuzhang/datafusion-comet that referenced this pull request Oct 20, 2026
…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 <noreply@anthropic.com>
manuzhang added a commit to manuzhang/datafusion-comet that referenced this pull request Oct 20, 2026
…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 <noreply@anthropic.com>
manuzhang added a commit to manuzhang/datafusion-comet that referenced this pull request Oct 20, 2026
…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 <noreply@anthropic.com>
manuzhang added a commit to manuzhang/datafusion-comet that referenced this pull request Oct 20, 2026
…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 <noreply@anthropic.com>
manuzhang added a commit to manuzhang/datafusion-comet that referenced this pull request Oct 20, 2026
…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 <noreply@anthropic.com>
manuzhang added a commit to manuzhang/datafusion-comet that referenced this pull request Oct 20, 2026
…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 <noreply@anthropic.com>
@manuzhang
manuzhang force-pushed the binary-as-string-native branch from 6c310a4 to 1f4fae6 Compare October 21, 2026 00:22
@github-actions github-actions Bot added the area:Delta Delta Lake contrib (native scan, deletion vectors) label Oct 21, 2026
manuzhang and others added 4 commits October 21, 2026 22:51
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
@manuzhang
manuzhang force-pushed the binary-as-string-native branch from 1f4fae6 to f5408c3 Compare October 21, 2026 14:59
@manuzhang manuzhang added the run-spark-4.2-tests Run the Spark 4.2 SQL tests on this PR label Oct 21, 2026
reach an Arrow string array unchecked. A debug build panics on them
such nested value. At the top level, `spark.sql.parquet.binaryAsString=true` replaces each
ill-formed sequence with `U+FFFD`, as described in
[Strings with non-UTF-8 bytes](compatibility/index.md#strings-with-non-utf-8-bytes). Through a

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The linked section still says non-UTF-8 strings don't occur in text read from Parquet. Could it mention this binaryAsString path too, so the two pages agree?

- Date: 2026-07-18
- `spark.sql.parquet.binaryAsString`
- Default: `false`
- Status: Native (see notes)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: the legend doesn't define Native. Would Partial fit here, given the documented U+FFFD differences?

Suggested change
- Status: Native (see notes)
- Status: Partial (see notes)

/// pruning can't see through the cast. The cast keeps the bytes, so `CAST(s AS BINARY) = X'..'`
/// holds exactly when `s` equals the same bytes in its own type, and likewise for `IN` and
/// `IS [NOT] NULL`. Compare the column directly so the file can still prune.
fn unwrap_cast_to_binary_comparison(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we add tests for NOT IN, and for literals this leaves unchanged (invalid UTF-8 or null binary)? I don't think those cases are covered yet.

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 <noreply@anthropic.com>

@comphead comphead left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

}
}

fn binary_as_string_projection(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.


// Folded names of the columns read as binary. Filters that reference one are rewritten for
// the binary read below, or kept out of the scan.
let binary_read_columns: HashSet<String> = if binary_as_string {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The planner binds data filters by position against required_schema followed by the partition columns, and filter_read_schema already relies on that. So reads_binary could check column.index() < required_schema.fields().len() and required_schema.field(column.index()).data_type() == &DataType::Utf8, and references_any_column could reuse that check over collect_columns. The folded-name HashSet and the new fold_name and fold_names imports would then go. Gating binary_projection and read_base_schema on at least one such column existing would also keep scans with no string column on the existing with_projection_indices path. I read the binding order in planner.rs and did not run it.

([#7030](https://github.com/apache/datafusion-comet/issues/7030)).
- **Nested BINARY values that are not valid UTF-8, read as STRING.** This applies to a BINARY
column without the UTF8 annotation that is read as `string`, through the read schema or
- **BINARY values that are not valid UTF-8, read as STRING.** This applies to a BINARY column

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This bullet and #7106 rewrite the same lines, and git merge-tree reports a conflict in parquet.md between the two heads. Each version describes the other PR's part as still open. This one keeps "Comet returns NULL for each such nested value", and #7106 keeps "Top-level values reach an Arrow string array unchecked" without the binaryAsString case. Whichever lands second could merge them: nested values and top-level values read with binaryAsString=true become U+FFFD, and top-level values read through a read schema still reach an Arrow string array unchecked (#7032). The list of decode sites at the top of "Strings with non-UTF-8 bytes" in compatibility/index.md could name this scan too, as #7106 does for nested values.

- `spark.sql.parquet.binaryAsString`
- Default: `false`
- Status: Partial (see notes)
- Affected operator: native Parquet scan

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The other two audited entries in this file record a Spark versions checked line. Could this one list the versions whose schema converter and ParquetFilters were compared, so the audit entries stay uniform?

@andygrove andygrove left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:Delta Delta Lake contrib (native scan, deletion vectors) area:joins Join operators and dynamic filter pushdown area:scan Parquet scan / data reading bug Something isn't working run-spark-4.2-tests Run the Spark 4.2 SQL tests on this PR

Projects

None yet

Development

Successfully merging this pull request may close these issues.

6 participants