manuzhang opened a new pull request, #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246
## Which issue does this PR close?
Part of #4180 (the `spark.sql.parquet.binaryAsString` audit item) and #4764.
Does not close #4121. Gap A is the same mechanism applied to annotated
`STRING`
columns; this PR enables the binary read schema only when
`spark.sql.parquet.binaryAsString` is set, so `hll.sql` stays disabled.
## Rationale for this change
When `spark.sql.parquet.binaryAsString` is enabled, Spark infers unannotated
Parquet `BINARY` columns as `StringType`. Those columns can hold arbitrary
bytes,
and Spark keeps them verbatim in `UTF8String`.
arrow-rs validates UTF-8 during Parquet decode whenever the target Arrow
field is
`Utf8`/`LargeUtf8` (`arrow/buffer/offset_buffer.rs::try_push`). So reading
such a
column natively failed with `Parquet error: encountered non UTF-8 data` where
Spark succeeds.
This follows the ingress policy the project already applies elsewhere:
decode at
the boundary rather than reinterpret or reject. `CAST(binary AS string)`
(#4763)
and the JVM-to-native FFI import boundary (#5310) both decode with
`decode_utf8_spark_lossy`, which matches the JVM's `new String(bytes,
UTF_8)`.
This extends the same treatment to the native scan.
The alternative, keeping affected data in Spark to preserve raw bytes, was
considered and rejected. It contradicts the documented policy in
`compatibility/index.md`, and it cannot coexist with the test added in #5234,
which asserts `CometNativeScanExec` under `binaryAsString=true`.
## What changes are included in this PR?
- `NativeScanCommon.binary_as_string` carries the configuration to the native
scan; `CometNativeScan` sets it from `SQLConf.PARQUET_BINARY_AS_STRING`.
- When it is set, `init_datasource_exec` builds the read schema with
top-level
`Utf8`/`LargeUtf8` fields replaced by `Binary`/`LargeBinary`, so arrow-rs
runs
no validation, and adds a scan-boundary projection that casts each such
column
back to the output type.
- `parquet_convert_array` handles `(Binary, Utf8)` and `(LargeBinary,
LargeUtf8)`
with `spark_cast`, placed ahead of the `can_cast_types` arm so arrow's
validating cast never sees malformed bytes. This covers nested values as
well
as the top-level projection.
- Documents the configuration in the Spark configuration audit guide,
including
the divergences that follow from lossy decoding.
### Known limitation
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. Results stay correct, since
Spark's
`Filter` above the scan re-evaluates every data filter, but a scan that would
otherwise prune row groups does not. Happy to split that out or address it
here,
whichever reviewers prefer.
### Divergence this does not remove
Decoding is not byte-preserving, so the differences already documented under
[Strings with non-UTF-8
bytes](https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)
apply to malformed values this configuration surfaces: byte-level round
trips,
`octet_length`, hashing, and value identity, where two distinct malformed
sequences both become `U+FFFD`. That is the same trade-off taken by #4763 and
#5310.
## How are these changes tested?
- `ParquetReadSuite`: a new `spark.sql.parquet.binaryAsString uses native
scan`
writes a raw Parquet file with no Spark schema metadata containing a valid
value and the malformed byte `0xff`, and asserts the inferred schema under
both
configuration values, Spark answer parity, `CometNativeScanExec` in both
cases,
and that a projection reading only the non-string column stays native.
- A Rust test covering a nested `List<Binary>` to `List<Utf8>` conversion,
which
exercises the new arm through `parquet_convert_array` rather than the scan
projection.
- The dictionary test added in #5234 passes unchanged, which is the case the
rejected fallback design would have broken.
Commands run locally:
```
make core # clean, no warnings
cargo test -p datafusion-comet --lib parquet # 197 passed, 0 failed, 5
ignored
./mvnw test -Dtest=none \
-Dsuites="org.apache.comet.parquet.ParquetReadV1Suite binary" -Pspark-4.1
# 4 passed, 0 failed
./mvnw spotless:apply -Pspark-4.1
```
Only Spark 4.1 has been exercised locally so far, and the broader Spark SQL
and
Iceberg suites have not been run. This touches the serde, the planner and a
native operator, so it needs `run-spark-sql-tests` before it goes near the
merge
queue.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
