Nested dictionary-encoded BINARY read as string overflows Utf8 offsets when flattened per batch #7254

Describe the bug

A nested string column read from a dictionary-encoded Parquet BINARY leaf is flattened into one Utf8 array per batch, with 32-bit offsets. A batch whose keys reference wide dictionary entries many times can fit in memory as a dictionary and still expand past i32::MAX bytes when flattened, so the scan fails where Spark, which keeps the dictionary and materializes a value only when a row reads it, succeeds. This affects any query over the column, including one that never reads the strings, such as SUM(size(arr)).

The flattening predates #7106: Arrow's cast of Dictionary<_, Binary> to Utf8 also goes through take. #7106 decodes ill-formed UTF-8 with U+FFFD, which expands a malformed byte to three, so it lowers the byte count that triggers the overflow by up to 3x and replaces the opaque Offset overflow error with an error naming the decoded byte count and suggesting a lower spark.comet.batchSize. The underlying limit is the same for valid data: a valid 262,144-byte entry referenced 8,192 times overflows on main too.

Found by sunchao while reviewing #7106: #7106 (comment)

Steps to reproduce

1. Write a Parquet file with a list<binary> column, unannotated, where 8,192 singleton lists all reference one dictionary entry of 88,000 FF bytes (264,786 bytes on disk).
2. Read it with spark.sql.parquet.binaryAsString=true and the native scan at the default spark.comet.batchSize=8192.
3. Run SELECT SUM(size(arr)) FROM t.

On main the ill-formed entry decodes to NULL so the batch flattens to zero bytes and the query returns 8192. On #7106 each decoded value is 264,000 bytes, 8,192 of them is 2,162,688,000 bytes, and the scan reports that the batch does not fit a string column. A valid entry of the same decoded width fails on both.

Expected behavior

The query returns 8192 and the scan does not need a smaller batch size.

Additional context

Options discussed on the thread:

• Keep the leaf dictionary-encoded past the scan and materialize in bounded batches. Comet's native plan never carries a dictionary type: ScanExec unpacks every dictionary column because some DataFusion operators and expressions do not accept them (native/core/src/execution/operators/scan.rs), the Parquet scan unwraps dictionary columns to the required types, and the conversion expression declares Spark's required type, which the batch is checked against. Retaining a nested dictionary means teaching the planner, every operator up to the JVM boundary, and the JVM vector import about nested dictionaries.
• Split the batch at parent-row boundaries using decoded widths. The conversion runs inside the scan's projection, which returns one batch per input batch, so splitting needs a separate operator after the scan that may emit several batches per input. The planner cannot tell unannotated BINARY from UTF8 without reading footers, so the scan would declare such leaves as binary and the operator would validate every nested string, including annotated ones that pass through untouched today.

Either way the fix is independent of the UTF-8 decoding and applies to valid wide values as well.

Willingness to contribute

I would be willing to contribute a fix for this bug with guidance from the DataFusion community