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

Đang mở
  • Có gì không ổn?
#27,379 1 bình luận 0 reaction 1 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

@lakshaydevoo-arch đang làm issue này rồi.

Từ ngày 12/11/2026.

Đánh giá

Issue này chưa được đánh giá.

Mô tả

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

Ngôn ngữ chính
Rust
Star
9.6k
Fork
2.6k
Merge trung bình
3 ngày 4 giờ
Pull request đã merge (30 ngày)
345

Chuẩn bị môi trường

Mở trong Codespaces

Khởi chạy dev container của dự án ngay trên trình duyệt, bằng tài khoản GitHub của bạn.

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/datafusion

Tất cả issue của apache/datafusion

Issue tương tự

Thêm issue về Rust

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.