feat: support nested loop join with the initial version #4562

liukun4515 · 2022-12-09T03:46:53Z

Which issue does this PR close?

Closes #4363

fix bug: join plan with nonequal condition will be converted to the NLJ instead of the error cross join

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

liukun4515 · 2022-12-09T03:48:21Z

datafusion/core/src/physical_plan/joins/nested_loop_join.rs

+}
+
+#[cfg(test)]
+mod tests {


there are som duplicated test code in the hashjoin and cross join.

I will refactor and clean up them in the followup pr

jackwener

That's great👍👍
I prepare to review this PR carefully in a few days.

mingmwang · 2022-12-09T11:04:47Z

I will review this PR tomorrow.

alamb

I took a quick scan through this code -- looks quite cool @liukun4515

Given @mingmwang and @jackwener plan to review I won't put this one on my deep review queue, unless there is something specific you would like me to look at

cc @Dandandan as I think he enjoys Join implementations

Dandandan · 2022-12-09T19:23:00Z

I wonder if it makes sense to also make use of NestedLoopJoin for cross joins (with empty filter) and get rid of CrossJoinExec, as that should have the same result while avoiding duplicate code. Does this make sense?

alamb · 2022-12-10T10:26:56Z

I wonder if it makes sense to also make use of NestedLoopJoin for cross joins (with empty filter) and get rid of CrossJoinExec

I would be in favor of this proposal (as a follow on PR) if possible. Joins are already (necessarily) complicated, so the more we can do to keep their complexity in check the better

jackwener · 2022-12-10T12:57:30Z

I wonder if it makes sense to also make use of NestedLoopJoin for cross joins (with empty filter) and get rid of CrossJoinExec, as that should have the same result while avoiding duplicate code. Does this make sense?

Yes, it make sense. I think CrossJoinExec should be a form of NLJ

mingmwang · 2022-12-12T05:57:41Z

datafusion/core/tests/sql/joins.rs

 async fn error_cross_join() -> Result<()> {
    let test_repartition_joins = vec![true, false];
    for repartition_joins in test_repartition_joins {


You can rename this UT's name to something else.

change to left_join_with_nonequal_condition

mingmwang · 2022-12-12T06:01:19Z

datafusion/core/src/physical_plan/planner.rs

-                    if session_state.config.target_partitions() > 1
+                    if join_on.is_empty() {
+                        // there is no equal join condition, use the nested loop join
+                        // TODO optimize the plan, and use the config of `target_partitions` and `repartition_joins`


I think for NLJ, there is no partitioned version.

A partitioned version is actually similar to CrossJoin/CartesianProduct.

In the left out join type, the distribution of the join is

vec![ Distribution::UnspecifiedDistribution, Distribution::SinglePartition, ]

can we partitioned the left side before NLJ exec to accelerate the execution?

mingmwang · 2022-12-12T07:37:25Z

datafusion/core/src/physical_plan/joins/nested_loop_join.rs

+fn distribution_from_join_type(join_type: &JoinType) -> Vec<Distribution> {
+    match join_type {
+        JoinType::Inner | JoinType::Left | JoinType::LeftSemi | JoinType::LeftAnti => {
+            // need the left data, and the right should be one partition
+            vec![
+                Distribution::UnspecifiedDistribution,
+                Distribution::SinglePartition,
+            ]
+        }
+        JoinType::Right | JoinType::RightSemi | JoinType::RightAnti => {
+            // need the right data, and the left should be one partition
+            vec![
+                Distribution::SinglePartition,
+                Distribution::UnspecifiedDistribution,
+            ]
+        }
+        JoinType::Full => {
+            // need the left and right data, and the left and right should be one partition
+            vec![Distribution::SinglePartition, Distribution::SinglePartition]
+        }
+    }
+}
+


The required distribution logic for the nested loop join is not consistent with the other physical join types(HashJoinExec::CollectLeft, CrossJoinExec), those existing joins do not consider the join types. For CrossJoinExec and HashJoinExec::CollectLeft, they always collect left. And there is a JoinSelection rule to adjust/reorder the joins sides to make sure the left side is the smaller side.

Especially for FullOut join, if we enforce the both input sides coalesced to the single partition, we might encounter performance issue.

In the current execution model, we need to decide the distribution by the join type

mingmwang · 2022-12-12T07:49:11Z

datafusion/core/src/physical_plan/joins/utils.rs

+pub(crate) fn need_produce_result_in_final(join_type: JoinType) -> bool {
+    matches!(
+        join_type,
+        JoinType::Left | JoinType::LeftAnti | JoinType::LeftSemi | JoinType::Full
+    )
+}


I would suggest to add more comments to those methods. Originally they are only used by HashJoin and they are private methods. Now they are used both by HashJoin and NestedLoopJoin.

mingmwang · 2022-12-12T07:49:38Z

datafusion/core/src/physical_plan/joins/utils.rs

+pub(crate) fn get_final_indices(
+    left_bit_map: &BooleanBufferBuilder,
+    join_type: JoinType,
+) -> (UInt64Array, UInt32Array) {


Same as above. Please add more comments to explain more clearly what's the expected input and what are the outputs.

jackwener · 2022-12-12T10:22:46Z

datafusion/core/src/physical_plan/joins/nested_loop_join.rs

+    /// right side
+    pub(crate) right: Arc<dyn ExecutionPlan>,
+    /// Filters which are applied while finding matching rows
+    pub(crate) filter: Option<JoinFilter>,


I am confused about this.
In my humble opinion, look like that JoinFilter in HashJoin isn't condition but predicate which is used for filtering result of hashjoin.
But JoinFilter here is as condition.
I don't know if my opinion is wrong.
If it's right, I recommend to name condition instead of filter, and add comment for it.

I think the naming of JoinFilter is consistent between HashJoinExec and NestedLoopJoinExec. And in the logical Join plan, it is also called filter.

mingmwang · 2022-12-12T10:31:40Z

datafusion/core/src/physical_plan/joins/nested_loop_join.rs

+        // right side
+        let right_side = if left_is_single_partition {
+            self.right.execute(partition, context)?
+        } else {
+            // the distribution of right is `SinglePartition`
+            self.right.execute(0, context)?
+        };


Is it correct here ? For FullOut join, both the left input and right input are single partition now.

Yes, if the join type is Full Join, the left and right is single partition.

fn execute( &self, partition: usize, context: Arc<TaskContext>, ) -> Result<SendableRecordBatchStream> {

The parameter of partition must be 0.

The parameter of partition is triggered by left side, if the left side is multi partition and have 4 partition,
the value of partition will be 0,1,2,3.

The implementation is right, but from your comments, I find it will make reviewer confused. I will add a new method to check if right is single partition.

jackwener · 2022-12-12T12:21:43Z

datafusion/core/src/physical_plan/joins/nested_loop_join.rs

+        let left_is_single_partition = self.is_single_partition_for_left();
+        // left side
+        let left_fut = if left_is_single_partition {
+            self.left_fut.once(|| {
+                // just one partition for the left side, and the first partition is all of data for left
+                load_left_specified_partition(0, self.left.clone(), context.clone())
+            })
+        } else {
+            // the distribution of left is not single partition, just need the specified partition for left
+            OnceFut::new(load_left_specified_partition(
+                partition,
+                self.left.clone(),
+                context.clone(),
+            ))
+        };


I'm not sure about how to design for different kind partition.
I need to investigate about this.

liukun4515 · 2022-12-13T07:35:37Z

I wonder if it makes sense to also make use of NestedLoopJoin for cross joins (with empty filter) and get rid of CrossJoinExec, as that should have the same result while avoiding duplicate code. Does this make sense?

In the implementation of Join, this is join type in current codebase:

pub enum JoinType {
    /// Inner Join
    Inner,
    /// Left Join
    Left,
    /// Right Join
    Right,
    /// Full Join
    Full,
    /// Left Semi Join
    LeftSemi,
    /// Right Semi Join
    RightSemi,
    /// Left Anti Join
    LeftAnti,
    /// Right Anti Join
    RightAnti,
}

In the logical phase, we split the join to cross join with other join type.

I think we can fill a new issue to discuss it.

I prefer merge cross join to the join type.

cc @Dandandan

mingmwang · 2022-12-13T07:48:37Z

I think CrossJoin is still required as a physical join plan implementation, because for Cross Join we can have different Partitioning/Shuffle strategies(M * N). Of course we can merge all the logic to NestedLoopJoin, but this will make the NestedLoopJoin's partitioning logic very complex(which is already quite complex).

liukun4515 · 2022-12-16T04:20:01Z

@mingmwang @jackwener @Dandandan PTAL

jackwener

LGTM, thanks @liukun4515 .

liukun4515 · 2022-12-19T05:41:35Z

will merge this pr tomorrow, if there are no other comments.

cc @mingmwang @alamb

alamb · 2022-12-20T18:27:48Z

Merging this one in so it doesn't accumulate conflicts. Thanks for all the work @liukun4515, @mingmwang and @jackwener

ursabot · 2022-12-20T18:32:58Z

Benchmark runs are scheduled for baseline = 2792113 and contender = fddb3d3. fddb3d3 is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ec2-t3-xlarge-us-east-2] ec2-t3-xlarge-us-east-2
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] test-mac-arm
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] ursa-i9-9960x
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] ursa-thinkcentre-m75q
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java

support on condition without equal condition

2d1e9ad

github-actions bot added core Core DataFusion crate sql SQL Planner labels Dec 9, 2022

liukun4515 commented Dec 9, 2022

View reviewed changes

refine code for hash join exec

18c6152

liukun4515 force-pushed the nlj_fix_bug branch from 731229e to 18c6152 Compare December 9, 2022 03:49

Merge remote-tracking branch 'upstream/master' into nlj_fix_bug

c7c6d9b

liukun4515 requested review from alamb and Dandandan December 9, 2022 03:59

jackwener reviewed Dec 9, 2022

View reviewed changes

alamb reviewed Dec 9, 2022

View reviewed changes

mingmwang reviewed Dec 12, 2022

View reviewed changes

jackwener reviewed Dec 12, 2022

View reviewed changes

mingmwang reviewed Dec 12, 2022

View reviewed changes

jackwener reviewed Dec 12, 2022

View reviewed changes

liukun4515 added 3 commits December 13, 2022 17:38

add method check distribution of right, make code clearly

d4528e5

fix failed case

03b0a48

add comments for pub(crate) method

2466633

liukun4515 requested review from mingmwang and removed request for mingmwang December 16, 2022 04:16

liukun4515 requested review from jackwener and mingmwang and removed request for jackwener and mingmwang December 16, 2022 04:16

jackwener approved these changes Dec 16, 2022

View reviewed changes

Merge remote-tracking branch 'upstream/master' into nlj_fix_bug

14f77af

liukun4515 mentioned this pull request Dec 20, 2022

Support type coercion for equijoin #4666

Merged

alamb merged commit fddb3d3 into apache:master Dec 20, 2022

Jefffrey mentioned this pull request Jan 8, 2023

Incorrect results for join condition against current master branch #4844

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: support nested loop join with the initial version #4562

feat: support nested loop join with the initial version #4562

liukun4515 commented Dec 9, 2022

liukun4515 Dec 9, 2022

jackwener left a comment

mingmwang commented Dec 9, 2022

alamb left a comment

Dandandan commented Dec 9, 2022

alamb commented Dec 10, 2022

jackwener commented Dec 10, 2022

mingmwang Dec 12, 2022

liukun4515 Dec 13, 2022

mingmwang Dec 12, 2022

mingmwang Dec 12, 2022

liukun4515 Dec 13, 2022 •

edited

Loading

mingmwang Dec 12, 2022

mingmwang Dec 12, 2022 •

edited

Loading

liukun4515 Dec 13, 2022

mingmwang Dec 12, 2022

mingmwang Dec 12, 2022 •

edited

Loading

jackwener Dec 12, 2022 •

edited

Loading

mingmwang Dec 12, 2022

mingmwang Dec 12, 2022

liukun4515 Dec 13, 2022

jackwener Dec 12, 2022

liukun4515 commented Dec 13, 2022

mingmwang commented Dec 13, 2022

liukun4515 commented Dec 16, 2022

jackwener left a comment

liukun4515 commented Dec 19, 2022

alamb commented Dec 20, 2022

ursabot commented Dec 20, 2022

feat: support nested loop join with the initial version #4562

feat: support nested loop join with the initial version #4562

Conversation

liukun4515 commented Dec 9, 2022

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Choose a reason for hiding this comment

jackwener left a comment

Choose a reason for hiding this comment

mingmwang commented Dec 9, 2022

alamb left a comment

Choose a reason for hiding this comment

Dandandan commented Dec 9, 2022

alamb commented Dec 10, 2022

jackwener commented Dec 10, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

liukun4515 Dec 13, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mingmwang Dec 12, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mingmwang Dec 12, 2022 • edited Loading

Choose a reason for hiding this comment

jackwener Dec 12, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

liukun4515 commented Dec 13, 2022

mingmwang commented Dec 13, 2022

liukun4515 commented Dec 16, 2022

jackwener left a comment

Choose a reason for hiding this comment

liukun4515 commented Dec 19, 2022

alamb commented Dec 20, 2022

ursabot commented Dec 20, 2022

liukun4515 Dec 13, 2022 •

edited

Loading

mingmwang Dec 12, 2022 •

edited

Loading

mingmwang Dec 12, 2022 •

edited

Loading

jackwener Dec 12, 2022 •

edited

Loading