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

オープン
  • 何か問題がありますか？
#27,379 コメント 1 件 リアクション 0 件 担当者 1 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

@lakshaydevoo-arch がすでに取り組んでいます。

2026年11月12日 から。

評価

この issue はまだ評価されていません。

説明

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

主要言語
Rust
スター
9.6k
フォーク
2.6k
平均マージ
3日 4時間
マージ済み PR（30日）
345

環境構築

Codespaces で開く

このプロジェクトの開発コンテナを、あなたの GitHub アカウントでブラウザ上に起動します。

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

apache/datafusion のほかの issue

apache/datafusion の issue をすべて見る

似ている issue

Rust の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。