Repository navigation
feat: Support null aware + equijoins for NestedLoopJoin - #16210
jonathanc-n wants to merge 11 commits into
Conversation
github-actions
Bot
added
core
| } | ||
| } | ||
|
|
||
| // Find matching indices based on join `on` predicates |
There was a problem hiding this comment.
Not sure if this is the most efficient way? I tried to add comments wherever possible to make the review process easier, and the logic is straight forward (should be simple to review 😄 ).
|
I believe a follow up pull request can be made to actually have this be used in the physical plan based on some statistics. @comphead Does this look fine for now? |
|
This feature is intended purely for performance, right? Not because there are cases that can only be run with NLJ. |
| // Compares values in the array, returns true if match, false otherwise | ||
| fn compare_arrays( | ||
| left_arrays: &[ArrayRef], | ||
| left: usize, |
There was a problem hiding this comment.
this reminds me of SMJ, the similar mechanism, probably we can factor it out?
| Arc::new(TestMemoryExec::update_cache(Arc::new(source))) | ||
| } | ||
|
|
||
| fn build_left_table_null() -> Arc<dyn ExecutionPlan> { |
There was a problem hiding this comment.
a nit: but I think it would be easier to show the table contents in the content.
The table can be built in text format like here https://www.tablesgenerator.com/text_tables
|
@2010YOUY01 I was doing some benchmarks on NLJs vs. HJs and it looked bad even for cases where one table is very small which is what NLJs should excel at. One thing is to note that it does seem slightly faster when both tables are very small, especially in the case when the outer table is the larger table (makes sense due to hashing the build side, I should probably test this with SMJ cases). Duckdb has something interesting where if one of the tables have a certain row threshold (5), they choose to use a nested loop join for BenchmarksHere is some experiment benchmarks that I had run if curious (one thing to note is that these benchmarks are just simulated on created row batches and are not read from disk + they are called directly from physical plan and not passed into the optimizer -- the batch sides would have been swapped): BenchmarkThe tests seem to be a bit inconsistent when I run locally so if someone would like to run it on their computer, here is the code (make a file in Codeuse std::sync::Arc;
use arrow::array::{Date32Builder, Decimal128Builder, Int32Builder, RecordBatch, StringBuilder};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use criterion::{criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion};
use datafusion_common::{Result};
use datafusion_execution::TaskContext;
use datafusion_expr::JoinType;
use datafusion_physical_expr::expressions::Column;
use datafusion_physical_plan::{collect, ExecutionPlan};
use tokio::runtime::Runtime;
use datafusion_physical_plan::joins::{HashJoinExec, NestedLoopJoinExec, PartitionMode};
use datafusion_physical_plan::test::TestMemoryExec;
pub fn create_batch(num_rows: usize, allow_nulls: bool) -> RecordBatch {
let schema = Arc::new(Schema::new(vec![
Field::new("c0", DataType::Int32, true),
Field::new("c1", DataType::Utf8, true),
Field::new("c2", DataType::Date32, true),
Field::new("c3", DataType::Decimal128(11, 2), true),
]));
let mut a = Int32Builder::new();
let mut b = StringBuilder::new();
let mut c = Date32Builder::new();
let mut d = Decimal128Builder::new().with_precision_and_scale(11, 2).unwrap();
for i in 0..num_rows {
a.append_value(i as i32);
c.append_value(i as i32);
d.append_value((i as i128) * 1_000_000);
if allow_nulls && i % 10 == 0 {
b.append_null();
} else {
b.append_value(format!("string_{i}"));
}
}
let a = Arc::new(a.finish());
let b = Arc::new(b.finish());
let c = Arc::new(c.finish());
let d = Arc::new(d.finish());
RecordBatch::try_new(
schema.clone(),
vec![a as _, b as _, c as _, d as _],
)
.unwrap()
}
fn make_memory_execs(
left_rows: usize,
right_rows: usize,
) -> (Arc<dyn ExecutionPlan>, Arc<dyn ExecutionPlan>, SchemaRef) {
let left_batch = create_batch(left_rows, true);
let schema = left_batch.schema();
let left_partitions = vec![vec![left_batch]];
let right_batch = create_batch(right_rows, true);
let right_partitions = vec![vec![right_batch]];
let left_mem = TestMemoryExec::try_new_exec(&left_partitions, schema.clone(), None).unwrap();
let right_mem = TestMemoryExec::try_new_exec(&right_partitions, schema.clone(), None).unwrap();
(left_mem, right_mem, schema)
}
fn build_two_joins(
left: Arc<dyn ExecutionPlan>,
right: Arc<dyn ExecutionPlan>,
) -> Result<(
Arc<dyn ExecutionPlan>, // hj
Arc<dyn ExecutionPlan>, // nlj
)> {
let hj = HashJoinExec::try_new(
left.clone(),
right.clone(),
vec![(Arc::new(Column::new("c0", 0)), Arc::new(Column::new("c0", 0)))],
None,
&JoinType::Inner,
None,
PartitionMode::Partitioned,
false, // null_aware = false
)?;
let nlj = NestedLoopJoinExec::try_new(
left,
right,
None,
&JoinType::Inner,
None,
false, // null_aware = false
Some(vec![(Arc::new(Column::new("c0", 0)), Arc::new(Column::new("c0", 0)))]),
)?;
Ok((Arc::new(hj), Arc::new(nlj)))
}
fn bench_joins(c: &mut Criterion) {
let rt = Runtime::new().unwrap();
let mut group = c.benchmark_group("joins");
// row pairs for each side in benchmarks
let size_pairs = &[
(5, 10_165_536),
(5, 1_165_536),
(5, 536),
(5, 36),
(1, 10_165_536),
(1, 1_165_536),
(1, 536),
(1, 36),
(10_165_536, 5),
(1_165_536, 5),
(536, 5),
(36, 5),
(10_165_536, 1),
(1_165_536, 1),
(536, 1),
(36, 1),
];
for &(left_rows, right_rows) in size_pairs.iter() {
let (left_mem, right_mem, _schema) = make_memory_execs(left_rows, right_rows);
let (hash_join, nested_loop_join) =
build_two_joins(left_mem.clone(), right_mem.clone()).unwrap();
group.bench_with_input(
BenchmarkId::new(
"HashJoin",
format!("l={}_r={}", left_rows, right_rows),
),
&hash_join,
|b, plan| {
b.iter_batched(
|| (),
|_setup| {
let ctx = TaskContext::default();
let fut = collect(plan.clone(), Arc::new(ctx));
rt.block_on(async {
let _ = fut.await.unwrap();
});
},
BatchSize::SmallInput,
)
},
);
group.bench_with_input(
BenchmarkId::new(
"NestedLoopJoin",
format!("l={}_r={}", left_rows, right_rows),
),
&nested_loop_join,
|b, plan| {
b.iter_batched(
|| (),
|_setup| {
let ctx = TaskContext::default();
let fut = collect(plan.clone(), Arc::new(ctx));
rt.block_on(async {
let _ = fut.await.unwrap();
});
},
BatchSize::SmallInput,
)
},
);
}
group.finish();
}
criterion_group!(benches, bench_joins);
criterion_main!(benches);I think this code should be run with different batches though. ConsensusI think we can either add this to Datafusion more as an feature users can experiment with (users can pass a flag and experiment -- |
|
There's an interesting implementation of index joins here, but this would involve creating indexes. What are the thoughts on supporting |
|
@comphead Pointers have been resolved, good for another review. Thanks! |
| /// in the case that a Hash Join cost is more expensive than a | ||
| /// nested loop join or when a user would like to pick nested loop | ||
| /// join by hint | ||
| on: Option<Vec<(PhysicalExprRef, PhysicalExprRef)>>, |
There was a problem hiding this comment.
Maybe we can change type to Vec<(PhysicalExprRef, PhysicalExprRef)> . on.is_empty() == true mean no equijoin columns.
There was a problem hiding this comment.
If we can merge on condition into filter, we can remove this field and reuse existing filter logic.
For example: filter: t1.a < t2.a, on: t1.c = t2.c -> filter: t1.a < t2.a and t1.c = t2.c
There was a problem hiding this comment.
Maybe we can change type to
Vec<(PhysicalExprRef, PhysicalExprRef)>.on.is_empty() == truemean no equijoin columns.
Because nested loop joins exclusively take in filter clauses more often than not, I decided it'd be easier to check if on.is_some(). Currently the physical planner doesnt take in nested loop join for equijoins on purpose, this will be added in another pr.
If we can merge
oncondition intofilter, we can remove this field and reuse existing filter logic. For example:filter: t1.a < t2.a, on: t1.c = t2.c->filter: t1.a < t2.a and t1.c = t2.c
I separated merging the filter + on condition to make sure that we can do the null_equals_null check (this is how it is implemented for the other joins as well).
There was a problem hiding this comment.
to make sure that we can do the null_equals_null check
We can also do this by merging on condition into filter.
For example: filter: t1.a < t2.a, on: t1.c = t2.c -> filter: t1.a < t2.a and (t1.c = t2.c or (is_null(t1.c) and (is_null(t2.c))))
this is how it is implemented for the other joins as well
Hash Joins and Sort-Merge Joins are fundamentally reliant on equality conditions. Their high-speed mechanisms (hashing, merging) are impossible without them.
But for nlj separating equality conditions doesn't improve performance
There was a problem hiding this comment.
But how would you suggest doing the null_equal_null check. I do think that it willl probably be faster to merge the two.
There was a problem hiding this comment.
Oh yeah that makes sense, just have it be merged in the join filter. 🤦 Thanks for finding this 😄. I think I'll just put out a pull request removing the todo statement, that statement led me to think that null aware wasn't supported yet.
There was a problem hiding this comment.
I am curious though, how were you able to have the query use a nested loop join with a on condition?
There was a problem hiding this comment.
I missed that we may convert EXCEPT to a null aware join. Sorry for that.
If null_equal_null is true, we can replace = in the ON clause with <=> and merge it into the filter in the planner. Otherwise, we merge the ON clause into the filter directly.
how were you able to have the query use a nested loop join with a on condition?
There's no equal(= not <=>) join condition in on, so the planner choose nlj
datafusion/datafusion/core/src/physical_planner.rs
Lines 1104 to 1111 in e5f596b
There was a problem hiding this comment.
If null_equal_null is true, we can replace = in the ON clause with <=> and merge it into the filter in the planner
This seems slower, we should just pass this to smj or hj. I can play around with the threshold logic that duckdb has for smaller tables where nljs may be faster.
There's no equal(
=not<=>) join condition inon, so the planner choose nlj
Since hash join and smj are usually faster for this equi conditions, instead of passing in <=> into the join filter we can flip the null equals null bit to true and run the <=> in the on clause; but this would clash with a = if null_equals_null is false. The solution I was thinking of is these cases:
null_equals_nullequals true +<=>condition +=condition -> we pass the<=>expressions into theonclausenull_equals_nullequals false +<=>condition +=condition -><=>stays in join filternull_equals_nullequals false +<=>condition + no=condition -> we flip thenull_equals_nullto true and pass the<=>expressions into theonclause
WDYT @UBarney
There was a problem hiding this comment.
If null_equal_null is true, we can replace = in the ON clause with <=> and merge it into the filter in the planner
This seems slower, we should just pass this to smj or hj. I can play around with the threshold logic that duckdb has for smaller tables where nljs may be faster.
Under the premise that we prefer NLJ
Rest lgtm
null_equals_null equals true + <=> condition -> we can always pass the <=> expressions into the on clause
Which issue does this PR close?
Rationale for this change
Supporting null aware loop join in the case that cost is cheaper than Hash join.
What changes are included in this PR?
Added support for null aware + equijoins for nested loop join.
Are these changes tested?
Yes