Works for cross to inner join

[yuuch]

Which issue does this PR close?

Closes#3840
Closes#3554.

Rationale for this change

I found that the rule has been implemented in PR #3482 , but the guy forget to remove some codes in planners.

What changes are included in this PR?

I just keep cross join in the planner.rs and PR #3482 will do crossJoin to InnerJoin Optimize.

Are there any user-facing changes?

[alamb]

Thank you @yuuch 😍 -- Love the cleanup!

It looks like one of the tests is failing -- can you please update the plan?

https://github.com/apache/arrow-datafusion/actions/runs/3352036848/jobs/5558342674

---- sql::subqueries::tpch_q2_correlated stdout ----
thread 'sql::subqueries::tpch_q2_correlated' panicked at 'assertion failed: `(left == right)`
left: `"Sort: supplier.s_acctbal DESC NULLS FIRST, nation.n_name ASC NULLS LAST, supplier.s_name ASC NULLS LAST, part.p_partkey ASC NULLS LAST\n Projection: supplier.s_acctbal, supplier.s_name, nation.n_name, part.p_partkey, part.p_mfgr, supplier.s_address, supplier.s_phone, supplier.s_comment\n Inner Join: part.p_partkey = __sq_1.ps_partkey, partsupp.ps_supplycost = __sq_1.__value\n Inner Join: nation.n_regionkey = region.r_regionkey\n Inner Join: supplier.s_nationkey = nation.n_nationkey\n Inner Join: part.p_partkey = partsupp.ps_partkey, supplier.s_suppkey = partsupp.ps_suppkey\n CrossJoin:\n Filter: part.p_size = Int32(15) AND part.p_type LIKE Utf8(\"%BRASS\")\n TableScan: part projection=[p_partkey, p_mfgr, p_type, p_size], partial_filters=[part.p_size = Int32(15), part.p_type LIKE Utf8(\"%BRASS\")]\n TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]\n TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]\n TableScan: nation projection=[n_nationkey, n_name, n_regionkey]\n Filter: region.r_name = Utf8(\"EUROPE\")\n TableScan: region projection=[r_regionkey, r_name], partial_filters=[region.r_name = Utf8(\"EUROPE\")]\n Projection: partsupp.ps_partkey, MIN(partsupp.ps_supplycost) AS __value, alias=__sq_1\n Aggregate: groupBy=[[partsupp.ps_partkey]], aggr=[[MIN(partsupp.ps_supplycost)]]\n Inner Join: nation.n_regionkey = region.r_regionkey\n Inner Join: supplier.s_nationkey = nation.n_nationkey\n Inner Join: partsupp.ps_suppkey = supplier.s_suppkey\n TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]\n TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]\n TableScan: nation projection=[n_nationkey, n_name, n_regionkey]\n Filter: region.r_name = Utf8(\"EUROPE\")\n TableScan: region projection=[r_regionkey, r_name], partial_filters=[region.r_name = Utf8(\"EUROPE\")]"`,
right: `"Sort: supplier.s_acctbal DESC NULLS FIRST, nation.n_name ASC NULLS LAST, supplier.s_name ASC NULLS LAST, part.p_partkey ASC NULLS LAST\n Projection: supplier.s_acctbal, supplier.s_name, nation.n_name, part.p_partkey, part.p_mfgr, supplier.s_address, supplier.s_phone, supplier.s_comment\n Inner Join: part.p_partkey = __sq_1.ps_partkey, partsupp.ps_supplycost = __sq_1.__value\n Inner Join: nation.n_regionkey = region.r_regionkey\n Inner Join: supplier.s_nationkey = nation.n_nationkey\n Inner Join: partsupp.ps_suppkey = supplier.s_suppkey\n Inner Join: part.p_partkey = partsupp.ps_partkey\n Filter: part.p_size = Int32(15) AND part.p_type LIKE Utf8(\"%BRASS\")\n TableScan: part projection=[p_partkey, p_mfgr, p_type, p_size], partial_filters=[part.p_size = Int32(15), part.p_type LIKE Utf8(\"%BRASS\")]\n TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]\n TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]\n TableScan: nation projection=[n_nationkey, n_name, n_regionkey]\n Filter: region.r_name = Utf8(\"EUROPE\")\n TableScan: region projection=[r_regionkey, r_name], partial_filters=[region.r_name = Utf8(\"EUROPE\")]\n Projection: partsupp.ps_partkey, MIN(partsupp.ps_supplycost) AS __value, alias=__sq_1\n Aggregate: groupBy=[[partsupp.ps_partkey]], aggr=[[MIN(partsupp.ps_supplycost)]]\n Inner Join: nation.n_regionkey = region.r_regionkey\n Inner Join: supplier.s_nationkey = nation.n_nationkey\n Inner Join: partsupp.ps_suppkey = supplier.s_suppkey\n TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]\n TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]\n TableScan: nation projection=[n_nationkey, n_name, n_regionkey]\n Filter: region.r_name = Utf8(\"EUROPE\")\n TableScan: region projection=[r_regionkey, r_name], partial_filters=[region.r_name = Utf8(\"EUROPE\")]"`', datafusion/core/tests/sql/subqueries.rs:167:5

[alamb]

Marking as a draft to get it off the review queue -- please fix CI and mark as ready for review. If you need help let me know. I love this type of cleanup 👍

[Dandandan]

Looking great.
FYI @yuuch I am getting this updated and fix test(s).

[Dandandan]

Some more plans are also failing (as more of the plans include cross join now).
This requires some more work to update the reduce_cross_join code.

[Dandandan]

FYI @xudong963@DhamoPS as you worked on this before

[Dandandan]

I believe the difference is: the earlier implementation looked at all of the plans in the from clause when selecting a join, while the optimization rule processes the tree in order and only replaces a cross join with an inner join when it's possible.

Some transformations from cross join to inner join however need reordering of the joins within the tree in order for this to be possible.

[mingmwang]

I think @jackwener 's implementation is more general and we should move the logic from planner to rule.
#4185