You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/19 03:27:01 UTC
(datafusion) branch gh-readonly-queue/main/pr-24192-76a45130a440b3f4238cfcbb211803d90565229a created (now 54355e1b7b)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-24192-76a45130a440b3f4238cfcbb211803d90565229a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 54355e1b7b minor(test): strengthen bitwise sort-merge join spill coverage (#24192)
This branch includes the following new commits:
new 54355e1b7b minor(test): strengthen bitwise sort-merge join spill coverage (#24192)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: minor(test): strengthen bitwise sort-merge join spill coverage (#24192)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24192-76a45130a440b3f4238cfcbb211803d90565229a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 54355e1b7b21f093cd7cb9e7a04865d7f2e6df9c
Author: Burak Şen <bu...@gmail.com>
AuthorDate: Wed Aug 19 03:26:52 2026 +0000
minor(test): strengthen bitwise sort-merge join spill coverage (#24192)
## Which issue does this PR close?
- Part of #13431
## Rationale for this change
Continuing the work after #23988 and #24056.
`BitwiseSortMergeJoinStream` (filtered semi/anti/mark joins) buffers the
inner key group and spills it under memory pressure, but existing tests
only cover the successful spill path:
- when the memory pool is exhausted and disk spilling is disabled, the
error path had no coverage
- the multi-spill slice retention regression test only asserted the semi
join outcome; a dropped slice resurrects the anti row and flips mark to
false, which row counts alone cannot catch
- no spill test exercised `NullEqualsNull`, so NULL-key group boundary
detection through the spill path was untested
## What changes are included in this PR?
- new `bitwise_filtered_no_spill`: all six semi/anti/mark join types
must surface a clear "Disk spilling disabled" error (not spill or panic)
when the pool is exhausted with the `DiskManager` disabled
- extend `bitwise_multi_spill_inner_key_group` over
`LeftAnti`/`LeftMark`, asserting the mark column value
- new `bitwise_spill_null_key_group`: under `NullEqualsNull`, a NULL-key
group spanning a batch boundary must spill as one group with one spill
file and match a NULL-key outer row
- extract shared setup of `bitwise_spill_with_filter` into a fixture
## Are these changes tested?
test only change and it can be tested via:
```
cargo test -p datafusion-physical-plan --lib -- sort_merge_join::tests::bitwise
```
## Are there any user-facing changes?
no test only change
---
.../src/joins/sort_merge_join/tests.rs | 284 ++++++++++++++++++---
1 file changed, 244 insertions(+), 40 deletions(-)
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs
index 1d29db20dc..91d1b893f1 100644
--- a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs
+++ b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs
@@ -4890,15 +4890,15 @@ async fn filtered_boundary_pending_outer_rows() -> Result<()> {
// ── Bitwise stream spill tests ─────────────────────────────────────────────
-/// Exercises inner key group spilling under memory pressure.
-///
-/// Uses a tiny memory limit (100 bytes) with disk spilling enabled. Since our
-/// operator only buffers inner rows when a filter is present, this test includes
-/// a filter (c1 < c2, always true). Verifies:
-/// 1. Spill metrics are recorded (spill_count, spilled_bytes, spilled_rows > 0)
-/// 2. Results match a non-spilled run
-#[tokio::test]
-async fn bitwise_spill_with_filter() -> Result<()> {
+struct FilteredBitwiseSpillFixture {
+ left: Arc<dyn ExecutionPlan>,
+ right: Arc<dyn ExecutionPlan>,
+ on: JoinOn,
+ sort_options: Vec<SortOptions>,
+ filter: JoinFilter,
+}
+
+fn filtered_bitwise_spill_fixture() -> Result<FilteredBitwiseSpillFixture> {
let left = build_table(
("a1", &vec![1, 2, 3, 4, 5, 6]),
("b1", &vec![1, 2, 3, 4, 5, 6]),
@@ -4914,29 +4914,33 @@ async fn bitwise_spill_with_filter() -> Result<()> {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
)];
let sort_options = vec![SortOptions::default(); on.len()];
+ let filter = build_c1_lt_c2_filter(left.schema().as_ref(), right.schema().as_ref());
- // c1 < c2 is always true for matching keys
- let filter = JoinFilter::new(
- Arc::new(BinaryExpr::new(
- Arc::new(Column::new("c1", 0)),
- Operator::Lt,
- Arc::new(Column::new("c2", 1)),
- )),
- vec![
- ColumnIndex {
- index: 2,
- side: JoinSide::Left,
- },
- ColumnIndex {
- index: 2,
- side: JoinSide::Right,
- },
- ],
- Arc::new(Schema::new(vec![
- Field::new("c1", DataType::Int32, false),
- Field::new("c2", DataType::Int32, false),
- ])),
- );
+ Ok(FilteredBitwiseSpillFixture {
+ left,
+ right,
+ on,
+ sort_options,
+ filter,
+ })
+}
+
+/// Exercises inner key group spilling under memory pressure.
+///
+/// Uses a tiny memory limit (100 bytes) with disk spilling enabled. Since our
+/// operator only buffers inner rows when a filter is present, this test includes
+/// a filter (c1 < c2, always true). Verifies:
+/// 1. Spill metrics are recorded (spill_count, spilled_bytes, spilled_rows > 0)
+/// 2. Results match a non-spilled run
+#[tokio::test]
+async fn bitwise_spill_with_filter() -> Result<()> {
+ let FilteredBitwiseSpillFixture {
+ left,
+ right,
+ on,
+ sort_options,
+ filter,
+ } = filtered_bitwise_spill_fixture()?;
let runtime = RuntimeEnvBuilder::new()
.with_memory_limit(100, 1.0)
@@ -5033,10 +5037,82 @@ async fn bitwise_spill_with_filter() -> Result<()> {
Ok(())
}
+/// Semi/anti/mark joins use `BitwiseSortMergeJoinStream`, which buffers the
+/// inner key group for filter evaluation. When that buffer exhausts the memory
+/// pool and the `DiskManager` has spilling disabled, the stream must surface a
+/// clear "Disk spilling disabled" error instead of spilling or panicking.
+#[tokio::test]
+async fn bitwise_filtered_no_spill() -> Result<()> {
+ let FilteredBitwiseSpillFixture {
+ left,
+ right,
+ on,
+ sort_options,
+ filter,
+ } = filtered_bitwise_spill_fixture()?;
+
+ // Tiny memory pool with the DiskManager disabled: spilling is impossible.
+ let runtime = RuntimeEnvBuilder::new()
+ .with_memory_limit(100, 1.0)
+ .with_disk_manager_builder(
+ DiskManagerBuilder::default().with_mode(DiskManagerMode::Disabled),
+ )
+ .build_arc()?;
+ let session_config = SessionConfig::default().with_batch_size(1);
+
+ for join_type in [
+ LeftSemi, LeftAnti, RightSemi, RightAnti, LeftMark, RightMark,
+ ] {
+ let task_ctx = Arc::new(
+ TaskContext::default()
+ .with_session_config(session_config.clone())
+ .with_runtime(Arc::clone(&runtime)),
+ );
+
+ let join = SortMergeJoinExec::try_new(
+ Arc::clone(&left),
+ Arc::clone(&right),
+ on.clone(),
+ Some(filter.clone()),
+ join_type,
+ sort_options.clone(),
+ NullEquality::NullEqualsNothing,
+ )?;
+ let stream = join.execute(0, task_ctx)?;
+ let err = common::collect(stream).await.unwrap_err();
+
+ assert_contains!(err.to_string(), "Failed to allocate additional");
+ assert_contains!(err.to_string(), "SMJStream[0]");
+ assert_contains!(err.to_string(), "Disk spilling disabled");
+ let metrics = join
+ .metrics()
+ .unwrap_or_else(|| panic!("metrics missing for {join_type:?}"));
+ assert_eq!(
+ metrics.spill_count(),
+ Some(0),
+ "unexpected spill for {join_type:?} with disk disabled",
+ );
+ assert_eq!(
+ metrics.spilled_bytes(),
+ Some(0),
+ "unexpected spilled bytes for {join_type:?} with disk disabled",
+ );
+ assert_eq!(
+ metrics.spilled_rows(),
+ Some(0),
+ "unexpected spilled rows for {join_type:?} with disk disabled",
+ );
+ }
+
+ Ok(())
+}
+
/// A single inner key group spanning several inner batches can spill more
/// than once under memory pressure. Every spilled slice must still be
/// evaluated against the outer rows — an earlier spill file must not be
-/// dropped when a later slice of the same group spills.
+/// dropped when a later slice of the same group spills. A dropped slice
+/// surfaces differently per join type: semi loses its row, anti resurrects
+/// it, and mark flips to `false`.
#[tokio::test]
async fn bitwise_multi_spill_inner_key_group() -> Result<()> {
// Outer: one row with key 1, c1 = 5.
@@ -5044,10 +5120,140 @@ async fn bitwise_multi_spill_inner_key_group() -> Result<()> {
// Inner: one key group (b2 = 1) spanning two batches. Only the first
// batch satisfies the filter c1 < c2 (5 < 10); the second (5 < 0) does
- // not, so dropping the first spilled slice flips the semi-join result.
- let right_batches = vec![
+ // not, so the outcome depends on the group's first spilled slice.
+ let right = build_table_from_batches(vec![
build_table_i32(("a2", &vec![10]), ("b2", &vec![1]), ("c2", &vec![10])),
build_table_i32(("a2", &vec![20]), ("b2", &vec![1]), ("c2", &vec![0])),
+ ]);
+
+ // Right variants stream right and buffer left, so mirror the two-batch
+ // key group onto left. Again, only its first slice satisfies c1 < c2.
+ let mirrored_left = build_table_from_batches(vec![
+ build_table_i32(("a1", &vec![10]), ("b1", &vec![1]), ("c1", &vec![0])),
+ build_table_i32(("a1", &vec![20]), ("b1", &vec![1]), ("c1", &vec![10])),
+ ]);
+ let mirrored_right =
+ build_table(("a2", &vec![1]), ("b2", &vec![1]), ("c2", &vec![5]));
+
+ let on: JoinOn = vec![(
+ Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
+ )];
+ let sort_options = vec![SortOptions::default(); on.len()];
+ let filter = build_c1_lt_c2_filter(left.schema().as_ref(), right.schema().as_ref());
+
+ // 100-byte pool: every buffered slice fails its reservation, so each
+ // inner batch of the key group spills separately.
+ let runtime = RuntimeEnvBuilder::new()
+ .with_memory_limit(100, 1.0)
+ .with_disk_manager_builder(
+ DiskManagerBuilder::default().with_mode(DiskManagerMode::OsTmpDirectory),
+ )
+ .build_arc()?;
+
+ for (join_type, expected_rows) in [
+ (LeftSemi, 1),
+ (LeftAnti, 0),
+ (LeftMark, 1),
+ (RightSemi, 1),
+ (RightAnti, 0),
+ (RightMark, 1),
+ ] {
+ let task_ctx = Arc::new(
+ TaskContext::default()
+ .with_session_config(SessionConfig::default().with_batch_size(1))
+ .with_runtime(Arc::clone(&runtime)),
+ );
+ let (join_left, join_right) =
+ if matches!(join_type, RightSemi | RightAnti | RightMark) {
+ (&mirrored_left, &mirrored_right)
+ } else {
+ (&left, &right)
+ };
+
+ let join = SortMergeJoinExec::try_new(
+ Arc::clone(join_left),
+ Arc::clone(join_right),
+ on.clone(),
+ Some(filter.clone()),
+ join_type,
+ sort_options.clone(),
+ NullEquality::NullEqualsNothing,
+ )?;
+ let stream = join.execute(0, task_ctx)?;
+ let batches = common::collect(stream).await?;
+
+ let output_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(
+ output_rows, expected_rows,
+ "unexpected output rows for {join_type:?}",
+ );
+ if matches!(join_type, LeftMark | RightMark) {
+ let batch = batches.iter().find(|b| b.num_rows() > 0).unwrap();
+ let mark = batch
+ .column_by_name("mark")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<BooleanArray>()
+ .unwrap();
+ assert!(
+ mark.value(0),
+ "mark must be true: the row matches the group's first spilled slice",
+ );
+ }
+
+ let metrics = join.metrics().expect("must have metrics");
+ assert_eq!(
+ metrics.spill_count(),
+ Some(1),
+ "all overflows of one key group must share a single spill file for {join_type:?}",
+ );
+ assert_eq!(
+ metrics.spilled_rows(),
+ Some(2),
+ "both inner slices of the group must be spilled for {join_type:?}",
+ );
+ }
+ Ok(())
+}
+
+/// Under `NullEqualsNull`, a NULL-key inner group spanning a batch boundary
+/// must behave like any equal-key group through the spill path: one group,
+/// one spill file, matched by a NULL-key outer row.
+#[tokio::test]
+async fn bitwise_spill_null_key_group() -> Result<()> {
+ // Outer: one row whose join key is NULL, c1 = 5.
+ let left = build_table_i32_nullable(
+ ("a1", &vec![Some(1)]),
+ ("b1", &vec![None]),
+ ("c1", &vec![Some(5)]),
+ );
+
+ // Inner: one NULL-key group spanning two batches; only the first batch
+ // satisfies the filter (5 < 10), so the result depends on the group's
+ // first spilled slice.
+ let right_schema = Arc::new(Schema::new(vec![
+ Field::new("a2", DataType::Int32, true),
+ Field::new("b2", DataType::Int32, true),
+ Field::new("c2", DataType::Int32, true),
+ ]));
+ let right_batches = vec![
+ RecordBatch::try_new(
+ Arc::clone(&right_schema),
+ vec![
+ Arc::new(Int32Array::from(vec![Some(10)])),
+ Arc::new(Int32Array::from(vec![None::<i32>])),
+ Arc::new(Int32Array::from(vec![Some(10)])),
+ ],
+ )?,
+ RecordBatch::try_new(
+ Arc::clone(&right_schema),
+ vec![
+ Arc::new(Int32Array::from(vec![Some(20)])),
+ Arc::new(Int32Array::from(vec![None::<i32>])),
+ Arc::new(Int32Array::from(vec![Some(0)])),
+ ],
+ )?,
];
let right = build_table_from_batches(right_batches);
@@ -5058,8 +5264,6 @@ async fn bitwise_multi_spill_inner_key_group() -> Result<()> {
let sort_options = vec![SortOptions::default(); on.len()];
let filter = build_c1_lt_c2_filter(left.schema().as_ref(), right.schema().as_ref());
- // 100-byte pool: every buffered slice fails its reservation, so each
- // inner batch of the key group spills separately.
let runtime = RuntimeEnvBuilder::new()
.with_memory_limit(100, 1.0)
.with_disk_manager_builder(
@@ -5079,7 +5283,7 @@ async fn bitwise_multi_spill_inner_key_group() -> Result<()> {
Some(filter),
LeftSemi,
sort_options,
- NullEquality::NullEqualsNothing,
+ NullEquality::NullEqualsNull,
)?;
let stream = join.execute(0, task_ctx)?;
let batches = common::collect(stream).await?;
@@ -5087,19 +5291,19 @@ async fn bitwise_multi_spill_inner_key_group() -> Result<()> {
let output_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(
output_rows, 1,
- "left row must match the group's first (spilled) inner slice",
+ "NULL-key outer row must match the spilled NULL-key group under NullEqualsNull",
);
let metrics = join.metrics().expect("must have metrics");
assert_eq!(
metrics.spill_count(),
Some(1),
- "all overflows of one key group must share a single spill file",
+ "the NULL-key group must span the batch boundary as one group with one spill file",
);
assert_eq!(
metrics.spilled_rows(),
Some(2),
- "both inner slices of the group must be spilled",
+ "both NULL-key slices must be spilled",
);
Ok(())
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org