Add rule to reimplement Eliminate cross join and remove it in planner (#4185)

* reimplement eliminate_cross_join

* add test for subquery alias and projection alias.

* add test

* fix fmt

* review

* fmt

* fix conflict

Author: jackwener

benchmarks/expected-plans/q2.txt

@@ -1,24 +1,25 @@
 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
   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
-    Inner Join: part.p_partkey = __sq_1.ps_partkey, partsupp.ps_supplycost = __sq_1.__value
-      Inner Join: nation.n_regionkey = region.r_regionkey
-        Inner Join: supplier.s_nationkey = nation.n_nationkey
-          Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
-            Inner Join: part.p_partkey = partsupp.ps_partkey
-              Filter: part.p_size = Int32(15) AND part.p_type LIKE Utf8("%BRASS")
-                TableScan: part projection=[p_partkey, p_mfgr, p_type, p_size]
-              TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]
-            TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]
-          TableScan: nation projection=[n_nationkey, n_name, n_regionkey]
-        Filter: region.r_name = Utf8("EUROPE")
-          TableScan: region projection=[r_regionkey, r_name]
-      Projection: partsupp.ps_partkey, MIN(partsupp.ps_supplycost) AS __value, alias=__sq_1
-        Aggregate: groupBy=[[partsupp.ps_partkey]], aggr=[[MIN(partsupp.ps_supplycost)]]
-          Inner Join: nation.n_regionkey = region.r_regionkey
-            Inner Join: supplier.s_nationkey = nation.n_nationkey
-              Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
+    Projection: part.p_partkey, part.p_mfgr, supplier.s_name, supplier.s_address, supplier.s_phone, supplier.s_acctbal, supplier.s_comment, nation.n_name
+      Inner Join: part.p_partkey = __sq_1.ps_partkey, partsupp.ps_supplycost = __sq_1.__value
+        Inner Join: nation.n_regionkey = region.r_regionkey
+          Inner Join: supplier.s_nationkey = nation.n_nationkey
+            Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
+              Inner Join: part.p_partkey = partsupp.ps_partkey
+                Filter: part.p_size = Int32(15) AND part.p_type LIKE Utf8("%BRASS")
+                  TableScan: part projection=[p_partkey, p_mfgr, p_type, p_size]
                 TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]
-                TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]
-              TableScan: nation projection=[n_nationkey, n_name, n_regionkey]
-            Filter: region.r_name = Utf8("EUROPE")
-              TableScan: region projection=[r_regionkey, r_name]
+              TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]
+            TableScan: nation projection=[n_nationkey, n_name, n_regionkey]
+          Filter: region.r_name = Utf8("EUROPE")
+            TableScan: region projection=[r_regionkey, r_name]
+        Projection: partsupp.ps_partkey, MIN(partsupp.ps_supplycost) AS __value, alias=__sq_1
+          Aggregate: groupBy=[[partsupp.ps_partkey]], aggr=[[MIN(partsupp.ps_supplycost)]]
+            Inner Join: nation.n_regionkey = region.r_regionkey
+              Inner Join: supplier.s_nationkey = nation.n_nationkey
+                Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
+                  TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]
+                  TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]
+                TableScan: nation projection=[n_nationkey, n_name, n_regionkey]
+              Filter: region.r_name = Utf8("EUROPE")
+                TableScan: region projection=[r_regionkey, r_name]

benchmarks/expected-plans/q8.txt

@@ -3,23 +3,24 @@ Sort: all_nations.o_year ASC NULLS LAST
     Aggregate: groupBy=[[all_nations.o_year]], aggr=[[SUM(CASE WHEN all_nations.nation = Utf8("BRAZIL") THEN all_nations.volume ELSE Decimal128(Some(0),38,4) END) AS SUM(CASE WHEN all_nations.nation = Utf8("BRAZIL") THEN all_nations.volume ELSE Int64(0) END), SUM(all_nations.volume)]]
       Projection: o_year, volume, nation, alias=all_nations
         Projection: datepart(Utf8("YEAR"), orders.o_orderdate) AS o_year, CAST(lineitem.l_extendedprice AS Decimal128(38, 4)) * CAST(Decimal128(Some(100),23,2) - CAST(lineitem.l_discount AS Decimal128(23, 2)) AS Decimal128(38, 4)) AS volume, n2.n_name AS nation
-          Inner Join: n1.n_regionkey = region.r_regionkey
-            Inner Join: supplier.s_nationkey = n2.n_nationkey
-              Inner Join: customer.c_nationkey = n1.n_nationkey
-                Inner Join: orders.o_custkey = customer.c_custkey
-                  Inner Join: lineitem.l_orderkey = orders.o_orderkey
-                    Inner Join: lineitem.l_suppkey = supplier.s_suppkey
-                      Inner Join: part.p_partkey = lineitem.l_partkey
-                        Filter: part.p_type = Utf8("ECONOMY ANODIZED STEEL")
-                          TableScan: part projection=[p_partkey, p_type]
-                        TableScan: lineitem projection=[l_orderkey, l_partkey, l_suppkey, l_extendedprice, l_discount]
-                      TableScan: supplier projection=[s_suppkey, s_nationkey]
-                    Filter: orders.o_orderdate >= Date32("9131") AND orders.o_orderdate <= Date32("9861")
-                      TableScan: orders projection=[o_orderkey, o_custkey, o_orderdate]
-                  TableScan: customer projection=[c_custkey, c_nationkey]
-                SubqueryAlias: n1
-                  TableScan: nation projection=[n_nationkey, n_regionkey]
-              SubqueryAlias: n2
-                TableScan: nation projection=[n_nationkey, n_name]
-            Filter: region.r_name = Utf8("AMERICA")
-              TableScan: region projection=[r_regionkey, r_name]
+          Projection: lineitem.l_extendedprice, lineitem.l_discount, orders.o_orderdate, n2.n_name
+            Inner Join: n1.n_regionkey = region.r_regionkey
+              Inner Join: supplier.s_nationkey = n2.n_nationkey
+                Inner Join: customer.c_nationkey = n1.n_nationkey
+                  Inner Join: orders.o_custkey = customer.c_custkey
+                    Inner Join: lineitem.l_orderkey = orders.o_orderkey
+                      Inner Join: lineitem.l_suppkey = supplier.s_suppkey
+                        Inner Join: part.p_partkey = lineitem.l_partkey
+                          Filter: part.p_type = Utf8("ECONOMY ANODIZED STEEL")
+                            TableScan: part projection=[p_partkey, p_type]
+                          TableScan: lineitem projection=[l_orderkey, l_partkey, l_suppkey, l_extendedprice, l_discount]
+                        TableScan: supplier projection=[s_suppkey, s_nationkey]
+                      Filter: orders.o_orderdate >= Date32("9131") AND orders.o_orderdate <= Date32("9861")
+                        TableScan: orders projection=[o_orderkey, o_custkey, o_orderdate]
+                    TableScan: customer projection=[c_custkey, c_nationkey]
+                  SubqueryAlias: n1
+                    TableScan: nation projection=[n_nationkey, n_regionkey]
+                SubqueryAlias: n2
+                  TableScan: nation projection=[n_nationkey, n_name]
+              Filter: region.r_name = Utf8("AMERICA")
+                TableScan: region projection=[r_regionkey, r_name]

benchmarks/expected-plans/q9.txt

@@ -3,15 +3,16 @@ Sort: profit.nation ASC NULLS LAST, profit.o_year DESC NULLS FIRST
     Aggregate: groupBy=[[profit.nation, profit.o_year]], aggr=[[SUM(profit.amount)]]
       Projection: nation, o_year, amount, alias=profit
         Projection: nation.n_name AS nation, datepart(Utf8("YEAR"), orders.o_orderdate) AS o_year, CAST(lineitem.l_extendedprice AS Decimal128(38, 4)) * CAST(Decimal128(Some(100),23,2) - CAST(lineitem.l_discount AS Decimal128(23, 2)) AS Decimal128(38, 4)) - CAST(partsupp.ps_supplycost * lineitem.l_quantity AS Decimal128(38, 4)) AS amount
-          Inner Join: supplier.s_nationkey = nation.n_nationkey
-            Inner Join: lineitem.l_orderkey = orders.o_orderkey
-              Inner Join: lineitem.l_suppkey = partsupp.ps_suppkey, lineitem.l_partkey = partsupp.ps_partkey
-                Inner Join: lineitem.l_suppkey = supplier.s_suppkey
-                  Inner Join: part.p_partkey = lineitem.l_partkey
-                    Filter: part.p_name LIKE Utf8("%green%")
-                      TableScan: part projection=[p_partkey, p_name]
-                    TableScan: lineitem projection=[l_orderkey, l_partkey, l_suppkey, l_quantity, l_extendedprice, l_discount]
-                  TableScan: supplier projection=[s_suppkey, s_nationkey]
-                TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]
-              TableScan: orders projection=[o_orderkey, o_orderdate]
-            TableScan: nation projection=[n_nationkey, n_name]
+          Projection: lineitem.l_quantity, lineitem.l_extendedprice, lineitem.l_discount, partsupp.ps_supplycost, orders.o_orderdate, nation.n_name
+            Inner Join: supplier.s_nationkey = nation.n_nationkey
+              Inner Join: lineitem.l_orderkey = orders.o_orderkey
+                Inner Join: lineitem.l_suppkey = partsupp.ps_suppkey, lineitem.l_partkey = partsupp.ps_partkey
+                  Inner Join: lineitem.l_suppkey = supplier.s_suppkey
+                    Inner Join: part.p_partkey = lineitem.l_partkey
+                      Filter: part.p_name LIKE Utf8("%green%")
+                        TableScan: part projection=[p_partkey, p_name]
+                      TableScan: lineitem projection=[l_orderkey, l_partkey, l_suppkey, l_quantity, l_extendedprice, l_discount]
+                    TableScan: supplier projection=[s_suppkey, s_nationkey]
+                  TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]
+                TableScan: orders projection=[o_orderkey, o_orderdate]
+              TableScan: nation projection=[n_nationkey, n_name]

datafusion/core/tests/sql/subqueries.rs

@@ -141,28 +141,29 @@ order by s_acctbal desc, n_name, s_name, p_partkey;"#;
     let actual = format!("{}", plan.display_indent());
     let expected = r#"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
   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
-    Inner Join: part.p_partkey = __sq_1.ps_partkey, partsupp.ps_supplycost = __sq_1.__value
-      Inner Join: nation.n_regionkey = region.r_regionkey
-        Inner Join: supplier.s_nationkey = nation.n_nationkey
-          Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
-            Inner Join: part.p_partkey = partsupp.ps_partkey
-              Filter: part.p_size = Int32(15) AND part.p_type LIKE Utf8("%BRASS")
-                TableScan: part projection=[p_partkey, p_mfgr, p_type, p_size], partial_filters=[part.p_size = Int32(15), part.p_type LIKE Utf8("%BRASS")]
-              TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]
-            TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]
-          TableScan: nation projection=[n_nationkey, n_name, n_regionkey]
-        Filter: region.r_name = Utf8("EUROPE")
-          TableScan: region projection=[r_regionkey, r_name], partial_filters=[region.r_name = Utf8("EUROPE")]
-      Projection: partsupp.ps_partkey, MIN(partsupp.ps_supplycost) AS __value, alias=__sq_1
-        Aggregate: groupBy=[[partsupp.ps_partkey]], aggr=[[MIN(partsupp.ps_supplycost)]]
-          Inner Join: nation.n_regionkey = region.r_regionkey
-            Inner Join: supplier.s_nationkey = nation.n_nationkey
-              Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
+    Projection: part.p_partkey, part.p_mfgr, supplier.s_name, supplier.s_address, supplier.s_phone, supplier.s_acctbal, supplier.s_comment, nation.n_name
+      Inner Join: part.p_partkey = __sq_1.ps_partkey, partsupp.ps_supplycost = __sq_1.__value
+        Inner Join: nation.n_regionkey = region.r_regionkey
+          Inner Join: supplier.s_nationkey = nation.n_nationkey
+            Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
+              Inner Join: part.p_partkey = partsupp.ps_partkey
+                Filter: part.p_size = Int32(15) AND part.p_type LIKE Utf8("%BRASS")
+                  TableScan: part projection=[p_partkey, p_mfgr, p_type, p_size], partial_filters=[part.p_size = Int32(15), part.p_type LIKE Utf8("%BRASS")]
                 TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]
-                TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]
-              TableScan: nation projection=[n_nationkey, n_name, n_regionkey]
-            Filter: region.r_name = Utf8("EUROPE")
-              TableScan: region projection=[r_regionkey, r_name], partial_filters=[region.r_name = Utf8("EUROPE")]"#
+              TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]
+            TableScan: nation projection=[n_nationkey, n_name, n_regionkey]
+          Filter: region.r_name = Utf8("EUROPE")
+            TableScan: region projection=[r_regionkey, r_name], partial_filters=[region.r_name = Utf8("EUROPE")]
+        Projection: partsupp.ps_partkey, MIN(partsupp.ps_supplycost) AS __value, alias=__sq_1
+          Aggregate: groupBy=[[partsupp.ps_partkey]], aggr=[[MIN(partsupp.ps_supplycost)]]
+            Inner Join: nation.n_regionkey = region.r_regionkey
+              Inner Join: supplier.s_nationkey = nation.n_nationkey
+                Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
+                  TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost]
+                  TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment]
+                TableScan: nation projection=[n_nationkey, n_name, n_regionkey]
+              Filter: region.r_name = Utf8("EUROPE")
+                TableScan: region projection=[r_regionkey, r_name], partial_filters=[region.r_name = Utf8("EUROPE")]"#
         .to_string();
     assert_eq!(actual, expected);