You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "manuzhang (via GitHub)" <gi...@apache.org> on 2026/09/26 16:21:49 UTC
[PR] fix: support binaryAsString in the native Parquet scan [datafusion-comet]
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: support binaryAsString in the native Parquet scan [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura commented on code in PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#discussion_r4299018024
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -52,6 +54,81 @@ mod variant_tests;
/// scan would.
pub(crate) const METADATA_SIZE_HINT: usize = 512 * 1024;
+/// Keep top-level Spark string fields as binary while arrow-rs decodes Parquet. arrow-rs only
+/// validates UTF-8 for columns annotated as strings, so decoding an unannotated BINARY column that
+/// `spark.sql.parquet.binaryAsString=true` types as a string straight into Arrow Utf8 would admit
+/// invalid UTF-8. A projection added at the scan boundary converts the binary arrays to valid Utf8
+/// with Spark-compatible replacement semantics.
+fn binary_as_string_read_schema(schema: &SchemaRef) -> SchemaRef {
+ let fields = schema.fields().iter().map(|field| {
+ binary_as_string_read_type(field.data_type()).map_or_else(
+ || Arc::clone(field),
+ |data_type| Arc::new(field.as_ref().clone().with_data_type(data_type)),
+ )
+ });
+ Arc::new(Schema::new_with_metadata(
+ fields.collect::<Vec<_>>(),
+ schema.metadata().clone(),
+ ))
+}
+
+/// The type `binary_as_string_read_schema` reads a top-level field of `data_type` as, or `None`
+/// when the field is read as declared.
+fn binary_as_string_read_type(data_type: &DataType) -> Option<DataType> {
+ match data_type {
+ DataType::Utf8 => Some(DataType::Binary),
+ DataType::LargeUtf8 => Some(DataType::LargeBinary),
+ _ => None,
+ }
+}
+
+/// Whether `expr` references a column whose folded name is in `folded_names`.
+fn references_any_column(
+ expr: &Arc<dyn PhysicalExpr>,
+ folded_names: &HashSet<String>,
+ case_sensitive: bool,
+) -> Result<bool, ExecutionError> {
+ for column in collect_columns(expr) {
+ if folded_names.contains(&fold_name(column.name(), case_sensitive)?) {
+ return Ok(true);
+ }
+ }
+ Ok(false)
+}
+
+fn binary_as_string_projection(
Review Comment:
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.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -218,6 +218,55 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
+ test("spark.sql.parquet.binaryAsString uses native scan") {
Review Comment:
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<string> and a struct<string> built from unannotated binary with a malformed byte.
- A partitioned or multi-file read
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -272,7 +381,22 @@ pub(crate) fn init_datasource_exec(
.filter(|filter| !calls_jvm(filter))
.collect::<Vec<_>>()
});
- let file_source: Arc<dyn FileSource> = match data_filters {
+ // 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 {
Review Comment:
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?
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -330,6 +330,24 @@ fn parquet_convert_array_impl(
Ok(Arc::new(string_array))
}
+ // Parquet's unannotated BINARY is read as Spark StringType when
+ // spark.sql.parquet.binaryAsString is enabled or a read schema declares a string.
+ // Arrow's cast below is safe, so it would turn malformed UTF-8 into NULL. Use Spark's
+ // cast so the scan's top-level projection and nested values replace malformed input
+ // with U+FFFD, as the JVM's `new String(bytes, UTF_8)` does.
+ (Binary, Utf8) | (LargeBinary, LargeUtf8) => {
Review Comment:
This arm also changes behavior with binaryAsString=false when a read schema declares a string over a binary column. Malformed bytes used to become NULL and now become U+FFFD. That looks like the right Spark behavior. Could you say so in the PR description, add a test for it, and consider whether this part should be backported?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: support binaryAsString in the native Parquet scan [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#issuecomment-6253072242
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?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: support binaryAsString in the native Parquet scan [datafusion-comet]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#issuecomment-6208133803
Hi @kazuyukitanimura, thanks for the reminder. It's ready now.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: support binaryAsString in the native Parquet scan [datafusion-comet]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
rich7420 commented on code in PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#discussion_r4349097021
##########
docs/source/user-guide/latest/parquet.md:
##########
@@ -264,11 +270,14 @@ for production, so do not turn it off to avoid an entry on this page.
and 4.0, and to Spark 4.1 and later with
`spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing=true`
([#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
+ without the UTF8 annotation that is read as `string`, through the read schema or
`spark.sql.parquet.binaryAsString=true`. Spark returns the bytes. Comet returns NULL for each
- such nested value. Top-level values match Spark in a release build, because the invalid bytes
- 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
Review Comment:
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?
##########
docs/source/contributor-guide/spark_configs_support.md:
##########
@@ -50,6 +50,11 @@ The status column uses these values:
- Affected expressions: `date_format`, `from_unixtime`, `unix_timestamp`, `to_unix_timestamp`, `to_timestamp`, `to_timestamp_ntz`, `to_date`, `try_to_date` (Spark 4.1+), `try_to_timestamp` (Spark 3.4+)
- Spark versions checked: 3.4.3, 3.5.8, 4.0.1, 4.1.2
- Date: 2026-07-18
+- `spark.sql.parquet.binaryAsString`
+ - Default: `false`
+ - Status: Native (see notes)
Review Comment:
nit: the legend doesn't define `Native`. Would `Partial` fit here, given the documented `U+FFFD` differences?
```suggestion
- Status: Partial (see notes)
```
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -1386,6 +1409,80 @@ impl SparkPhysicalExprAdapter {
.data()
}
+ /// A `binaryAsString` scan reads string columns as Binary, so for a file that annotates one
+ /// as a string, the default adapter compares `CAST(s AS BINARY)` in the pushed filters, and
+ /// 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(
Review Comment:
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.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: support binaryAsString in the native Parquet scan [datafusion-comet]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on code in PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#discussion_r4338615754
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -218,6 +218,55 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
+ test("spark.sql.parquet.binaryAsString uses native scan") {
Review Comment:
Done.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: support binaryAsString in the native Parquet scan [datafusion-comet]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on code in PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#discussion_r4315293290
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -330,6 +330,24 @@ fn parquet_convert_array_impl(
Ok(Arc::new(string_array))
}
+ // Parquet's unannotated BINARY is read as Spark StringType when
+ // spark.sql.parquet.binaryAsString is enabled or a read schema declares a string.
+ // Arrow's cast below is safe, so it would turn malformed UTF-8 into NULL. Use Spark's
+ // cast so the scan's top-level projection and nested values replace malformed input
+ // with U+FFFD, as the JVM's `new String(bytes, UTF_8)` does.
+ (Binary, Utf8) | (LargeBinary, LargeUtf8) => {
Review Comment:
I've split this out to #7106 while holding off the rest for [apache/datafusion#25802](https://github.com/apache/datafusion/pull/25802), which adds an option to turn off the coercion from a file's BINARY to the table's Utf8. Rebuilt on that support,
- The table schema stays Utf8 and filters are pushed unchanged. Annotated files prune as they do today, with no double cast.
- Only an unannotated column is decoded, per file, by the existing adapter's Spark cast. Pruning is lost only for that column in that file. Spark doesn't push `=` or `IN` there either; `IS NULL` is the one case to follow up on.
- There's no skipped filter pushdown, so no need to document.
cc @andygrove
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: support binaryAsString in the native Parquet scan [datafusion-comet]
Posted by "kazuyukitanimura (via GitHub)" <gi...@apache.org>.
kazuyukitanimura commented on PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#issuecomment-6199907692
Hi @manuzhang is this ready for review?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: support binaryAsString in the native Parquet scan [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#issuecomment-6253003412
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? That entry also describes the old behavior (NULL for invalid UTF-8), so it may need updating.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: support binaryAsString in the native Parquet scan [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#discussion_r4354854700
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -197,10 +384,41 @@ pub(crate) fn init_datasource_exec(
.flat_map(|s| s.fields().iter())
.map(|f| Arc::new(Field::new(f.name(), f.data_type().clone(), f.is_nullable())))
.collect();
- let table_schema = TableSchema::builder(base_schema)
- .with_table_partition_cols(partition_fields)
+ let output_table_schema = TableSchema::builder(Arc::clone(&base_schema))
+ .with_table_partition_cols(partition_fields.clone())
+ .build();
+ let read_base_schema = if binary_as_string {
+ binary_as_string_read_schema(&base_schema)
+ } else {
+ Arc::clone(&base_schema)
+ };
+ let table_schema = TableSchema::builder(read_base_schema)
+ .with_table_partition_cols(partition_fields.clone())
.build();
+ // 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 {
Review Comment:
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.
##########
docs/source/user-guide/latest/parquet.md:
##########
@@ -264,11 +270,14 @@ for production, so do not turn it off to avoid an entry on this page.
and 4.0, and to Spark 4.1 and later with
`spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing=true`
([#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
Review Comment:
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.
##########
docs/source/contributor-guide/spark_configs_support.md:
##########
@@ -50,6 +50,11 @@ The status column uses these values:
- Affected expressions: `date_format`, `from_unixtime`, `unix_timestamp`, `to_unix_timestamp`, `to_timestamp`, `to_timestamp_ntz`, `to_date`, `try_to_date` (Spark 4.1+), `try_to_timestamp` (Spark 3.4+)
- Spark versions checked: 3.4.3, 3.5.8, 4.0.1, 4.1.2
- Date: 2026-07-18
+- `spark.sql.parquet.binaryAsString`
+ - Default: `false`
+ - Status: Partial (see notes)
+ - Affected operator: native Parquet scan
Review Comment:
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?
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -63,6 +68,187 @@ pub(crate) struct CometParquetScanOptions {
pub prefetch_next_file_metadata: bool,
}
+/// Keep top-level Spark string fields as binary while arrow-rs decodes Parquet. arrow-rs only
+/// validates UTF-8 for columns annotated as strings, so decoding an unannotated BINARY column that
+/// `spark.sql.parquet.binaryAsString=true` types as a string straight into Arrow Utf8 would admit
+/// invalid UTF-8. A projection added at the scan boundary converts the binary arrays to valid Utf8
+/// with Spark-compatible replacement semantics.
+fn binary_as_string_read_schema(schema: &SchemaRef) -> SchemaRef {
+ Arc::new(Schema::new_with_metadata(
+ schema
+ .fields()
+ .iter()
+ .map(binary_as_string_read_field)
+ .collect::<Vec<_>>(),
+ schema.metadata().clone(),
+ ))
+}
+
+/// The field `binary_as_string_read_schema` reads a top-level `field` as. Spark's StringType is
+/// 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))
+ } else {
+ Arc::clone(field)
+ }
+}
+
+/// Whether `binary_as_string_read_schema` reads a top-level field of `data_type` as Binary.
+fn is_binary_as_string_read(data_type: &DataType) -> bool {
+ data_type == &DataType::Utf8
+}
+
+/// Whether `expr` references a column whose folded name is in `folded_names`.
+fn references_any_column(
+ expr: &Arc<dyn PhysicalExpr>,
+ folded_names: &HashSet<String>,
+ case_sensitive: bool,
+) -> Result<bool, ExecutionError> {
+ for column in collect_columns(expr) {
+ if folded_names.contains(&fold_name(column.name(), case_sensitive)?) {
+ return Ok(true);
+ }
+ }
+ Ok(false)
+}
+
+/// Rewrites a data filter for a scan that reads string columns as Binary (see
+/// `binary_as_string_read_schema`), or returns `None` when it can't go into the scan.
+///
+/// The scan decodes each such column with replacement: a valid value decodes to itself, and an
+/// invalid one to a string containing U+FFFD. So for a literal without U+FFFD, `s = 'lit'` on the
+/// decoded string holds exactly when `s = X'<bytes of lit>'` holds on the raw bytes, and the same
+/// goes for `IN`. `IS [NOT] NULL` holds on both, because decoding keeps nulls. Other filters on
+/// these columns stay above the scan. A range can disagree because decoding can reorder values:
+/// `"a\x80"` sorts below `"aé"` as bytes, but `"a\u{FFFD}"` sorts above it.
+///
+/// Spark's Filter above the scan re-evaluates every data filter, so a pushed filter only has to
+/// keep every row the original keeps. That's why an `AND` keeps whichever side can be pushed.
+fn binary_as_string_pushdown_filter(
+ filter: &Arc<dyn PhysicalExpr>,
+ binary_read_columns: &HashSet<String>,
+ case_sensitive: bool,
+ filter_read_schema: &Schema,
+) -> Result<Option<Arc<dyn PhysicalExpr>>, ExecutionError> {
+ if !references_any_column(filter, binary_read_columns, case_sensitive)? {
+ return Ok(Some(Arc::clone(filter)));
+ }
+ let reads_binary = |expr: &Arc<dyn PhysicalExpr>| -> Result<bool, ExecutionError> {
+ match expr.downcast_ref::<Column>() {
+ Some(column) => {
+ Ok(binary_read_columns.contains(&fold_name(column.name(), case_sensitive)?))
+ }
+ None => Ok(false),
+ }
+ };
+ let pushdown = |expr: &Arc<dyn PhysicalExpr>| {
+ binary_as_string_pushdown_filter(
+ expr,
+ binary_read_columns,
+ case_sensitive,
+ filter_read_schema,
+ )
+ };
+
+ if let Some(binary) = filter.downcast_ref::<BinaryExpr>() {
+ let (left, right) = (binary.left(), binary.right());
+ match binary.op() {
+ Operator::And => {
+ return Ok(match (pushdown(left)?, pushdown(right)?) {
+ (Some(left), Some(right)) => {
+ Some(Arc::new(BinaryExpr::new(left, Operator::And, right)))
+ }
+ (side, None) | (None, side) => side,
+ });
+ }
+ Operator::Or => {
+ return Ok(match (pushdown(left)?, pushdown(right)?) {
+ (Some(left), Some(right)) => {
+ Some(Arc::new(BinaryExpr::new(left, Operator::Or, right)))
+ }
+ _ => None,
+ });
+ }
+ Operator::Eq => {
+ if reads_binary(left)? {
+ if let Some(literal) = binary_literal(right) {
+ let eq = BinaryExpr::new(Arc::clone(left), Operator::Eq, literal);
+ return Ok(Some(Arc::new(eq)));
+ }
+ }
+ if reads_binary(right)? {
+ if let Some(literal) = binary_literal(left) {
+ let eq = BinaryExpr::new(literal, Operator::Eq, Arc::clone(right));
+ return Ok(Some(Arc::new(eq)));
+ }
+ }
+ }
+ _ => {}
+ }
+ } else if let Some(list) = filter.downcast_ref::<InListExpr>() {
+ if !list.negated() && reads_binary(list.expr())? {
+ if let Some(values) = list.list().iter().map(binary_literal).collect() {
+ let expr = Arc::clone(list.expr());
+ return Ok(Some(in_list(expr, values, &false, filter_read_schema)?));
+ }
+ }
+ } else if let Some(is_null) = filter.downcast_ref::<IsNullExpr>() {
+ if reads_binary(is_null.arg())? {
+ return Ok(Some(Arc::clone(filter)));
+ }
+ } else if let Some(is_not_null) = filter.downcast_ref::<IsNotNullExpr>() {
+ if reads_binary(is_not_null.arg())? {
+ return Ok(Some(Arc::clone(filter)));
+ }
+ }
+ Ok(None)
+}
+
+/// `expr` as a Binary literal holding the same bytes, if it is a string literal without U+FFFD.
+fn binary_literal(expr: &Arc<dyn PhysicalExpr>) -> Option<Arc<dyn PhysicalExpr>> {
+ match expr.downcast_ref::<Literal>()?.value() {
+ ScalarValue::Utf8(Some(value)) if !value.contains(char::REPLACEMENT_CHARACTER) => {
+ let bytes = value.as_bytes().to_vec();
+ Some(Arc::new(Literal::new(ScalarValue::Binary(Some(bytes)))))
+ }
+ _ => None,
+ }
+}
+
+fn binary_as_string_projection(
Review Comment:
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 reac
hes 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.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org