Order-preserving RepartitionExec fails with ResourcesExhausted on non-spillable merge reservation for Utf8View data

Offen
  • Stimmt etwas nicht?
#27,323 2 Kommentare 0 Reaktionen 1 zugewiesene Person Auf GitHub ansehen

Maintainer antworten meist innerhalb von 1 Tag

@Kristofferjoha arbeitet bereits daran.

Seit 11.11.2026.

  • #27360 von @Kristofferjoha — offen

Bewertung

Dieses Issue wurde noch nicht bewertet.

Beschreibung

bug
Describe the bug

An order-preserving hash RepartitionExec fails with ResourcesExhausted on RepartitionExec[Merge N] reservations, which report can spill: false. The query returns only 7000 rows. The merge reservations grow until they fill the pool, whatever the pool size.

Commit: 34eb6dc5bf057a35c471a4fbfd2eece2a7937afe (main).

To Reproduce

Run with datafusion-cli -m 200M -f repro.sql:

COPY (SELECT value % 1000 AS k, repeat('x', 100) || (value % 7)::varchar AS s, value AS v
      FROM generate_series(1, 2000000) ORDER BY k)
TO 'data.parquet' STORED AS PARQUET;

SET datafusion.execution.target_partitions = 8;
CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION 'data.parquet' WITH ORDER (k ASC);
SELECT count(*) FROM (SELECT k, s, count(*) FROM t GROUP BY k, s ORDER BY k);

The file is 5.6 MB. Error:

Resources exhausted: Additional allocation failed for RepartitionExec[Merge 6] with top memory consumers (across reservations) as:
  RepartitionExec[Merge 3]#44(can spill: false) consumed 25.5 MB, peak 25.5 MB,
  RepartitionExec[Merge 5]#43(can spill: false) consumed 25.3 MB, peak 25.3 MB,
  RepartitionExec[Merge 0]#35(can spill: false) consumed 25.2 MB, peak 25.2 MB.
Error: Failed to allocate additional 640.0 B for RepartitionExec[Merge 6] with 24.8 MB already allocated for this reservation - 6.0 B remain available for the total memory pool: greedy(used: 200.0 MB, pool_size: 200.0 MB)

With -m 50M each merge holds about 6 MB. With -m 20M each holds about 2.5 MB. The query passes at -m 400M.

Plan (relevant part):

AggregateExec: mode=FinalPartitioned, gby=[k, s]
  RepartitionExec: partitioning=Hash([k@0, s@1], 8), input_partitions=8, preserve_order=true, sort_exprs=k@0 ASC NULLS LAST
    AggregateExec: mode=Partial, gby=[k, s], group_clustering_mode=Partial([0])
      DataSourceExec: file_groups=8, format=parquet

EXPLAIN ANALYZE metrics without a limit:

operator output_rows output_batches output_bytes
DataSourceExec 2.00 M 245 46.1 MB
AggregateExec Partial 7.01 K 246 42.7 MB
RepartitionExec 7.01 K 8 330.1 MB

If s is Utf8 (arrow_cast(s, 'Utf8')), the partial aggregate reports 1.7 MB, the repartition reports 828 KB, and the query passes at -m 10M.

Expected behavior

The query succeeds with a small limit. The data that the merge must hold is a few KB per output partition.

Additional context

Three effects combine.

  1. The partial aggregate emits Utf8View batches of about 28 rows that report about 174 KB each. The view data buffers dominate that size.
  2. The hash partitioner uses take_arrays (https://github.com/apache/datafusion/blob/34eb6dc5bf057a35c471a4fbfd2eece2a7937afe/datafusion/physical-plan/src/repartition/mod.rs#L1376). For Utf8View columns, take copies only the views and shares the data buffers. Each of the 8 output batches reports the full buffer size. The sum over all merges is about 8 times the real memory.
  3. BatchBuilder keeps every input batch that has a row in indices until it builds an output batch (https://github.com/apache/datafusion/blob/34eb6dc5bf057a35c471a4fbfd2eece2a7937afe/datafusion/physical-plan/src/sorts/builder.rs#L89-L95, https://github.com/apache/datafusion/blob/34eb6dc5bf057a35c471a4fbfd2eece2a7937afe/datafusion/physical-plan/src/sorts/builder.rs#L163-L170). Each output partition has fewer than batch_size rows, so the merge never emits before the end and keeps all of its input batches. The reservation grows with the number of input batches. It is not bounded by N inputs x batch size.

The merge reservation is created at https://github.com/apache/datafusion/blob/34eb6dc5bf057a35c471a4fbfd2eece2a7937afe/datafusion/physical-plan/src/repartition/mod.rs#L1857-L1869 without with_can_spill(true). A refused try_grow in BatchBuilder::push_batch fails the query. The repartition channel spill path (https://github.com/apache/datafusion/blob/34eb6dc5bf057a35c471a4fbfd2eece2a7937afe/datafusion/physical-plan/src/repartition/mod.rs#L218-L230) does not help, because the memory is held after the channel. SortPreservingMergeExec uses the same StreamingMergeBuilder and BatchBuilder, so it has the same retention behavior.

Possible fixes:

  • In BatchBuilder, when push_batch cannot grow the reservation, emit the rows in indices as a short batch, then release the dead batches and retry. This bounds the retained batches to about one per input stream.
  • Avoid the shared-buffer over-count: compact (gc) view arrays after take in the hash partitioner, or account shared buffers once.
Vorherrschende Sprache
Rust
Sterne
9.6k
Forks
2.6k
Ø Merge
3 T. 4 Std.
Gemergte PRs (30 T.)
345

Entwicklungsumgebung

In Codespaces öffnen

Startet den Dev-Container des Projekts im Browser, mit Ihrem eigenen GitHub-Konto.

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus apache/datafusion

Alle Issues in apache/datafusion

Ähnliche Issues

Weitere Issues zu Rust

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.