TPC-H Q2 subquery plan optimization

[skyzh]

select
        s_acctbal,
        s_name,
        n_name,
        p_partkey,
        p_mfgr,
        s_address,
        s_phone,
        s_comment
from
        part,
        supplier,
        partsupp,
        nation,
        region
where
        p_partkey = ps_partkey
        and s_suppkey = ps_suppkey
        and p_size = 4
        and p_type like '%TIN'
        and s_nationkey = n_nationkey
        and n_regionkey = r_regionkey
        and r_name = 'AFRICA'
        and ps_supplycost = (
                select
                        min(ps_supplycost)
                from
                        partsupp,
                        supplier,
                        nation,
                        region
                where
                        p_partkey = ps_partkey
                        and s_suppkey = ps_suppkey
                        and s_nationkey = n_nationkey
                        and n_regionkey = r_regionkey
                        and r_name = 'AFRICA'
        )
order by
    s_acctbal desc,
    n_name,
    s_name,
    p_partkey
limit 100;

PhysicalLimit { skip: 0(u64), fetch: 100(u64) }
└── PhysicalSort
    ├── exprs:
    │   ┌── SortOrder { order: Desc }
    │   │   └── #0
    │   ├── SortOrder { order: Asc }
    │   │   └── #2
    │   ├── SortOrder { order: Asc }
    │   │   └── #1
    │   └── SortOrder { order: Asc }
    │       └── #3
    └── PhysicalProjection { exprs: [ #21, #17, #4, #7, #9, #18, #20, #22 ] }
        └── PhysicalHashJoin { join_type: Inner, left_keys: [ #26, #7 ], right_keys: [ #1, #0 ] }
            ├── PhysicalHashJoin { join_type: Inner, left_keys: [ #7, #16 ], right_keys: [ #0, #1 ] }
            │   ├── PhysicalHashJoin { join_type: Inner, left_keys: [ #3 ], right_keys: [ #12 ] }
            │   │   ├── PhysicalHashJoin { join_type: Inner, left_keys: [ #0 ], right_keys: [ #2 ] }
            │   │   │   ├── PhysicalFilter
            │   │   │   │   ├── cond:Eq
            │   │   │   │   │   ├── #1
            │   │   │   │   │   └── "AFRICA"
            │   │   │   │   └── PhysicalScan { table: region }
            │   │   │   └── PhysicalScan { table: nation }
            │   │   └── PhysicalNestedLoopJoin { join_type: Cross, cond: true }
            │   │       ├── PhysicalFilter
            │   │       │   ├── cond:And
            │   │       │   │   ├── Eq
            │   │       │   │   │   ├── Cast { cast_to: Int64, child: #5 }
            │   │       │   │   │   └── 4(i64)
            │   │       │   │   └── Like { expr: #4, pattern: "%TIN", negated: false, case_insensitive: false }
            │   │       │   └── PhysicalScan { table: part }
            │   │       └── PhysicalScan { table: supplier }
            │   └── PhysicalScan { table: partsupp }
            └── PhysicalAgg
                ├── aggrs:Agg(Min)
                │   └── [ #4 ]
                ├── groups: [ #1 ]
                └── PhysicalFilter
                    ├── cond:And
                    │   ├── Eq
                    │   │   ├── #0
                    │   │   └── #1
                    │   ├── Eq
                    │   │   ├── #6
                    │   │   └── #2
                    │   ├── Eq
                    │   │   ├── #9
                    │   │   └── #13
                    │   ├── Eq
                    │   │   ├── #15
                    │   │   └── #17
                    │   └── Eq
                    │       ├── #18
                    │       └── "AFRICA"
                    └── PhysicalNestedLoopJoin { join_type: Inner, cond: true }
                        ├── PhysicalAgg { aggrs: [], groups: [ #0 ] }
                        │   └── PhysicalNestedLoopJoin { join_type: Cross, cond: true }
                        │       ├── PhysicalNestedLoopJoin { join_type: Cross, cond: true }
                        │       │   ├── PhysicalNestedLoopJoin { join_type: Cross, cond: true }
                        │       │   │   ├── PhysicalNestedLoopJoin { join_type: Cross, cond: true }
                        │       │   │   │   ├── PhysicalScan { table: part }
                        │       │   │   │   └── PhysicalScan { table: supplier }
                        │       │   │   └── PhysicalScan { table: partsupp }
                        │       │   └── PhysicalScan { table: nation }
                        │       └── PhysicalScan { table: region }
                        └── PhysicalNestedLoopJoin { join_type: Cross, cond: true }
                            ├── PhysicalNestedLoopJoin { join_type: Cross, cond: true }
                            │   ├── PhysicalNestedLoopJoin { join_type: Cross, cond: true }
                            │   │   ├── PhysicalScan { table: partsupp }
                            │   │   └── PhysicalScan { table: supplier }
                            │   └── PhysicalScan { table: nation }
                            └── PhysicalScan { table: region }

the problem is the generated distinct aggregation from the initial depjoin step consists of 5-way nested loop join without any filter within the aggregation child, which cannot be executed efficiently. either something wrong with the depjoin rules, or we need to implement pushdown across aggregation nodes?

[skyzh]

probably we need a new set of rules there -- the aggregation only needs the first column of the part table, so we can convert most the joins into semi joins. otherwise, datafusion will fail with:

attempt to multiply with overflow

when computing the statistics for the cross join operator

[skyzh]

this seems related to the problem of "how to do column pruning / projection pushdown in cascades?"