[manuzhang]

Which issue does this PR close?

Part of #7032 (the nested values) and #4764. Split out of #6246.

Rationale for this change

When a Parquet BINARY column without the UTF8 annotation is read as a string, through a read schema or spark.sql.parquet.binaryAsString=true, the native scan converts nested values (struct fields, list elements, map keys and values) in parquet_convert_array. That fell through to Arrow's safe cast, which turns invalid UTF-8 into NULL, so those values silently became NULL where Spark keeps the bytes.

This decodes them the way CAST(binary AS string) and native shuffle already do (#4764), replacing each ill-formed sequence with U+FFFD. Collected values now match Spark. Byte-level results such as hex still differ, as described under "Strings with non-UTF-8 bytes" in the compatibility guide.

It is split out of #6246, which also changes the top-level path and is waiting on a decision about filter pushdown. This part doesn't affect pushdown, doesn't depend on spark.sql.parquet.binaryAsString, and applies to branch-1.1 as well.

What changes are included in this PR?

• parquet_convert_array_impl converts Binary to Utf8 with Spark's cast instead of Arrow's safe cast. The reader keeps the types a file's ARROW:schema stores, so an Arrow writer's LargeBinary or BinaryView values arrive as those types. They're converted to Binary first, which for LargeBinary keeps the value buffer. A dictionary of Binary or LargeBinary values decodes each dictionary value once, then expands the keys.
• cast_binary_to_string checks the bytes the rows span once for the default and UTF8 styles. Valid input reuses the input's value buffer, narrowed to those bytes for a slice, which keeps the zero-copy conversion Arrow's cast gave valid nested values. When those bytes fill less than half of the buffer's allocation, as in a batch sliced from a large aggregate output, they are copied into a compact buffer instead, so the batch doesn't keep the whole allocation alive. The offsets are shared on the same terms when they start at zero, and rebuilt otherwise. The validity bitmap follows the same rule: a slice whose bits fill less than half of the bitmap's allocation gets a compact copy of them. For invalid input, rows in the valid prefix are copied as is and only the rest are decoded. When null rows span more than half of those bytes, for example after nullif, the check is skipped and the live rows are decoded one at a time as before, so their payload isn't scanned or kept alive. The decoded output is sized by the live rows' bytes rather than the whole span. CAST(binary AS string) uses the same function.
• The new behavior also applies with spark.sql.parquet.binaryAsString=false when a read schema declares a string over a BINARY column: a malformed nested value used to become NULL and now becomes U+FFFD.
• Docs: the Native Parquet scan returns NULL for nested BINARY values that are not valid UTF-8 when they are read as STRING #7032 entry under the Parquet known issues, and "Strings with non-UTF-8 bytes" in the compatibility guide, which now lists Parquet BINARY columns read as strings among the ways invalid UTF-8 reaches Comet.

Top-level columns are unchanged. DataFusion's file-schema coercion decodes them straight into Utf8 without validation, so they never reach this conversion. That is the rest of #7032.

How are these changes tested?

• Rust unit tests:
  • parquet_convert_array converts a List<Binary> holding 0xFF to List<Utf8> with U+FFFD.
  • A scan of a file whose ARROW:schema stores list elements as LargeBinary, BinaryView, or a dictionary of Binary or LargeBinary reads 0xFF as U+FFFD. Each came back NULL before.
  • cast_binary_to_string reuses the value buffer for a valid array and for a valid slice of an array with invalid bytes elsewhere. Invalid bytes under a null or inside a slice, and a code point split across two rows, decode the same way as before. A slice whose null rows hide most of its bytes gets a compact copy of the live rows, while the same slice with a quarter of its rows null still reuses the buffer. A valid slice filling an eighth of its parent's allocation gets a compact copy, and one filling more than half reuses it. A large array's first slice gets fresh offsets too, while an unsliced array with an oversized value buffer keeps its offsets. A small slice gets a compact copy of its validity bitmap, whether or not it starts on a byte boundary, while a slice filling most of the bitmap shares it.
• A new ParquetReadSuite test writes a raw Parquet file with unannotated BINARY in a list, a struct and a map, holding 41 FF 42, beside a row with a null list element, struct field and map value and a row whose list, struct and map are null. It reads the file with binaryAsString=true and, separately, with a read schema, and checks:
  • the inferred schema,
  • Spark answer parity with a CometNativeScanExec, which fails without this change because the nested values come back NULL,
  • that hex of each nested value is 41EFBFBD42, which shows the value was decoded rather than passed through unchecked.
• The hex check also guards the DataFusion 56 upgrade. fix: make Parquet file schema type coercion work on nested schemas datafusion#25193 extends DataFusion's file-schema coercion into struct and list children, which would send these values around this conversion and into unchecked Utf8 arrays.

🤖 Generated with Claude Code

[sunchao]

Reviewed the full four-file diff against 4a8a9496d119341b6f8ea94959312340c353583c at head 0e8d1dcd0b036ca94ba3423e9e89a011d06a72ee. The PR is not a draft. The snapshot and live discussion contained no existing review concerns. Found one introduced P2 performance regression.

Routed skills: review-comet-pr and review-comet-expression-pr, including the reused binary-cast semantics.

Summary

• Prior state and problem: Nested unannotated Parquet BINARY values requested as strings passed through Arrow’s safe cast, which replaced malformed values with NULL. The relevant conversion behaves identically on the latest release branch, branch-1.1 at 89c729a9697467e8f8d89e3eadf1797a75e2328c.
• Design approach: Route nested binary-to-string conversion through the existing spark_cast implementation and its JVM-compatible replacement decoder.
• Correctness: Checked Spark’s Parquet readers, binary casts and UTF8String implementation. Spark preserves the bytes internally and uses JVM replacement decoding when rendering strings. The exercised Binary → Utf8 path matches that rendered result and preserves nulls. No introduced correctness issue meeting the P1/P2 bar was identified.
• Compatibility analysis: Checked sources for Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. The relevant byte-preservation/rendering distinction holds across these versions. Replacement remains different from Spark for byte-sensitive operations, as the updated documentation explains.
• Key design decisions: Reuse the existing decoder, retain the recursive struct/list/map conversion, and leave top-level file-schema coercion unchanged. Both inferred strings and explicitly requested string schemas receive the intended fix.
• Implementation sketch: Adds the binary conversion match arm, a Rust list regression, and a Scala fixture covering struct fields, list elements, map keys and map values. The Scala test checks native execution, Spark answer parity and replacement bytes through hex.
• Performance: The new route also rebuilds entirely valid arrays, losing Arrow’s validated zero-copy conversion. A release-mode kernel probe measured 7.07×, 4.46× and 1.79× conversion overhead for 8-, 32- and 1,024-byte values respectively. This is the P2 finding below. These are kernel measurements, not end-to-end scan timings.
• Design: The localized fix is straightforward and avoids changing unrelated scan behavior. Keeping the existing valid-data fast path would preserve that simplicity while avoiding the regression.
• Abstraction & complexity: Reusing spark_cast avoids another UTF-8 decoder. The concern is its unconditional array rebuilding on this newly routed path, rather than a need for additional abstraction.
• Behavioral changes worth calling out: Malformed nested values now contain U+FFFD instead of becoming NULL, including when binaryAsString=false and a read schema requests strings. This intended change also applies relative to branch-1.1. The documented top-level limitation remains unchanged.
• Suggested improvements: Preserve validated buffer reuse for valid UTF-8 and invoke replacement decoding only when validation fails. Cover this distinction with a valid-data conversion benchmark when addressing the finding.

Exact-head CI: All executed checks succeeded, including native build/Rust tests, Spark 4.1 scan/expression/execution/shuffle suites, TPC-H/TPC-DS verification and profile compilation checks. Logs from run 39201318203 confirm both new regression tests passed. Spark SQL suites, Iceberg, Delta and macOS checks were skipped.

Validation limits: The local full native test build failed because jni.h is unavailable. A disposable release-mode harness using Arrow 59.3.0 and the exact head’s binary-cast helper and decoder passed valid, empty, null, malformed, surrogate and sliced-input checks and measured the regression. No local JVM suite or end-to-end scan benchmark was run. Project source was unchanged and nothing was published.

[sunchao]

Reviewed commit 491d88bfe1adff80905795ab3abd289563933aed. New P1/P2 findings are detailed inline.

[sunchao]

Reviewed commit 79b864eb4c6b81005f3bee0ed1f3ef0ba71a3a0d. New P1/P2 findings are detailed inline.

[andygrove]

Thanks for splitting this out of #6246. The nested NULLs were a real correctness gap, and reusing the input buffer when the bytes are valid is a nice improvement to CAST(binary AS string) as well. I walked through the fast path and the fallback and they look right to me, including the slice rebasing and the case where an offset splits a code point.

I'd like to see sunchao's last point about bytes under null rows addressed before this merges. Validating data[start..end] scans the payload of every null row, which the row-by-row decoder skipped, and reusing the buffer then keeps that payload alive downstream too. Could the fast path apply only when the live rows cover most of the span? When input.null_count() > 0, one pass over the offsets and the null buffer gives the live byte count, and below some fraction of end - start the existing row path is cheaper and returns a compact buffer. A unit test with a mostly-null slice over a large buffer would pin that choice down.

[sunchao]

Reviewed commit 4a62c5dd9d7151225047de3bdd9bdcf85aea8dea. New P1/P2 findings are detailed inline.

[comphead]

Compacting a slice that fills little of its buffer and scanning only the slice's own bytes are both in the head, and the thread's earlier points are settled. Three comments.

[andygrove]

Thanks @manuzhang. Everything from the last round is in at d2974af65. A mostly-null slice now decodes only its live rows, a slice that fills little of its buffer gets a compact copy, and the dictionary, LargeBinary and BinaryView leaves all reach the decoder, with tests for each. Since 1.1 returns NULL for these nested values where Spark returns them, I think this should go to branch-1.1 too, and I'll add the backport-1.1 label.

[sunchao]

Reviewed commit d2974af65803bc2d975185f49f27ca46f28a09ed. New P1/P2 findings are detailed inline.

[sunchao]

Reviewed commit ab2d1d2c44b485ddc8fe3bc87186604e96be6c03. New P1/P2 findings are detailed inline.

[sunchao]

[P2] take flattens the decoded dictionary into one Utf8 array with signed 32-bit offsets. Repeated references can fit in the input batch and exceed that limit after replacement. This aborts even SUM(size(arr)), which only needs list lengths. Split the enclosing batch at parent-row boundaries using decoded widths, or retain the dictionary until it can be materialized in bounded batches.

Evidence: A 264,786-byte Parquet file held 8,192 singleton lists referencing one 88,000-byte dictionary value of FF bytes. Each decoded value was 264,000 bytes. At the default batch size of 8,192, Spark 4.1.3 and the base returned 8192 for SUM(size(arr)). The head’s native scan raised Offset overflow error: 2147640000. An Arrow probe independently confirmed the failure during offset calculation.

[manuzhang]

I don't think it's feasible in this PR so a clearer error message has been added with suggestion of a smaller spark.comet.batchSize.

[manuzhang]

I opened #7254 to track.

[andygrove]

Thanks @manuzhang. The nested decode matches Spark, and the earlier rounds on slices, hidden null payloads and the dictionary, LargeBinary and BinaryView leaves look settled. I approved at d2974af65, but I'm changing that to request changes. sunchao's last round on ab2d1d2c4 is still open, and two of those threads change answers or fail queries where Spark and the base succeed.

The first is map keys. FF and EF BF BD both decode to U+FFFD. Two struct keys that differ only in those bytes collapse into one key, so a lookup can return the other entry's value. Spark keeps the raw bytes, so the keys stay distinct there. Could the map-key case fall back to Spark, or keep the bytes through key evaluation?

The second is the dictionary path. take flattens the decoded dictionary into one Utf8 array with 32-bit offsets. In the repro in the thread, 8,192 singleton lists reference one 88,000-byte entry, and the scan fails with Offset overflow error: 2147640000. Spark and the base return 8192 for SUM(size(arr)). Could this decode in bounded batches, or split the batch at parent-row boundaries?

The remaining threads are about memory and CPU on existing paths. Please address or answer each one.

• A fallback buffer that grows past its initial capacity is handed to the sorter at twice the size it needs, which caused two spills where the base had none.
• A shared buffer imported over the Arrow C Stream pins the wide sibling columns of its batch. Peak CometArrowAllocator use rose from 0.85 MB to 37.8 MB in the repro.
• The row-group dictionary is decoded again for every output batch, even when a batch references only a short valid value.
• valid_indices() visits nearly every row when nulls are sparse, where summing offset differences over contiguous validity runs would not.
• A truncated wide multibyte row restarts decoding from byte zero instead of reusing the validated prefix.

The outdated thread on compacting the validity bitmap looks fixed by ab2d1d2c4. Could you confirm and resolve it?

This is labeled backport-1.1, so whatever lands here will also ship on that branch. That is another reason to settle these first.

[andygrove]

Thanks @manuzhang. The map-key fix, the per-batch dictionary decode, the sparse-null count, the prefix reuse and the compact fallback buffer all look right to me at bfbf75894, and the extended fixture now covers nulls through the list, struct and map plumbing. I checked out the head and the 13 cast_binary_to_string tests in datafusion-comet-spark-expr pass. Two of sunchao's threads are still open, and I'd like both settled before this merges.

The first is the dictionary overflow. A batch whose decoded dictionary values pass i32::MAX bytes still fails in take, where Spark and the release succeed. I understand a bounded split isn't simple here, and the clearer error is a help. The window is narrow, since valid data hits the same limit in Arrow's cast and only malformed bytes add the 3x growth. Could you add the limit to the known-issues entry in parquet.md, naming the batch size and the growth for malformed bytes? Could you also file a tracking issue and link it from the code comment? Without both, a query that runs fine on the release fails with no record of why.

The second is the imported buffer ownership thread on cast_binary_to_string, which was resolved without a reply, and the code hasn't changed. fills() compares a buffer against its own capacity. With an Arrow C Stream input, a child buffer still keeps the whole root batch alive, including wide sibling columns. sunchao measured peak CometArrowAllocator use going from 0.85 MB to 37.8 MB in the repro. I haven't reproduced that myself. The old nested scan path already shared buffers through Arrow's cast, but plain CAST(binary AS string) always copied before, so this is a new retention. Do you have a view on it? Would it work to share only when the input isn't an imported array, or to copy in that case?

One test ask. nested_binary_to_string_keeps_map_key_bytes covers Binary and struct keys. Could you add a Dictionary(Int32, Binary) key, so the Dictionary arm with in_map_key set is covered, including take over the unchecked values?

This is labeled backport-1.1. Since branch-1.2 has the same gap, should it go there too?

CI is still running on this head, so the extended ParquetReadSuite test hasn't run on the new commit yet.

[andygrove]

Thanks @manuzhang. Sharing the input's buffers is now the caller's decision, so a plain CAST(binary AS string) copies the validated bytes once and a scan shares only buffers the reader owns. That settles the imported-buffer question. The dictionary limit is in the known issues with the batch size and the growth for ill-formed bytes, and the code comment links #7254. The Dictionary(Int32, Binary) map key is covered too.

Everything I raised is in at b679144763, and CI is green on this head, including the extended ParquetReadSuite test. I'm switching my review to approve.

[manuzhang]

Thanks everyone for review!