support simple/cross lateral joins

[skyzh]

Which issue does this PR close?

• part of Feature request: Support for lateral joins #10048

Rationale for this change

Add partial lateral join support.

What changes are included in this PR?

The cross lateral join logical plan is like:

CrossJoin
  Left
  Right (with outer column reference)

We added a new rule DecorrelateLateralJoin to handle this case. We reuse the scalar/predicate subquery handling code for decorrelating lateral joins.

Are these changes tested?

New test case in sqllogictest.

Are there any user-facing changes?

Some lateral joins are now supported. Note that we still have some of them in joins.slt that cannot be planned, and can be fixed gradually.

[skyzh]

cc @alamb would you please help take a look as you were engaged on lateral join discussions before? Thanks :)

[skyzh]

should pass all sqllogictests :)

[alamb]

Thank you very much for starting this work @skyzh 🙏

Do you know of a reference / description somewhere for how to plan LATERAL joins? I am not familar with their semantics nor how we can rewrite them to subqueries.

My one smoke test was to get an explain plan in DuckDB and it produced a plan with 2 joins which seems diffrent than what this PR does 🤔

I also left some code suggestions, which are minor compared to the plan

Basically I don't know what the plans for such queries should look like so I am not sure if this code does the right thing

[alamb]

I think you can do this more simply using apply_subqueries as described here: https://datafusion.apache.org/library-user-guide/query-optimizer.html#recursively-walk-an-expression-tree

Something like (untested):

fn plan_contains_outer_reference(plan: &LogicalPlan) -> bool {
  let mut contains = false;
  plan.apply_subqueries(|plan| {
            if plan.contains_outer_reference() {
                self.contains = true;
                Ok(TreeNodeRecursion::Stop)
            } else {
                Ok(TreeNodeRecursion::Continue)
            }
  }).unwrap();
  contains
}

[skyzh]

This is not equivalent to the current visitor unfortunately because apply_subqueries only visits the subqueries 🤣

[peter-toth]

You can use apply_with_subqueries().

[alamb]

I believe you can cause rewrite to be called in a bottom up fashion by using:
https://docs.rs/datafusion/latest/datafusion/optimizer/trait.OptimizerRule.html#method.apply_order

Then you could maybe add a contains_outer_reference type flag to DecorrelateLateralJoin

[skyzh]

24a375d I decided to use TreeNodeRecursion::Jump to skip all children to make it O(n).

[alamb]

Thanks!

Could we please add some negative cases to these tests too such as:

1. When there are two subqueries
2. When the subqueries have no correlation

[skyzh]

only added (2) in c353246, I have some test cases for (1) in the comments but I don't plan to support them right now.

[skyzh]

Okay, I just realized that the plan is not correct. I'll need to do more transformation to make it work :(

So back to the semantics of the lateral join,

SELECT *
  FROM t0,
  LATERAL (SELECT sum(v1) FROM t1 WHERE t0.v0 = t1.v0);

is basically saying that "for every row in t0, run the lateral query", which is equivalent to the following program:

for row in t0:
  lateral_result = run_query(SELECT sum(v1) FROM t1 WHERE t0.v0 = t1.v0)
  for row2 in lateral_result:
    output.append(row ++ row2) 

Note that sum will produce 1 row of null even when the right table doesn't contain any matching row.

The produced plan with this patch is:

Inner Join t0.v0 = t1.v0
  Scan t0
  Aggregate: groupBy=[[t1.v0]], aggr=[[sum(t1.v1)]]
    Scan t1

If t0 = (0, 0), (1, 1) and t1 = (0, 0), the output should have 2 rows, while this plan will only produce 1 row.

The correct plan should be:

Left Outer Join t0.v0 = t1.v0
  Scan t0
  Aggregate: groupBy=[[t1.v0]], aggr=[[sum(t1.v1)]]
    Scan t1

And aggregations like count(*) needs to be handled separately, which I already see some code on that in the codebase.

Back to the DuckDB plan. They are using the Hyper subquery unnesting. The idea of Hyper is that it will first eliminate duplicates on the join column. Therefore, in the case where the left table has a lot of duplicated rows, it only needs to compute the subquery once for all such rows (if it turns out to be a nested loop join). So what they are doing is that: eliminate the duplicates from the left table, then compute the subquery result, and finally join it back to the original one, which will be planned like:

^1:
Left Outer Join t0.v0 = t1.v0
  DistinctScan/DelimScan t0
  Aggregate: groupBy=[[t1.v0]], aggr=[[sum(t1.v1)]]
    Scan t1

which computes the right side row for every value in the left table. We will get (distinct t0.*, sum(t1.v1)) at this step. And then, deduplicate the result of the subquery (this step seems unnecessary?)

^2:
Dedup (HashJoin)
  ^1

And finally, join the result back to the original table,

HashJoin
  Scan t0
  ^2

The result should be the same as the corrected plan I mentioned before.

[alamb]

So back to the semantics of the lateral join,

Thank you @skyzh -- this is an amazing description. THank you -- hopefully we can include that in the source code eventually once we work through this issue

Ideally in my mind when someone found the lateral join support in DataFusion they would also have enough background via comments / links to understand what was going on.

[skyzh]

The current one-pass subquery unnesting implementation in PullUpCorrelatedExpr does not handle cases such as when we have joins in subqueries:

SELECT * FROM t0, LATERAL (SELECT 1), LATERAL (SELECT * FROM t1 WHERE t0.v0 = t1.v0);

Just note it down here and we can improve it in the future. We could implement a rule-based unnesting to push down the lateral join operator.

[skyzh]

For multiple subqueries like:

SELECT * FROM t0, LATERAL (SELECT * FROM t1 WHERE t0.v0 = t1.v0 AND t1.v1 > (SELECT SUM(v1) FROM t2 WHERE t1.v0 = t2.v0));

I'm not planning to support it in this patch now as I don't think the current one-pass de-correlate process can handle it correctly without significant refactor :(

[skyzh]

I think this patch will not regress existing supported cases as they get fully decorrelated with the scalar query rule and the predicate subquery rule. The patch unlocks a new set of plans produced by lateral joins, though it is possible that we could have missed some cases and generated incorrect plans for those newly-supported one. For queries that can be optimized into a physical plan: count(*) is not handled correctly. For queries they we know we don't support for now: as in the above comments.

[skyzh]

And ready for review again :)

After trying understanding what's happening in decorrelate.rs, I think we need new code path to support a variety of logical plans produced by lateral joins. The key is that we should make the decorrelation code aware of the joins while de-correlating, instead of first gathering the information and then generate a join at the top.

If we are going towards making the optimizer able to unnest any subqueries and lateral joins, then we will likely have a meta rule that recursively apply the following rules top-down:

• Convert join operators into LogicalApply if the right side of the join contains outer column reference. This can also be done in the SQL->logical phase when we encounter a lateral join. For the correlated filter predicate and scalar subqueries (exists/in), we can also convert them into the apply operator in the future.
• Have a set of rules like: push down apply->join, push down apply->aggregation, push down apply->filter, etc.
• Apply these rules top-down until no outer column reference is in the plan tree.

We can either use the Hyper unnesting rules (we implemented it in CMU-DB's optd optimizer) or the SQL server unnesting rules (which we've implemented in risinglight).

This meta unnesting rule is more powerful than what we have right now (decorrelate predicate subquery + scalar subquery unnesting rule) and we can eventually replace these two rules with the new meta unnesting rule in the future.

[skyzh]

And FYI, the Hyper paper https://dl.gi.de/server/api/core/bitstreams/10f3ad20-e9ae-4876-abdf-5c8e83e4c595/content, and the SQL server paper https://www.microsoft.com/en-us/research/wp-content/uploads/2016/02/tr-2000-31.pdf