Uh oh!
There was an error while loading. Please reload this page.
Basic support for IN and NOT IN Subqueries by rewriting them to SEMI / ANTI Join - #2421
Conversation
There was a problem hiding this comment.
Thanks @korowa this is looking good. This actually adds support for benchmark queries 16 and 18, which previously were not supported.
If you want to, you could enable these queries in the benchmark test at benchmarks/src/bin/tpch.rs by adding these tests. This could also be done as a separate PR.
#[tokio::test]asyncfnrun_q16() -> Result<()>{run_query(16).await}#[tokio::test]asyncfnrun_q18() -> Result<()>{run_query(18).await}| })?; | ||
| if !subqueries_in_regular.is_empty() { | ||
| return Err(DataFusionError::NotImplemented( |
There was a problem hiding this comment.
Can we just revert to the original query here rather than fail?
There was a problem hiding this comment.
All fixed. My idea was to give user some explanation about why query is "incorrect" by throwing errors.
Now I see that if DF is able to produce this kind of logical plan, then it's valid (at least for some purposes maybe), even if we don't have physical implementation for some of its parts yet.
| let right_key = right_schema.field(0).qualified_column(); | ||
| let left_key = match *expr.clone() { | ||
| Expr::Column(col) => col, | ||
| _ => return Err(DataFusionError::NotImplemented( |
There was a problem hiding this comment.
Same comment here. Can we just abort the optimization attempt rather than fail.
| Ok(()) | ||
| } | ||
| _ => Err(DataFusionError::Plan( | ||
| "Unknown expression while rewriting subquery to joins" |
| )), | ||
| }; | ||
| let join_type = match negated { |
| }; | ||
| let schema = build_join_schema( | ||
| new_input.schema(), |
There was a problem hiding this comment.
Better to use optimized_input.schema() and avoid creating a mut
There was a problem hiding this comment.
Switched to try_fold, no mutable variables required now
| null_equals_null: false, | ||
| }); | ||
| Ok(()) |
There was a problem hiding this comment.
and return result here to use below
IN and NOT IN Subqueries by rewriting them to SEMI / ANTI| run_query(14).await | ||
| } | ||
| #[tokio::test] |
| LogicalPlan::Filter(Filter { input, predicate }) => { | ||
| let mut predicates = vec![]; | ||
| split_members(predicate, &mut predicates); | ||
| utils::split_conjunction(predicate, &mut predicates); |
| } | ||
| } | ||
| /// converts "A AND B AND C" => [A, B, C] |
There was a problem hiding this comment.
👍 for moving these functions into utils
| )), | ||
| }; | ||
| let join_type = if *negated { |
| .build()?; | ||
| let expected = "Projection: #test.b [b:UInt32]\ | ||
| \n Anti Join: #test.c = #test.c [a:UInt32, b:UInt32, c:UInt32]\ |
There was a problem hiding this comment.
I can't tell from this plan if the predicate is correct (because #test is used as the relation name in both the inner and outer query.
It might make these tests more readable if the relation name in the subquery was something different (like test_sq) so that this join predicate appears as #test.c = #sq_ test.c
There was a problem hiding this comment.
Done - now table names in tests are different, so it should by much easier to read
| .build()?; | ||
| let expected = "Projection: #test.b [b:UInt32]\ | ||
| \n Semi Join: #test.b = #test.a [a:UInt32, b:UInt32, c:UInt32]\ |
| assert_optimized_plan_eq(&plan, expected); | ||
| Ok(()) | ||
| } | ||
| } |
There was a problem hiding this comment.
I think it would also be helpful to coverage of the negative cases (aka cases that can't be rewritten like x IN (select ...) OR y = 5?
There was a problem hiding this comment.
Added cases for unsupported filter expression and for filters input being rewritten while filter remains untouched (checks that falling back to original query doesn't affect its recursive call results)
| let expected = "Projection: #test.b [b:UInt32]\ | ||
| \n Semi Join: #test.c = #test.c [a:UInt32, b:UInt32, c:UInt32]\ | ||
| \n Semi Join: #test.c = #sq.c [a:UInt32, b:UInt32, c:UInt32]\ |
| } | ||
| /// Test for filter input modification in case filter not supported | ||
| /// Outer filter expression not modified while inner converted to join |
korowa
commented
May 4, 2022
Thanks for reviews! I hope I'll follow up with PR(s?) for currently unsupported cases soon. |
IN and NOT IN Subqueries by rewriting them to SEMI / ANTIIN and NOT IN Subqueries by rewriting them to SEMI / ANTI Join
Which issue does this PR close?
Partially #488.
Rationale for this change
Naive implementation of optimizer rule for replacing InSubquery with join, which allows to execute queries with IN (subquery) in case of proper WHERE condition.
What changes are included in this PR?
SubqueryFilterToJoin rule is able to replace Filter input in logical plan with Sem/AntiJoin in case InSubquery is a part of logical conjunction - this precondition allows to pushdown IN predicate before other predicates in Filter.
Cases when IN (subquery) cannot be pushed to Filters input due to its result being required for predicate evaluation, are handled by returning NotImplemented error for now.
Are there any user-facing changes?
Queries with IN (subquery) predicates start executing for described above filter combinations