fix: decode invalid UTF-8 in nested Parquet BINARY values read as strings

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

@@ -79,7 +79,8 @@ codegen dispatcher covers only scalar expressions; see the
 Spark's `StringType` can hold arbitrary bytes, including sequences that are not valid UTF-8 (for
 example `CAST(X'FF' AS STRING)`). Arrow's string type requires valid UTF-8, so Comet cannot store
 the raw bytes natively. When Comet produces a string from arbitrary bytes (such as
-`CAST(binary AS string)` or a columnar shuffle), it decodes them the same way the JVM does
+`CAST(binary AS string)`, a columnar shuffle, or a nested Parquet BINARY value read as a string),
+it decodes them the same way the JVM does
 (`new String(bytes, UTF_8)`), replacing each ill-formed sequence with the Unicode replacement
 character `U+FFFD`. Spark itself applies the identical replacement whenever such a string is
 materialized (collected, printed, or passed to most string functions), so the rendered result
@@ -98,9 +99,15 @@ 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. Comet's native Parquet scan never
-produces such strings (it rejects them, see below), and neither do most string expressions. They can
-reach Comet in these ways, which are the ones to check before relying on a result:
-
+produces such strings from UTF8-annotated columns (it rejects them, see below), and neither do most
+string expressions. They can reach Comet in these ways, which are the ones to check before relying on
+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
+  [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
   them when it converts a Text scan to Arrow, which it does only when

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

@@ -280,12 +280,14 @@ for production, so do not turn it off to avoid an entry on this page.
   read back wrong inside lists written in a legacy layout, as Hive and
   `spark.sql.parquet.writeLegacyFormat=true` write them
   ([#7029](https://github.com/apache/datafusion-comet/issues/7029)).
-- **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
-  `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
-  ([#7032](https://github.com/apache/datafusion-comet/issues/7032), part of
+- **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
   [#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
@@ -354,6 +356,15 @@ With `spark.comet.write.parquet.enabled=true`:
   ([#7035](https://github.com/apache/datafusion-comet/issues/7035)).
 - **A `DATE` column read as `INTERVAL YEAR TO MONTH`.** Spark reads it, and Comet fails the read
   ([#5188](https://github.com/apache/datafusion-comet/issues/5188)).
+- **A nested dictionary-encoded BINARY column, read as STRING, whose batch decodes to more than
+  2 GiB.** In a struct field, list element, map key or map value, Comet expands the dictionary
+  into one string array per batch, whose offsets hold at most 2,147,483,647 bytes. A batch of
+  `spark.comet.batchSize` rows (8192 by default) fails the scan when the values its rows reference
+  add up to more, counting each reference of a shared entry, even for a query that never reads the
+  strings. Values that are not valid UTF-8 decode to up to three times their size, since each
+  ill-formed sequence becomes `U+FFFD`. Spark keeps the dictionary and reads it row by row. Lower
+  `spark.comet.batchSize` so that a batch holds fewer rows
+  ([#7254](https://github.com/apache/datafusion-comet/issues/7254)).
 
 ### Writes that fail where Spark succeeds
 

native/core/src/parquet/parquet_exec.rs

@@ -834,6 +834,91 @@ mod tests {
         assert!(err.contains("rebase"), "unexpected error: {err}");
     }
 
+    /// The reader keeps the nested types of a file's ARROW:schema, so an Arrow writer's
+    /// `LargeBinary`, `BinaryView` or dictionary-encoded list elements reach the conversion to
+    /// Spark's string type as those types, not as `Binary`. A malformed value must still decode
+    /// to U+FFFD.
+    #[tokio::test]
+    async fn nested_arrow_binary_types_read_as_string_decode_invalid_utf8() {
+        use arrow::array::{Array, AsArray, BinaryArray, ListArray};
+        use arrow::buffer::OffsetBuffer;
+
+        for binary_type in [
+            DataType::LargeBinary,
+            DataType::BinaryView,
+            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Binary)),
+            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::LargeBinary)),
+        ] {
+            let element = Arc::new(Field::new("element", binary_type.clone(), true));
+            let values = arrow::compute::cast(
+                &BinaryArray::from(vec![b"one".as_slice(), &[0xff]]),
+                &binary_type,
+            )
+            .unwrap();
+            let list = ListArray::new(element, OffsetBuffer::new(vec![0, 2].into()), values, None);
+            let file_schema = Arc::new(Schema::new(vec![Field::new(
+                "s",
+                list.data_type().clone(),
+                true,
+            )]));
+            let batch =
+                RecordBatch::try_new(Arc::clone(&file_schema), vec![Arc::new(list)]).unwrap();
+
+            let filename = get_temp_filename()
+                .as_path()
+                .as_os_str()
+                .to_str()
+                .unwrap()
+                .to_string();
+            let file = File::create(&filename).unwrap();
+            let mut writer = ArrowWriter::try_new(file, file_schema, None).unwrap();
+            writer.write(&batch).unwrap();
+            writer.close().unwrap();
+
+            let string_list = DataType::List(Arc::new(Field::new("element", DataType::Utf8, true)));
+            let required_schema = Arc::new(Schema::new(vec![Field::new("s", string_list, true)]));
+            let session_ctx = Arc::new(SessionContext::new());
+            let scan = init_datasource_exec(
+                required_schema,
+                None,
+                None,
+                ObjectStoreUrl::local_filesystem(),
+                ObjectStoreBackend::Local,
+                vec![vec![PartitionedFile::from_path(filename).unwrap()]],
+                None,
+                None,
+                None,
+                "UTC",
+                true,
+                false,
+                false,
+                false,
+                &session_ctx,
+                false,
+                false,
+                false,
+                true,
+                "CORRECTED",
+                "CORRECTED",
+            )
+            .unwrap();
+
+            let mut strings = Vec::new();
+            let mut stream = scan.execute(0, session_ctx.task_ctx()).unwrap();
+            while let Some(batch) = stream.next().await {
+                let batch = batch.unwrap();
+                let values = batch.column(0).as_list::<i32>().values();
+                let values = values.as_string::<i32>();
+                strings.extend(values.iter().map(|v| v.map(str::to_string)));
+            }
+            assert_eq!(
+                strings,
+                vec![Some("one".to_string()), Some("\u{fffd}".to_string())],
+                "{binary_type}"
+            );
+        }
+    }
+
     /// Writes a metadata-free parquet file with one INT64 `TIMESTAMP_MICROS` column (`ts`) and
     /// one INT96 column (`ts96`) through the low-level writer (arrow's writer cannot emit
     /// INT96), then scans it with the given session read modes. `int96_days` is the day count