feature: reimplement push_down_projection and prune_column.

Author: jackwener

benchmarks/expected-plans/q10.txt

@@ -1,12 +1,17 @@
 Sort: revenue DESC NULLS FIRST
   Projection: customer.c_custkey, customer.c_name, SUM(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount) AS revenue, customer.c_acctbal, nation.n_name, customer.c_address, customer.c_phone, customer.c_comment
     Aggregate: groupBy=[[customer.c_custkey, customer.c_name, customer.c_acctbal, customer.c_phone, nation.n_name, customer.c_address, customer.c_comment]], aggr=[[SUM(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 SUM(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]]
-      Inner Join: customer.c_nationkey = nation.n_nationkey
-        Inner Join: orders.o_orderkey = lineitem.l_orderkey
-          Inner Join: customer.c_custkey = orders.o_custkey
-            TableScan: customer projection=[c_custkey, c_name, c_address, c_nationkey, c_phone, c_acctbal, c_comment]
-            Filter: orders.o_orderdate >= Date32("8674") AND orders.o_orderdate < Date32("8766")
-              TableScan: orders projection=[o_orderkey, o_custkey, o_orderdate]
-          Filter: lineitem.l_returnflag = Utf8("R")
-            TableScan: lineitem projection=[l_orderkey, l_extendedprice, l_discount, l_returnflag]
-        TableScan: nation projection=[n_nationkey, n_name]
+      Projection: customer.c_custkey, customer.c_name, customer.c_address, customer.c_phone, customer.c_acctbal, customer.c_comment, lineitem.l_extendedprice, lineitem.l_discount, nation.n_name
+        Inner Join: customer.c_nationkey = nation.n_nationkey
+          Projection: customer.c_custkey, customer.c_name, customer.c_address, customer.c_nationkey, customer.c_phone, customer.c_acctbal, customer.c_comment, lineitem.l_extendedprice, lineitem.l_discount
+            Inner Join: orders.o_orderkey = lineitem.l_orderkey
+              Projection: customer.c_custkey, customer.c_name, customer.c_address, customer.c_nationkey, customer.c_phone, customer.c_acctbal, customer.c_comment, orders.o_orderkey
+                Inner Join: customer.c_custkey = orders.o_custkey
+                  TableScan: customer projection=[c_custkey, c_name, c_address, c_nationkey, c_phone, c_acctbal, c_comment]
+                  Projection: orders.o_orderkey, orders.o_custkey
+                    Filter: orders.o_orderdate >= Date32("8674") AND orders.o_orderdate < Date32("8766")
+                      TableScan: orders projection=[o_orderkey, o_custkey, o_orderdate]
+              Projection: lineitem.l_orderkey, lineitem.l_extendedprice, lineitem.l_discount
+                Filter: lineitem.l_returnflag = Utf8("R")
+                  TableScan: lineitem projection=[l_orderkey, l_extendedprice, l_discount, l_returnflag]
+          TableScan: nation projection=[n_nationkey, n_name]

benchmarks/expected-plans/q11.txt

@@ -3,18 +3,24 @@ Sort: value DESC NULLS FIRST
     Filter: CAST(SUM(partsupp.ps_supplycost * partsupp.ps_availqty) AS Decimal128(38, 15)) > CAST(__scalar_sq_1.__value AS Decimal128(38, 15))
       CrossJoin:
         Aggregate: groupBy=[[partsupp.ps_partkey]], aggr=[[SUM(CAST(partsupp.ps_supplycost AS Decimal128(26, 2)) * CAST(partsupp.ps_availqty AS Decimal128(26, 2)))]]
-          Inner Join: supplier.s_nationkey = nation.n_nationkey
-            Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
-              TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_availqty, ps_supplycost]
-              TableScan: supplier projection=[s_suppkey, s_nationkey]
-            Filter: nation.n_name = Utf8("GERMANY")
-              TableScan: nation projection=[n_nationkey, n_name]
-        SubqueryAlias: __scalar_sq_1
-          Projection: CAST(SUM(partsupp.ps_supplycost * partsupp.ps_availqty) AS Float64) * Float64(0.0001) AS __value
-            Aggregate: groupBy=[[]], aggr=[[SUM(CAST(partsupp.ps_supplycost AS Decimal128(26, 2)) * CAST(partsupp.ps_availqty AS Decimal128(26, 2)))]]
-              Inner Join: supplier.s_nationkey = nation.n_nationkey
+          Projection: partsupp.ps_partkey, partsupp.ps_availqty, partsupp.ps_supplycost
+            Inner Join: supplier.s_nationkey = nation.n_nationkey
+              Projection: partsupp.ps_partkey, partsupp.ps_availqty, partsupp.ps_supplycost, supplier.s_nationkey
                 Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
-                  TableScan: partsupp projection=[ps_suppkey, ps_availqty, ps_supplycost]
+                  TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_availqty, ps_supplycost]
                   TableScan: supplier projection=[s_suppkey, s_nationkey]
+              Projection: nation.n_nationkey
                 Filter: nation.n_name = Utf8("GERMANY")
-                  TableScan: nation projection=[n_nationkey, n_name]
+                  TableScan: nation projection=[n_nationkey, n_name]
+        SubqueryAlias: __scalar_sq_1
+          Projection: CAST(SUM(partsupp.ps_supplycost * partsupp.ps_availqty) AS Float64) * Float64(0.0001) AS __value
+            Aggregate: groupBy=[[]], aggr=[[SUM(CAST(partsupp.ps_supplycost AS Decimal128(26, 2)) * CAST(partsupp.ps_availqty AS Decimal128(26, 2)))]]
+              Projection: partsupp.ps_availqty, partsupp.ps_supplycost
+                Inner Join: supplier.s_nationkey = nation.n_nationkey
+                  Projection: partsupp.ps_availqty, partsupp.ps_supplycost, supplier.s_nationkey
+                    Inner Join: partsupp.ps_suppkey = supplier.s_suppkey
+                      TableScan: partsupp projection=[ps_suppkey, ps_availqty, ps_supplycost]
+                      TableScan: supplier projection=[s_suppkey, s_nationkey]
+                  Projection: nation.n_nationkey
+                    Filter: nation.n_name = Utf8("GERMANY")
+                      TableScan: nation projection=[n_nationkey, n_name]

benchmarks/expected-plans/q12.txt

@@ -1,7 +1,9 @@
 Sort: lineitem.l_shipmode ASC NULLS LAST
   Projection: lineitem.l_shipmode, SUM(CASE WHEN orders.o_orderpriority = Utf8("1-URGENT") OR orders.o_orderpriority = Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END) AS high_line_count, SUM(CASE WHEN orders.o_orderpriority != Utf8("1-URGENT") AND orders.o_orderpriority != Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END) AS low_line_count
     Aggregate: groupBy=[[lineitem.l_shipmode]], aggr=[[SUM(CASE WHEN orders.o_orderpriority = Utf8("1-URGENT") OR orders.o_orderpriority = Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END), SUM(CASE WHEN orders.o_orderpriority != Utf8("1-URGENT") AND orders.o_orderpriority != Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)]]
-      Inner Join: lineitem.l_orderkey = orders.o_orderkey
-        Filter: (lineitem.l_shipmode = Utf8("SHIP") OR lineitem.l_shipmode = Utf8("MAIL")) AND lineitem.l_commitdate < lineitem.l_receiptdate AND lineitem.l_shipdate < lineitem.l_commitdate AND lineitem.l_receiptdate >= Date32("8766") AND lineitem.l_receiptdate < Date32("9131")
-          TableScan: lineitem projection=[l_orderkey, l_shipdate, l_commitdate, l_receiptdate, l_shipmode]
-        TableScan: orders projection=[o_orderkey, o_orderpriority]
+      Projection: lineitem.l_shipmode, orders.o_orderpriority
+        Inner Join: lineitem.l_orderkey = orders.o_orderkey
+          Projection: lineitem.l_orderkey, lineitem.l_shipmode
+            Filter: (lineitem.l_shipmode = Utf8("SHIP") OR lineitem.l_shipmode = Utf8("MAIL")) AND lineitem.l_commitdate < lineitem.l_receiptdate AND lineitem.l_shipdate < lineitem.l_commitdate AND lineitem.l_receiptdate >= Date32("8766") AND lineitem.l_receiptdate < Date32("9131")
+              TableScan: lineitem projection=[l_orderkey, l_shipdate, l_commitdate, l_receiptdate, l_shipmode]
+          TableScan: orders projection=[o_orderkey, o_orderpriority]

benchmarks/expected-plans/q13.txt

@@ -5,7 +5,9 @@ Sort: custdist DESC NULLS FIRST, c_count DESC NULLS FIRST
         SubqueryAlias: c_orders
           Projection: COUNT(orders.o_orderkey)
             Aggregate: groupBy=[[customer.c_custkey]], aggr=[[COUNT(orders.o_orderkey)]]
-              Left Join: customer.c_custkey = orders.o_custkey
-                TableScan: customer projection=[c_custkey]
-                Filter: orders.o_comment NOT LIKE Utf8("%special%requests%")
-                  TableScan: orders projection=[o_orderkey, o_custkey, o_comment]
+              Projection: customer.c_custkey, orders.o_orderkey
+                Left Join: customer.c_custkey = orders.o_custkey
+                  TableScan: customer projection=[c_custkey]
+                  Projection: orders.o_orderkey, orders.o_custkey
+                    Filter: orders.o_comment NOT LIKE Utf8("%special%requests%")
+                      TableScan: orders projection=[o_orderkey, o_custkey, o_comment]

benchmarks/expected-plans/q14.txt

@@ -2,6 +2,7 @@ Projection: Float64(100) * CAST(SUM(CASE WHEN part.p_type LIKE Utf8("PROMO%")  T
   Aggregate: groupBy=[[]], aggr=[[SUM(CASE WHEN part.p_type LIKE Utf8("PROMO%") THEN 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(Decimal128(Some(100),23,2) - CAST(lineitem.l_discount AS Decimal128(23, 2)) AS Decimal128(38, 4))Decimal128(Some(100),23,2) - CAST(lineitem.l_discount AS Decimal128(23, 2))CAST(lineitem.l_discount AS Decimal128(23, 2))lineitem.l_discountDecimal128(Some(100),23,2)CAST(lineitem.l_extendedprice AS Decimal128(38, 4))lineitem.l_extendedprice AS lineitem.l_extendedprice * Decimal128(Some(100),23,2) - lineitem.l_discount ELSE Decimal128(Some(0),38,4) END) AS SUM(CASE WHEN part.p_type LIKE Utf8("PROMO%")  THEN lineitem.l_extendedprice * Int64(1) - lineitem.l_discount ELSE Int64(0) END), SUM(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(Decimal128(Some(100),23,2) - CAST(lineitem.l_discount AS Decimal128(23, 2)) AS Decimal128(38, 4))Decimal128(Some(100),23,2) - CAST(lineitem.l_discount AS Decimal128(23, 2))CAST(lineitem.l_discount AS Decimal128(23, 2))lineitem.l_discountDecimal128(Some(100),23,2)CAST(lineitem.l_extendedprice AS Decimal128(38, 4))lineitem.l_extendedprice AS lineitem.l_extendedprice * Decimal128(Some(100),23,2) - lineitem.l_discount) AS SUM(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]]
     Projection: 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 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(Decimal128(Some(100),23,2) - CAST(lineitem.l_discount AS Decimal128(23, 2)) AS Decimal128(38, 4))Decimal128(Some(100),23,2) - CAST(lineitem.l_discount AS Decimal128(23, 2))CAST(lineitem.l_discount AS Decimal128(23, 2))lineitem.l_discountDecimal128(Some(100),23,2)CAST(lineitem.l_extendedprice AS Decimal128(38, 4))lineitem.l_extendedprice, part.p_type
       Inner Join: lineitem.l_partkey = part.p_partkey
-        Filter: lineitem.l_shipdate >= Date32("9374") AND lineitem.l_shipdate < Date32("9404")
-          TableScan: lineitem projection=[l_partkey, l_extendedprice, l_discount, l_shipdate]
+        Projection: lineitem.l_partkey, lineitem.l_extendedprice, lineitem.l_discount
+          Filter: lineitem.l_shipdate >= Date32("9374") AND lineitem.l_shipdate < Date32("9404")
+            TableScan: lineitem projection=[l_partkey, l_extendedprice, l_discount, l_shipdate]
         TableScan: part projection=[p_partkey, p_type]

benchmarks/expected-plans/q15.txt

@@ -2,22 +2,22 @@ EmptyRelation
 Sort: supplier.s_suppkey ASC NULLS LAST
   Projection: supplier.s_suppkey, supplier.s_name, supplier.s_address, supplier.s_phone, revenue0.total_revenue
     Inner Join: revenue0.total_revenue = __scalar_sq_1.__value
-      Inner Join: supplier.s_suppkey = revenue0.supplier_no
-        TableScan: supplier projection=[s_suppkey, s_name, s_address, s_phone]
-        SubqueryAlias: revenue0
-          Projection: supplier_no, total_revenue
+      Projection: supplier.s_suppkey, supplier.s_name, supplier.s_address, supplier.s_phone, revenue0.total_revenue
+        Inner Join: supplier.s_suppkey = revenue0.supplier_no
+          TableScan: supplier projection=[s_suppkey, s_name, s_address, s_phone]
+          SubqueryAlias: revenue0
             Projection: lineitem.l_suppkey AS supplier_no, SUM(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount) AS total_revenue
               Aggregate: groupBy=[[lineitem.l_suppkey]], aggr=[[SUM(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 SUM(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]]
-                Filter: lineitem.l_shipdate >= Date32("9496") AND lineitem.l_shipdate < Date32("9587")
-                  TableScan: lineitem projection=[l_suppkey, l_extendedprice, l_discount, l_shipdate]
+                Projection: lineitem.l_suppkey, lineitem.l_extendedprice, lineitem.l_discount
+                  Filter: lineitem.l_shipdate >= Date32("9496") AND lineitem.l_shipdate < Date32("9587")
+                    TableScan: lineitem projection=[l_suppkey, l_extendedprice, l_discount, l_shipdate]
       SubqueryAlias: __scalar_sq_1
         Projection: MAX(revenue0.total_revenue) AS __value
           Aggregate: groupBy=[[]], aggr=[[MAX(revenue0.total_revenue)]]
             SubqueryAlias: revenue0
-              Projection: total_revenue
-                Projection: SUM(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount) AS total_revenue
-                  Projection: SUM(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)
-                    Aggregate: groupBy=[[lineitem.l_suppkey]], aggr=[[SUM(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 SUM(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]]
-                      Filter: lineitem.l_shipdate >= Date32("9496") AND lineitem.l_shipdate < Date32("9587")
-                        TableScan: lineitem projection=[l_suppkey, l_extendedprice, l_discount, l_shipdate]
+              Projection: SUM(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount) AS total_revenue
+                Aggregate: groupBy=[[lineitem.l_suppkey]], aggr=[[SUM(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 SUM(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]]
+                  Projection: lineitem.l_suppkey, lineitem.l_extendedprice, lineitem.l_discount
+                    Filter: lineitem.l_shipdate >= Date32("9496") AND lineitem.l_shipdate < Date32("9587")
+                      TableScan: lineitem projection=[l_suppkey, l_extendedprice, l_discount, l_shipdate]
 EmptyRelation

benchmarks/expected-plans/q16.txt

@@ -1,14 +1,14 @@
 Sort: supplier_cnt DESC NULLS FIRST, part.p_brand ASC NULLS LAST, part.p_type ASC NULLS LAST, part.p_size ASC NULLS LAST
-  Projection: part.p_brand, part.p_type, part.p_size, COUNT(DISTINCT partsupp.ps_suppkey) AS supplier_cnt
-    Projection: group_alias_0 AS part.p_brand, group_alias_1 AS part.p_type, group_alias_2 AS part.p_size, COUNT(alias1) AS COUNT(DISTINCT partsupp.ps_suppkey)
-      Aggregate: groupBy=[[group_alias_0, group_alias_1, group_alias_2]], aggr=[[COUNT(alias1)]]
-        Aggregate: groupBy=[[part.p_brand AS group_alias_0, part.p_type AS group_alias_1, part.p_size AS group_alias_2, partsupp.ps_suppkey AS alias1]], aggr=[[]]
-          LeftAnti Join: partsupp.ps_suppkey = __correlated_sq_1.s_suppkey
+  Projection: group_alias_0 AS part.p_brand, group_alias_1 AS part.p_type, group_alias_2 AS part.p_size, COUNT(alias1) AS supplier_cnt
+    Aggregate: groupBy=[[group_alias_0, group_alias_1, group_alias_2]], aggr=[[COUNT(alias1)]]
+      Aggregate: groupBy=[[part.p_brand AS group_alias_0, part.p_type AS group_alias_1, part.p_size AS group_alias_2, partsupp.ps_suppkey AS alias1]], aggr=[[]]
+        LeftAnti Join: partsupp.ps_suppkey = __correlated_sq_1.s_suppkey
+          Projection: partsupp.ps_suppkey, part.p_brand, part.p_type, part.p_size
             Inner Join: partsupp.ps_partkey = part.p_partkey
               TableScan: partsupp projection=[ps_partkey, ps_suppkey]
               Filter: part.p_brand != Utf8("Brand#45") AND part.p_type NOT LIKE Utf8("MEDIUM POLISHED%") AND part.p_size IN ([Int32(49), Int32(14), Int32(23), Int32(45), Int32(19), Int32(3), Int32(36), Int32(9)])
                 TableScan: part projection=[p_partkey, p_brand, p_type, p_size]
-            SubqueryAlias: __correlated_sq_1
-              Projection: supplier.s_suppkey AS s_suppkey
-                Filter: supplier.s_comment LIKE Utf8("%Customer%Complaints%")
-                  TableScan: supplier projection=[s_suppkey, s_comment]
+          SubqueryAlias: __correlated_sq_1
+            Projection: supplier.s_suppkey AS s_suppkey
+              Filter: supplier.s_comment LIKE Utf8("%Customer%Complaints%")
+                TableScan: supplier projection=[s_suppkey, s_comment]

benchmarks/expected-plans/q17.txt

@@ -1,12 +1,15 @@
 Projection: CAST(SUM(lineitem.l_extendedprice) AS Float64) / Float64(7) AS avg_yearly
   Aggregate: groupBy=[[]], aggr=[[SUM(lineitem.l_extendedprice)]]
-    Filter: CAST(lineitem.l_quantity AS Decimal128(30, 15)) < CAST(__scalar_sq_1.__value AS Decimal128(30, 15))
-      Inner Join: part.p_partkey = __scalar_sq_1.l_partkey, lineitem.l_partkey = __scalar_sq_1.l_partkey
-        Inner Join: lineitem.l_partkey = part.p_partkey
-          TableScan: lineitem projection=[l_partkey, l_quantity, l_extendedprice]
-          Filter: part.p_brand = Utf8("Brand#23") AND part.p_container = Utf8("MED BOX")
-            TableScan: part projection=[p_partkey, p_brand, p_container]
-        SubqueryAlias: __scalar_sq_1
-          Projection: lineitem.l_partkey, Float64(0.2) * CAST(AVG(lineitem.l_quantity) AS Float64) AS __value
-            Aggregate: groupBy=[[lineitem.l_partkey]], aggr=[[AVG(lineitem.l_quantity)]]
-              TableScan: lineitem projection=[l_partkey, l_quantity]
+    Projection: lineitem.l_extendedprice
+      Filter: CAST(lineitem.l_quantity AS Decimal128(30, 15)) < CAST(__scalar_sq_1.__value AS Decimal128(30, 15))
+        Projection: lineitem.l_quantity, lineitem.l_extendedprice, __scalar_sq_1.__value
+          Inner Join: part.p_partkey = __scalar_sq_1.l_partkey, lineitem.l_partkey = __scalar_sq_1.l_partkey
+            Inner Join: lineitem.l_partkey = part.p_partkey
+              TableScan: lineitem projection=[l_partkey, l_quantity, l_extendedprice]
+              Projection: part.p_partkey
+                Filter: part.p_brand = Utf8("Brand#23") AND part.p_container = Utf8("MED BOX")
+                  TableScan: part projection=[p_partkey, p_brand, p_container]
+            SubqueryAlias: __scalar_sq_1
+              Projection: lineitem.l_partkey, Float64(0.2) * CAST(AVG(lineitem.l_quantity) AS Float64) AS __value
+                Aggregate: groupBy=[[lineitem.l_partkey]], aggr=[[AVG(lineitem.l_quantity)]]
+                  TableScan: lineitem projection=[l_partkey, l_quantity]