andygrove opened a new issue, #2554:
URL: https://github.com/apache/datafusion-ballista/issues/2554
**Describe the bug**
When a Parquet table decodes to `Utf8View` columns, small outputs written by
`ShuffleWriterExec` keep the full data buffers of every batch their rows came
from. TPC-H Q10 at SF10 returns 20 rows, but its final stage writes a 25 MB
shuffle file, and the client can't fetch it at the default limits:
```
ArrowError(ExternalError(Status { code: OutOfRange, message: "Error, encoded
message length too large: found 25080937 bytes, the limit is: 16777216 bytes",
source: None }), Some(""))
```
The job itself succeeds (`Completed. Produced 1 partition containing 20
rows.`). Only the result fetch fails.
`SessionConfig::new_with_ballista()` sets
`datafusion.execution.parquet.schema_force_view_types = false` because of this
IPC problem (#1182, #1294). That setting only stops DataFusion from converting
strings to views. Files whose Parquet schema already stores `Utf8View` still
decode as views, and datafusion-cli writes files like that by default.
The sort-based shuffle writer compacts view columns before it writes them
(`compact_view_columns` in `sort_shuffle/partitioned_batch_iterator.rs`).
`ShuffleWriterExec` doesn't. It writes each partition's stream with
`utils::write_stream_to_disk` as is. In Q10 that writer runs stage 4
(`TopK(fetch=20)` per partition) and stage 5 (`SortPreservingMergeExec` with
`fetch=20`):
| Stage | Output with `tpchgen-cli` files | Output with `customer` rewritten
by datafusion-cli |
| ----- | ------------------------------: |
-------------------------------------------------: |
| 4 | 0.1 MB |
56.9 MB |
| 5 | ~0 MB |
25.1 MB |
The final file holds one batch of 20 rows, which takes 46 MB in memory:
| Column | Type | Buffers | Bytes |
| ----------- | ------------- | ------: | ---------: |
| `c_name` | `string_view` | 16 | 6,427,220 |
| `c_address` | `string_view` | 16 | 8,543,377 |
| `c_phone` | `string_view` | 16 | 5,356,070 |
| `c_comment` | `string_view` | 16 | 25,861,736 |
| others | | | 1,032 |
**To Reproduce**
1. Generate SF10 data:
```sh
tpchgen-cli parquet -s 10 --parts 16 --output-dir sf10
```
2. Make a copy where only `customer` is rewritten by datafusion-cli (I used
datafusion-cli 53.0.0):
```sh
mkdir sf10-view
for t in lineitem orders part partsupp supplier nation region; do
ln -s "$PWD/sf10/$t" "sf10-view/$t"
done
datafusion-cli \
-c "CREATE EXTERNAL TABLE customer STORED AS PARQUET LOCATION
'$PWD/sf10/customer/';" \
-c "COPY (SELECT * FROM customer) TO '$PWD/sf10-view/customer/' STORED
AS PARQUET;"
```
3. Start a scheduler and one executor with default settings.
4. Run Q10:
```sh
cargo run --release --bin tpch -- benchmark ballista --host localhost
--port 50050 \
--path /path/to/sf10-view --format parquet --partitions 16 --query 10
```
The same command with `--path /path/to/sf10` returns 20 rows. I reproduced
this on `main` at `60cd1a52f` (DataFusion 55.1.0).
**Expected behavior**
A 20-row result should be a few KB, and fetching it should work at the
default limits, as it does for the `tpchgen-cli` files.
**Additional context**
- Raising the limits to 128 MiB works around it. That means
`--grpc-server-max-decoding-message-size` and
`--grpc-server-max-encoding-message-size` on the scheduler and executor, plus
`ballista.client.grpc_max_message_size` on the client.
- Q10 is where I noticed it, but any small output picked out of larger view
batches should be affected, for example a `LIMIT` or a selective filter.
- Two possible fixes. `ShuffleWriterExec` could compact view columns as
well, perhaps with the threshold proposed in #2521. Or, while view types are
disabled (#1294), the scan could convert `Utf8View` back to `Utf8`.
- #2165 shows the same error on Q21 and Q22, and may have the same cause.
- I found this while benchmarking #2553 on a copy of TPC-H that
datafusion-cli had partitioned. Partitioning turned out not to matter.
--
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]
