SubhamSinghal opened a new issue, #26381:
URL: https://github.com/apache/datafusion/issues/26381
### Describe the bug
With `datafusion.optimizer.enable_window_topn = true`, `PartitionedTopKExec`
keeps the whole data buffers of
`Utf8View` / `BinaryView` input columns alive for the few rows it retains,
and charges them to the memory pool.
Each input batch's admitted rows are gathered into the operator's
`RecordBatchStore` with `take`. For view
types `take` copies the views but shares every one of the input batch's data
buffers, so retaining 10 rows of an
8192-row batch keeps that batch's whole string buffer resident. Store
compaction can't release it:
- it is triggered by a row ratio (`total_rows > 2 × live rows`), and the
store already holds about one row
per live row;
- when it does run, it rewrites with `interleave`, which shares the same
buffers again.
Parquet scans produce `Utf8View` by default (`schema_force_view_types =
true`), so string payloads read from
Parquet hit this without any extra configuration.
### To Reproduce
In `datafusion-cli`, create 4.1M rows in which each 8192-row batch
introduces 10 new partitions, with a
114-byte string payload:
COPY (
SELECT CAST((i / 8192) * 10 + (i % 10) AS BIGINT) AS pk,
CAST(i AS BIGINT) AS v,
concat(md5(CAST(i AS VARCHAR)), md5(CAST(i + 1 AS VARCHAR)),
md5(CAST(i + 2 AS VARCHAR)), substr(md5(CAST(i + 3 AS
VARCHAR)), 1, 18)) AS s
FROM generate_series(0, 4095999) AS t(i)
ORDER BY i
) TO '/tmp/ptk.parquet' STORED AS PARQUET;
CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION '/tmp/ptk.parquet';
SET datafusion.execution.target_partitions = 1;
SET datafusion.optimizer.enable_window_topn = true;
SELECT count(*), sum(length(s)), sum(pk), sum(v), sum(rn)
FROM (SELECT pk, v, s, row_number() OVER (PARTITION BY pk ORDER BY v) AS
rn FROM t)
WHERE rn <= 1;
The query retains 5,000 rows (~570 KB of payload). Smallest `--memory-limit`
it passes with:
| | memory limit |
|---|---|
| `enable_window_topn = true` | **920M** (fails at 919M: `Resources
exhausted ... PartitionedTopK[0]`) |
| `enable_window_topn = true`, payload written as `Utf8` | 3M |
| `enable_window_topn = false` (sort plan, spills) | 17M |
The memory needed grows with the input, not with the rows retained: 409K
rows of the same shape need 92M.
With more retained rows the over-charging gets worse. For `rn <= 10` over
100K partitions (4.1M rows,
`pk = i % 100000`), the reservation asks for 36.7 GB in a single allocation
while peak RSS is 729 MB, so the
query cannot run under any memory limit. `rank()` behaves the same way.
### Expected behavior
Retained memory proportional to the retained rows, as with `Utf8`. Enabling
the optimization shouldn't need
more memory than the sort plan it replaces.
### Additional context
_No response_
### Willingness to contribute
None
--
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]
