xudong963 opened a new issue, #26785:
URL: https://github.com/apache/datafusion/issues/26785
### Is your feature request related to a problem or challenge?
`COPY ... STORED AS ARROW` (any write through `ArrowFileSink`) can produce
files many times larger than the data when the input has `Utf8View` /
`BinaryView` columns.
Kernels such as `take`, `interleave`, `filter` and `slice` copy only the
16-byte views of a view array and keep referencing the original data buffers.
After a sort or a merge, each output batch therefore references the data
buffers of many input batches. The Arrow IPC writer writes every data buffer of
an array in full, so each batch written carries most of the string data of the
whole input.
Reproducer with `datafusion-cli` 55.0.0:
```sql
-- 200k rows of 64-character strings (about 12.8 MB of string data), as
Utf8View
CREATE TABLE t AS
SELECT arrow_cast(concat(md5(value::varchar), md5((value + 1)::varchar)),
'Utf8View') AS s,
random() AS r
FROM generate_series(1, 200000);
COPY (SELECT s FROM t) TO 'out/unsorted.arrow' STORED AS ARROW;
COPY (SELECT s FROM t ORDER BY r) TO 'out/sorted.arrow' STORED AS ARROW;
COPY (SELECT arrow_cast(s, 'Utf8') AS s FROM t ORDER BY r) TO
'out/sorted_utf8.arrow' STORED AS ARROW;
COPY (SELECT s FROM t ORDER BY r) TO 'out/sorted.parquet' STORED AS PARQUET;
```
| Output | File size |
|---|---|
| `unsorted.arrow` (Utf8View) | 9.1 MB |
| `sorted.arrow` (Utf8View) | **193.5 MB** |
| `sorted_utf8.arrow` (Utf8) | 13.6 MB |
| `sorted.parquet` (Utf8View) | 7.0 MB |
`sorted.arrow` reads back correctly (200,000 rows), but it is about 14x
larger than the same rows as `Utf8`, and the sorted `Utf8View` write takes
0.3-0.5 s instead of 0.01-0.05 s.
The effect grows with the number of input batches. Two larger cases I
measured on `main`:
- TPC-H SF1 `lineitem`, `COPY (SELECT l_orderkey, l_partkey, l_comment,
l_shipinstruct FROM lineitem WHERE l_orderkey <= 300000 ORDER BY l_partkey) TO
'...' STORED AS ARROW`: 593 MB, 2.47 s
- a merge of 17 presorted files (10M rows) written to one Arrow file: 3.9
GB, 9.1 s
Reading Parquet produces view types by default
(`datafusion.execution.parquet.schema_force_view_types = true`), so ordinary
`COPY (SELECT ... ORDER BY ...) TO ... STORED AS ARROW` queries hit this.
The spill writer has the same problem and already compacts view arrays
before writing them (`gc_view_arrays`, see #23565).
### Describe the solution you'd like
Compact view arrays in `ArrowFileSink` before writing each batch, reusing
`datafusion_physical_plan::spill::gc_view_arrays` (made public). It only copies
an array when its data buffers hold more than its views reference (and more
than 10 KB), so batches without waste are written as before.
With a local prototype of this change:
| Case | Before | After |
|---|---|---|
| TPC-H SF1 query above | 593 MB, 2.47 s | 9.1 MB, 0.08 s |
| 17-file merge, 10M rows | 3.9 GB, 9.1 s | 246 MB, 0.76 s |
| Writes without view types | | unchanged |
### Describe alternatives you've considered
- Compacting in the arrow-rs IPC writer. apache/arrow-rs#10800 discussed
this; the conclusion was that kernels stay zero-copy and callers compact
explicitly at the IPC boundary.
- Compacting in the sort and merge operators. This would add copies for
consumers that never serialize the batches, and would not cover `filter` or
`slice`.
### Additional context
The Parquet, CSV and JSON writers are not affected, as they encode values
one by one.
### Willingness to contribute
I plan to contribute to this feature independently.
--
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]
