manuzhang opened a new issue, #7254: URL: https://github.com/apache/datafusion-comet/issues/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: https://github.com/apache/datafusion-comet/pull/7106#discussion_r4367567424 ### 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 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
