[opt](nereids) adjust cost model for BroadCastJoin and PartitionJoin (#20713)

we add penalty for broadcast join (bc for brief in the following). the intuition of penalty is as follow: 1. if the build side is very small (< 1M), we prefer bc, and set `penalty=1`, which means no penalty 2. if build side is more than 1M, we consider the ratio of the probe row count to the build row count. the less the ratio is, the higher penalty is. this pr has positive impact on tpch queries. Only q3 is changed. in out test (tpch 1T, 3BE) q3 improved from 5.1sec to 2.5 sec. this pr has positive impact on tpcds queries. test on tpcds sf100 (3BE), cold run improve from 163 sec to 156 sec, hot run improves from 155 sec to 149 sec
2023-06-16 22:49:04 +08:00
parent bb4f10b457
commit bf197ee8d2
31 changed files with 598 additions and 630 deletions
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModelV1.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModelV1.java
@ -59,6 +59,13 @@ class CostModelV1 extends PlanVisitor<Cost, PlanContext> {
     */
    static final double HEAVY_OPERATOR_PUNISH_FACTOR = 0.0;

+    // for a join, skew = leftRowCount/rightRowCount
+    // the higher skew is, the more we prefer broadcast join than shuffle join
+    // if skew < BROADCAST_JOIN_SKEW_RATIO, broadcast join will be punished,
+    // the penalty factor is no more than BROADCAST_JOIN_SKEW_PENALTY_LIMIT
+    static final double BROADCAST_JOIN_SKEW_RATIO = 30.0;
+    static final double BROADCAST_JOIN_SKEW_PENALTY_LIMIT = 2.0;
+
    public static Cost addChildCost(Plan plan, Cost planCost, Cost childCost, int index) {
        Preconditions.checkArgument(childCost instanceof CostV1 && planCost instanceof CostV1);
        CostV1 childCostV1 = (CostV1) childCost;
@ -161,27 +168,28 @@ class CostModelV1 extends PlanVisitor<Cost, PlanContext> {
    @Override
    public Cost visitPhysicalDistribute(
            PhysicalDistribute<? extends Plan> distribute, PlanContext context) {
-        int kBytes = 1024;
        Statistics childStatistics = context.getChildStatistics(0);
+        double intputRowCount = childStatistics.getRowCount();
        DistributionSpec spec = distribute.getDistributionSpec();
        int beNumber = ConnectContext.get().getEnv().getClusterInfo().getBackendsNumber(true);
        beNumber = Math.max(1, beNumber);
-        double dataSize = childStatistics.computeSize() / kBytes; // in K bytes
+
        // shuffle
        if (spec instanceof DistributionSpecHash) {
            return CostV1.of(
                    0,
                    0,
-                    dataSize / beNumber);
+                    intputRowCount * childStatistics.dataSizeFactor() / beNumber);
        }

        // replicate
        if (spec instanceof DistributionSpecReplicated) {
+            double dataSize = childStatistics.computeSize();
            double memLimit = ConnectContext.get().getSessionVariable().getMaxExecMemByte();
            //if build side is big, avoid use broadcast join
            double rowsLimit = ConnectContext.get().getSessionVariable().getBroadcastRowCountLimit();
            double brMemlimit = ConnectContext.get().getSessionVariable().getBroadcastHashtableMemLimitPercentage();
-            if (dataSize > memLimit * brMemlimit / kBytes
+            if (dataSize > memLimit * brMemlimit
                    || childStatistics.getRowCount() > rowsLimit) {
                return CostV1.of(Double.MAX_VALUE, Double.MAX_VALUE, Double.MAX_VALUE);
            }
@ -191,7 +199,7 @@ class CostModelV1 extends PlanVisitor<Cost, PlanContext> {
            return CostV1.of(
                    0,
                    0,
-                    dataSize, 0.0);
+                    intputRowCount * childStatistics.dataSizeFactor());

        }

@ -200,12 +208,12 @@ class CostModelV1 extends PlanVisitor<Cost, PlanContext> {
            return CostV1.of(
                    0,
                    0,
-                    dataSize / beNumber);
+                    intputRowCount * childStatistics.dataSizeFactor() / beNumber);
        }

        // any
        return CostV1.of(
-                childStatistics.getRowCount(),
+                intputRowCount,
                0,
                0);
    }
@ -220,6 +228,17 @@ class CostModelV1 extends PlanVisitor<Cost, PlanContext> {
        return CostV1.of(inputStatistics.getRowCount(), statistics.getRowCount(), 0);
    }

+    private double broadCastJoinBalancePenalty(Statistics probeStats, Statistics buildStats) {
+        // if build side is small enough (<1M), broadcast is also good, no penalty
+        if (buildStats.computeSize() < 1024 * 1024) {
+            return 1;
+        }
+        double broadcastJoinPenalty = (BROADCAST_JOIN_SKEW_RATIO * buildStats.getRowCount()) / probeStats.getRowCount();
+        broadcastJoinPenalty = Math.max(1, broadcastJoinPenalty);
+        broadcastJoinPenalty = Math.min(BROADCAST_JOIN_SKEW_PENALTY_LIMIT, broadcastJoinPenalty);
+        return broadcastJoinPenalty;
+    }
+
    @Override
    public Cost visitPhysicalHashJoin(
            PhysicalHashJoin<? extends Plan, ? extends Plan> physicalHashJoin, PlanContext context) {
@ -252,10 +271,19 @@ class CostModelV1 extends PlanVisitor<Cost, PlanContext> {
                    leftRowCount + rightRowCount,
                    penalty);
        }
+
+        if (context.isBroadcastJoin()) {
+            double broadcastJoinPenalty = broadCastJoinBalancePenalty(probeStats, buildStats);
+            return CostV1.of(leftRowCount * broadcastJoinPenalty + rightRowCount + outputRowCount,
+                    rightRowCount,
+                    0,
+                    0
+            );
+        }
        return CostV1.of(leftRowCount + rightRowCount + outputRowCount,
                rightRowCount,
                0,
-                penalty
+                0
        );
    }

--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/ColumnStatistic.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/ColumnStatistic.java
@ -381,4 +381,8 @@ public class ColumnStatistic {
    public boolean rangeChanged() {
        return original != null && (minValue != original.minValue || maxValue != original.maxValue);
    }
+
+    public boolean isUnKnown() {
+        return isUnKnown;
+    }
 }
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/Statistics.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/Statistics.java
@ -27,11 +27,14 @@ import java.util.Map;
 import java.util.Map.Entry;

 public class Statistics {
+    private static int K_BYTES = 1024;
+
    private final double rowCount;

    private final Map<Expression, ColumnStatistic> expressionToColumnStats;

-    private double computeSize;
+    // the byte size of one tuple
+    private double tupleSize;

    @Deprecated
    private double width;
@ -146,13 +149,21 @@ public class Statistics {
        return this;
    }

-    public double computeSize() {
-        if (computeSize <= 0) {
-            computeSize = Math.max(1, expressionToColumnStats.values().stream()
-                    .map(s -> s.avgSizeByte).reduce(0D, Double::sum)
-            ) * rowCount;
+    private double computeTupleSize() {
+        if (tupleSize <= 0) {
+            tupleSize = expressionToColumnStats.values().stream()
+                    .map(s -> s.avgSizeByte).reduce(0D, Double::sum);
+            tupleSize = Math.max(1, tupleSize);
        }
-        return computeSize;
+        return tupleSize;
+    }
+
+    public double computeSize() {
+        return computeTupleSize() * rowCount;
+    }
+
+    public double dataSizeFactor() {
+        return computeTupleSize() / K_BYTES;
    }

    @Override
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query1.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query1.out
@ -18,8 +18,9 @@ CteAnchor[cteId= ( CTEId#2=] )
 ------PhysicalTopN
 --------PhysicalProject
 ----------hashJoin[INNER_JOIN](ctr1.ctr_customer_sk = customer.c_customer_sk)
------------PhysicalProject
--------------PhysicalOlapScan[customer]
+------------PhysicalDistribute
+--------------PhysicalProject
+----------------PhysicalOlapScan[customer]
 ------------PhysicalDistribute
 --------------PhysicalProject
 ----------------hashJoin[LEFT_SEMI_JOIN](ctr1.ctr_store_sk = ctr2.ctr_store_sk)(cast(ctr_total_return as DOUBLE) > cast((avg(ctr_total_return) * 1.2) as DOUBLE))
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query11.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query11.out
@ -28,14 +28,15 @@ CteAnchor[cteId= ( CTEId#4=] )
 --------------hashAgg[LOCAL]
 ----------------PhysicalProject
 ------------------hashJoin[INNER_JOIN](customer.c_customer_sk = web_sales.ws_bill_customer_sk)
--------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
----------------------PhysicalProject
------------------------filter((('w' = 'w') OR ('w' = 's')))
--------------------------PhysicalOlapScan[web_sales]
----------------------PhysicalDistribute
+--------------------PhysicalDistribute
+----------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
 ------------------------PhysicalProject
--------------------------filter(((date_dim.d_year = 2001) OR (date_dim.d_year = 2002))(('w' = 'w') OR ('w' = 's')))
----------------------------PhysicalOlapScan[date_dim]
+--------------------------filter((('w' = 'w') OR ('w' = 's')))
+----------------------------PhysicalOlapScan[web_sales]
+------------------------PhysicalDistribute
+--------------------------PhysicalProject
+----------------------------filter(((date_dim.d_year = 2001) OR (date_dim.d_year = 2002))(('w' = 'w') OR ('w' = 's')))
+------------------------------PhysicalOlapScan[date_dim]
 --------------------PhysicalDistribute
 ----------------------PhysicalProject
 ------------------------filter((('w' = 'w') OR ('w' = 's')))
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query12.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query12.out
@ -12,15 +12,15 @@ PhysicalTopN
 ------------------hashAgg[LOCAL]
 --------------------PhysicalProject
 ----------------------hashJoin[INNER_JOIN](web_sales.ws_item_sk = item.i_item_sk)
------------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
--------------------------PhysicalProject
----------------------------PhysicalOlapScan[web_sales]
--------------------------PhysicalDistribute
----------------------------PhysicalProject
------------------------------filter((date_dim.d_date <= 1998-05-06)(date_dim.d_date >= 1998-04-06))
--------------------------------PhysicalOlapScan[date_dim]
 ------------------------PhysicalDistribute
--------------------------PhysicalProject
----------------------------filter(i_category IN ('Books', 'Sports', 'Men'))
------------------------------PhysicalOlapScan[item]
+--------------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[web_sales]
+----------------------------PhysicalDistribute
+------------------------------PhysicalProject
+--------------------------------filter((date_dim.d_date <= 1998-05-06)(date_dim.d_date >= 1998-04-06))
+----------------------------------PhysicalOlapScan[date_dim]
+------------------------PhysicalProject
+--------------------------filter(i_category IN ('Books', 'Sports', 'Men'))
+----------------------------PhysicalOlapScan[item]

--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query15.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query15.out
@ -8,13 +8,14 @@ PhysicalTopN
 ----------hashAgg[LOCAL]
 ------------PhysicalProject
 --------------hashJoin[INNER_JOIN](catalog_sales.cs_bill_customer_sk = customer.c_customer_sk)((substring(ca_zip, 1, 5) IN ('85669', '86197', '88274', '83405', '86475', '85392', '85460', '80348', '81792') OR ca_state IN ('CA', 'WA', 'GA')) OR (catalog_sales.cs_sales_price > 500.00))
----------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
------------------PhysicalProject
--------------------PhysicalOlapScan[catalog_sales]
------------------PhysicalDistribute
+----------------PhysicalDistribute
+------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
 --------------------PhysicalProject
----------------------filter((date_dim.d_qoy = 1)(date_dim.d_year = 2001))
------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalOlapScan[catalog_sales]
+--------------------PhysicalDistribute
+----------------------PhysicalProject
+------------------------filter((date_dim.d_qoy = 1)(date_dim.d_year = 2001))
+--------------------------PhysicalOlapScan[date_dim]
 ----------------PhysicalDistribute
 ------------------PhysicalProject
 --------------------hashJoin[INNER_JOIN](customer.c_current_addr_sk = customer_address.ca_address_sk)
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query18.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query18.out
@ -15,8 +15,9 @@ PhysicalTopN
 --------------------PhysicalDistribute
 ----------------------PhysicalProject
 ------------------------hashJoin[INNER_JOIN](customer.c_current_cdemo_sk = cd2.cd_demo_sk)
--------------------------PhysicalProject
----------------------------PhysicalOlapScan[customer_demographics]
+--------------------------PhysicalDistribute
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[customer_demographics]
 --------------------------PhysicalDistribute
 ----------------------------PhysicalProject
 ------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
@ -28,13 +29,13 @@ PhysicalTopN
 ------------------------------------------PhysicalOlapScan[catalog_sales]
 ----------------------------------------PhysicalDistribute
 ------------------------------------------hashJoin[INNER_JOIN](customer.c_current_addr_sk = customer_address.ca_address_sk)
--------------------------------------------PhysicalProject
----------------------------------------------filter(c_birth_month IN (1, 2, 4, 7, 8, 10))
------------------------------------------------PhysicalOlapScan[customer]
 --------------------------------------------PhysicalDistribute
 ----------------------------------------------PhysicalProject
------------------------------------------------filter(ca_state IN ('WA', 'GA', 'NC', 'ME', 'WY', 'OK', 'IN'))
--------------------------------------------------PhysicalOlapScan[customer_address]
+------------------------------------------------filter(c_birth_month IN (1, 2, 4, 7, 8, 10))
+--------------------------------------------------PhysicalOlapScan[customer]
+--------------------------------------------PhysicalProject
+----------------------------------------------filter(ca_state IN ('WA', 'GA', 'NC', 'ME', 'WY', 'OK', 'IN'))
+------------------------------------------------PhysicalOlapScan[customer_address]
 ------------------------------------PhysicalDistribute
 --------------------------------------PhysicalProject
 ----------------------------------------filter((cast(cd_gender as VARCHAR(*)) = 'F')(cast(cd_education_status as VARCHAR(*)) = 'Advanced Degree'))
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query27.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query27.out
@ -11,23 +11,23 @@ PhysicalTopN
 ----------------PhysicalRepeat
 ------------------PhysicalProject
 --------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
----------------------PhysicalProject
------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = item.i_item_sk)
--------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
----------------------------hashJoin[INNER_JOIN](store_sales.ss_cdemo_sk = customer_demographics.cd_demo_sk)
------------------------------PhysicalProject
--------------------------------PhysicalOlapScan[store_sales]
+----------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = item.i_item_sk)
+------------------------PhysicalDistribute
+--------------------------PhysicalProject
+----------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
+------------------------------hashJoin[INNER_JOIN](store_sales.ss_cdemo_sk = customer_demographics.cd_demo_sk)
+--------------------------------PhysicalProject
+----------------------------------PhysicalOlapScan[store_sales]
+--------------------------------PhysicalDistribute
+----------------------------------PhysicalProject
+------------------------------------filter((cast(cd_gender as VARCHAR(*)) = 'F')(cast(cd_education_status as VARCHAR(*)) = 'Secondary')(cast(cd_marital_status as VARCHAR(*)) = 'D'))
+--------------------------------------PhysicalOlapScan[customer_demographics]
 ------------------------------PhysicalDistribute
 --------------------------------PhysicalProject
----------------------------------filter((cast(cd_gender as VARCHAR(*)) = 'F')(cast(cd_education_status as VARCHAR(*)) = 'Secondary')(cast(cd_marital_status as VARCHAR(*)) = 'D'))
------------------------------------PhysicalOlapScan[customer_demographics]
----------------------------PhysicalDistribute
------------------------------PhysicalProject
--------------------------------filter((date_dim.d_year = 1999))
----------------------------------PhysicalOlapScan[date_dim]
--------------------------PhysicalDistribute
----------------------------PhysicalProject
------------------------------PhysicalOlapScan[item]
+----------------------------------filter((date_dim.d_year = 1999))
+------------------------------------PhysicalOlapScan[date_dim]
+------------------------PhysicalProject
+--------------------------PhysicalOlapScan[item]
 ----------------------PhysicalDistribute
 ------------------------PhysicalProject
 --------------------------filter(s_state IN ('MO', 'AL', 'MI', 'TN', 'LA', 'SC'))
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query28.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query28.out
@ -2,31 +2,20 @@
 -- !ds_shape_28 --
 PhysicalLimit
 --PhysicalLimit
----NestedLoopJoin[CROSS_JOIN]
------PhysicalLimit
+----PhysicalProject
+------NestedLoopJoin[CROSS_JOIN]
 --------PhysicalLimit
----------NestedLoopJoin[CROSS_JOIN]
------------PhysicalLimit
+----------PhysicalLimit
+------------NestedLoopJoin[CROSS_JOIN]
 --------------PhysicalLimit
----------------NestedLoopJoin[CROSS_JOIN]
------------------PhysicalLimit
+----------------PhysicalLimit
+------------------NestedLoopJoin[CROSS_JOIN]
 --------------------PhysicalLimit
----------------------NestedLoopJoin[CROSS_JOIN]
------------------------PhysicalLimit
+----------------------PhysicalLimit
+------------------------NestedLoopJoin[CROSS_JOIN]
 --------------------------PhysicalLimit
----------------------------NestedLoopJoin[CROSS_JOIN]
------------------------------PhysicalLimit
--------------------------------PhysicalLimit
----------------------------------hashAgg[DISTINCT_GLOBAL]
------------------------------------PhysicalDistribute
--------------------------------------hashAgg[DISTINCT_LOCAL]
----------------------------------------hashAgg[GLOBAL]
------------------------------------------PhysicalDistribute
--------------------------------------------hashAgg[LOCAL]
----------------------------------------------PhysicalProject
------------------------------------------------filter((store_sales.ss_quantity <= 5)((((store_sales.ss_list_price >= 131.00) AND (store_sales.ss_list_price <= 141.00)) OR ((store_sales.ss_coupon_amt >= 16798.00) AND (store_sales.ss_coupon_amt <= 17798.00))) OR ((store_sales.ss_wholesale_cost >= 25.00) AND (store_sales.ss_wholesale_cost <= 45.00)))(store_sales.ss_quantity >= 0))
--------------------------------------------------PhysicalOlapScan[store_sales]
------------------------------PhysicalDistribute
+----------------------------PhysicalLimit
+------------------------------NestedLoopJoin[CROSS_JOIN]
 --------------------------------PhysicalLimit
 ----------------------------------PhysicalLimit
 ------------------------------------hashAgg[DISTINCT_GLOBAL]
@ -36,54 +25,66 @@ PhysicalLimit
 --------------------------------------------PhysicalDistribute
 ----------------------------------------------hashAgg[LOCAL]
 ------------------------------------------------PhysicalProject
--------------------------------------------------filter((store_sales.ss_quantity <= 10)((((store_sales.ss_list_price >= 145.00) AND (store_sales.ss_list_price <= 155.00)) OR ((store_sales.ss_coupon_amt >= 14792.00) AND (store_sales.ss_coupon_amt <= 15792.00))) OR ((store_sales.ss_wholesale_cost >= 46.00) AND (store_sales.ss_wholesale_cost <= 66.00)))(store_sales.ss_quantity >= 6))
+--------------------------------------------------filter((store_sales.ss_quantity <= 5)((((store_sales.ss_list_price >= 131.00) AND (store_sales.ss_list_price <= 141.00)) OR ((store_sales.ss_coupon_amt >= 16798.00) AND (store_sales.ss_coupon_amt <= 17798.00))) OR ((store_sales.ss_wholesale_cost >= 25.00) AND (store_sales.ss_wholesale_cost <= 45.00)))(store_sales.ss_quantity >= 0))
 ----------------------------------------------------PhysicalOlapScan[store_sales]
------------------------PhysicalDistribute
--------------------------PhysicalLimit
+--------------------------------PhysicalDistribute
+----------------------------------PhysicalLimit
+------------------------------------PhysicalLimit
+--------------------------------------hashAgg[DISTINCT_GLOBAL]
+----------------------------------------PhysicalDistribute
+------------------------------------------hashAgg[DISTINCT_LOCAL]
+--------------------------------------------hashAgg[GLOBAL]
+----------------------------------------------PhysicalDistribute
+------------------------------------------------hashAgg[LOCAL]
+--------------------------------------------------PhysicalProject
+----------------------------------------------------filter((store_sales.ss_quantity <= 10)((((store_sales.ss_list_price >= 145.00) AND (store_sales.ss_list_price <= 155.00)) OR ((store_sales.ss_coupon_amt >= 14792.00) AND (store_sales.ss_coupon_amt <= 15792.00))) OR ((store_sales.ss_wholesale_cost >= 46.00) AND (store_sales.ss_wholesale_cost <= 66.00)))(store_sales.ss_quantity >= 6))
+------------------------------------------------------PhysicalOlapScan[store_sales]
+--------------------------PhysicalDistribute
 ----------------------------PhysicalLimit
------------------------------hashAgg[DISTINCT_GLOBAL]
--------------------------------PhysicalDistribute
----------------------------------hashAgg[DISTINCT_LOCAL]
------------------------------------hashAgg[GLOBAL]
--------------------------------------PhysicalDistribute
----------------------------------------hashAgg[LOCAL]
------------------------------------------PhysicalProject
--------------------------------------------filter(((((store_sales.ss_list_price >= 1.5E+2) AND (store_sales.ss_list_price <= 1.6E+2)) OR ((store_sales.ss_coupon_amt >= 6.6E+3) AND (store_sales.ss_coupon_amt <= 7.6E+3))) OR ((store_sales.ss_wholesale_cost >= 9.00) AND (store_sales.ss_wholesale_cost <= 29.00)))(store_sales.ss_quantity >= 11)(store_sales.ss_quantity <= 15))
----------------------------------------------PhysicalOlapScan[store_sales]
------------------PhysicalDistribute
--------------------PhysicalLimit
+------------------------------PhysicalLimit
+--------------------------------hashAgg[DISTINCT_GLOBAL]
+----------------------------------PhysicalDistribute
+------------------------------------hashAgg[DISTINCT_LOCAL]
+--------------------------------------hashAgg[GLOBAL]
+----------------------------------------PhysicalDistribute
+------------------------------------------hashAgg[LOCAL]
+--------------------------------------------PhysicalProject
+----------------------------------------------filter(((((store_sales.ss_list_price >= 1.5E+2) AND (store_sales.ss_list_price <= 1.6E+2)) OR ((store_sales.ss_coupon_amt >= 6.6E+3) AND (store_sales.ss_coupon_amt <= 7.6E+3))) OR ((store_sales.ss_wholesale_cost >= 9.00) AND (store_sales.ss_wholesale_cost <= 29.00)))(store_sales.ss_quantity >= 11)(store_sales.ss_quantity <= 15))
+------------------------------------------------PhysicalOlapScan[store_sales]
+--------------------PhysicalDistribute
 ----------------------PhysicalLimit
------------------------hashAgg[DISTINCT_GLOBAL]
--------------------------PhysicalDistribute
----------------------------hashAgg[DISTINCT_LOCAL]
------------------------------hashAgg[GLOBAL]
--------------------------------PhysicalDistribute
----------------------------------hashAgg[LOCAL]
------------------------------------PhysicalProject
--------------------------------------filter((store_sales.ss_quantity <= 20)((((store_sales.ss_list_price >= 91.00) AND (store_sales.ss_list_price <= 101.00)) OR ((store_sales.ss_coupon_amt >= 13493.00) AND (store_sales.ss_coupon_amt <= 14493.00))) OR ((store_sales.ss_wholesale_cost >= 36.00) AND (store_sales.ss_wholesale_cost <= 56.00)))(store_sales.ss_quantity >= 16))
----------------------------------------PhysicalOlapScan[store_sales]
------------PhysicalDistribute
--------------PhysicalLimit
----------------PhysicalLimit
------------------hashAgg[DISTINCT_GLOBAL]
--------------------PhysicalDistribute
----------------------hashAgg[DISTINCT_LOCAL]
------------------------hashAgg[GLOBAL]
--------------------------PhysicalDistribute
----------------------------hashAgg[LOCAL]
------------------------------PhysicalProject
--------------------------------filter(((((store_sales.ss_list_price >= 0.00) AND (store_sales.ss_list_price <= 10.00)) OR ((store_sales.ss_coupon_amt >= 7629.00) AND (store_sales.ss_coupon_amt <= 8629.00))) OR ((store_sales.ss_wholesale_cost >= 6.00) AND (store_sales.ss_wholesale_cost <= 26.00)))(store_sales.ss_quantity <= 25)(store_sales.ss_quantity >= 21))
----------------------------------PhysicalOlapScan[store_sales]
------PhysicalDistribute
--------PhysicalLimit
----------PhysicalLimit
------------hashAgg[DISTINCT_GLOBAL]
+------------------------PhysicalLimit
+--------------------------hashAgg[DISTINCT_GLOBAL]
+----------------------------PhysicalDistribute
+------------------------------hashAgg[DISTINCT_LOCAL]
+--------------------------------hashAgg[GLOBAL]
+----------------------------------PhysicalDistribute
+------------------------------------hashAgg[LOCAL]
+--------------------------------------PhysicalProject
+----------------------------------------filter((store_sales.ss_quantity <= 20)((((store_sales.ss_list_price >= 91.00) AND (store_sales.ss_list_price <= 101.00)) OR ((store_sales.ss_coupon_amt >= 13493.00) AND (store_sales.ss_coupon_amt <= 14493.00))) OR ((store_sales.ss_wholesale_cost >= 36.00) AND (store_sales.ss_wholesale_cost <= 56.00)))(store_sales.ss_quantity >= 16))
+------------------------------------------PhysicalOlapScan[store_sales]
 --------------PhysicalDistribute
----------------hashAgg[DISTINCT_LOCAL]
------------------hashAgg[GLOBAL]
--------------------PhysicalDistribute
----------------------hashAgg[LOCAL]
------------------------PhysicalProject
--------------------------filter((store_sales.ss_quantity >= 26)((((store_sales.ss_list_price >= 89.00) AND (store_sales.ss_list_price <= 99.00)) OR ((store_sales.ss_coupon_amt >= 15257.00) AND (store_sales.ss_coupon_amt <= 16257.00))) OR ((store_sales.ss_wholesale_cost >= 31.00) AND (store_sales.ss_wholesale_cost <= 51.00)))(store_sales.ss_quantity <= 30))
----------------------------PhysicalOlapScan[store_sales]
+----------------PhysicalLimit
+------------------PhysicalLimit
+--------------------hashAgg[DISTINCT_GLOBAL]
+----------------------PhysicalDistribute
+------------------------hashAgg[DISTINCT_LOCAL]
+--------------------------hashAgg[GLOBAL]
+----------------------------PhysicalDistribute
+------------------------------hashAgg[LOCAL]
+--------------------------------PhysicalProject
+----------------------------------filter(((((store_sales.ss_list_price >= 0.00) AND (store_sales.ss_list_price <= 10.00)) OR ((store_sales.ss_coupon_amt >= 7629.00) AND (store_sales.ss_coupon_amt <= 8629.00))) OR ((store_sales.ss_wholesale_cost >= 6.00) AND (store_sales.ss_wholesale_cost <= 26.00)))(store_sales.ss_quantity <= 25)(store_sales.ss_quantity >= 21))
+------------------------------------PhysicalOlapScan[store_sales]
+--------PhysicalDistribute
+----------PhysicalLimit
+------------PhysicalLimit
+--------------hashAgg[DISTINCT_GLOBAL]
+----------------PhysicalDistribute
+------------------hashAgg[DISTINCT_LOCAL]
+--------------------hashAgg[GLOBAL]
+----------------------PhysicalDistribute
+------------------------hashAgg[LOCAL]
+--------------------------PhysicalProject
+----------------------------filter((store_sales.ss_quantity >= 26)((((store_sales.ss_list_price >= 89.00) AND (store_sales.ss_list_price <= 99.00)) OR ((store_sales.ss_coupon_amt >= 15257.00) AND (store_sales.ss_coupon_amt <= 16257.00))) OR ((store_sales.ss_wholesale_cost >= 31.00) AND (store_sales.ss_wholesale_cost <= 51.00)))(store_sales.ss_quantity <= 30))
+------------------------------PhysicalOlapScan[store_sales]

--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query30.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query30.out
@ -24,7 +24,8 @@ CteAnchor[cteId= ( CTEId#2=] )
 --------PhysicalProject
 ----------hashJoin[LEFT_SEMI_JOIN](ctr1.ctr_state = ctr2.ctr_state)(cast(ctr_total_return as DOUBLE) > cast((avg(ctr_total_return) * 1.2) as DOUBLE))
 ------------hashJoin[INNER_JOIN](ctr1.ctr_customer_sk = customer.c_customer_sk)
--------------CteConsumer[cteId= ( CTEId#2=] )
+--------------PhysicalDistribute
+----------------CteConsumer[cteId= ( CTEId#2=] )
 --------------PhysicalDistribute
 ----------------hashJoin[INNER_JOIN](customer_address.ca_address_sk = customer.c_current_addr_sk)
 ------------------PhysicalProject
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query31.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query31.out
@ -27,17 +27,17 @@ CteAnchor[cteId= ( CTEId#6=] )
 ------------hashAgg[LOCAL]
 --------------PhysicalProject
 ----------------hashJoin[INNER_JOIN](web_sales.ws_bill_addr_sk = customer_address.ca_address_sk)
------------------PhysicalProject
--------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
----------------------PhysicalProject
------------------------PhysicalOlapScan[web_sales]
----------------------PhysicalDistribute
------------------------PhysicalProject
--------------------------filter((ws.d_year = 2000)d_qoy IN (1, 2, 3))
----------------------------PhysicalOlapScan[date_dim]
 ------------------PhysicalDistribute
 --------------------PhysicalProject
----------------------PhysicalOlapScan[customer_address]
+----------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
+------------------------PhysicalProject
+--------------------------PhysicalOlapScan[web_sales]
+------------------------PhysicalDistribute
+--------------------------PhysicalProject
+----------------------------filter((ws.d_year = 2000)d_qoy IN (1, 2, 3))
+------------------------------PhysicalOlapScan[date_dim]
+------------------PhysicalProject
+--------------------PhysicalOlapScan[customer_address]
 ----PhysicalQuickSort
 ------PhysicalDistribute
 --------PhysicalQuickSort
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query33.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query33.out
@ -14,22 +14,23 @@ PhysicalTopN
 ----------------------hashAgg[LOCAL]
 ------------------------PhysicalProject
 --------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = item.i_item_sk)
----------------------------PhysicalProject
------------------------------hashJoin[INNER_JOIN](store_sales.ss_addr_sk = customer_address.ca_address_sk)
--------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
----------------------------------PhysicalProject
------------------------------------PhysicalOlapScan[store_sales]
+----------------------------PhysicalDistribute
+------------------------------PhysicalProject
+--------------------------------hashJoin[INNER_JOIN](store_sales.ss_addr_sk = customer_address.ca_address_sk)
 ----------------------------------PhysicalDistribute
 ------------------------------------PhysicalProject
--------------------------------------filter((date_dim.d_moy = 1)(date_dim.d_year = 2002))
----------------------------------------PhysicalOlapScan[date_dim]
--------------------------------PhysicalDistribute
+--------------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
+----------------------------------------PhysicalProject
+------------------------------------------PhysicalOlapScan[store_sales]
+----------------------------------------PhysicalDistribute
+------------------------------------------PhysicalProject
+--------------------------------------------filter((date_dim.d_moy = 1)(date_dim.d_year = 2002))
+----------------------------------------------PhysicalOlapScan[date_dim]
 ----------------------------------PhysicalProject
 ------------------------------------filter((customer_address.ca_gmt_offset = -5.00))
 --------------------------------------PhysicalOlapScan[customer_address]
----------------------------PhysicalDistribute
------------------------------PhysicalProject
--------------------------------PhysicalOlapScan[item]
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[item]
 ------------------PhysicalDistribute
 --------------------PhysicalProject
 ----------------------filter((item.i_category = 'Home'))
@ -44,17 +45,18 @@ PhysicalTopN
 ----------------------------PhysicalDistribute
 ------------------------------PhysicalProject
 --------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_bill_addr_sk = customer_address.ca_address_sk)
----------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
------------------------------------PhysicalProject
--------------------------------------PhysicalOlapScan[catalog_sales]
------------------------------------PhysicalDistribute
--------------------------------------PhysicalProject
----------------------------------------filter((date_dim.d_moy = 1)(date_dim.d_year = 2002))
------------------------------------------PhysicalOlapScan[date_dim]
 ----------------------------------PhysicalDistribute
 ------------------------------------PhysicalProject
--------------------------------------filter((customer_address.ca_gmt_offset = -5.00))
----------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
+----------------------------------------PhysicalProject
+------------------------------------------PhysicalOlapScan[catalog_sales]
+----------------------------------------PhysicalDistribute
+------------------------------------------PhysicalProject
+--------------------------------------------filter((date_dim.d_moy = 1)(date_dim.d_year = 2002))
+----------------------------------------------PhysicalOlapScan[date_dim]
+----------------------------------PhysicalProject
+------------------------------------filter((customer_address.ca_gmt_offset = -5.00))
+--------------------------------------PhysicalOlapScan[customer_address]
 ----------------------------PhysicalProject
 ------------------------------PhysicalOlapScan[item]
 ------------------PhysicalDistribute
@ -73,17 +75,18 @@ PhysicalTopN
 ----------------------------PhysicalDistribute
 ------------------------------PhysicalProject
 --------------------------------hashJoin[INNER_JOIN](web_sales.ws_bill_addr_sk = customer_address.ca_address_sk)
----------------------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
------------------------------------PhysicalProject
--------------------------------------PhysicalOlapScan[web_sales]
------------------------------------PhysicalDistribute
--------------------------------------PhysicalProject
----------------------------------------filter((date_dim.d_moy = 1)(date_dim.d_year = 2002))
------------------------------------------PhysicalOlapScan[date_dim]
 ----------------------------------PhysicalDistribute
 ------------------------------------PhysicalProject
--------------------------------------filter((customer_address.ca_gmt_offset = -5.00))
----------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
+----------------------------------------PhysicalProject
+------------------------------------------PhysicalOlapScan[web_sales]
+----------------------------------------PhysicalDistribute
+------------------------------------------PhysicalProject
+--------------------------------------------filter((date_dim.d_moy = 1)(date_dim.d_year = 2002))
+----------------------------------------------PhysicalOlapScan[date_dim]
+----------------------------------PhysicalProject
+------------------------------------filter((customer_address.ca_gmt_offset = -5.00))
+--------------------------------------PhysicalOlapScan[customer_address]
 ------------------PhysicalDistribute
 --------------------PhysicalProject
 ----------------------filter((item.i_category = 'Home'))
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query38.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query38.out
@ -13,13 +13,15 @@ PhysicalLimit
 --------------------hashAgg[LOCAL]
 ----------------------PhysicalProject
 ------------------------hashJoin[INNER_JOIN](store_sales.ss_customer_sk = customer.c_customer_sk)
--------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
+--------------------------PhysicalDistribute
 ----------------------------PhysicalProject
------------------------------PhysicalOlapScan[store_sales]
----------------------------PhysicalDistribute
------------------------------PhysicalProject
--------------------------------filter((date_dim.d_month_seq <= 1194)(date_dim.d_month_seq >= 1183))
----------------------------------PhysicalOlapScan[date_dim]
+------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
+--------------------------------PhysicalProject
+----------------------------------PhysicalOlapScan[store_sales]
+--------------------------------PhysicalDistribute
+----------------------------------PhysicalProject
+------------------------------------filter((date_dim.d_month_seq <= 1194)(date_dim.d_month_seq >= 1183))
+--------------------------------------PhysicalOlapScan[date_dim]
 --------------------------PhysicalDistribute
 ----------------------------PhysicalProject
 ------------------------------PhysicalOlapScan[customer]
@ -29,13 +31,15 @@ PhysicalLimit
 --------------------hashAgg[LOCAL]
 ----------------------PhysicalProject
 ------------------------hashJoin[INNER_JOIN](catalog_sales.cs_bill_customer_sk = customer.c_customer_sk)
--------------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
+--------------------------PhysicalDistribute
 ----------------------------PhysicalProject
------------------------------PhysicalOlapScan[catalog_sales]
----------------------------PhysicalDistribute
------------------------------PhysicalProject
--------------------------------filter((date_dim.d_month_seq <= 1194)(date_dim.d_month_seq >= 1183))
----------------------------------PhysicalOlapScan[date_dim]
+------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
+--------------------------------PhysicalProject
+----------------------------------PhysicalOlapScan[catalog_sales]
+--------------------------------PhysicalDistribute
+----------------------------------PhysicalProject
+------------------------------------filter((date_dim.d_month_seq <= 1194)(date_dim.d_month_seq >= 1183))
+--------------------------------------PhysicalOlapScan[date_dim]
 --------------------------PhysicalDistribute
 ----------------------------PhysicalProject
 ------------------------------PhysicalOlapScan[customer]
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query39.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query39.out
@ -1,48 +1,38 @@
 -- This file is automatically generated. You should know what you did if you want to edit this
 -- !ds_shape_39 --
-PhysicalQuickSort
--PhysicalDistribute
----PhysicalQuickSort
------hashJoin[INNER_JOIN](inv1.i_item_sk = inv2.i_item_sk)(inv1.w_warehouse_sk = inv2.w_warehouse_sk)
--------PhysicalProject
----------filter((CASE WHEN (mean = 0.0) THEN 0.0 ELSE (stdev / mean) END > 1.0))
------------hashAgg[GLOBAL]
--------------PhysicalDistribute
----------------hashAgg[LOCAL]
------------------PhysicalProject
--------------------hashJoin[INNER_JOIN](inventory.inv_warehouse_sk = warehouse.w_warehouse_sk)
----------------------hashJoin[INNER_JOIN](inventory.inv_item_sk = item.i_item_sk)
------------------------hashJoin[INNER_JOIN](inventory.inv_date_sk = date_dim.d_date_sk)
--------------------------PhysicalOlapScan[inventory]
--------------------------PhysicalDistribute
----------------------------PhysicalProject
------------------------------filter((inv1.d_moy = 1)(date_dim.d_year = 1998))
--------------------------------PhysicalOlapScan[date_dim]
------------------------PhysicalDistribute
--------------------------PhysicalProject
----------------------------PhysicalOlapScan[item]
+CteAnchor[cteId= ( CTEId#3=] )
+--CteProducer[cteId= ( CTEId#3=] )
+----PhysicalProject
+------filter((CASE WHEN (mean = 0.0) THEN 0.0 ELSE (stdev / mean) END > 1.0))
+--------hashAgg[GLOBAL]
+----------PhysicalDistribute
+------------hashAgg[LOCAL]
+--------------PhysicalProject
+----------------hashJoin[INNER_JOIN](inventory.inv_warehouse_sk = warehouse.w_warehouse_sk)
+------------------hashJoin[INNER_JOIN](inventory.inv_item_sk = item.i_item_sk)
+--------------------hashJoin[INNER_JOIN](inventory.inv_date_sk = date_dim.d_date_sk)
+----------------------PhysicalOlapScan[inventory]
 ----------------------PhysicalDistribute
 ------------------------PhysicalProject
--------------------------PhysicalOlapScan[warehouse]
--------PhysicalDistribute
----------PhysicalProject
------------filter((CASE WHEN (mean = 0.0) THEN 0.0 ELSE (stdev / mean) END > 1.0))
--------------hashAgg[GLOBAL]
----------------PhysicalDistribute
------------------hashAgg[LOCAL]
+--------------------------filter(((inv.d_moy = 1) OR (inv.d_moy = 2))(date_dim.d_year = 1998))
+----------------------------PhysicalOlapScan[date_dim]
+--------------------PhysicalDistribute
+----------------------PhysicalProject
+------------------------PhysicalOlapScan[item]
+------------------PhysicalDistribute
 --------------------PhysicalProject
----------------------hashJoin[INNER_JOIN](inventory.inv_warehouse_sk = warehouse.w_warehouse_sk)
------------------------hashJoin[INNER_JOIN](inventory.inv_item_sk = item.i_item_sk)
--------------------------hashJoin[INNER_JOIN](inventory.inv_date_sk = date_dim.d_date_sk)
----------------------------PhysicalOlapScan[inventory]
----------------------------PhysicalDistribute
------------------------------PhysicalProject
--------------------------------filter((date_dim.d_year = 1998)(inv2.d_moy = 2))
----------------------------------PhysicalOlapScan[date_dim]
--------------------------PhysicalDistribute
----------------------------PhysicalProject
------------------------------PhysicalOlapScan[item]
------------------------PhysicalDistribute
--------------------------PhysicalProject
----------------------------PhysicalOlapScan[warehouse]
+----------------------PhysicalOlapScan[warehouse]
+--PhysicalQuickSort
+----PhysicalDistribute
+------PhysicalQuickSort
+--------PhysicalProject
+----------hashJoin[INNER_JOIN](inv1.w_warehouse_sk = inv2.w_warehouse_sk)(inv1.i_item_sk = inv2.i_item_sk)
+------------PhysicalDistribute
+--------------PhysicalProject
+----------------filter((inv1.d_moy = 1))
+------------------CteConsumer[cteId= ( CTEId#3=] )
+------------PhysicalDistribute
+--------------PhysicalProject
+----------------filter((inv2.d_moy = 2))
+------------------CteConsumer[cteId= ( CTEId#3=] )

--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query45.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query45.out
@ -14,16 +14,16 @@ PhysicalTopN
 ----------------------PhysicalDistribute
 ------------------------PhysicalProject
 --------------------------hashJoin[INNER_JOIN](web_sales.ws_item_sk = item.i_item_sk)
----------------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
------------------------------PhysicalProject
--------------------------------PhysicalOlapScan[web_sales]
------------------------------PhysicalDistribute
--------------------------------PhysicalProject
----------------------------------filter((date_dim.d_qoy = 2)(date_dim.d_year = 2000))
------------------------------------PhysicalOlapScan[date_dim]
 ----------------------------PhysicalDistribute
------------------------------PhysicalProject
--------------------------------PhysicalOlapScan[item]
+------------------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
+--------------------------------PhysicalProject
+----------------------------------PhysicalOlapScan[web_sales]
+--------------------------------PhysicalDistribute
+----------------------------------PhysicalProject
+------------------------------------filter((date_dim.d_qoy = 2)(date_dim.d_year = 2000))
+--------------------------------------PhysicalOlapScan[date_dim]
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[item]
 ----------------------PhysicalDistribute
 ------------------------PhysicalProject
 --------------------------hashJoin[INNER_JOIN](customer.c_current_addr_sk = customer_address.ca_address_sk)
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query46.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query46.out
@ -14,30 +14,28 @@ PhysicalTopN
 ------------------PhysicalOlapScan[customer]
 ----------PhysicalDistribute
 ------------PhysicalProject
--------------hashAgg[GLOBAL]
----------------PhysicalDistribute
------------------hashAgg[LOCAL]
--------------------PhysicalProject
----------------------hashJoin[INNER_JOIN](store_sales.ss_addr_sk = customer_address.ca_address_sk)
------------------------PhysicalProject
--------------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
----------------------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = household_demographics.hd_demo_sk)
------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[store_sales]
--------------------------------PhysicalDistribute
----------------------------------PhysicalProject
------------------------------------filter(((date_dim.d_dow = 0) OR (date_dim.d_dow = 6))d_year IN (1999, 2000, 2001))
--------------------------------------PhysicalOlapScan[date_dim]
+--------------hashAgg[LOCAL]
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN](store_sales.ss_addr_sk = customer_address.ca_address_sk)
+--------------------PhysicalDistribute
+----------------------PhysicalProject
+------------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
+--------------------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = household_demographics.hd_demo_sk)
+----------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
+------------------------------PhysicalProject
+--------------------------------PhysicalOlapScan[store_sales]
 ------------------------------PhysicalDistribute
 --------------------------------PhysicalProject
----------------------------------filter(((household_demographics.hd_dep_count = 6) OR (household_demographics.hd_vehicle_count = 0)))
------------------------------------PhysicalOlapScan[household_demographics]
+----------------------------------filter(((date_dim.d_dow = 0) OR (date_dim.d_dow = 6))d_year IN (1999, 2000, 2001))
+------------------------------------PhysicalOlapScan[date_dim]
 ----------------------------PhysicalDistribute
 ------------------------------PhysicalProject
--------------------------------filter(s_city IN ('Five Points', 'Centerville', 'Oak Grove', 'Fairview', 'Liberty'))
----------------------------------PhysicalOlapScan[store]
------------------------PhysicalDistribute
--------------------------PhysicalProject
----------------------------PhysicalOlapScan[customer_address]
+--------------------------------filter(((household_demographics.hd_dep_count = 6) OR (household_demographics.hd_vehicle_count = 0)))
+----------------------------------PhysicalOlapScan[household_demographics]
+--------------------------PhysicalDistribute
+----------------------------PhysicalProject
+------------------------------filter(s_city IN ('Five Points', 'Centerville', 'Oak Grove', 'Fairview', 'Liberty'))
+--------------------------------PhysicalOlapScan[store]
+--------------------PhysicalProject
+----------------------PhysicalOlapScan[customer_address]

--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query56.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query56.out
@ -13,54 +13,54 @@ PhysicalTopN
 --------------------hashAgg[LOCAL]
 ----------------------PhysicalProject
 ------------------------hashJoin[INNER_JOIN](store_sales.ss_addr_sk = customer_address.ca_address_sk)
--------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = item.i_item_sk)
----------------------------PhysicalProject
------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[store_sales]
--------------------------------PhysicalDistribute
----------------------------------PhysicalProject
------------------------------------filter((date_dim.d_year = 2000)(date_dim.d_moy = 2))
--------------------------------------PhysicalOlapScan[date_dim]
----------------------------PhysicalDistribute
------------------------------hashJoin[LEFT_SEMI_JOIN](item.i_item_id = item.i_item_id)
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[item]
--------------------------------PhysicalDistribute
----------------------------------PhysicalProject
------------------------------------filter(i_color IN ('powder', 'green', 'cyan'))
--------------------------------------PhysicalOlapScan[item]
 --------------------------PhysicalDistribute
----------------------------PhysicalProject
------------------------------filter((customer_address.ca_gmt_offset = -6.00))
--------------------------------PhysicalOlapScan[customer_address]
+----------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = item.i_item_sk)
+------------------------------PhysicalProject
+--------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
+----------------------------------PhysicalProject
+------------------------------------PhysicalOlapScan[store_sales]
+----------------------------------PhysicalDistribute
+------------------------------------PhysicalProject
+--------------------------------------filter((date_dim.d_year = 2000)(date_dim.d_moy = 2))
+----------------------------------------PhysicalOlapScan[date_dim]
+------------------------------PhysicalDistribute
+--------------------------------hashJoin[LEFT_SEMI_JOIN](item.i_item_id = item.i_item_id)
+----------------------------------PhysicalProject
+------------------------------------PhysicalOlapScan[item]
+----------------------------------PhysicalDistribute
+------------------------------------PhysicalProject
+--------------------------------------filter(i_color IN ('powder', 'green', 'cyan'))
+----------------------------------------PhysicalOlapScan[item]
+--------------------------PhysicalProject
+----------------------------filter((customer_address.ca_gmt_offset = -6.00))
+------------------------------PhysicalOlapScan[customer_address]
 --------------PhysicalProject
 ----------------hashAgg[GLOBAL]
 ------------------PhysicalDistribute
 --------------------hashAgg[LOCAL]
 ----------------------PhysicalProject
 ------------------------hashJoin[INNER_JOIN](catalog_sales.cs_bill_addr_sk = customer_address.ca_address_sk)
--------------------------hashJoin[INNER_JOIN](catalog_sales.cs_item_sk = item.i_item_sk)
----------------------------PhysicalProject
------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[catalog_sales]
--------------------------------PhysicalDistribute
----------------------------------PhysicalProject
------------------------------------filter((date_dim.d_year = 2000)(date_dim.d_moy = 2))
--------------------------------------PhysicalOlapScan[date_dim]
----------------------------PhysicalDistribute
------------------------------hashJoin[LEFT_SEMI_JOIN](item.i_item_id = item.i_item_id)
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[item]
--------------------------------PhysicalDistribute
----------------------------------PhysicalProject
------------------------------------filter(i_color IN ('powder', 'green', 'cyan'))
--------------------------------------PhysicalOlapScan[item]
 --------------------------PhysicalDistribute
----------------------------PhysicalProject
------------------------------filter((customer_address.ca_gmt_offset = -6.00))
--------------------------------PhysicalOlapScan[customer_address]
+----------------------------hashJoin[INNER_JOIN](catalog_sales.cs_item_sk = item.i_item_sk)
+------------------------------PhysicalProject
+--------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
+----------------------------------PhysicalProject
+------------------------------------PhysicalOlapScan[catalog_sales]
+----------------------------------PhysicalDistribute
+------------------------------------PhysicalProject
+--------------------------------------filter((date_dim.d_year = 2000)(date_dim.d_moy = 2))
+----------------------------------------PhysicalOlapScan[date_dim]
+------------------------------PhysicalDistribute
+--------------------------------hashJoin[LEFT_SEMI_JOIN](item.i_item_id = item.i_item_id)
+----------------------------------PhysicalProject
+------------------------------------PhysicalOlapScan[item]
+----------------------------------PhysicalDistribute
+------------------------------------PhysicalProject
+--------------------------------------filter(i_color IN ('powder', 'green', 'cyan'))
+----------------------------------------PhysicalOlapScan[item]
+--------------------------PhysicalProject
+----------------------------filter((customer_address.ca_gmt_offset = -6.00))
+------------------------------PhysicalOlapScan[customer_address]
 --------------PhysicalProject
 ----------------hashAgg[GLOBAL]
 ------------------PhysicalDistribute
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query59.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query59.out
@ -24,8 +24,9 @@ CteAnchor[cteId= ( CTEId#4=] )
 ------------------PhysicalOlapScan[date_dim]
 ------------PhysicalDistribute
 --------------hashJoin[INNER_JOIN](y.s_store_id1 = x.s_store_id2)(wss.ss_store_sk = store.s_store_sk)
----------------PhysicalProject
------------------PhysicalOlapScan[store]
+----------------PhysicalDistribute
+------------------PhysicalProject
+--------------------PhysicalOlapScan[store]
 ----------------PhysicalDistribute
 ------------------hashJoin[INNER_JOIN](expr_cast(d_week_seq1 as BIGINT) = expr_(d_week_seq2 - 52))
 --------------------PhysicalProject
@ -33,8 +34,9 @@ CteAnchor[cteId= ( CTEId#4=] )
 --------------------PhysicalDistribute
 ----------------------PhysicalProject
 ------------------------hashJoin[INNER_JOIN](wss.ss_store_sk = store.s_store_sk)
--------------------------PhysicalProject
----------------------------PhysicalOlapScan[store]
+--------------------------PhysicalDistribute
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[store]
 --------------------------PhysicalDistribute
 ----------------------------hashJoin[INNER_JOIN](d.d_week_seq = d_week_seq1)
 ------------------------------PhysicalProject
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query6.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query6.out
@ -11,26 +11,28 @@ PhysicalTopN
 ----------------hashJoin[LEFT_SEMI_JOIN](j.i_category = i.i_category)(cast(i_current_price as DECIMALV3(38, 5)) > (1.2 * avg(i_current_price)))
 ------------------PhysicalProject
 --------------------hashJoin[INNER_JOIN](c.c_customer_sk = s.ss_customer_sk)
----------------------hashJoin[INNER_JOIN](s.ss_item_sk = i.i_item_sk)
+----------------------PhysicalDistribute
 ------------------------PhysicalProject
--------------------------hashJoin[INNER_JOIN](s.ss_sold_date_sk = d.d_date_sk)
----------------------------PhysicalProject
------------------------------PhysicalOlapScan[store_sales]
+--------------------------hashJoin[INNER_JOIN](s.ss_item_sk = i.i_item_sk)
 ----------------------------PhysicalDistribute
------------------------------hashJoin[INNER_JOIN](d.d_month_seq = date_dim.d_month_seq)
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[date_dim]
--------------------------------PhysicalDistribute
----------------------------------PhysicalAssertNumRows
------------------------------------hashAgg[GLOBAL]
+------------------------------PhysicalProject
+--------------------------------hashJoin[INNER_JOIN](s.ss_sold_date_sk = d.d_date_sk)
+----------------------------------PhysicalProject
+------------------------------------PhysicalOlapScan[store_sales]
+----------------------------------PhysicalDistribute
+------------------------------------hashJoin[INNER_JOIN](d.d_month_seq = date_dim.d_month_seq)
+--------------------------------------PhysicalProject
+----------------------------------------PhysicalOlapScan[date_dim]
 --------------------------------------PhysicalDistribute
----------------------------------------hashAgg[LOCAL]
------------------------------------------PhysicalProject
--------------------------------------------filter((date_dim.d_year = 2002)(date_dim.d_moy = 3))
----------------------------------------------PhysicalOlapScan[date_dim]
------------------------PhysicalDistribute
--------------------------PhysicalProject
----------------------------PhysicalOlapScan[item]
+----------------------------------------PhysicalAssertNumRows
+------------------------------------------hashAgg[GLOBAL]
+--------------------------------------------PhysicalDistribute
+----------------------------------------------hashAgg[LOCAL]
+------------------------------------------------PhysicalProject
+--------------------------------------------------filter((date_dim.d_year = 2002)(date_dim.d_moy = 3))
+----------------------------------------------------PhysicalOlapScan[date_dim]
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[item]
 ----------------------PhysicalDistribute
 ------------------------PhysicalProject
 --------------------------hashJoin[INNER_JOIN](a.ca_address_sk = c.c_current_addr_sk)
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query60.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query60.out
@ -13,19 +13,19 @@ PhysicalTopN
 --------------------hashAgg[LOCAL]
 ----------------------PhysicalProject
 ------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = item.i_item_sk)
--------------------------PhysicalProject
----------------------------hashJoin[INNER_JOIN](store_sales.ss_addr_sk = customer_address.ca_address_sk)
------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[store_sales]
--------------------------------PhysicalDistribute
+--------------------------hashJoin[INNER_JOIN](store_sales.ss_addr_sk = customer_address.ca_address_sk)
+----------------------------PhysicalDistribute
+------------------------------PhysicalProject
+--------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
 ----------------------------------PhysicalProject
------------------------------------filter((date_dim.d_year = 2000)(date_dim.d_moy = 8))
--------------------------------------PhysicalOlapScan[date_dim]
------------------------------PhysicalDistribute
--------------------------------PhysicalProject
----------------------------------filter((customer_address.ca_gmt_offset = -7.00))
------------------------------------PhysicalOlapScan[customer_address]
+------------------------------------PhysicalOlapScan[store_sales]
+----------------------------------PhysicalDistribute
+------------------------------------PhysicalProject
+--------------------------------------filter((date_dim.d_year = 2000)(date_dim.d_moy = 8))
+----------------------------------------PhysicalOlapScan[date_dim]
+----------------------------PhysicalProject
+------------------------------filter((customer_address.ca_gmt_offset = -7.00))
+--------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalDistribute
 ----------------------------hashJoin[LEFT_SEMI_JOIN](item.i_item_id = item.i_item_id)
 ------------------------------PhysicalProject
@ -40,19 +40,19 @@ PhysicalTopN
 --------------------hashAgg[LOCAL]
 ----------------------PhysicalProject
 ------------------------hashJoin[INNER_JOIN](catalog_sales.cs_item_sk = item.i_item_sk)
--------------------------PhysicalProject
----------------------------hashJoin[INNER_JOIN](catalog_sales.cs_bill_addr_sk = customer_address.ca_address_sk)
------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[catalog_sales]
--------------------------------PhysicalDistribute
+--------------------------hashJoin[INNER_JOIN](catalog_sales.cs_bill_addr_sk = customer_address.ca_address_sk)
+----------------------------PhysicalDistribute
+------------------------------PhysicalProject
+--------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
 ----------------------------------PhysicalProject
------------------------------------filter((date_dim.d_moy = 8)(date_dim.d_year = 2000))
--------------------------------------PhysicalOlapScan[date_dim]
------------------------------PhysicalDistribute
--------------------------------PhysicalProject
----------------------------------filter((customer_address.ca_gmt_offset = -7.00))
------------------------------------PhysicalOlapScan[customer_address]
+------------------------------------PhysicalOlapScan[catalog_sales]
+----------------------------------PhysicalDistribute
+------------------------------------PhysicalProject
+--------------------------------------filter((date_dim.d_moy = 8)(date_dim.d_year = 2000))
+----------------------------------------PhysicalOlapScan[date_dim]
+----------------------------PhysicalProject
+------------------------------filter((customer_address.ca_gmt_offset = -7.00))
+--------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalDistribute
 ----------------------------hashJoin[LEFT_SEMI_JOIN](item.i_item_id = item.i_item_id)
 ------------------------------PhysicalProject
@ -67,19 +67,19 @@ PhysicalTopN
 --------------------hashAgg[LOCAL]
 ----------------------PhysicalProject
 ------------------------hashJoin[INNER_JOIN](web_sales.ws_item_sk = item.i_item_sk)
--------------------------PhysicalProject
----------------------------hashJoin[INNER_JOIN](web_sales.ws_bill_addr_sk = customer_address.ca_address_sk)
------------------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[web_sales]
--------------------------------PhysicalDistribute
+--------------------------hashJoin[INNER_JOIN](web_sales.ws_bill_addr_sk = customer_address.ca_address_sk)
+----------------------------PhysicalDistribute
+------------------------------PhysicalProject
+--------------------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
 ----------------------------------PhysicalProject
------------------------------------filter((date_dim.d_year = 2000)(date_dim.d_moy = 8))
--------------------------------------PhysicalOlapScan[date_dim]
------------------------------PhysicalDistribute
--------------------------------PhysicalProject
----------------------------------filter((customer_address.ca_gmt_offset = -7.00))
------------------------------------PhysicalOlapScan[customer_address]
+------------------------------------PhysicalOlapScan[web_sales]
+----------------------------------PhysicalDistribute
+------------------------------------PhysicalProject
+--------------------------------------filter((date_dim.d_year = 2000)(date_dim.d_moy = 8))
+----------------------------------------PhysicalOlapScan[date_dim]
+----------------------------PhysicalProject
+------------------------------filter((customer_address.ca_gmt_offset = -7.00))
+--------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalDistribute
 ----------------------------hashJoin[LEFT_SEMI_JOIN](item.i_item_id = item.i_item_id)
 ------------------------------PhysicalProject
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query64.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query64.out
@ -1,209 +1,118 @@
 -- This file is automatically generated. You should know what you did if you want to edit this
 -- !ds_shape_64 --
-PhysicalQuickSort
--PhysicalDistribute
----PhysicalQuickSort
------PhysicalProject
--------hashJoin[INNER_JOIN](cs1.item_sk = cs2.item_sk)(cs1.store_name = cs2.store_name)(cs1.store_zip = cs2.store_zip)(cs2.cnt <= cs1.cnt)
----------PhysicalProject
------------hashAgg[GLOBAL]
--------------PhysicalDistribute
----------------hashAgg[LOCAL]
+CteAnchor[cteId= ( CTEId#14=] )
+--CteProducer[cteId= ( CTEId#14=] )
+----PhysicalProject
+------hashAgg[GLOBAL]
+--------PhysicalDistribute
+----------hashAgg[LOCAL]
+------------PhysicalProject
+--------------hashJoin[INNER_JOIN](customer.c_current_addr_sk = ad2.ca_address_sk)
+----------------PhysicalProject
+------------------PhysicalOlapScan[customer_address]
+----------------PhysicalDistribute
 ------------------PhysicalProject
--------------------hashJoin[INNER_JOIN](customer.c_first_shipto_date_sk = d3.d_date_sk)
+--------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = cs_ui.cs_item_sk)
 ----------------------PhysicalProject
------------------------PhysicalOlapScan[date_dim]
+------------------------filter((sale > (2 * refund)))
+--------------------------hashAgg[GLOBAL]
+----------------------------PhysicalDistribute
+------------------------------hashAgg[LOCAL]
+--------------------------------PhysicalProject
+----------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_item_sk = catalog_returns.cr_item_sk)(catalog_sales.cs_order_number = catalog_returns.cr_order_number)
+------------------------------------PhysicalProject
+--------------------------------------PhysicalOlapScan[catalog_sales]
+------------------------------------PhysicalProject
+--------------------------------------PhysicalOlapScan[catalog_returns]
 ----------------------PhysicalDistribute
 ------------------------PhysicalProject
--------------------------hashJoin[INNER_JOIN](hd1.hd_income_band_sk = ib1.ib_income_band_sk)
+--------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = d1.d_date_sk)
 ----------------------------PhysicalProject
------------------------------PhysicalOlapScan[income_band]
+------------------------------filter(((d1.d_year = 2001) OR (d1.d_year = 2002)))
+--------------------------------PhysicalOlapScan[date_dim]
 ----------------------------PhysicalDistribute
 ------------------------------PhysicalProject
--------------------------------hashJoin[INNER_JOIN](hd2.hd_income_band_sk = ib2.ib_income_band_sk)
+--------------------------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = hd1.hd_demo_sk)
 ----------------------------------PhysicalProject
------------------------------------PhysicalOlapScan[income_band]
+------------------------------------hashJoin[INNER_JOIN](hd1.hd_income_band_sk = ib1.ib_income_band_sk)
+--------------------------------------PhysicalProject
+----------------------------------------PhysicalOlapScan[household_demographics]
+--------------------------------------PhysicalDistribute
+----------------------------------------PhysicalProject
+------------------------------------------PhysicalOlapScan[income_band]
 ----------------------------------PhysicalDistribute
 ------------------------------------PhysicalProject
--------------------------------------hashJoin[INNER_JOIN](customer.c_current_addr_sk = ad2.ca_address_sk)
+--------------------------------------hashJoin[INNER_JOIN](store_sales.ss_promo_sk = promotion.p_promo_sk)
 ----------------------------------------PhysicalProject
------------------------------------------PhysicalOlapScan[customer_address]
+------------------------------------------PhysicalOlapScan[promotion]
 ----------------------------------------PhysicalDistribute
 ------------------------------------------PhysicalProject
--------------------------------------------hashJoin[INNER_JOIN](customer.c_first_sales_date_sk = d2.d_date_sk)
+--------------------------------------------hashJoin[INNER_JOIN](customer.c_first_shipto_date_sk = d3.d_date_sk)
 ----------------------------------------------PhysicalProject
 ------------------------------------------------PhysicalOlapScan[date_dim]
 ----------------------------------------------PhysicalDistribute
 ------------------------------------------------PhysicalProject
--------------------------------------------------hashJoin[INNER_JOIN](customer.c_current_hdemo_sk = hd2.hd_demo_sk)
----------------------------------------------------PhysicalProject
------------------------------------------------------PhysicalOlapScan[household_demographics]
+--------------------------------------------------hashJoin[INNER_JOIN](hd2.hd_income_band_sk = ib2.ib_income_band_sk)
 ----------------------------------------------------PhysicalDistribute
 ------------------------------------------------------PhysicalProject
--------------------------------------------------------hashJoin[INNER_JOIN](customer.c_current_cdemo_sk = cd2.cd_demo_sk)( not (cd_marital_status = cd_marital_status))
+--------------------------------------------------------PhysicalOlapScan[income_band]
+----------------------------------------------------PhysicalDistribute
+------------------------------------------------------PhysicalProject
+--------------------------------------------------------hashJoin[INNER_JOIN](customer.c_first_sales_date_sk = d2.d_date_sk)
 ----------------------------------------------------------PhysicalProject
------------------------------------------------------------PhysicalOlapScan[customer_demographics]
+------------------------------------------------------------PhysicalOlapScan[date_dim]
 ----------------------------------------------------------PhysicalDistribute
 ------------------------------------------------------------PhysicalProject
--------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_customer_sk = customer.c_customer_sk)
+--------------------------------------------------------------hashJoin[INNER_JOIN](customer.c_current_hdemo_sk = hd2.hd_demo_sk)
 ----------------------------------------------------------------PhysicalProject
------------------------------------------------------------------PhysicalOlapScan[customer]
+------------------------------------------------------------------PhysicalOlapScan[household_demographics]
 ----------------------------------------------------------------PhysicalDistribute
 ------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_cdemo_sk = cd1.cd_demo_sk)
+--------------------------------------------------------------------hashJoin[INNER_JOIN](customer.c_current_cdemo_sk = cd2.cd_demo_sk)( not (cd_marital_status = cd_marital_status))
 ----------------------------------------------------------------------PhysicalProject
 ------------------------------------------------------------------------PhysicalOlapScan[customer_demographics]
 ----------------------------------------------------------------------PhysicalDistribute
 ------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = cs_ui.cs_item_sk)
+--------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_customer_sk = customer.c_customer_sk)
 ----------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------filter((sale > (2 * refund)))
--------------------------------------------------------------------------------hashAgg[GLOBAL]
----------------------------------------------------------------------------------PhysicalDistribute
------------------------------------------------------------------------------------hashAgg[LOCAL]
--------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_item_sk = catalog_returns.cr_item_sk)(catalog_sales.cs_order_number = catalog_returns.cr_order_number)
------------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------------PhysicalOlapScan[catalog_sales]
------------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------------PhysicalOlapScan[catalog_returns]
+------------------------------------------------------------------------------PhysicalOlapScan[customer]
 ----------------------------------------------------------------------------PhysicalDistribute
 ------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = d1.d_date_sk)
+--------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_cdemo_sk = cd1.cd_demo_sk)
 ----------------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = hd1.hd_demo_sk)
--------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------PhysicalOlapScan[household_demographics]
--------------------------------------------------------------------------------------PhysicalDistribute
----------------------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_promo_sk = promotion.p_promo_sk)
--------------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
------------------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_addr_sk = ad1.ca_address_sk)
----------------------------------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------------------------------PhysicalOlapScan[customer_address]
----------------------------------------------------------------------------------------------------PhysicalDistribute
------------------------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = store_returns.sr_item_sk)(store_sales.ss_ticket_number = store_returns.sr_ticket_number)
----------------------------------------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------------------------------------PhysicalOlapScan[store_returns]
----------------------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = item_sk)
------------------------------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------------------------------PhysicalOlapScan[store_sales]
------------------------------------------------------------------------------------------------------------PhysicalDistribute
--------------------------------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------------------------------filter((item.i_current_price >= 24.00)(item.i_current_price <= 33.00)i_color IN ('blanched', 'medium', 'brown', 'chocolate', 'burlywood', 'drab'))
------------------------------------------------------------------------------------------------------------------PhysicalOlapScan[item]
------------------------------------------------------------------------------------------------PhysicalDistribute
--------------------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------------------PhysicalOlapScan[store]
--------------------------------------------------------------------------------------------PhysicalDistribute
----------------------------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------------------------PhysicalOlapScan[promotion]
+------------------------------------------------------------------------------------PhysicalOlapScan[customer_demographics]
 ----------------------------------------------------------------------------------PhysicalDistribute
 ------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------filter((d1.d_year = 2001))
----------------------------------------------------------------------------------------PhysicalOlapScan[date_dim]
----------PhysicalDistribute
------------PhysicalProject
--------------hashAgg[GLOBAL]
----------------PhysicalDistribute
------------------hashAgg[LOCAL]
--------------------PhysicalProject
----------------------hashJoin[INNER_JOIN](customer.c_first_shipto_date_sk = d3.d_date_sk)
------------------------PhysicalProject
--------------------------PhysicalOlapScan[date_dim]
------------------------PhysicalDistribute
--------------------------PhysicalProject
----------------------------hashJoin[INNER_JOIN](hd1.hd_income_band_sk = ib1.ib_income_band_sk)
------------------------------PhysicalProject
--------------------------------PhysicalOlapScan[income_band]
------------------------------PhysicalDistribute
--------------------------------PhysicalProject
----------------------------------hashJoin[INNER_JOIN](hd2.hd_income_band_sk = ib2.ib_income_band_sk)
------------------------------------PhysicalProject
--------------------------------------PhysicalOlapScan[income_band]
------------------------------------PhysicalDistribute
--------------------------------------PhysicalProject
----------------------------------------hashJoin[INNER_JOIN](customer.c_current_addr_sk = ad2.ca_address_sk)
------------------------------------------PhysicalProject
--------------------------------------------PhysicalOlapScan[customer_address]
------------------------------------------PhysicalDistribute
--------------------------------------------PhysicalProject
----------------------------------------------hashJoin[INNER_JOIN](customer.c_first_sales_date_sk = d2.d_date_sk)
------------------------------------------------PhysicalProject
--------------------------------------------------PhysicalOlapScan[date_dim]
------------------------------------------------PhysicalDistribute
--------------------------------------------------PhysicalProject
----------------------------------------------------hashJoin[INNER_JOIN](customer.c_current_hdemo_sk = hd2.hd_demo_sk)
------------------------------------------------------PhysicalProject
--------------------------------------------------------PhysicalOlapScan[household_demographics]
------------------------------------------------------PhysicalDistribute
--------------------------------------------------------PhysicalProject
----------------------------------------------------------hashJoin[INNER_JOIN](customer.c_current_cdemo_sk = cd2.cd_demo_sk)( not (cd_marital_status = cd_marital_status))
------------------------------------------------------------PhysicalProject
--------------------------------------------------------------PhysicalOlapScan[customer_demographics]
------------------------------------------------------------PhysicalDistribute
--------------------------------------------------------------PhysicalProject
----------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_customer_sk = customer.c_customer_sk)
------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------PhysicalOlapScan[customer]
------------------------------------------------------------------PhysicalDistribute
--------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_cdemo_sk = cd1.cd_demo_sk)
------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------PhysicalOlapScan[customer_demographics]
------------------------------------------------------------------------PhysicalDistribute
--------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = cs_ui.cs_item_sk)
------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------filter((sale > (2 * refund)))
----------------------------------------------------------------------------------hashAgg[GLOBAL]
------------------------------------------------------------------------------------PhysicalDistribute
--------------------------------------------------------------------------------------hashAgg[LOCAL]
+--------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
 ----------------------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](catalog_sales.cs_item_sk = catalog_returns.cr_item_sk)(catalog_sales.cs_order_number = catalog_returns.cr_order_number)
+------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_addr_sk = ad1.ca_address_sk)
 --------------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------------PhysicalOlapScan[catalog_sales]
--------------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------------PhysicalOlapScan[catalog_returns]
------------------------------------------------------------------------------PhysicalDistribute
--------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = d1.d_date_sk)
------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = hd1.hd_demo_sk)
----------------------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------------------PhysicalOlapScan[household_demographics]
+----------------------------------------------------------------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------------------------------------------------------------------PhysicalDistribute
+----------------------------------------------------------------------------------------------PhysicalProject
+------------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = store_returns.sr_item_sk)(store_sales.ss_ticket_number = store_returns.sr_ticket_number)
+--------------------------------------------------------------------------------------------------PhysicalProject
+----------------------------------------------------------------------------------------------------PhysicalOlapScan[store_returns]
+--------------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = item_sk)
+----------------------------------------------------------------------------------------------------PhysicalProject
+------------------------------------------------------------------------------------------------------PhysicalOlapScan[store_sales]
+----------------------------------------------------------------------------------------------------PhysicalDistribute
+------------------------------------------------------------------------------------------------------PhysicalProject
+--------------------------------------------------------------------------------------------------------filter((item.i_current_price >= 24.00)(item.i_current_price <= 33.00)i_color IN ('blanched', 'medium', 'brown', 'chocolate', 'burlywood', 'drab'))
+----------------------------------------------------------------------------------------------------------PhysicalOlapScan[item]
 ----------------------------------------------------------------------------------------PhysicalDistribute
 ------------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_promo_sk = promotion.p_promo_sk)
----------------------------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
--------------------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_addr_sk = ad1.ca_address_sk)
------------------------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------------------------PhysicalOlapScan[customer_address]
------------------------------------------------------------------------------------------------------PhysicalDistribute
--------------------------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = store_returns.sr_item_sk)(store_sales.ss_ticket_number = store_returns.sr_ticket_number)
------------------------------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------------------------------PhysicalOlapScan[store_returns]
------------------------------------------------------------------------------------------------------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = item_sk)
--------------------------------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------------------------------PhysicalOlapScan[store_sales]
--------------------------------------------------------------------------------------------------------------PhysicalDistribute
----------------------------------------------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------------------------------------------filter((item.i_current_price >= 24.00)(item.i_current_price <= 33.00)i_color IN ('blanched', 'medium', 'brown', 'chocolate', 'burlywood', 'drab'))
--------------------------------------------------------------------------------------------------------------------PhysicalOlapScan[item]
--------------------------------------------------------------------------------------------------PhysicalDistribute
----------------------------------------------------------------------------------------------------PhysicalProject
------------------------------------------------------------------------------------------------------PhysicalOlapScan[store]
----------------------------------------------------------------------------------------------PhysicalDistribute
------------------------------------------------------------------------------------------------PhysicalProject
--------------------------------------------------------------------------------------------------PhysicalOlapScan[promotion]
------------------------------------------------------------------------------------PhysicalDistribute
--------------------------------------------------------------------------------------PhysicalProject
----------------------------------------------------------------------------------------filter((d1.d_year = 2002))
------------------------------------------------------------------------------------------PhysicalOlapScan[date_dim]
+--------------------------------------------------------------------------------------------PhysicalOlapScan[store]
+--PhysicalQuickSort
+----PhysicalDistribute
+------PhysicalQuickSort
+--------PhysicalProject
+----------hashJoin[INNER_JOIN](cs1.item_sk = cs2.item_sk)(cs1.store_name = cs2.store_name)(cs1.store_zip = cs2.store_zip)(cs2.cnt <= cs1.cnt)
+------------PhysicalProject
+--------------filter((cs1.syear = 2001))
+----------------CteConsumer[cteId= ( CTEId#14=] )
+------------PhysicalDistribute
+--------------PhysicalProject
+----------------filter((cs2.syear = 2002))
+------------------CteConsumer[cteId= ( CTEId#14=] )

--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query68.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query68.out
@ -10,8 +10,9 @@ PhysicalTopN
 ----------PhysicalDistribute
 ------------PhysicalProject
 --------------hashJoin[INNER_JOIN](dn.ss_customer_sk = customer.c_customer_sk)
----------------PhysicalProject
------------------PhysicalOlapScan[customer]
+----------------PhysicalDistribute
+------------------PhysicalProject
+--------------------PhysicalOlapScan[customer]
 ----------------PhysicalDistribute
 ------------------PhysicalProject
 --------------------hashAgg[LOCAL]
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query7.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query7.out
@ -8,25 +8,25 @@ PhysicalTopN
 ----------hashAgg[LOCAL]
 ------------PhysicalProject
 --------------hashJoin[INNER_JOIN](store_sales.ss_item_sk = item.i_item_sk)
----------------PhysicalProject
+----------------PhysicalDistribute
 ------------------hashJoin[INNER_JOIN](store_sales.ss_promo_sk = promotion.p_promo_sk)
--------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
----------------------hashJoin[INNER_JOIN](store_sales.ss_cdemo_sk = customer_demographics.cd_demo_sk)
------------------------PhysicalProject
--------------------------PhysicalOlapScan[store_sales]
+--------------------PhysicalProject
+----------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
+------------------------hashJoin[INNER_JOIN](store_sales.ss_cdemo_sk = customer_demographics.cd_demo_sk)
+--------------------------PhysicalProject
+----------------------------PhysicalOlapScan[store_sales]
+--------------------------PhysicalDistribute
+----------------------------PhysicalProject
+------------------------------filter((cast(cd_marital_status as VARCHAR(*)) = 'W')(cast(cd_education_status as VARCHAR(*)) = 'College')(cast(cd_gender as VARCHAR(*)) = 'F'))
+--------------------------------PhysicalOlapScan[customer_demographics]
 ------------------------PhysicalDistribute
 --------------------------PhysicalProject
----------------------------filter((cast(cd_marital_status as VARCHAR(*)) = 'W')(cast(cd_education_status as VARCHAR(*)) = 'College')(cast(cd_gender as VARCHAR(*)) = 'F'))
------------------------------PhysicalOlapScan[customer_demographics]
----------------------PhysicalDistribute
------------------------PhysicalProject
--------------------------filter((date_dim.d_year = 2001))
----------------------------PhysicalOlapScan[date_dim]
+----------------------------filter((date_dim.d_year = 2001))
+------------------------------PhysicalOlapScan[date_dim]
 --------------------PhysicalDistribute
 ----------------------PhysicalProject
 ------------------------filter(((cast(p_channel_email as VARCHAR(*)) = 'N') OR (cast(p_channel_event as VARCHAR(*)) = 'N')))
 --------------------------PhysicalOlapScan[promotion]
----------------PhysicalDistribute
------------------PhysicalProject
--------------------PhysicalOlapScan[item]
+----------------PhysicalProject
+------------------PhysicalOlapScan[item]

--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query74.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query74.out
@ -27,14 +27,15 @@ CteAnchor[cteId= ( CTEId#4=] )
 ------------hashAgg[LOCAL]
 --------------PhysicalProject
 ----------------hashJoin[INNER_JOIN](customer.c_customer_sk = web_sales.ws_bill_customer_sk)
------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
--------------------PhysicalProject
----------------------filter((('w' = 's') OR ('w' = 'w')))
------------------------PhysicalOlapScan[web_sales]
--------------------PhysicalDistribute
+------------------PhysicalDistribute
+--------------------hashJoin[INNER_JOIN](web_sales.ws_sold_date_sk = date_dim.d_date_sk)
 ----------------------PhysicalProject
------------------------filter(((date_dim.d_year = 1999) OR (date_dim.d_year = 2000))(('w' = 's') OR ('w' = 'w')))
--------------------------PhysicalOlapScan[date_dim]
+------------------------filter((('w' = 's') OR ('w' = 'w')))
+--------------------------PhysicalOlapScan[web_sales]
+----------------------PhysicalDistribute
+------------------------PhysicalProject
+--------------------------filter(((date_dim.d_year = 1999) OR (date_dim.d_year = 2000))(('w' = 's') OR ('w' = 'w')))
+----------------------------PhysicalOlapScan[date_dim]
 ------------------PhysicalDistribute
 --------------------PhysicalProject
 ----------------------filter((('w' = 's') OR ('w' = 'w')))
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query79.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query79.out
@ -5,28 +5,29 @@ PhysicalTopN
 ----PhysicalTopN
 ------PhysicalProject
 --------hashJoin[INNER_JOIN](ms.ss_customer_sk = customer.c_customer_sk)
----------PhysicalProject
------------hashAgg[GLOBAL]
--------------PhysicalDistribute
----------------hashAgg[LOCAL]
------------------PhysicalProject
--------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
----------------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = household_demographics.hd_demo_sk)
------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
--------------------------PhysicalProject
----------------------------PhysicalOlapScan[store_sales]
+----------PhysicalDistribute
+------------PhysicalProject
+--------------hashAgg[GLOBAL]
+----------------PhysicalDistribute
+------------------hashAgg[LOCAL]
+--------------------PhysicalProject
+----------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
+------------------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = household_demographics.hd_demo_sk)
+--------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[store_sales]
+----------------------------PhysicalDistribute
+------------------------------PhysicalProject
+--------------------------------filter(d_year IN (1998, 1999, 2000)(date_dim.d_dow = 1))
+----------------------------------PhysicalOlapScan[date_dim]
 --------------------------PhysicalDistribute
 ----------------------------PhysicalProject
------------------------------filter(d_year IN (1998, 1999, 2000)(date_dim.d_dow = 1))
--------------------------------PhysicalOlapScan[date_dim]
+------------------------------filter(((household_demographics.hd_dep_count = 5) OR (household_demographics.hd_vehicle_count > 4)))
+--------------------------------PhysicalOlapScan[household_demographics]
 ------------------------PhysicalDistribute
 --------------------------PhysicalProject
----------------------------filter(((household_demographics.hd_dep_count = 5) OR (household_demographics.hd_vehicle_count > 4)))
------------------------------PhysicalOlapScan[household_demographics]
----------------------PhysicalDistribute
------------------------PhysicalProject
--------------------------filter((store.s_number_employees >= 200)(store.s_number_employees <= 295))
----------------------------PhysicalOlapScan[store]
+----------------------------filter((store.s_number_employees >= 200)(store.s_number_employees <= 295))
+------------------------------PhysicalOlapScan[store]
 ----------PhysicalDistribute
 ------------PhysicalProject
 --------------PhysicalOlapScan[customer]
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query87.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query87.out
@ -11,13 +11,15 @@ hashAgg[GLOBAL]
 ----------------hashAgg[LOCAL]
 ------------------PhysicalProject
 --------------------hashJoin[INNER_JOIN](store_sales.ss_customer_sk = customer.c_customer_sk)
----------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
+----------------------PhysicalDistribute
 ------------------------PhysicalProject
--------------------------PhysicalOlapScan[store_sales]
------------------------PhysicalDistribute
--------------------------PhysicalProject
----------------------------filter((date_dim.d_month_seq >= 1184)(date_dim.d_month_seq <= 1195))
------------------------------PhysicalOlapScan[date_dim]
+--------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_date_sk = date_dim.d_date_sk)
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[store_sales]
+----------------------------PhysicalDistribute
+------------------------------PhysicalProject
+--------------------------------filter((date_dim.d_month_seq >= 1184)(date_dim.d_month_seq <= 1195))
+----------------------------------PhysicalOlapScan[date_dim]
 ----------------------PhysicalDistribute
 ------------------------PhysicalProject
 --------------------------PhysicalOlapScan[customer]
@ -27,13 +29,15 @@ hashAgg[GLOBAL]
 ----------------hashAgg[LOCAL]
 ------------------PhysicalProject
 --------------------hashJoin[INNER_JOIN](catalog_sales.cs_bill_customer_sk = customer.c_customer_sk)
----------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
+----------------------PhysicalDistribute
 ------------------------PhysicalProject
--------------------------PhysicalOlapScan[catalog_sales]
------------------------PhysicalDistribute
--------------------------PhysicalProject
----------------------------filter((date_dim.d_month_seq >= 1184)(date_dim.d_month_seq <= 1195))
------------------------------PhysicalOlapScan[date_dim]
+--------------------------hashJoin[INNER_JOIN](catalog_sales.cs_sold_date_sk = date_dim.d_date_sk)
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[catalog_sales]
+----------------------------PhysicalDistribute
+------------------------------PhysicalProject
+--------------------------------filter((date_dim.d_month_seq >= 1184)(date_dim.d_month_seq <= 1195))
+----------------------------------PhysicalOlapScan[date_dim]
 ----------------------PhysicalDistribute
 ------------------------PhysicalProject
 --------------------------PhysicalOlapScan[customer]
--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query88.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query88.out
@ -1,33 +1,56 @@
 -- This file is automatically generated. You should know what you did if you want to edit this
 -- !ds_shape_88 --
-NestedLoopJoin[CROSS_JOIN]
+PhysicalProject
 --NestedLoopJoin[CROSS_JOIN]
 ----NestedLoopJoin[CROSS_JOIN]
 ------NestedLoopJoin[CROSS_JOIN]
 --------NestedLoopJoin[CROSS_JOIN]
 ----------NestedLoopJoin[CROSS_JOIN]
 ------------NestedLoopJoin[CROSS_JOIN]
--------------hashAgg[GLOBAL]
----------------PhysicalDistribute
------------------hashAgg[LOCAL]
--------------------PhysicalProject
----------------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = household_demographics.hd_demo_sk)
------------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
--------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_time_sk = time_dim.t_time_sk)
----------------------------PhysicalProject
------------------------------PhysicalOlapScan[store_sales]
+--------------NestedLoopJoin[CROSS_JOIN]
+----------------hashAgg[GLOBAL]
+------------------PhysicalDistribute
+--------------------hashAgg[LOCAL]
+----------------------PhysicalProject
+------------------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = household_demographics.hd_demo_sk)
+--------------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
+----------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_time_sk = time_dim.t_time_sk)
+------------------------------PhysicalProject
+--------------------------------PhysicalOlapScan[store_sales]
+------------------------------PhysicalDistribute
+--------------------------------PhysicalProject
+----------------------------------filter((time_dim.t_minute >= 30)(time_dim.t_hour = 8))
+------------------------------------PhysicalOlapScan[time_dim]
 ----------------------------PhysicalDistribute
 ------------------------------PhysicalProject
--------------------------------filter((time_dim.t_minute >= 30)(time_dim.t_hour = 8))
----------------------------------PhysicalOlapScan[time_dim]
+--------------------------------filter((store.s_store_name = 'ese'))
+----------------------------------PhysicalOlapScan[store]
 --------------------------PhysicalDistribute
 ----------------------------PhysicalProject
------------------------------filter((store.s_store_name = 'ese'))
--------------------------------PhysicalOlapScan[store]
------------------------PhysicalDistribute
--------------------------PhysicalProject
----------------------------filter(((((household_demographics.hd_dep_count = -1) AND (household_demographics.hd_vehicle_count <= 1)) OR ((household_demographics.hd_dep_count = 4) AND (household_demographics.hd_vehicle_count <= 6))) OR ((household_demographics.hd_dep_count = 3) AND (household_demographics.hd_vehicle_count <= 5))))
------------------------------PhysicalOlapScan[household_demographics]
+------------------------------filter(((((household_demographics.hd_dep_count = -1) AND (household_demographics.hd_vehicle_count <= 1)) OR ((household_demographics.hd_dep_count = 4) AND (household_demographics.hd_vehicle_count <= 6))) OR ((household_demographics.hd_dep_count = 3) AND (household_demographics.hd_vehicle_count <= 5))))
+--------------------------------PhysicalOlapScan[household_demographics]
+----------------PhysicalDistribute
+------------------hashAgg[GLOBAL]
+--------------------PhysicalDistribute
+----------------------hashAgg[LOCAL]
+------------------------PhysicalProject
+--------------------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = household_demographics.hd_demo_sk)
+----------------------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
+------------------------------hashJoin[INNER_JOIN](store_sales.ss_sold_time_sk = time_dim.t_time_sk)
+--------------------------------PhysicalProject
+----------------------------------PhysicalOlapScan[store_sales]
+--------------------------------PhysicalDistribute
+----------------------------------PhysicalProject
+------------------------------------filter((time_dim.t_hour = 9)(time_dim.t_minute < 30))
+--------------------------------------PhysicalOlapScan[time_dim]
+------------------------------PhysicalDistribute
+--------------------------------PhysicalProject
+----------------------------------filter((store.s_store_name = 'ese'))
+------------------------------------PhysicalOlapScan[store]
+----------------------------PhysicalDistribute
+------------------------------PhysicalProject
+--------------------------------filter(((((household_demographics.hd_dep_count = -1) AND (household_demographics.hd_vehicle_count <= 1)) OR ((household_demographics.hd_dep_count = 4) AND (household_demographics.hd_vehicle_count <= 6))) OR ((household_demographics.hd_dep_count = 3) AND (household_demographics.hd_vehicle_count <= 5))))
+----------------------------------PhysicalOlapScan[household_demographics]
 --------------PhysicalDistribute
 ----------------hashAgg[GLOBAL]
 ------------------PhysicalDistribute
@ -40,7 +63,7 @@ NestedLoopJoin[CROSS_JOIN]
 --------------------------------PhysicalOlapScan[store_sales]
 ------------------------------PhysicalDistribute
 --------------------------------PhysicalProject
----------------------------------filter((time_dim.t_hour = 9)(time_dim.t_minute < 30))
+----------------------------------filter((time_dim.t_minute >= 30)(time_dim.t_hour = 9))
 ------------------------------------PhysicalOlapScan[time_dim]
 ----------------------------PhysicalDistribute
 ------------------------------PhysicalProject
@ -62,7 +85,7 @@ NestedLoopJoin[CROSS_JOIN]
 ------------------------------PhysicalOlapScan[store_sales]
 ----------------------------PhysicalDistribute
 ------------------------------PhysicalProject
--------------------------------filter((time_dim.t_minute >= 30)(time_dim.t_hour = 9))
+--------------------------------filter((time_dim.t_hour = 10)(time_dim.t_minute < 30))
 ----------------------------------PhysicalOlapScan[time_dim]
 --------------------------PhysicalDistribute
 ----------------------------PhysicalProject
@ -84,7 +107,7 @@ NestedLoopJoin[CROSS_JOIN]
 ----------------------------PhysicalOlapScan[store_sales]
 --------------------------PhysicalDistribute
 ----------------------------PhysicalProject
------------------------------filter((time_dim.t_hour = 10)(time_dim.t_minute < 30))
+------------------------------filter((time_dim.t_minute >= 30)(time_dim.t_hour = 10))
 --------------------------------PhysicalOlapScan[time_dim]
 ------------------------PhysicalDistribute
 --------------------------PhysicalProject
@ -106,7 +129,7 @@ NestedLoopJoin[CROSS_JOIN]
 --------------------------PhysicalOlapScan[store_sales]
 ------------------------PhysicalDistribute
 --------------------------PhysicalProject
----------------------------filter((time_dim.t_minute >= 30)(time_dim.t_hour = 10))
+----------------------------filter((time_dim.t_minute < 30)(time_dim.t_hour = 11))
 ------------------------------PhysicalOlapScan[time_dim]
 ----------------------PhysicalDistribute
 ------------------------PhysicalProject
@ -128,7 +151,7 @@ NestedLoopJoin[CROSS_JOIN]
 ------------------------PhysicalOlapScan[store_sales]
 ----------------------PhysicalDistribute
 ------------------------PhysicalProject
--------------------------filter((time_dim.t_minute < 30)(time_dim.t_hour = 11))
+--------------------------filter((time_dim.t_hour = 11)(time_dim.t_minute >= 30))
 ----------------------------PhysicalOlapScan[time_dim]
 --------------------PhysicalDistribute
 ----------------------PhysicalProject
@ -150,7 +173,7 @@ NestedLoopJoin[CROSS_JOIN]
 ----------------------PhysicalOlapScan[store_sales]
 --------------------PhysicalDistribute
 ----------------------PhysicalProject
------------------------filter((time_dim.t_hour = 11)(time_dim.t_minute >= 30))
+------------------------filter((time_dim.t_hour = 12)(time_dim.t_minute < 30))
 --------------------------PhysicalOlapScan[time_dim]
 ------------------PhysicalDistribute
 --------------------PhysicalProject
@ -160,26 +183,4 @@ NestedLoopJoin[CROSS_JOIN]
 ------------------PhysicalProject
 --------------------filter(((((household_demographics.hd_dep_count = -1) AND (household_demographics.hd_vehicle_count <= 1)) OR ((household_demographics.hd_dep_count = 4) AND (household_demographics.hd_vehicle_count <= 6))) OR ((household_demographics.hd_dep_count = 3) AND (household_demographics.hd_vehicle_count <= 5))))
 ----------------------PhysicalOlapScan[household_demographics]
--PhysicalDistribute
----hashAgg[GLOBAL]
------PhysicalDistribute
--------hashAgg[LOCAL]
----------PhysicalProject
------------hashJoin[INNER_JOIN](store_sales.ss_hdemo_sk = household_demographics.hd_demo_sk)
--------------hashJoin[INNER_JOIN](store_sales.ss_store_sk = store.s_store_sk)
----------------hashJoin[INNER_JOIN](store_sales.ss_sold_time_sk = time_dim.t_time_sk)
------------------PhysicalProject
--------------------PhysicalOlapScan[store_sales]
------------------PhysicalDistribute
--------------------PhysicalProject
----------------------filter((time_dim.t_hour = 12)(time_dim.t_minute < 30))
------------------------PhysicalOlapScan[time_dim]
----------------PhysicalDistribute
------------------PhysicalProject
--------------------filter((store.s_store_name = 'ese'))
----------------------PhysicalOlapScan[store]
--------------PhysicalDistribute
----------------PhysicalProject
------------------filter(((((household_demographics.hd_dep_count = -1) AND (household_demographics.hd_vehicle_count <= 1)) OR ((household_demographics.hd_dep_count = 4) AND (household_demographics.hd_vehicle_count <= 6))) OR ((household_demographics.hd_dep_count = 3) AND (household_demographics.hd_vehicle_count <= 5))))
--------------------PhysicalOlapScan[household_demographics]

--- a/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query91.out
+++ b/regression-test/data/nereids_tpcds_shape_sf100_p0/shape/query91.out
@ -23,8 +23,9 @@ PhysicalQuickSort
 --------------------------hashJoin[INNER_JOIN](household_demographics.hd_demo_sk = customer.c_current_hdemo_sk)
 ----------------------------PhysicalProject
 ------------------------------hashJoin[INNER_JOIN](catalog_returns.cr_returning_customer_sk = customer.c_customer_sk)
--------------------------------PhysicalProject
----------------------------------PhysicalOlapScan[customer]
+--------------------------------PhysicalDistribute
+----------------------------------PhysicalProject
+------------------------------------PhysicalOlapScan[customer]
 --------------------------------PhysicalDistribute
 ----------------------------------PhysicalProject
 ------------------------------------hashJoin[INNER_JOIN](catalog_returns.cr_returned_date_sk = date_dim.d_date_sk)
--- a/regression-test/data/nereids_tpch_shape_sf1000_p0/shape/q3.out
+++ b/regression-test/data/nereids_tpch_shape_sf1000_p0/shape/q3.out
@ -10,12 +10,13 @@ PhysicalTopN
 --------------PhysicalProject
 ----------------filter((lineitem.l_shipdate > 1995-03-15))
 ------------------PhysicalOlapScan[lineitem]
--------------PhysicalProject
----------------hashJoin[INNER_JOIN](customer.c_custkey = orders.o_custkey)
------------------PhysicalProject
--------------------filter((orders.o_orderdate < 1995-03-15))
----------------------PhysicalOlapScan[orders]
------------------PhysicalDistribute
+--------------PhysicalDistribute
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN](customer.c_custkey = orders.o_custkey)
+--------------------PhysicalDistribute
+----------------------PhysicalProject
+------------------------filter((orders.o_orderdate < 1995-03-15))
+--------------------------PhysicalOlapScan[orders]
 --------------------PhysicalProject
 ----------------------filter((customer.c_mktsegment = 'BUILDING'))
 ------------------------PhysicalOlapScan[customer]
--- a/regression-test/data/nereids_tpch_shape_sf500_p0/shape/q3.out
+++ b/regression-test/data/nereids_tpch_shape_sf500_p0/shape/q3.out
@ -10,12 +10,13 @@ PhysicalTopN
 --------------PhysicalProject
 ----------------filter((lineitem.l_shipdate > 1995-03-15))
 ------------------PhysicalOlapScan[lineitem]
--------------PhysicalProject
----------------hashJoin[INNER_JOIN](customer.c_custkey = orders.o_custkey)
------------------PhysicalProject
--------------------filter((orders.o_orderdate < 1995-03-15))
----------------------PhysicalOlapScan[orders]
------------------PhysicalDistribute
+--------------PhysicalDistribute
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN](customer.c_custkey = orders.o_custkey)
+--------------------PhysicalDistribute
+----------------------PhysicalProject
+------------------------filter((orders.o_orderdate < 1995-03-15))
+--------------------------PhysicalOlapScan[orders]
 --------------------PhysicalProject
 ----------------------filter((customer.c_mktsegment = 'BUILDING'))
 ------------------------PhysicalOlapScan[customer]