feat: rewrite subquery into dependent join logical plan

[duongcongtoai]

Which issue does this PR close?

This PR is a part of a long story for a general purpose subquery decorrelation, after some discussion in #5492, this PR proposes adding the followings:

• an optimizor that convert all the subqueries into dependent join logical plan (this is only a temporary plan) <---- This is what this PR trying to achieve
• an optimizor to decorrelate dependent join logical plan, POC is in this working branch

To avoid breaking existing tests and smoother collaboration, the changes should happen in the following sequence

1. Merge item 1 without integrate the new optimizor to the main flow (behavior is tested in-code instead of sqllogictests)
2. Start implement more rewriting rules for different query plan (aggregate, projection, filter ...) using the a new optimizer
We keep the working of this new optimizor in a working branch, and if the implementation can fully support existing subqueries,
we make the following change to the main branch

From

            Arc::new(DecorrelatePredicateSubquery::new()),
            Arc::new(ScalarSubqueryToJoin::new()),
            Arc::new(DecorrelateLateralJoin::new()),

Into

            Arc::new(DependentJoinRewriter::new()),
            Arc::new(DependentJoinDecorrelator::new()),

Or we can even combine the 2 optimizors into one, into

            Arc::new(DependentJoinDecorrelator::new()),

The following works are needed after merging this PR

• Implement DelimGet logical plan and physical plan
• Implement DelimGetRemove optimizor similar to DuckDB
• Implement JoinType::Single similar to duckdb. This operator is needed to support this type of query

select * from outer where outer.a = (select inner.b from inner where inner.c=outer.c)

Rationale for this change

Are these changes tested?

Are there any user-facing changes?

[alamb]

FYI @irenjj

[irenjj]

Thanks @duongcongtoai , Here are some of my review comments. There are a few things I haven't fully figured out yet and will need some more time.

[irenjj]

Maybe we can add an empty json!() here.

[irenjj]

Ditto

[duongcongtoai]

not sure what you mean, but i also updated the implementation here

[irenjj]

It seems that break may prevent mark_outer_column_access.🤔

[duongcongtoai]

right, i didn't test this logic, so didn't catch this issue 👍

[duongcongtoai]

done, problem addressed, i also handle f_up for the case of projecion, (also added a test case)

[irenjj]

If it's a dependent join, the depth should be incremented by 1, right?

[duongcongtoai]

the depth is not incremented until this match block finishes, and function predicate.apply will not traverse down to the expr under a Subquery expr, for example:
Given a predicate: where col_level1=1 and col_level2=(select count(*) from inner where inner_col=col_level1)
This function only visits:

• col_level1=1, col_level1, 1
• col_level2=ScalarSubqueryExpr, col_level2, ScalarSubqueryExpr

It will not visit the expr inner_col=col_level1. This happen when f_down function visits the children of this node (at that time depth has already been incremented)

[irenjj]

I think more considerations are needed for the subquery case, such as OuterRefColumn, depth, etc. How does DuckDB handle this?

[duongcongtoai]

we don't do much for LogicalPlan::Subquery here, because it contains little information of the OuterRefColumn or depth (notice how i create an empty entry for this node for the parent dependent_join_node

                parent_node
                    .columns_accesses_by_subquery_id
                    .insert(new_id, vec![]);

The same entry will be updated with OuterRefColumn and depth should any such expr is found during the downard traversal (f_down). I think the diagram mentioned here describe how the traversal works

                  ↓1   
                  ↑12   
             ┌────────────┐
             │  FILTER    │<--- DependentJoin rewrite
             │   (1)      │     happens here
             └─────┬────┬─┘     Here we already have enough information
             |     |    |       of which node is accessing which column
             |     |    |       provided by "Table Scan t1" node        
             │     |    |       (for example node (6) below )           
             │     |    |       
             │     |    |       
             │     |    |
       ↓2────┘     ↓6   └────↓10
       ↑5          ↑11         ↑11
   ┌───▼───┐    ┌──▼───┐   ┌───▼───────┐
   │SUBQ1  │    │SUBQ2 │   │TABLE SCAN │
   └──┬────┘    └──┬───┘   │    t1     │
      |            |       └───────────┘
      |            |
      |            |
      |            ↓7
      |            ↑10
      |         ┌──▼───────┐
      |         │Filter    │----> mark_outer_column_access(outer_ref)
      |         │outer_ref |
      |         │ (6)      |
      |         └──┬───────┘  
      |            |
     ↓3           ↓8
     ↑4           ↑9
  ┌──▼────┐    ┌──▼────┐
  │SCAN t2│    │SCAN t2│
  └───────┘    └───────┘

[duongcongtoai]

I can relate this section to a codeblock in duckdb: https://github.com/duckdb/duckdb/blob/95d71ea4ec0fb3a151b18943fb532f7b5131dc95/src/planner/binder/query_node/plan_subquery.cpp#L294

unique_ptr<Expression> Binder::PlanSubquery(BoundSubqueryExpression &expr, unique_ptr<LogicalOperator> &root) {
	...
	auto plan = std::move(subquery_root);
	unique_ptr<Expression> result_expression;
	if (!expr.IsCorrelated()) {
		result_expression = PlanUncorrelatedSubquery(*this, expr, root, std::move(plan));
	} else {
		result_expression = PlanCorrelatedSubquery(*this, expr, root, std::move(plan));
	}
	// finally, we recursively plan the nested subqueries (if there are any)
	if (sub_binder->has_unplanned_dependent_joins) {
		RecursiveDependentJoinPlanner plan(*this);
		plan.VisitOperator(*root);
	}

Let's use the SUBQ2 on the above diagram as demonstration:
for duckdb, plan=std::move(subquery_root) is equivalent to node (6), and root is equivalent to node (1) (I think the usage of root is quite misleading, it should be named "subquery_parent" instead, because it is used to reference the direct parent node of subquery instead of the "root" of the whole query plan.

what PlanCorrelatedSubquery does is convert node (6) into DelimJoin given all the information of correlated_columns.

static unique_ptr<Expression> PlanCorrelatedSubquery(Binder &binder, BoundSubqueryExpression &expr,
                                                     unique_ptr<LogicalOperator> &root,
                                                     unique_ptr<LogicalOperator> plan) {
	auto &correlated_columns = expr.binder->correlated_columns;
	// FIXME: there should be a way of disabling decorrelation for ANY queries as well, but not for now...
	bool perform_delim =
	    expr.subquery_type == SubqueryType::ANY ? true : PerformDuplicateElimination(binder, correlated_columns);

however, in our context, because we does not have such information provided by the planner, we have to perform an initial traversal to get them.

After we are done with the traversal, we have a tree which looks exactly like what duckdb has, and the recursion can imitate the same logic

[irenjj]

nit: unwrap() is not allowed

[irenjj]

Maybe the shared parts among rewrite_* functions can be extracted