Memory limited nest loop join

[2010YOUY01]

Is your feature request related to a problem or challenge?

The common NLJ implementation consumes constant memory. However, DataFusion's implementation is optimized for execution time, which requires it to buffer all input data on one side, making it possible to fail under memory-constrained cases.

The following pseudocode explains the current implementation:

Normal NLJ

for l_batch in nlj.left.get_next_batch():
    for r_batch in nlj.right.get_next_batch():
        let matched_entries = match(l_batch, r_batch);
        output(matched_entries)
    nlj.right.reset(); // Let right side of the join restart and scan again

DataFusion's NLJ implementation

for l_batch in nlj.left.get_next_batch():
    buffered_left_batches.push(l_batch);

for r_batch in nlj.right.get_next_batch():
    let matched_entries = match(buffered_left_batches, r_batch);
    output(matched_batches);

Related code:

datafusion/datafusion/physical-plan/src/joins/nested_loop_join.rs

Line 499 in 1282170

let inner_table = self.inner_table.try_once(|| {

Describe the solution you'd like

When the memory limit is reached when buffering the left side input, start probing the right side. After it's done, collect the remaining left side entries, and let the right side scan from the beginning and probe again, until finished.

Describe alternatives you've considered

No response

Additional context

No response

[shruti2522]

take

[UBarney]

let the right side scan from the beginning

As I understand it, the only way to achieve this is by read the entire right side all at once, either buffering it entirely in memory or spilling it to disk.
But if the entire right side is read at once, then streaming the left side would be the same as the current approach, right?

[2010YOUY01]

let the right side scan from the beginning

As I understand it, the only way to achieve this is by read the entire right side all at once, either buffering it entirely in memory or spilling it to disk. But if the entire right side is read at once, then streaming the left side would be the same as the current approach, right?

If the right side is a table backed by a local file, the operator can scan it again without buffering or spilling. I was thinking the right side should be changed to an ExecutionPlan instead of the existing RecordBatchStream, to let it able to be executed multiple times.

[UBarney]

If the right side is a table backed by a local file, the operator can scan it again without buffering or spilling.

I previously thought that 'backed by a local file' and 'spilling' were the same thing.

I find that the current implementation may load all results of join(all_row_in_left_table, current_right_batch) into memory at once. This could potentially contain up to all_row_in_left_table.row_count() * current_right_batch.row_count() rows in the worst case.

datafusion/datafusion/physical-plan/src/joins/nested_loop_join.rs

Lines 885 to 895 in 3fe300c

	let result = join_left_and_right_batch(
	left_data.batch(),
	batch,
	self.join_type,
	self.filter.as_ref(),
	&self.column_indices,
	&self.schema,
	visited_left_side,
	&mut self.indices_cache,
	self.right_side_ordered,
	);

I added some log to confirm that:

+                let batch1 = result?;
+                dbg!(&batch1.num_rows());
+
+                self.batch_transformer.set_batch(batch1);

output

> select count(*) from (select * from  generate_series(10000,20000) as t1 join generate_series(8194)  as t2 on t1.value != t2.value);

[datafusion/physical-plan/src/joins/nested_loop_join.rs:900:17] &batch1.num_rows() = 30003
[datafusion/physical-plan/src/joins/nested_loop_join.rs:900:17] &batch1.num_rows() = 81928192

Maybe we can reduce avoid get join result of all_row_in_left_table, current_right_batch at once