A FULL or LEFT hash join whose probe side gives no row emits its unmatched build rows as one batch, ignoring batch_size

Offen
  • Stimmt etwas nicht?
#27,379 1 Kommentar 0 Reaktionen 1 zugewiesene Person Auf GitHub ansehen

Maintainer antworten meist innerhalb von 1 Tag

@lakshaydevoo-arch arbeitet bereits daran.

Seit 12.11.2026.

Bewertung

Dieses Issue wurde noch nicht bewertet.

Beschreibung

bug
Describe the bug

A FULL or LEFT join planned as HashJoinExec: mode=CollectLeft whose right (probe) side yields no row emits every unmatched row of its left (build) side as a single batch, whatever datafusion.execution.batch_size is. With one probe row, matching or not, the same rows come in batches of batch_size.

This matters under a memory pool. A SortExec above the join reserves about twice the batch's size (9.2 MB below, for a batch of about 4.6 MB) before it holds anything it could spill, so the statement fails with Resources exhausted ... for ExternalSorter[0] with 0.0 B already allocated, where the same rows in batches of 8192 sort and spill within the pool.

Settings that did not change it: datafusion.execution.batch_size = 1000 (still one batch of 200000), enforce_batch_size_in_joins = true, coalesce_batches = false, hash_join_buffering_capacity = 0 and 1048576.

To Reproduce

With datafusion-python 55.0.0:

import pyarrow as pa
from datafusion import RuntimeEnvBuilder, SessionConfig, SessionContext

rows = 200_000
table = pa.table({"k": pa.array(range(rows), pa.int64()), "v": pa.array(range(rows), pa.float64())})


def context(pool_bytes=None):
    config = SessionConfig().with_target_partitions(1).set("datafusion.execution.sort_spill_reservation_bytes", str(2**20))
    runtime = RuntimeEnvBuilder().with_disk_manager_os()
    if pool_bytes:
        runtime = runtime.with_fair_spill_pool(pool_bytes)
    ctx = SessionContext(config, runtime)
    ctx.register_record_batches("a", [table.to_batches(max_chunksize=8192)])
    ctx.register_record_batches("b", [table.to_batches(max_chunksize=8192)])
    return ctx


for where in ("k < 0", "k < 1"):  # the probe side gives no row, then one row
    sql = f"SELECT a.k, a.v, b.v AS w FROM a FULL JOIN (SELECT * FROM b WHERE {where}) b ON a.k = b.k"
    sizes = [batch.to_pyarrow().num_rows for batch in context().sql(sql).execute_stream()]
    print(f"WHERE {where}: {len(sizes)} batches, largest {max(sizes)}")
    try:
        sorted_rows = sum(b.to_pyarrow().num_rows for b in context(8 * 2**20).sql(sql + " ORDER BY a.v DESC").execute_stream())
        print(f"  ORDER BY under an 8 MB fair spill pool: sorted {sorted_rows} rows")
    except Exception as error:
        print(f"  ORDER BY under an 8 MB fair spill pool: {str(error).splitlines()[-1]}")

Output:

WHERE k < 0: 1 batches, largest 200000
  ORDER BY under an 8 MB fair spill pool: Resources exhausted: Failed to allocate additional 9.2 MB for ExternalSorter[0] with 0.0 B already allocated for this reservation - 5.3 MB remain available for the total memory pool: fair(pool_size: 8.0 MB)
WHERE k < 1: 25 batches, largest 8192
  ORDER BY under an 8 MB fair spill pool: sorted 200000 rows

The full error starts with Not enough memory to continue external sort. Consider increasing the memory limit config .... The plan:

SortExec: expr=[v@1 DESC], preserve_partitioning=[false]
  ProjectionExec: expr=[k@0 as k, v@1 as v, v@2 as w]
    HashJoinExec: mode=CollectLeft, join_type=Full, on=[(k@0, k@0)], projection=[k@0, v@1, v@3]
      DataSourceExec: partitions=1, partition_sizes=[25]
      FilterExec: k@0 < 0
        DataSourceExec: partitions=1, partition_sizes=[25]

LEFT JOIN gives the same batches ([200000] against batches of 8192). Reading the two tables from Parquet does too; there, with collect_statistics = true, the planner makes the side without rows the build side and the rows come in batches.

Expected behavior

The unmatched build rows emitted in batches of batch_size, as they are once the probe side has given a row, so the ORDER BY above sorts and spills within the pool in both cases.

Additional context

The probe side giving no row is common in practice: a filter on the joined table that matches nothing, such as a key range one table holds no row of. A query then fails or succeeds depending on whether one table happens to be empty for it, with the same memory pool.

Willingness to contribute

I cannot contribute a fix for this bug at this time

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.