fix: support binaryAsString in the native Parquet scan

docs/source/contributor-guide/spark_configs_support.md

@@ -50,6 +50,12 @@ 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
+  - Spark versions checked: 3.4.3, 3.5.9, 4.0.4, 4.1.3, 4.2.0
+  - Date: 2026-10-21
 
 ## Audit Notes
 
@@ -151,3 +157,57 @@ 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. The inference in
+`ParquetToSparkSchemaConverter` and the `BINARY` cases of `ParquetFilters` are the same in every
+Spark version listed above.
+
+**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

@@ -106,7 +106,8 @@ a result:
 - **Parquet BINARY columns read as strings.** A BINARY column without the UTF8 annotation can hold
   bytes that are not valid UTF-8 when it is read as `string`, through the read schema or
   `spark.sql.parquet.binaryAsString=true`. Comet decodes such a value in a struct field, list element
-  or map value. Top-level values and map keys are not checked; see
+  or map value, and a top-level value when `spark.sql.parquet.binaryAsString=true`. A top-level
+  value read through the read schema, and a map key, are not checked; see
   [Parquet](../parquet.md#reads-that-differ-from-spark).
 - **Text files.** Spark's Text reader keeps the bytes of each line as they are, so a Text file in an
   encoding other than UTF-8, such as Latin-1, holds strings that are not valid UTF-8. Comet decodes

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

@@ -211,6 +211,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
@@ -291,11 +297,13 @@ for production, so do not turn it off to avoid an entry on this page.
 - **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 keeps the bytes. In a struct field, list element
-  or map value, Comet 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). Top-level
-  values, and map keys at any depth, reach an Arrow string array unchecked, which keeps keys that
-  differ only in such bytes distinct: they match Spark in a release build, and a debug build
-  panics on top-level values ([#7032](https://github.com/apache/datafusion-comet/issues/7032), part of
+  or map value, and at the top level with `spark.sql.parquet.binaryAsString=true`, Comet 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, as do map keys at any depth,
+  which keeps keys that differ only in such bytes distinct: they match Spark in a release build,
+  and a debug build panics on top-level values
+  ([#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
   `spark.sql.parquet.fieldId.read.enabled=true`, Comet returns NULL for the sibling at the top

native/core/src/execution/delta_dv.rs

@@ -1585,6 +1585,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

@@ -607,6 +607,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,

native/core/src/parquet/eager_page_index_reader_factory.rs

@@ -272,9 +272,9 @@ impl EagerPageIndexReaderFactory {
         }
     }
 
-    /// Retain raw LIST encoding and `TIMESTAMP(NANOS)` leaves for open-time validation. When row
-    /// filters are enabled, or the file has a `TIMESTAMP(NANOS)` leaf, enforce deferred failures
-    /// at the first data-page read, after format pruning.
+    /// Retain raw LIST encoding and `TIMESTAMP(NANOS)` leaves for open-time validation.
+    /// Row filters, binary-as-string checks and `TIMESTAMP(NANOS)` defer failures until
+    /// the first data-page read, after format pruning.
     pub(crate) fn with_conversion_check(
         mut self,
         required: SchemaRef,
@@ -778,6 +778,7 @@ impl AsyncFileReader for EagerPageIndexReader {
                 // `TIMESTAMP(NANOS)` rejection is deferred from here even without row filters.
                 let has_nanos = columns.iter().any(is_nanos_timestamp);
                 let needs_check = *row_filters
+                    || conversion_options.reject_fixed_binary_as_string
                     || has_nanos
                     || columns.iter().any(|column| {
                         column.self_type().get_basic_info().repetition()
@@ -800,7 +801,7 @@ impl AsyncFileReader for EagerPageIndexReader {
                     DataFusionError::External(source) => ParquetError::External(source),
                     other => ParquetError::External(Box::new(other)),
                 })?;
-                if *row_filters || has_nanos {
+                if *row_filters || conversion_options.reject_fixed_binary_as_string || has_nanos {
                     if let Some(rejection) = &rejection {
                         self.deferred_rejections.lock().unwrap().insert(
                             object_meta.location.clone(),