diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModel.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModel.java
index 15b8acc7fece97..839f9390f7344d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModel.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModel.java
@@ -359,7 +359,9 @@ public Cost visitPhysicalHashAggregate(
                     exprCost / 100 + inputStatistics.getRowCount() / beNumber,
                     inputStatistics.getRowCount() / beNumber, 0);
         } else {
-            int factor = aggregate.getGroupByExpressions().isEmpty() ? 1 : beNumber;
+            boolean isPartitioned = !aggregate.getGroupByExpressions().isEmpty()
+                    || aggregate.getPartitionExpressions().filter(expressions -> !expressions.isEmpty()).isPresent();
+            int factor = isPartitioned ? beNumber : 1;
             // global
             return Cost.of(context.getCostWeight(),
                     exprCost / 100 + inputStatistics.getRowCount() / factor,
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/cost/CostModelV1Test.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/cost/CostModelV1Test.java
index fd4785ecae0150..47a77c5a74d150 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/cost/CostModelV1Test.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/cost/CostModelV1Test.java
@@ -17,15 +17,28 @@
 
 package org.apache.doris.nereids.cost;
 
+import org.apache.doris.nereids.PlanContext;
 import org.apache.doris.nereids.StatementContext;
 import org.apache.doris.nereids.sqltest.SqlTestBase;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
+import org.apache.doris.nereids.trees.plans.AggMode;
+import org.apache.doris.nereids.trees.plans.AggPhase;
 import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
 import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.nereids.util.PlanConstructor;
 import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.statistics.Statistics;
+import org.apache.doris.statistics.StatisticsBuilder;
 
+import com.google.common.collect.ImmutableList;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Optional;
 
 class CostModelV1Test extends SqlTestBase {
 
@@ -87,4 +100,41 @@ void testMaterializingCost() {
                 .getBestPlanTree();
         p.anyMatch(j -> j instanceof PhysicalHashJoin && ((PhysicalHashJoin<?, ?>) j).getJoinType().isRightJoin());
     }
+
+    @Test
+    void testPartitionedScalarAggregateCostUsesClusterScale() {
+        int originBeNumberForTest = connectContext.getSessionVariable().getBeNumberForTest();
+        connectContext.getSessionVariable().setBeNumberForTest(4);
+        try {
+            Plan child = PlanConstructor.newLogicalOlapScan(101, "partitioned_scalar_agg_t", 0);
+            Slot partitionKey = child.getOutput().get(0);
+            PhysicalHashAggregate<Plan> aggregate = new PhysicalHashAggregate<Plan>(
+                    ImmutableList.of(), ImmutableList.of(partitionKey), Optional.of(ImmutableList.of(partitionKey)),
+                    new AggregateParam(AggPhase.GLOBAL, AggMode.INPUT_TO_RESULT), false, null, false, child);
+            PhysicalHashAggregate<Plan> singlePointAggregate = new PhysicalHashAggregate<Plan>(
+                    ImmutableList.of(), ImmutableList.of(partitionKey), Optional.empty(),
+                    new AggregateParam(AggPhase.GLOBAL, AggMode.INPUT_TO_RESULT), false, null, false, child);
+            PhysicalHashAggregate<Plan> groupByAggregate = new PhysicalHashAggregate<Plan>(
+                    ImmutableList.of(partitionKey), ImmutableList.of(partitionKey), Optional.empty(),
+                    new AggregateParam(AggPhase.GLOBAL, AggMode.INPUT_TO_RESULT), false, null, false, child);
+            Statistics childStats = new StatisticsBuilder().setRowCount(1000).build();
+            PlanContext context = Mockito.mock(PlanContext.class);
+            Mockito.when(context.getChildStatistics(0)).thenReturn(childStats);
+            Mockito.when(context.getCostWeight()).thenReturn(new CostWeight(1, 1, 1.5, 1));
+
+            Cost cost = new CostModel(connectContext).visitPhysicalHashAggregate(aggregate, context);
+            Cost singlePointCost = new CostModel(connectContext).visitPhysicalHashAggregate(singlePointAggregate,
+                    context);
+            Cost groupByCost = new CostModel(connectContext).visitPhysicalHashAggregate(groupByAggregate, context);
+
+            Assertions.assertEquals(250, cost.getCpuCost(), 1e-9);
+            Assertions.assertEquals(250, cost.getMemoryCost(), 1e-9);
+            Assertions.assertEquals(1000, singlePointCost.getCpuCost(), 1e-9);
+            Assertions.assertEquals(1000, singlePointCost.getMemoryCost(), 1e-9);
+            Assertions.assertEquals(250, groupByCost.getCpuCost(), 1e-9);
+            Assertions.assertEquals(250, groupByCost.getMemoryCost(), 1e-9);
+        } finally {
+            connectContext.getSessionVariable().setBeNumberForTest(originBeNumberForTest);
+        }
+    }
 }
diff --git a/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query16.out b/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query16.out
index 02559f87cb35c3..15c0bb5e5ce1cf 100644
--- a/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query16.out
+++ b/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query16.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF3 cc_call_center_sk->[cs_call_center_sk]
 ----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF3 cc_call_center_sk->[cs_call_center_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[cs_ship_addr_sk]
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[cs_ship_addr_sk]
-------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
-----------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
+------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
+--------------------------PhysicalProject
+----------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF0 cs_order_number->[cs_order_number]
 ------------------------------PhysicalProject
---------------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF0 cs_order_number->[cs_order_number]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF1 RF2 RF3
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0
 ------------------------------PhysicalProject
---------------------------------PhysicalOlapScan[catalog_returns]
-----------------------------PhysicalProject
-------------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
---------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF1 RF2 RF3
+--------------------------PhysicalProject
+----------------------------PhysicalOlapScan[catalog_returns]
 ------------------------PhysicalProject
---------------------------filter((customer_address.ca_state = 'WV'))
-----------------------------PhysicalOlapScan[customer_address]
+--------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
+----------------------------PhysicalOlapScan[date_dim]
 --------------------PhysicalProject
-----------------------filter(cc_county IN ('Barrow County', 'Daviess County', 'Luce County', 'Richland County', 'Ziebach County'))
-------------------------PhysicalOlapScan[call_center]
+----------------------filter((customer_address.ca_state = 'WV'))
+------------------------PhysicalOlapScan[customer_address]
+----------------PhysicalProject
+------------------filter(cc_county IN ('Barrow County', 'Daviess County', 'Luce County', 'Richland County', 'Ziebach County'))
+--------------------PhysicalOlapScan[call_center]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query94.out b/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query94.out
index d26128a78f3b6a..a840ab389fd620 100644
--- a/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query94.out
+++ b/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query94.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
 ----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[ws_ship_addr_sk]
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[ws_ship_addr_sk]
-------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
-----------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
+------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
+--------------------------PhysicalProject
+----------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF0 ws_order_number->[ws_order_number]
 ------------------------------PhysicalProject
---------------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF0 ws_order_number->[ws_order_number]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0
 ------------------------------PhysicalProject
---------------------------------PhysicalOlapScan[web_returns]
-----------------------------PhysicalProject
-------------------------------filter((date_dim.d_date <= '2000-04-01') and (date_dim.d_date >= '2000-02-01'))
---------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
+--------------------------PhysicalProject
+----------------------------PhysicalOlapScan[web_returns]
 ------------------------PhysicalProject
---------------------------filter((customer_address.ca_state = 'OK'))
-----------------------------PhysicalOlapScan[customer_address]
+--------------------------filter((date_dim.d_date <= '2000-04-01') and (date_dim.d_date >= '2000-02-01'))
+----------------------------PhysicalOlapScan[date_dim]
 --------------------PhysicalProject
-----------------------filter((web_site.web_company_name = 'pri'))
-------------------------PhysicalOlapScan[web_site]
+----------------------filter((customer_address.ca_state = 'OK'))
+------------------------PhysicalOlapScan[customer_address]
+----------------PhysicalProject
+------------------filter((web_site.web_company_name = 'pri'))
+--------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query95.out b/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query95.out
index 4435e9fb23dcd5..804e6335b457b7 100644
--- a/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query95.out
+++ b/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query95.out
@@ -11,34 +11,32 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
 --PhysicalResultSink
 ----PhysicalLimit[GLOBAL]
 ------PhysicalLimit[LOCAL]
---------hashAgg[DISTINCT_GLOBAL]
+--------hashAgg[GLOBAL]
 ----------PhysicalDistribute[DistributionSpecGather]
-------------hashAgg[DISTINCT_LOCAL]
---------------hashAgg[GLOBAL]
-----------------hashAgg[LOCAL]
+------------hashAgg[GLOBAL]
+--------------PhysicalProject
+----------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF6 web_site_sk->[ws_web_site_sk]
 ------------------PhysicalProject
---------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF6 web_site_sk->[ws_web_site_sk]
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF5 ca_address_sk->[ws_ship_addr_sk]
 ----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF5 ca_address_sk->[ws_ship_addr_sk]
---------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF4 d_date_sk->[ws_ship_date_sk]
-------------------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF3 ws_order_number->[ws_order_number]
---------------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF3
---------------------------------hashJoin[RIGHT_SEMI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF2 ws_order_number->[wr_order_number];RF7 ws_order_number->[ws_order_number,ws_order_number]
-----------------------------------PhysicalProject
-------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=()
---------------------------------------PhysicalCteConsumer ( cteId=CTEId#0 )
---------------------------------------PhysicalProject
-----------------------------------------PhysicalOlapScan[web_returns] apply RFs: RF2
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF4 d_date_sk->[ws_ship_date_sk]
+--------------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF3 ws_order_number->[ws_order_number]
+----------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF3
+----------------------------hashJoin[RIGHT_SEMI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF2 ws_order_number->[wr_order_number];RF7 ws_order_number->[ws_order_number,ws_order_number]
+------------------------------PhysicalProject
+--------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=()
+----------------------------------PhysicalCteConsumer ( cteId=CTEId#0 )
 ----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF4 RF5 RF6
+------------------------------------PhysicalOlapScan[web_returns] apply RFs: RF2
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '1999-04-02') and (date_dim.d_date >= '1999-02-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF4 RF5 RF6
 --------------------------PhysicalProject
-----------------------------filter((customer_address.ca_state = 'NC'))
-------------------------------PhysicalOlapScan[customer_address]
+----------------------------filter((date_dim.d_date <= '1999-04-02') and (date_dim.d_date >= '1999-02-01'))
+------------------------------PhysicalOlapScan[date_dim]
 ----------------------PhysicalProject
-------------------------filter((web_site.web_company_name = 'pri'))
---------------------------PhysicalOlapScan[web_site]
+------------------------filter((customer_address.ca_state = 'NC'))
+--------------------------PhysicalOlapScan[customer_address]
+------------------PhysicalProject
+--------------------filter((web_site.web_company_name = 'pri'))
+----------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query16.out b/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query16.out
index 02559f87cb35c3..15c0bb5e5ce1cf 100644
--- a/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query16.out
+++ b/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query16.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF3 cc_call_center_sk->[cs_call_center_sk]
 ----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF3 cc_call_center_sk->[cs_call_center_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[cs_ship_addr_sk]
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[cs_ship_addr_sk]
-------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
-----------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
+------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
+--------------------------PhysicalProject
+----------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF0 cs_order_number->[cs_order_number]
 ------------------------------PhysicalProject
---------------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF0 cs_order_number->[cs_order_number]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF1 RF2 RF3
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0
 ------------------------------PhysicalProject
---------------------------------PhysicalOlapScan[catalog_returns]
-----------------------------PhysicalProject
-------------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
---------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF1 RF2 RF3
+--------------------------PhysicalProject
+----------------------------PhysicalOlapScan[catalog_returns]
 ------------------------PhysicalProject
---------------------------filter((customer_address.ca_state = 'WV'))
-----------------------------PhysicalOlapScan[customer_address]
+--------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
+----------------------------PhysicalOlapScan[date_dim]
 --------------------PhysicalProject
-----------------------filter(cc_county IN ('Barrow County', 'Daviess County', 'Luce County', 'Richland County', 'Ziebach County'))
-------------------------PhysicalOlapScan[call_center]
+----------------------filter((customer_address.ca_state = 'WV'))
+------------------------PhysicalOlapScan[customer_address]
+----------------PhysicalProject
+------------------filter(cc_county IN ('Barrow County', 'Daviess County', 'Luce County', 'Richland County', 'Ziebach County'))
+--------------------PhysicalOlapScan[call_center]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query94.out b/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query94.out
index d26128a78f3b6a..a840ab389fd620 100644
--- a/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query94.out
+++ b/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query94.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
 ----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[ws_ship_addr_sk]
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[ws_ship_addr_sk]
-------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
-----------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
+------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
+--------------------------PhysicalProject
+----------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF0 ws_order_number->[ws_order_number]
 ------------------------------PhysicalProject
---------------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF0 ws_order_number->[ws_order_number]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0
 ------------------------------PhysicalProject
---------------------------------PhysicalOlapScan[web_returns]
-----------------------------PhysicalProject
-------------------------------filter((date_dim.d_date <= '2000-04-01') and (date_dim.d_date >= '2000-02-01'))
---------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
+--------------------------PhysicalProject
+----------------------------PhysicalOlapScan[web_returns]
 ------------------------PhysicalProject
---------------------------filter((customer_address.ca_state = 'OK'))
-----------------------------PhysicalOlapScan[customer_address]
+--------------------------filter((date_dim.d_date <= '2000-04-01') and (date_dim.d_date >= '2000-02-01'))
+----------------------------PhysicalOlapScan[date_dim]
 --------------------PhysicalProject
-----------------------filter((web_site.web_company_name = 'pri'))
-------------------------PhysicalOlapScan[web_site]
+----------------------filter((customer_address.ca_state = 'OK'))
+------------------------PhysicalOlapScan[customer_address]
+----------------PhysicalProject
+------------------filter((web_site.web_company_name = 'pri'))
+--------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query95.out b/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query95.out
index 253f14febe60d7..bfe61e266278c4 100644
--- a/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query95.out
+++ b/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query95.out
@@ -11,34 +11,32 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
 --PhysicalResultSink
 ----PhysicalLimit[GLOBAL]
 ------PhysicalLimit[LOCAL]
---------hashAgg[DISTINCT_GLOBAL]
+--------hashAgg[GLOBAL]
 ----------PhysicalDistribute[DistributionSpecGather]
-------------hashAgg[DISTINCT_LOCAL]
---------------hashAgg[GLOBAL]
-----------------hashAgg[LOCAL]
+------------hashAgg[GLOBAL]
+--------------PhysicalProject
+----------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF12 web_site_sk->[ws_web_site_sk];RF13 web_site_sk->[ws_web_site_sk]
 ------------------PhysicalProject
---------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF12 web_site_sk->[ws_web_site_sk];RF13 web_site_sk->[ws_web_site_sk]
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF10 ca_address_sk->[ws_ship_addr_sk];RF11 ca_address_sk->[ws_ship_addr_sk]
 ----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF10 ca_address_sk->[ws_ship_addr_sk];RF11 ca_address_sk->[ws_ship_addr_sk]
---------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF8 d_date_sk->[ws_ship_date_sk];RF9 d_date_sk->[ws_ship_date_sk]
-------------------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[ws_order_number];RF7 ws_order_number->[ws_order_number]
---------------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF6 RF7
---------------------------------hashJoin[RIGHT_SEMI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF4 ws_order_number->[wr_order_number];RF5 ws_order_number->[wr_order_number];RF14 ws_order_number->[ws_order_number,ws_order_number];RF15 ws_order_number->[ws_order_number,ws_order_number]
-----------------------------------PhysicalProject
-------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF2 wr_order_number->[ws_order_number];RF3 wr_order_number->[ws_order_number]
---------------------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF2 RF3
---------------------------------------PhysicalProject
-----------------------------------------PhysicalOlapScan[web_returns] apply RFs: RF4 RF5
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF8 d_date_sk->[ws_ship_date_sk];RF9 d_date_sk->[ws_ship_date_sk]
+--------------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[ws_order_number];RF7 ws_order_number->[ws_order_number]
+----------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF6 RF7
+----------------------------hashJoin[RIGHT_SEMI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF4 ws_order_number->[wr_order_number];RF5 ws_order_number->[wr_order_number];RF14 ws_order_number->[ws_order_number,ws_order_number];RF15 ws_order_number->[ws_order_number,ws_order_number]
+------------------------------PhysicalProject
+--------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF2 wr_order_number->[ws_order_number];RF3 wr_order_number->[ws_order_number]
+----------------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF2 RF3
 ----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF8 RF9 RF10 RF11 RF12 RF13
+------------------------------------PhysicalOlapScan[web_returns] apply RFs: RF4 RF5
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '1999-04-02') and (date_dim.d_date >= '1999-02-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF8 RF9 RF10 RF11 RF12 RF13
 --------------------------PhysicalProject
-----------------------------filter((customer_address.ca_state = 'NC'))
-------------------------------PhysicalOlapScan[customer_address]
+----------------------------filter((date_dim.d_date <= '1999-04-02') and (date_dim.d_date >= '1999-02-01'))
+------------------------------PhysicalOlapScan[date_dim]
 ----------------------PhysicalProject
-------------------------filter((web_site.web_company_name = 'pri'))
---------------------------PhysicalOlapScan[web_site]
+------------------------filter((customer_address.ca_state = 'NC'))
+--------------------------PhysicalOlapScan[customer_address]
+------------------PhysicalProject
+--------------------filter((web_site.web_company_name = 'pri'))
+----------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/rf_prune/query16.out b/regression-test/data/shape_check/tpcds_sf100/rf_prune/query16.out
index b1b5037f23d714..2fe109e08e093b 100644
--- a/regression-test/data/shape_check/tpcds_sf100/rf_prune/query16.out
+++ b/regression-test/data/shape_check/tpcds_sf100/rf_prune/query16.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF3 cs_order_number->[cs_order_number]
 ----------------PhysicalProject
-------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF3 cs_order_number->[cs_order_number]
---------------------PhysicalProject
-----------------------PhysicalOlapScan[catalog_sales] apply RFs: RF3
+------------------PhysicalOlapScan[catalog_sales] apply RFs: RF3
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF2 cc_call_center_sk->[cs_call_center_sk]
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF2 cc_call_center_sk->[cs_call_center_sk]
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
 ------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[cs_ship_addr_sk]
+----------------------------hashJoin[LEFT_ANTI_JOIN broadcast] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
+------------------------------PhysicalProject
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
+------------------------------PhysicalProject
+--------------------------------PhysicalOlapScan[catalog_returns]
 ----------------------------PhysicalProject
-------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[cs_ship_addr_sk]
---------------------------------hashJoin[LEFT_ANTI_JOIN broadcast] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_returns]
---------------------------------PhysicalProject
-----------------------------------filter((customer_address.ca_state = 'WV'))
-------------------------------------PhysicalOlapScan[customer_address]
-----------------------------PhysicalProject
-------------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
---------------------------------PhysicalOlapScan[date_dim]
+------------------------------filter((customer_address.ca_state = 'WV'))
+--------------------------------PhysicalOlapScan[customer_address]
 ------------------------PhysicalProject
---------------------------filter(cc_county IN ('Barrow County', 'Daviess County', 'Luce County', 'Richland County', 'Ziebach County'))
-----------------------------PhysicalOlapScan[call_center]
+--------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
+----------------------------PhysicalOlapScan[date_dim]
+--------------------PhysicalProject
+----------------------filter(cc_county IN ('Barrow County', 'Daviess County', 'Luce County', 'Richland County', 'Ziebach County'))
+------------------------PhysicalOlapScan[call_center]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/rf_prune/query94.out b/regression-test/data/shape_check/tpcds_sf100/rf_prune/query94.out
index df854654271e34..effa9e30aaa869 100644
--- a/regression-test/data/shape_check/tpcds_sf100/rf_prune/query94.out
+++ b/regression-test/data/shape_check/tpcds_sf100/rf_prune/query94.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF3 ws_order_number->[ws_order_number]
 ----------------PhysicalProject
-------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF3 ws_order_number->[ws_order_number]
---------------------PhysicalProject
-----------------------PhysicalOlapScan[web_sales] apply RFs: RF3
+------------------PhysicalOlapScan[web_sales] apply RFs: RF3
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
 ------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[ws_ship_addr_sk]
+----------------------------hashJoin[LEFT_ANTI_JOIN broadcast] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
+------------------------------PhysicalProject
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
+------------------------------PhysicalProject
+--------------------------------PhysicalOlapScan[web_returns]
 ----------------------------PhysicalProject
-------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[ws_ship_addr_sk]
---------------------------------hashJoin[LEFT_ANTI_JOIN broadcast] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_returns]
---------------------------------PhysicalProject
-----------------------------------filter((customer_address.ca_state = 'OK'))
-------------------------------------PhysicalOlapScan[customer_address]
-----------------------------PhysicalProject
-------------------------------filter((date_dim.d_date <= '2000-04-01') and (date_dim.d_date >= '2000-02-01'))
---------------------------------PhysicalOlapScan[date_dim]
+------------------------------filter((customer_address.ca_state = 'OK'))
+--------------------------------PhysicalOlapScan[customer_address]
 ------------------------PhysicalProject
---------------------------filter((web_site.web_company_name = 'pri'))
-----------------------------PhysicalOlapScan[web_site]
+--------------------------filter((date_dim.d_date <= '2000-04-01') and (date_dim.d_date >= '2000-02-01'))
+----------------------------PhysicalOlapScan[date_dim]
+--------------------PhysicalProject
+----------------------filter((web_site.web_company_name = 'pri'))
+------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/rf_prune/query95.out b/regression-test/data/shape_check/tpcds_sf100/rf_prune/query95.out
index 623fe7152eccbb..81a72d6f435874 100644
--- a/regression-test/data/shape_check/tpcds_sf100/rf_prune/query95.out
+++ b/regression-test/data/shape_check/tpcds_sf100/rf_prune/query95.out
@@ -11,34 +11,32 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
 --PhysicalResultSink
 ----PhysicalLimit[GLOBAL]
 ------PhysicalLimit[LOCAL]
---------hashAgg[DISTINCT_GLOBAL]
+--------hashAgg[GLOBAL]
 ----------PhysicalDistribute[DistributionSpecGather]
-------------hashAgg[DISTINCT_LOCAL]
---------------hashAgg[GLOBAL]
-----------------hashAgg[LOCAL]
-------------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[wr_order_number,ws_order_number]
+------------hashAgg[GLOBAL]
+--------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[wr_order_number,ws_order_number]
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=()
+--------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF6
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=()
-------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF6
-------------------------PhysicalProject
---------------------------PhysicalOlapScan[web_returns] apply RFs: RF6
---------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 ws_order_number->[ws_order_number,ws_order_number]
-----------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+----------------------PhysicalOlapScan[web_returns] apply RFs: RF6
+----------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 ws_order_number->[ws_order_number,ws_order_number]
+------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'NC'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '1999-04-02') and (date_dim.d_date >= '1999-02-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'NC'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
+----------------------------filter((date_dim.d_date <= '1999-04-02') and (date_dim.d_date >= '1999-02-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/shape/query16.out b/regression-test/data/shape_check/tpcds_sf100/shape/query16.out
index b1b5037f23d714..2fe109e08e093b 100644
--- a/regression-test/data/shape_check/tpcds_sf100/shape/query16.out
+++ b/regression-test/data/shape_check/tpcds_sf100/shape/query16.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF3 cs_order_number->[cs_order_number]
 ----------------PhysicalProject
-------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF3 cs_order_number->[cs_order_number]
---------------------PhysicalProject
-----------------------PhysicalOlapScan[catalog_sales] apply RFs: RF3
+------------------PhysicalOlapScan[catalog_sales] apply RFs: RF3
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF2 cc_call_center_sk->[cs_call_center_sk]
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF2 cc_call_center_sk->[cs_call_center_sk]
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
 ------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[cs_ship_addr_sk]
+----------------------------hashJoin[LEFT_ANTI_JOIN broadcast] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
+------------------------------PhysicalProject
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
+------------------------------PhysicalProject
+--------------------------------PhysicalOlapScan[catalog_returns]
 ----------------------------PhysicalProject
-------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[cs_ship_addr_sk]
---------------------------------hashJoin[LEFT_ANTI_JOIN broadcast] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_returns]
---------------------------------PhysicalProject
-----------------------------------filter((customer_address.ca_state = 'WV'))
-------------------------------------PhysicalOlapScan[customer_address]
-----------------------------PhysicalProject
-------------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
---------------------------------PhysicalOlapScan[date_dim]
+------------------------------filter((customer_address.ca_state = 'WV'))
+--------------------------------PhysicalOlapScan[customer_address]
 ------------------------PhysicalProject
---------------------------filter(cc_county IN ('Barrow County', 'Daviess County', 'Luce County', 'Richland County', 'Ziebach County'))
-----------------------------PhysicalOlapScan[call_center]
+--------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
+----------------------------PhysicalOlapScan[date_dim]
+--------------------PhysicalProject
+----------------------filter(cc_county IN ('Barrow County', 'Daviess County', 'Luce County', 'Richland County', 'Ziebach County'))
+------------------------PhysicalOlapScan[call_center]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/shape/query94.out b/regression-test/data/shape_check/tpcds_sf100/shape/query94.out
index df854654271e34..effa9e30aaa869 100644
--- a/regression-test/data/shape_check/tpcds_sf100/shape/query94.out
+++ b/regression-test/data/shape_check/tpcds_sf100/shape/query94.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF3 ws_order_number->[ws_order_number]
 ----------------PhysicalProject
-------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF3 ws_order_number->[ws_order_number]
---------------------PhysicalProject
-----------------------PhysicalOlapScan[web_sales] apply RFs: RF3
+------------------PhysicalOlapScan[web_sales] apply RFs: RF3
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
 ------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[ws_ship_addr_sk]
+----------------------------hashJoin[LEFT_ANTI_JOIN broadcast] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
+------------------------------PhysicalProject
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
+------------------------------PhysicalProject
+--------------------------------PhysicalOlapScan[web_returns]
 ----------------------------PhysicalProject
-------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[ws_ship_addr_sk]
---------------------------------hashJoin[LEFT_ANTI_JOIN broadcast] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_returns]
---------------------------------PhysicalProject
-----------------------------------filter((customer_address.ca_state = 'OK'))
-------------------------------------PhysicalOlapScan[customer_address]
-----------------------------PhysicalProject
-------------------------------filter((date_dim.d_date <= '2000-04-01') and (date_dim.d_date >= '2000-02-01'))
---------------------------------PhysicalOlapScan[date_dim]
+------------------------------filter((customer_address.ca_state = 'OK'))
+--------------------------------PhysicalOlapScan[customer_address]
 ------------------------PhysicalProject
---------------------------filter((web_site.web_company_name = 'pri'))
-----------------------------PhysicalOlapScan[web_site]
+--------------------------filter((date_dim.d_date <= '2000-04-01') and (date_dim.d_date >= '2000-02-01'))
+----------------------------PhysicalOlapScan[date_dim]
+--------------------PhysicalProject
+----------------------filter((web_site.web_company_name = 'pri'))
+------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/shape/query95.out b/regression-test/data/shape_check/tpcds_sf100/shape/query95.out
index 2e4d4278eaaed8..950bd921d5a0c5 100644
--- a/regression-test/data/shape_check/tpcds_sf100/shape/query95.out
+++ b/regression-test/data/shape_check/tpcds_sf100/shape/query95.out
@@ -11,34 +11,32 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
 --PhysicalResultSink
 ----PhysicalLimit[GLOBAL]
 ------PhysicalLimit[LOCAL]
---------hashAgg[DISTINCT_GLOBAL]
+--------hashAgg[GLOBAL]
 ----------PhysicalDistribute[DistributionSpecGather]
-------------hashAgg[DISTINCT_LOCAL]
---------------hashAgg[GLOBAL]
-----------------hashAgg[LOCAL]
-------------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF12 ws_order_number->[wr_order_number,ws_order_number];RF13 ws_order_number->[wr_order_number,ws_order_number]
+------------hashAgg[GLOBAL]
+--------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF12 ws_order_number->[wr_order_number,ws_order_number];RF13 ws_order_number->[wr_order_number,ws_order_number]
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF10 wr_order_number->[ws_order_number];RF11 wr_order_number->[ws_order_number]
+--------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF10 RF11 RF12 RF13
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF10 wr_order_number->[ws_order_number];RF11 wr_order_number->[ws_order_number]
-------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF10 RF11 RF12 RF13
-------------------------PhysicalProject
---------------------------PhysicalOlapScan[web_returns] apply RFs: RF12 RF13
---------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF14 ws_order_number->[ws_order_number,ws_order_number];RF15 ws_order_number->[ws_order_number,ws_order_number]
-----------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+----------------------PhysicalOlapScan[web_returns] apply RFs: RF12 RF13
+----------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF14 ws_order_number->[ws_order_number,ws_order_number];RF15 ws_order_number->[ws_order_number,ws_order_number]
+------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF6 web_site_sk->[ws_web_site_sk];RF7 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF6 web_site_sk->[ws_web_site_sk];RF7 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF4 d_date_sk->[ws_ship_date_sk];RF5 d_date_sk->[ws_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF4 d_date_sk->[ws_ship_date_sk];RF5 d_date_sk->[ws_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[ws_ship_addr_sk];RF3 ca_address_sk->[ws_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[ws_ship_addr_sk];RF3 ca_address_sk->[ws_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF2 RF3 RF4 RF5 RF6 RF7
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'NC'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF2 RF3 RF4 RF5 RF6 RF7
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '1999-04-02') and (date_dim.d_date >= '1999-02-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'NC'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
+----------------------------filter((date_dim.d_date <= '1999-04-02') and (date_dim.d_date >= '1999-02-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query95.out b/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query95.out
index 84bdd6cf3d8f4a..ceed9dc1377f18 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query95.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query95.out
@@ -11,34 +11,32 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
 --PhysicalResultSink
 ----PhysicalLimit[GLOBAL]
 ------PhysicalLimit[LOCAL]
---------hashAgg[DISTINCT_GLOBAL]
+--------hashAgg[GLOBAL]
 ----------PhysicalDistribute[DistributionSpecGather]
-------------hashAgg[DISTINCT_LOCAL]
---------------hashAgg[GLOBAL]
-----------------hashAgg[LOCAL]
-------------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[wr_order_number,ws_order_number]
+------------hashAgg[GLOBAL]
+--------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[wr_order_number,ws_order_number]
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF5 wr_order_number->[ws_order_number]
+--------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF5 RF6
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF5 wr_order_number->[ws_order_number]
-------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF5 RF6
-------------------------PhysicalProject
---------------------------PhysicalOlapScan[web_returns] apply RFs: RF6
---------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 ws_order_number->[ws_order_number,ws_order_number]
-----------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+----------------------PhysicalOlapScan[web_returns] apply RFs: RF6
+----------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 ws_order_number->[ws_order_number,ws_order_number]
+------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'VA'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '2001-05-31') and (date_dim.d_date >= '2001-04-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'VA'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
+----------------------------filter((date_dim.d_date <= '2001-05-31') and (date_dim.d_date >= '2001-04-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000/hint/query16.out b/regression-test/data/shape_check/tpcds_sf1000/hint/query16.out
index 391b7d7cc41ba7..fd98fe166ca273 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/hint/query16.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/hint/query16.out
@@ -3,35 +3,33 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF4 cs_order_number->[cs_order_number]
 ----------------PhysicalProject
-------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF4 cs_order_number->[cs_order_number]
---------------------PhysicalProject
-----------------------PhysicalOlapScan[catalog_sales] apply RFs: RF4
---------------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=() build RFs:RF3 cs_order_number->[cr_order_number]
+------------------PhysicalOlapScan[catalog_sales] apply RFs: RF4
+----------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=() build RFs:RF3 cs_order_number->[cr_order_number]
+------------------PhysicalProject
+--------------------PhysicalOlapScan[catalog_returns] apply RFs: RF3
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF2 cc_call_center_sk->[cs_call_center_sk]
 ----------------------PhysicalProject
-------------------------PhysicalOlapScan[catalog_returns] apply RFs: RF3
-----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF2 cc_call_center_sk->[cs_call_center_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[cs_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[cs_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'PA'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'PA'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((call_center.cc_county = 'Williamson County'))
-------------------------------PhysicalOlapScan[call_center]
+----------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((call_center.cc_county = 'Williamson County'))
+--------------------------PhysicalOlapScan[call_center]
 
 Hint log:
 Used: leading(catalog_sales { cs1 customer_address date_dim call_center } )
diff --git a/regression-test/data/shape_check/tpcds_sf1000/hint/query94.out b/regression-test/data/shape_check/tpcds_sf1000/hint/query94.out
index bd89dd58effa9b..1b02e371805b5b 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/hint/query94.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/hint/query94.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF4 ws_order_number->[ws_order_number]
 ----------------PhysicalProject
-------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF4 ws_order_number->[ws_order_number]
---------------------PhysicalProject
-----------------------PhysicalOlapScan[web_sales] apply RFs: RF4
---------------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=() build RFs:RF3 ws_order_number->[wr_order_number]
+------------------PhysicalOlapScan[web_sales] apply RFs: RF4
+----------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=() build RFs:RF3 ws_order_number->[wr_order_number]
+------------------PhysicalProject
+--------------------PhysicalOlapScan[web_returns] apply RFs: RF3
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------PhysicalOlapScan[web_returns] apply RFs: RF3
-----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[ws_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[ws_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'OK'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '2002-06-30') and (date_dim.d_date >= '2002-05-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'OK'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
+----------------------------filter((date_dim.d_date <= '2002-06-30') and (date_dim.d_date >= '2002-05-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000/hint/query95.out b/regression-test/data/shape_check/tpcds_sf1000/hint/query95.out
index 84bdd6cf3d8f4a..ceed9dc1377f18 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/hint/query95.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/hint/query95.out
@@ -11,34 +11,32 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
 --PhysicalResultSink
 ----PhysicalLimit[GLOBAL]
 ------PhysicalLimit[LOCAL]
---------hashAgg[DISTINCT_GLOBAL]
+--------hashAgg[GLOBAL]
 ----------PhysicalDistribute[DistributionSpecGather]
-------------hashAgg[DISTINCT_LOCAL]
---------------hashAgg[GLOBAL]
-----------------hashAgg[LOCAL]
-------------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[wr_order_number,ws_order_number]
+------------hashAgg[GLOBAL]
+--------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[wr_order_number,ws_order_number]
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF5 wr_order_number->[ws_order_number]
+--------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF5 RF6
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF5 wr_order_number->[ws_order_number]
-------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF5 RF6
-------------------------PhysicalProject
---------------------------PhysicalOlapScan[web_returns] apply RFs: RF6
---------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 ws_order_number->[ws_order_number,ws_order_number]
-----------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+----------------------PhysicalOlapScan[web_returns] apply RFs: RF6
+----------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 ws_order_number->[ws_order_number,ws_order_number]
+------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'VA'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '2001-05-31') and (date_dim.d_date >= '2001-04-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'VA'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
+----------------------------filter((date_dim.d_date <= '2001-05-31') and (date_dim.d_date >= '2001-04-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000/shape/query16.out b/regression-test/data/shape_check/tpcds_sf1000/shape/query16.out
index 21dcd5cf6df0ae..60ab2881290509 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/shape/query16.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/shape/query16.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF4 cs_order_number->[cs_order_number]
 ----------------PhysicalProject
-------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF4 cs_order_number->[cs_order_number]
---------------------PhysicalProject
-----------------------PhysicalOlapScan[catalog_sales] apply RFs: RF4
---------------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=() build RFs:RF3 cs_order_number->[cr_order_number]
+------------------PhysicalOlapScan[catalog_sales] apply RFs: RF4
+----------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=() build RFs:RF3 cs_order_number->[cr_order_number]
+------------------PhysicalProject
+--------------------PhysicalOlapScan[catalog_returns] apply RFs: RF3
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF2 cc_call_center_sk->[cs_call_center_sk]
 ----------------------PhysicalProject
-------------------------PhysicalOlapScan[catalog_returns] apply RFs: RF3
-----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF2 cc_call_center_sk->[cs_call_center_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[cs_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[cs_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'PA'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'PA'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((call_center.cc_county = 'Williamson County'))
-------------------------------PhysicalOlapScan[call_center]
+----------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((call_center.cc_county = 'Williamson County'))
+--------------------------PhysicalOlapScan[call_center]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000/shape/query94.out b/regression-test/data/shape_check/tpcds_sf1000/shape/query94.out
index bd89dd58effa9b..1b02e371805b5b 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/shape/query94.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/shape/query94.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF4 ws_order_number->[ws_order_number]
 ----------------PhysicalProject
-------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF4 ws_order_number->[ws_order_number]
---------------------PhysicalProject
-----------------------PhysicalOlapScan[web_sales] apply RFs: RF4
---------------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=() build RFs:RF3 ws_order_number->[wr_order_number]
+------------------PhysicalOlapScan[web_sales] apply RFs: RF4
+----------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=() build RFs:RF3 ws_order_number->[wr_order_number]
+------------------PhysicalProject
+--------------------PhysicalOlapScan[web_returns] apply RFs: RF3
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------PhysicalOlapScan[web_returns] apply RFs: RF3
-----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[ws_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[ws_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'OK'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '2002-06-30') and (date_dim.d_date >= '2002-05-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'OK'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
+----------------------------filter((date_dim.d_date <= '2002-06-30') and (date_dim.d_date >= '2002-05-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000/shape/query95.out b/regression-test/data/shape_check/tpcds_sf1000/shape/query95.out
index 84bdd6cf3d8f4a..ceed9dc1377f18 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/shape/query95.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/shape/query95.out
@@ -11,34 +11,32 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
 --PhysicalResultSink
 ----PhysicalLimit[GLOBAL]
 ------PhysicalLimit[LOCAL]
---------hashAgg[DISTINCT_GLOBAL]
+--------hashAgg[GLOBAL]
 ----------PhysicalDistribute[DistributionSpecGather]
-------------hashAgg[DISTINCT_LOCAL]
---------------hashAgg[GLOBAL]
-----------------hashAgg[LOCAL]
-------------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[wr_order_number,ws_order_number]
+------------hashAgg[GLOBAL]
+--------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[wr_order_number,ws_order_number]
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF5 wr_order_number->[ws_order_number]
+--------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF5 RF6
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF5 wr_order_number->[ws_order_number]
-------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF5 RF6
-------------------------PhysicalProject
---------------------------PhysicalOlapScan[web_returns] apply RFs: RF6
---------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 ws_order_number->[ws_order_number,ws_order_number]
-----------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+----------------------PhysicalOlapScan[web_returns] apply RFs: RF6
+----------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 ws_order_number->[ws_order_number,ws_order_number]
+------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'VA'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '2001-05-31') and (date_dim.d_date >= '2001-04-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'VA'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
+----------------------------filter((date_dim.d_date <= '2001-05-31') and (date_dim.d_date >= '2001-04-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query16.out b/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query16.out
index 2ef4689fc02b94..b02c96877900a1 100644
--- a/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query16.out
+++ b/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query16.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF4 cs_order_number->[cs_order_number]
 ----------------PhysicalProject
-------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF4 cs_order_number->[cs_order_number]
---------------------PhysicalProject
-----------------------PhysicalOlapScan[catalog_sales] apply RFs: RF4
---------------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=() build RFs:RF3 cs_order_number->[cr_order_number]
+------------------PhysicalOlapScan[catalog_sales] apply RFs: RF4
+----------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=() build RFs:RF3 cs_order_number->[cr_order_number]
+------------------PhysicalProject
+--------------------PhysicalOlapScan[catalog_returns] apply RFs: RF3
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF2 cc_call_center_sk->[cs_call_center_sk]
 ----------------------PhysicalProject
-------------------------PhysicalOlapScan[catalog_returns] apply RFs: RF3
-----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF2 cc_call_center_sk->[cs_call_center_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[cs_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[cs_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'PA'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'PA'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((call_center.cc_county = 'Williamson County'))
-------------------------------PhysicalOlapScan[call_center]
+----------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((call_center.cc_county = 'Williamson County'))
+--------------------------PhysicalOlapScan[call_center]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query94.out b/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query94.out
index d2e32764ac35a6..7bd136893b1cdf 100644
--- a/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query94.out
+++ b/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query94.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF4 ws_order_number->[ws_order_number]
 ----------------PhysicalProject
-------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF4 ws_order_number->[ws_order_number]
---------------------PhysicalProject
-----------------------PhysicalOlapScan[web_sales] apply RFs: RF4
---------------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=() build RFs:RF3 ws_order_number->[wr_order_number]
+------------------PhysicalOlapScan[web_sales] apply RFs: RF4
+----------------hashJoin[RIGHT_ANTI_JOIN shuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=() build RFs:RF3 ws_order_number->[wr_order_number]
+------------------PhysicalProject
+--------------------PhysicalOlapScan[web_returns] apply RFs: RF3
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------PhysicalOlapScan[web_returns] apply RFs: RF3
-----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[ws_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[ws_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF0 ca_address_sk->[ws_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'OK'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '2002-06-30') and (date_dim.d_date >= '2002-05-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'OK'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
+----------------------------filter((date_dim.d_date <= '2002-06-30') and (date_dim.d_date >= '2002-05-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query95.out b/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query95.out
index 6f2387bf13d7ba..26d7769756d961 100644
--- a/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query95.out
+++ b/regression-test/data/shape_check/tpcds_sf1000_constraints/shape/query95.out
@@ -11,34 +11,32 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
 --PhysicalResultSink
 ----PhysicalLimit[GLOBAL]
 ------PhysicalLimit[LOCAL]
---------hashAgg[DISTINCT_GLOBAL]
+--------hashAgg[GLOBAL]
 ----------PhysicalDistribute[DistributionSpecGather]
-------------hashAgg[DISTINCT_LOCAL]
---------------hashAgg[GLOBAL]
-----------------hashAgg[LOCAL]
-------------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[wr_order_number,ws_order_number]
+------------hashAgg[GLOBAL]
+--------------hashJoin[RIGHT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 ws_order_number->[wr_order_number,ws_order_number]
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF5 wr_order_number->[ws_order_number]
+--------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF5 RF6
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF5 wr_order_number->[ws_order_number]
-------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF5 RF6
-------------------------PhysicalProject
---------------------------PhysicalOlapScan[web_returns] apply RFs: RF6
---------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 ws_order_number->[ws_order_number,ws_order_number]
-----------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+----------------------PhysicalOlapScan[web_returns] apply RFs: RF6
+----------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 ws_order_number->[ws_order_number,ws_order_number]
+------------------PhysicalCteConsumer ( cteId=CTEId#0 )
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
-----------------------------------PhysicalProject
-------------------------------------filter((customer_address.ca_state = 'VA'))
---------------------------------------PhysicalOlapScan[customer_address]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF1 RF2 RF3
 ------------------------------PhysicalProject
---------------------------------filter((date_dim.d_date <= '2001-05-31') and (date_dim.d_date >= '2001-04-01'))
-----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------filter((customer_address.ca_state = 'VA'))
+----------------------------------PhysicalOlapScan[customer_address]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
+----------------------------filter((date_dim.d_date <= '2001-05-31') and (date_dim.d_date >= '2001-04-01'))
+------------------------------PhysicalOlapScan[date_dim]
+----------------------PhysicalProject
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
 
diff --git a/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query16.out b/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query16.out
index ff30193149f0de..fca21d674db536 100644
--- a/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query16.out
+++ b/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query16.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
+----------hashAgg[GLOBAL]
+------------PhysicalProject
+--------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF3 cc_call_center_sk->[cs_call_center_sk]
 ----------------PhysicalProject
-------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF3 cc_call_center_sk->[cs_call_center_sk]
+------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[cs_ship_addr_sk]
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->[cs_ship_addr_sk]
-------------------------PhysicalProject
---------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
-----------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_ship_date_sk]
+------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
+--------------------------PhysicalProject
+----------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF0 cs_order_number->[cs_order_number]
 ------------------------------PhysicalProject
---------------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs_warehouse_sk = cs_warehouse_sk))) build RFs:RF0 cs_order_number->[cs_order_number]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF1 RF2 RF3
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0
 ------------------------------PhysicalProject
---------------------------------PhysicalOlapScan[catalog_returns]
-----------------------------PhysicalProject
-------------------------------filter((date_dim.d_date <= '1999-05-31') and (date_dim.d_date >= '1999-04-01'))
---------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF1 RF2 RF3
+--------------------------PhysicalProject
+----------------------------PhysicalOlapScan[catalog_returns]
 ------------------------PhysicalProject
---------------------------filter((customer_address.ca_state = 'IL'))
-----------------------------PhysicalOlapScan[customer_address]
+--------------------------filter((date_dim.d_date <= '1999-05-31') and (date_dim.d_date >= '1999-04-01'))
+----------------------------PhysicalOlapScan[date_dim]
 --------------------PhysicalProject
-----------------------filter(cc_county IN ('Bronx County', 'Maverick County', 'Mesa County', 'Raleigh County', 'Richland County'))
-------------------------PhysicalOlapScan[call_center]
+----------------------filter((customer_address.ca_state = 'IL'))
+------------------------PhysicalOlapScan[customer_address]
+----------------PhysicalProject
+------------------filter(cc_county IN ('Bronx County', 'Maverick County', 'Mesa County', 'Raleigh County', 'Richland County'))
+--------------------PhysicalOlapScan[call_center]
 
diff --git a/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query94.out b/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query94.out
index 2f12dce9e26230..8cd6a619b2b49e 100644
--- a/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query94.out
+++ b/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query94.out
@@ -3,33 +3,31 @@
 PhysicalResultSink
 --PhysicalLimit[GLOBAL]
 ----PhysicalLimit[LOCAL]
-------hashAgg[DISTINCT_GLOBAL]
+------hashAgg[GLOBAL]
 --------PhysicalDistribute[DistributionSpecGather]
-----------hashAgg[DISTINCT_LOCAL]
-------------hashAgg[GLOBAL]
---------------hashAgg[LOCAL]
-----------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
+----------hashAgg[GLOBAL]
+------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
+--------------PhysicalProject
+----------------hashJoin[LEFT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF3 ws_order_number->[ws_order_number]
 ------------------PhysicalProject
---------------------hashJoin[LEFT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws_warehouse_sk = ws_warehouse_sk))) build RFs:RF3 ws_order_number->[ws_order_number]
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF2 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF1 ca_address_sk->[ws_ship_addr_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF0 d_date_sk->[ws_ship_date_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF0 d_date_sk->[ws_ship_date_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2 RF3
-----------------------------------PhysicalProject
-------------------------------------filter((date_dim.d_date <= '1999-05-31') and (date_dim.d_date >= '1999-04-01'))
---------------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2 RF3
 ------------------------------PhysicalProject
---------------------------------filter((customer_address.ca_state = 'NE'))
-----------------------------------PhysicalOlapScan[customer_address]
+--------------------------------filter((date_dim.d_date <= '1999-05-31') and (date_dim.d_date >= '1999-04-01'))
+----------------------------------PhysicalOlapScan[date_dim]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
+----------------------------filter((customer_address.ca_state = 'NE'))
+------------------------------PhysicalOlapScan[customer_address]
 ----------------------PhysicalProject
-------------------------PhysicalOlapScan[web_sales]
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
 ------------------PhysicalProject
---------------------PhysicalOlapScan[web_returns]
+--------------------PhysicalOlapScan[web_sales]
+--------------PhysicalProject
+----------------PhysicalOlapScan[web_returns]
 
diff --git a/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query95.out b/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query95.out
index 140ec5cb678d55..91f8fc1a213ea9 100644
--- a/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query95.out
+++ b/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query95.out
@@ -11,34 +11,32 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
 --PhysicalResultSink
 ----PhysicalLimit[GLOBAL]
 ------PhysicalLimit[LOCAL]
---------hashAgg[DISTINCT_GLOBAL]
+--------hashAgg[GLOBAL]
 ----------PhysicalDistribute[DistributionSpecGather]
-------------hashAgg[DISTINCT_LOCAL]
---------------hashAgg[GLOBAL]
-----------------hashAgg[LOCAL]
-------------------hashJoin[LEFT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 wr_order_number->[ws_order_number,ws_order_number]
---------------------hashJoin[LEFT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF5 ws_order_number->[ws_order_number]
+------------hashAgg[GLOBAL]
+--------------hashJoin[LEFT_SEMI_JOIN colocated] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF6 wr_order_number->[ws_order_number,ws_order_number]
+----------------hashJoin[LEFT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF5 ws_order_number->[ws_order_number]
+------------------PhysicalProject
+--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF4 web_site_sk->[ws_web_site_sk]
 ----------------------PhysicalProject
-------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF4 web_site_sk->[ws_web_site_sk]
+------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 ca_address_sk->[ws_ship_addr_sk]
 --------------------------PhysicalProject
-----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 ca_address_sk->[ws_ship_addr_sk]
+----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
 ------------------------------PhysicalProject
---------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[ws_ship_date_sk]
-----------------------------------PhysicalProject
-------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF2 RF3 RF4 RF5 RF6
-----------------------------------PhysicalProject
-------------------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
---------------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalOlapScan[web_sales] apply RFs: RF2 RF3 RF4 RF5 RF6
 ------------------------------PhysicalProject
---------------------------------filter((customer_address.ca_state = 'AL'))
-----------------------------------PhysicalOlapScan[customer_address]
+--------------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
+----------------------------------PhysicalOlapScan[date_dim]
 --------------------------PhysicalProject
-----------------------------filter((web_site.web_company_name = 'pri'))
-------------------------------PhysicalOlapScan[web_site]
-----------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF6
+----------------------------filter((customer_address.ca_state = 'AL'))
+------------------------------PhysicalOlapScan[customer_address]
+----------------------PhysicalProject
+------------------------filter((web_site.web_company_name = 'pri'))
+--------------------------PhysicalOlapScan[web_site]
+------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF6
+----------------PhysicalProject
+------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 wr_order_number->[ws_order_number,ws_order_number]
+--------------------PhysicalCteConsumer ( cteId=CTEId#0 )
 --------------------PhysicalProject
-----------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF7 wr_order_number->[ws_order_number,ws_order_number]
-------------------------PhysicalCteConsumer ( cteId=CTEId#0 )
-------------------------PhysicalProject
---------------------------PhysicalOlapScan[web_returns]
+----------------------PhysicalOlapScan[web_returns]
 
diff --git a/regression-test/suites/shape_check/clickbench/query1.groovy b/regression-test/suites/shape_check/clickbench/query1.groovy
index 032ff1ac59da95..074ba260437d1a 100644
--- a/regression-test/suites/shape_check/clickbench/query1.groovy
+++ b/regression-test/suites/shape_check/clickbench/query1.groovy
@@ -21,7 +21,8 @@ suite("query1") {
     sql 'set enable_nereids_planner=true'
     sql 'set enable_nereids_distribute_planner=false'
     sql 'set enable_fallback_to_original_planner=false'
-    sql "set disable_nereids_rules=PRUNE_EMPTY_PARTITION"
+    // Keep the aggregate shape stable after the exact row-count cache is populated.
+    sql "set disable_nereids_rules='PRUNE_EMPTY_PARTITION,REWRITE_SIMPLE_AGG_TO_CONSTANT'"
     sql 'set topn_opt_limit_threshold = 1024'
     def ckBench = """SELECT COUNT(*) FROM hits"""
     qt_ckbench_shape_1 """