fix: support binaryAsString in the native Parquet scan

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
+  - Date: 2026-10-21
 
 ## Audit Notes
 
@@ -151,3 +156,55 @@ Comet bugs were uncovered by the original audit. The `try_to_timestamp` policy t
 use the default `query` mode to enforce both result correctness and execution within
 the Comet pipeline. Tests for expressions with known fallback paths continue to use
 `query spark_answer_only`.
+
+### `spark.sql.parquet.binaryAsString`
+
+**Source.** When enabled, Spark infers unannotated Parquet `BINARY` columns as
+`StringType`. Parquet files written by Spark carry Spark schema metadata, so this
+configuration does not change the types inferred for those files.
+
+**Comet status.** Comet honors the configuration in the native Parquet scan. When it is
+enabled, the scan reads every top-level string column as Arrow `Binary` and decodes it to valid
+UTF-8 at the scan boundary using Spark-compatible replacement semantics. Spark's schema doesn't
+say which file columns carry the UTF8 annotation, so annotated columns take the same path.
+arrow-rs validates UTF-8 only for Parquet columns annotated as strings, so decoding unannotated
+columns straight into Arrow `Utf8` would put invalid UTF-8 into a string array, which is undefined
+behavior. The default `false` value continues to return unannotated `BINARY` columns as
+`BinaryType`.
+
+**Filter pushdown.** While the configuration is enabled, the scan rewrites `=`, `IN` and
+`IS [NOT] NULL` on a top-level string column to compare raw bytes, 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 rewritten filter keeps exactly the rows the
+original keeps. For a file that annotates the column as a string, the expression adapter compares
+the column with the string literal directly, so the file's statistics still prune. Other filters
+on these columns stay above the scan, because decoding can reorder values: `"a\x80"` sorts below
+`"aé"` as bytes, but `"a\uFFFD"` sorts above it. That covers ranges, `LIKE`, function calls and
+literals containing `U+FFFD`. Spark's `Filter` above the scan re-evaluates every data filter, so
+results don't depend on what is pushed. Join and TopK dynamic filters accept only integer keys,
+so they never reach these columns.
+
+Compared with Spark, Comet 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 Comet keeps above the scan while the configuration is on.
+
+**Test coverage.** `ParquetReadV1Suite` writes raw Parquet files without Spark schema metadata
+and verifies the inferred schema under both configuration values, Spark answer parity and
+`CometNativeScanExec`, including a row containing invalid UTF-8 bytes and a projection that reads
+only a non-string column. A second test reads two partition directories holding a valid value,
+`0xff` and a null, and a Spark-written file, with row-filter pushdown off and on. It compares `=`,
+`IN`, `NOT IN`, `IS [NOT] NULL`, a range, an `OR`, a self-join on the string column and an
+`ORDER BY ... LIMIT` against Spark, and checks that `value = 'one'` reaches the row filter. Native
+tests in `parquet_exec.rs` check which row groups each filter prunes, for an unannotated and an
+annotated column, with and without row filters. They include `NOT IN`, a range and `U+FFFD`
+literals, none of which may prune. A native test in `schema_adapter.rs` checks that the adapter
+compares an annotated column directly for `NOT IN` as for `IN`, and leaves a filter with a null or
+invalid UTF-8 literal unchanged.
+
+**Known divergences.** Decoding is not byte-preserving, so the differences described in
+[Strings with non-UTF-8 bytes](../user-guide/latest/compatibility/index.md) apply to any
+malformed value this configuration surfaces: byte-level round trips, `octet_length`,
+hashing, and value identity, where two distinct malformed sequences both become `U+FFFD`.
+This is the same trade-off taken by `CAST(binary AS string)` (#4763) and by the FFI import
+boundary (#5310).

docs/source/user-guide/latest/compatibility/index.md

@@ -98,7 +98,9 @@ underlying bytes rather than on the rendered text:
   such as `contains` can therefore disagree with Spark when non-UTF-8 bytes are involved.
 
 Both differences require string data that is not valid UTF-8, which does not occur for text read from
-Parquet or produced by string expressions. Consistent handling of invalid UTF-8 across all native
+UTF8-annotated Parquet columns or produced by string expressions. A Parquet BINARY column without
+that annotation can hold it when it is read as a string, through the read schema or
+`spark.sql.parquet.binaryAsString=true`. Consistent handling of invalid UTF-8 across all native
 string paths is tracked by
 [#4764](https://github.com/apache/datafusion-comet/issues/4764).
 

docs/source/user-guide/latest/parquet.md

@@ -182,6 +182,12 @@ page-index, and bloom-filter pruning happen regardless of this flag whenever Spa
 `spark.sql.parquet.filterPushdown` is on, except for filters that compare `FLOAT` or `DOUBLE` values,
 which do not prune while this flag is on (see [Floating-point Number Comparison](compatibility/floating-point.md)).
 
+With `spark.sql.parquet.binaryAsString=true`, the native scan reads every top-level string column
+as binary and decodes it after reading. Filters on those columns prune only for `=`, `IN` and
+`IS [NOT] NULL` against a literal without `U+FFFD`, which the scan compares as raw bytes. Other
+filters on them, such as ranges and `LIKE`, don't prune while the setting is on, even for columns
+the file annotates as strings.
+
 ### Row Group Assignment to Splits
 
 Spark and DataFusion's native Parquet scans use different rules to decide which row groups belong to a
@@ -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
+  read schema, top-level values reach an Arrow string array unchecked: they match Spark in a
+  release build, and a debug build panics on them
   ([#7032](https://github.com/apache/datafusion-comet/issues/7032), part of
   [#4764](https://github.com/apache/datafusion-comet/issues/4764)).
 - **Field-id reads where one file column matches a field by id and a sibling by name.** With

native/core/src/execution/delta_dv.rs

@@ -1543,6 +1543,7 @@ mod tests {
                 false,
                 false,
                 false,
+                false,
                 &session,
                 false,
                 false,

native/core/src/execution/operators/dynamic_filter/join/tests.rs

@@ -666,6 +666,7 @@ fn parquet_probe(
         false,
         false,
         false,
+        false,
         session,
         false,
         false,
@@ -820,6 +821,7 @@ async fn reader_filter_crosses_null_check_conjunction_and_retains_residual() {
             false,
             false,
             false,
+            false,
             &session,
             false,
             false,

native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors.rs

@@ -82,6 +82,7 @@ fn scan(
         false,
         allow_type_promotion,
         false,
+        false,
         session,
         false,
         false,

native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors/partition_columns.rs

@@ -67,6 +67,7 @@ fn partitioned_scan(
         false,
         false,
         false,
+        false,
         session,
         false,
         false,

native/core/src/execution/operators/dynamic_filter/join/tests/timestamp_errors.rs

@@ -101,6 +101,7 @@ async fn assert_timestamp_overflow_preserved(nested: bool) {
             false,
             false,
             false,
+            false,
             &session,
             false,
             false,

native/core/src/execution/operators/dynamic_filter/topk/tests.rs

@@ -192,6 +192,7 @@ fn parquet_scan(
         false,
         allow_type_promotion,
         false,
+        false,
         session,
         false,
         false,

native/core/src/execution/operators/dynamic_filter/topk/tests/timestamp.rs

@@ -91,6 +91,7 @@ fn timestamp_input(
         false,
         false,
         false,
+        false,
         session,
         false,
         false,

native/core/src/execution/planner.rs

@@ -461,6 +461,7 @@ impl PhysicalPlanner {
             common.return_null_struct_if_all_fields_missing,
             common.allow_type_promotion,
             common.allow_timestamp_ltz_to_ntz,
+            common.binary_as_string,
             self.session_ctx(),
             common.encryption_enabled,
             common.use_field_id,