diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java
index 7eed68241f534e..29f0b0c79a8ae1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java
@@ -132,6 +132,7 @@
 import org.apache.doris.nereids.rules.rewrite.PushDownEncodeSlot;
 import org.apache.doris.nereids.rules.rewrite.PushDownFilterIntoSchemaScan;
 import org.apache.doris.nereids.rules.rewrite.PushDownFilterThroughProject;
+import org.apache.doris.nereids.rules.rewrite.PushDownJoinOnAssertNumRows;
 import org.apache.doris.nereids.rules.rewrite.PushDownLimit;
 import org.apache.doris.nereids.rules.rewrite.PushDownLimitDistinctThroughJoin;
 import org.apache.doris.nereids.rules.rewrite.PushDownLimitDistinctThroughUnion;
@@ -736,6 +737,7 @@ public class Rewriter extends AbstractBatchJobExecutor {
                 ),
                 topic("set initial join order",
                         bottomUp(ImmutableList.of(new InitJoinOrder())),
+                        bottomUp(ImmutableList.of(new PushDownJoinOnAssertNumRows(), new MergeProjectable())),
                         topDown(new SkewJoin())),
                 topic("agg rewrite",
                     // these rules should be put after mv optimization to avoid mv matching fail
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleSet.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleSet.java
index 45cb22e6b5209f..7e3e710eee2359 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleSet.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleSet.java
@@ -176,7 +176,7 @@ public class RuleSet {
             new PushDownFilterThroughPartitionTopN(),
             ExpressionNormalizationAndOptimization.NO_MIN_MAX_RANGE_INSTANCE,
             new EliminateFilter()
-    );
+            );
 
     public static final List<Rule> IMPLEMENTATION_RULES = planRuleFactories()
             .add(new LogicalCTEProducerToPhysicalCTEProducer())
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java
index a8bc20523486a0..f1b4c3e91f2160 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java
@@ -233,6 +233,7 @@ public enum RuleType {
     LOGICAL_SEMI_JOIN_COMMUTE(RuleTypeClass.REWRITE),
     TRANSPOSE_LOGICAL_SEMI_JOIN_AGG(RuleTypeClass.REWRITE),
     TRANSPOSE_LOGICAL_SEMI_JOIN_AGG_PROJECT(RuleTypeClass.REWRITE),
+    PUSH_DOWN_JOIN_ON_ASSERT_NUM_ROWS(RuleTypeClass.REWRITE),
 
     // expression of plan rewrite
     EXTRACT_IN_PREDICATE_FROM_OR(RuleTypeClass.REWRITE),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownJoinOnAssertNumRows.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownJoinOnAssertNumRows.java
new file mode 100644
index 00000000000000..aad46430c3e5be
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownJoinOnAssertNumRows.java
@@ -0,0 +1,229 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite;
+
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.AssertNumRowsElement;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAssertNumRows;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Push down join when one child is LogicalAssertNumRows.
+ * select * from T1 join T2 where T1.b=T2.b and T1.a > (select x from T3 ...)
+ *
+ * <pre>
+ * Case 1: Push to left child
+ * Before:
+ *     topJoin(T1.a > x)
+ *       |-- Project
+ *       |     `-- bottomJoin(T1.b = T2.b)
+ *       |           |-- Scan(T1)
+ *       |           `-- Scan(T2)
+ *       `-- LogicalAssertNumRows(output=(x, ...))
+ *
+ * After:
+ *     Project
+ *       |-- topJoin(T1.b = T2.b)
+ *               |-- bottomJoin(T1.a > x)
+ *                   |-- Scan(T1)
+ *                   `-- LogicalAssertNumRows(output=(x, ...))
+ *               `-- Scan(T2)
+ *
+ * Case 2: Push to right child
+ * Before:
+ *     topJoin(T2.a > x)
+ *       |-- Project
+ *       |     `-- bottomJoin(T1.b = T2.b)
+ *       |           |-- Scan(T1)
+ *       |           `-- Scan(T2)
+ *       `-- LogicalAssertNumRows(output=(x, ...))
+ *
+ * After:
+ *     Project
+ *       |-- topJoin(T1.b = T2.b)
+ *              |--Scan(T1)
+ *             `-- bottomJoin(T2.a > x)
+ *                   |-- Scan(T2)
+ *                   `-- LogicalAssertNumRows(output=(x, ...))
+ * </pre>
+ */
+public class PushDownJoinOnAssertNumRows extends OneRewriteRuleFactory {
+    @Override
+    public Rule build() {
+        return logicalJoin()
+                .when(topJoin -> pattenCheck(topJoin))
+                .then(topJoin -> pushDownAssertNumRowsJoin(topJoin))
+                .toRule(RuleType.PUSH_DOWN_JOIN_ON_ASSERT_NUM_ROWS);
+    }
+
+    private boolean pattenCheck(LogicalJoin topJoin) {
+        // 1. right is LogicalAssertNumRows or LogicalProject->LogicalAssertNumRows
+        // 2. left is join or project->join
+        // 3. only one join condition.
+        if (topJoin.getJoinType() != JoinType.INNER_JOIN && topJoin.getJoinType() != JoinType.CROSS_JOIN) {
+            return false;
+        }
+        LogicalJoin bottomJoin;
+        Plan left = topJoin.left();
+        Plan right = topJoin.right();
+        if (!isAssertOneRowEqOrProjectAssertOneRowEq(right)) {
+            return false;
+        }
+        if (left instanceof LogicalJoin) {
+            bottomJoin = (LogicalJoin) left;
+        } else if (left instanceof LogicalProject && left.child(0) instanceof LogicalJoin) {
+            bottomJoin = (LogicalJoin) left.child(0);
+        } else {
+            return false;
+        }
+
+        if (bottomJoin.getJoinType() != JoinType.INNER_JOIN && bottomJoin.getJoinType() != JoinType.CROSS_JOIN) {
+            return false;
+        }
+
+        if (joinOnAssertOneRowEq(bottomJoin)) {
+            return false;
+        }
+
+        if (topJoin.getHashJoinConjuncts().isEmpty()) {
+            return topJoin.getOtherJoinConjuncts().size() == 1;
+        }
+        return false;
+    }
+
+    private boolean isAssertOneRowEqOrProjectAssertOneRowEq(Plan plan) {
+        if (plan instanceof LogicalProject) {
+            plan = plan.child(0);
+        }
+        if (plan instanceof LogicalAssertNumRows) {
+            AssertNumRowsElement assertNumRowsElement = ((LogicalAssertNumRows) plan).getAssertNumRowsElement();
+            if (assertNumRowsElement.getAssertion() == AssertNumRowsElement.Assertion.EQ
+                    || assertNumRowsElement.getDesiredNumOfRows() == 1L) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private boolean joinOnAssertOneRowEq(LogicalJoin join) {
+        return isAssertOneRowEqOrProjectAssertOneRowEq(join.right())
+                || isAssertOneRowEqOrProjectAssertOneRowEq(join.left());
+    }
+
+    private Plan pushDownAssertNumRowsJoin(LogicalJoin topJoin) {
+        Plan assertBranch = topJoin.right();
+        Expression condition = (Expression) topJoin.getOtherJoinConjuncts().get(0);
+        List<Alias> aliasUsedInConditionFromLeftProject = new ArrayList<>();
+        LogicalJoin<? extends Plan, ? extends Plan> bottomJoin;
+        if (topJoin.left() instanceof LogicalProject) {
+            LogicalProject<? extends Plan> leftProject = (LogicalProject<? extends Plan>) topJoin.left();
+            for (NamedExpression namedExpression : leftProject.getProjects()) {
+                if (namedExpression instanceof Alias && condition.getInputSlots().contains(namedExpression.toSlot())) {
+                    aliasUsedInConditionFromLeftProject.add((Alias) namedExpression);
+                }
+            }
+            condition = leftProject.pushDownExpressionPastProject(condition);
+            bottomJoin = (LogicalJoin<? extends Plan, ? extends Plan>) leftProject.child();
+        } else {
+            bottomJoin = (LogicalJoin<? extends Plan, ? extends Plan>) topJoin.left();
+        }
+        Plan bottomLeft = bottomJoin.left();
+        Plan bottomRight = bottomJoin.right();
+
+        List<Slot> conditionSlotsFromTopLeft = condition.getInputSlots().stream()
+                .filter(slot -> topJoin.left().getOutputSet().contains(slot))
+                .collect(Collectors.toList());
+        if (bottomLeft.getOutputSet().containsAll(conditionSlotsFromTopLeft)) {
+            // push to bottomLeft
+            Plan newBottomLeft;
+            if (aliasUsedInConditionFromLeftProject.isEmpty()) {
+                newBottomLeft = bottomLeft;
+            } else {
+                newBottomLeft = projectAliasOnPlan(aliasUsedInConditionFromLeftProject, bottomLeft);
+            }
+            LogicalJoin<? extends Plan, ? extends Plan> newBottomJoin = new LogicalJoin<>(
+                    topJoin.getJoinType(),
+                    topJoin.getHashJoinConjuncts(),
+                    topJoin.getOtherJoinConjuncts(),
+                    newBottomLeft,
+                    assertBranch,
+                    topJoin.getJoinReorderContext());
+            LogicalJoin<? extends Plan, ? extends Plan> newTopJoin = (LogicalJoin<? extends Plan, ? extends Plan>)
+                    bottomJoin.withChildren(newBottomJoin, bottomRight);
+            if (topJoin.left() instanceof LogicalProject) {
+                LogicalProject<? extends Plan> upperProject = projectAliasOnPlan(
+                        aliasUsedInConditionFromLeftProject, topJoin.left());
+                return upperProject.withChildren(newTopJoin);
+            } else {
+                return newTopJoin;
+            }
+        } else if (bottomRight.getOutputSet().containsAll(conditionSlotsFromTopLeft)) {
+            Plan newBottomRight;
+            if (aliasUsedInConditionFromLeftProject.isEmpty()) {
+                newBottomRight = bottomRight;
+            } else {
+                newBottomRight = projectAliasOnPlan(aliasUsedInConditionFromLeftProject, bottomRight);
+            }
+            LogicalJoin<? extends Plan, ? extends Plan> newBottomJoin = new LogicalJoin<>(
+                    topJoin.getJoinType(),
+                    topJoin.getHashJoinConjuncts(),
+                    topJoin.getOtherJoinConjuncts(),
+                    newBottomRight,
+                    assertBranch,
+                    topJoin.getJoinReorderContext());
+            LogicalJoin<? extends Plan, ? extends Plan> newTopJoin = (LogicalJoin<? extends Plan, ? extends Plan>)
+                    bottomJoin.withChildren(bottomLeft, newBottomJoin);
+            if (topJoin.left() instanceof LogicalProject) {
+                LogicalProject<? extends Plan> upperProject = projectAliasOnPlan(
+                        aliasUsedInConditionFromLeftProject, topJoin.left());
+                return upperProject.withChildren(newTopJoin);
+            } else {
+                return newTopJoin;
+            }
+        }
+        return null;
+    }
+
+    private LogicalProject<? extends Plan> projectAliasOnPlan(List<Alias> projections, Plan child) {
+        if (child instanceof LogicalProject) {
+            LogicalProject<? extends Plan> project = (LogicalProject<? extends Plan>) child;
+            List<NamedExpression> newProjections =
+                    Lists.newArrayList(project.getProjects());
+            newProjections.addAll(projections);
+            return project.withProjects(newProjections);
+        } else {
+            List<NamedExpression> newProjections = Lists.newArrayList(child.getOutput());
+            newProjections.addAll(projections);
+            return new LogicalProject<>(newProjections, child);
+        }
+    }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalProject.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalProject.java
index 6925bcf8fe7f9b..fef5e46f2c07a7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalProject.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalProject.java
@@ -305,4 +305,23 @@ public void computeFd(DataTrait.Builder builder) {
             builder.addDeps(expr.getInputSlots(), ImmutableSet.of(expr.toSlot()));
         }
     }
+
+    /**
+     *
+     * example:
+     * expression: x + 1
+     * project(a+b as x)
+     * then before project, the expression is a+b+1
+     *
+     */
+    public Expression pushDownExpressionPastProject(Expression expression) {
+        HashMap projectMap = new HashMap();
+        for (NamedExpression namedExpression : projects) {
+            if (namedExpression instanceof Alias) {
+                Alias alias = (Alias) namedExpression;
+                projectMap.put(alias.toSlot(), alias.child());
+            }
+        }
+        return ExpressionUtils.replace(expression, projectMap);
+    }
 }
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PushDownJoinOnAssertNumRowsTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PushDownJoinOnAssertNumRowsTest.java
new file mode 100644
index 00000000000000..f59a700c61dcb3
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PushDownJoinOnAssertNumRowsTest.java
@@ -0,0 +1,454 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite;
+
+import org.apache.doris.nereids.trees.expressions.Add;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.AssertNumRowsElement;
+import org.apache.doris.nereids.trees.expressions.AssertNumRowsElement.Assertion;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAssertNumRows;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.util.LogicalPlanBuilder;
+import org.apache.doris.nereids.util.MemoPatternMatchSupported;
+import org.apache.doris.nereids.util.MemoTestUtils;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.nereids.util.PlanConstructor;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+import java.util.List;
+
+/**
+ * Test for PushDownJoinOnAssertNumRows rule.
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class PushDownJoinOnAssertNumRowsTest implements MemoPatternMatchSupported {
+
+    private LogicalOlapScan t1;
+    private LogicalOlapScan t2;
+    private LogicalOlapScan t3;
+
+    private List<Slot> t1Slots;
+    private List<Slot> t2Slots;
+    private List<Slot> t3Slots;
+
+    /**
+     * Setup test fixtures.
+     */
+    @BeforeAll
+    final void beforeAll() {
+        t1 = new LogicalOlapScan(PlanConstructor.getNextRelationId(), PlanConstructor.student,
+        ImmutableList.of(""));
+        t2 = new LogicalOlapScan(PlanConstructor.getNextRelationId(), PlanConstructor.score,
+        ImmutableList.of(""));
+        t3 = new LogicalOlapScan(PlanConstructor.getNextRelationId(), PlanConstructor.course,
+        ImmutableList.of(""));
+        t1Slots = t1.getOutput();
+        t2Slots = t2.getOutput();
+        t3Slots = t3.getOutput();
+    }
+
+    /**
+     * Test basic push down to left child scenario:
+     * Before:
+     * topJoin(T1.a > x)
+     * |-- bottomJoin(T1.b = T2.b)
+     * | |-- Scan(T1)
+     * | `-- Scan(T2)
+     * `-- LogicalAssertNumRows(output=(x, ...))
+     *
+     * After:
+     * bottomJoin(T1.b = T2.b)
+     * |-- topJoin(T1.a > x)
+     * | |-- Scan(T1)
+     * | `-- LogicalAssertNumRows(output=(x, ...))
+     * `-- Scan(T2)
+     */
+    @Test
+    void testPushDownToLeftChild() {
+        // Create a one-row relation wrapped in LogicalAssertNumRows
+        Plan oneRowRelation = new LogicalPlanBuilder(t3)
+                .limit(1)
+                .build();
+
+        AssertNumRowsElement assertElement = new AssertNumRowsElement(1, "", Assertion.EQ);
+        LogicalAssertNumRows<Plan> assertNumRows = new LogicalAssertNumRows<>(assertElement, oneRowRelation);
+
+        // Create bottom join: T1 JOIN T2 on T1.id = T2.sid
+        Expression bottomJoinCondition = new EqualTo(t1Slots.get(0), t2Slots.get(1));
+
+        LogicalPlan bottomJoin = new LogicalPlanBuilder(t1)
+                .join(t2, JoinType.INNER_JOIN, ImmutableList.of(bottomJoinCondition),
+                                ImmutableList.of())
+                .build();
+
+        // Create top join: (T1 JOIN T2) JOIN assertNumRows on T1.age > course.id
+        Expression topJoinCondition = new GreaterThan(t1Slots.get(1), t3Slots.get(0));
+
+        LogicalPlan root = new LogicalPlanBuilder(bottomJoin)
+                .join(assertNumRows, JoinType.INNER_JOIN, ImmutableList.of(),
+                                ImmutableList.of(topJoinCondition))
+                .build();
+
+        // Apply the rule
+        PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+                .applyTopDown(new PushDownJoinOnAssertNumRows())
+                .matches(logicalJoin(
+                                logicalJoin(
+                                                        logicalOlapScan(),
+                                                        logicalAssertNumRows()),
+                                logicalOlapScan()));
+    }
+
+    /**
+     * Test push down to right child scenario:
+     * Before:
+     * topJoin(T2.a > x)
+     * |-- bottomJoin(T1.b = T2.b)
+     * | |-- Scan(T1)
+     * | `-- Scan(T2)
+     * `-- LogicalAssertNumRows(output=(x, ...))
+     *
+     * After:
+     * bottomJoin(T1.b = T2.b)
+     * |-- Scan(T1)
+     * `-- topJoin(T2.a > x)
+     * |-- Scan(T2)
+     * `-- LogicalAssertNumRows(output=(x, ...))
+     */
+    @Test
+    void testPushDownToRightChild() {
+        // Create a one-row relation wrapped in LogicalAssertNumRows
+        Plan oneRowRelation = new LogicalPlanBuilder(t3)
+                .limit(1)
+                .build();
+
+        AssertNumRowsElement assertElement = new AssertNumRowsElement(1, "", Assertion.EQ);
+        LogicalAssertNumRows<Plan> assertNumRows = new LogicalAssertNumRows<>(assertElement, oneRowRelation);
+
+        // Create bottom join: T1 JOIN T2 on T1.id = T2.sid
+        Expression bottomJoinCondition = new EqualTo(t1Slots.get(0), t2Slots.get(1));
+
+        LogicalPlan bottomJoin = new LogicalPlanBuilder(t1)
+                .join(t2, JoinType.INNER_JOIN, ImmutableList.of(bottomJoinCondition),
+                                ImmutableList.of())
+                .build();
+
+        // Create top join: (T1 JOIN T2) JOIN assertNumRows on T2.name > course.name
+        // This references T2 (right child of bottom join) and assertNumRows
+        Expression topJoinCondition = new GreaterThan(t2Slots.get(2), t3Slots.get(1));
+
+        LogicalPlan root = new LogicalPlanBuilder(bottomJoin)
+                .join(assertNumRows, JoinType.INNER_JOIN, ImmutableList.of(),
+                                ImmutableList.of(topJoinCondition))
+                .build();
+
+        // Apply the rule
+        PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+                .applyTopDown(new PushDownJoinOnAssertNumRows())
+                .matches(logicalJoin(
+                                logicalOlapScan(),
+                                logicalJoin(
+                                                        logicalOlapScan(),
+                                                        logicalAssertNumRows())));
+    }
+
+    /**
+     * Test with project node between top join and bottom join:
+     * Before:
+     * topJoin(alias_col > x)
+     * |-- Project(T1.id, T1.age + 1 as alias_col, ...)
+     * | `-- bottomJoin(T1.id = T2.sid)
+     * | |-- Scan(T1)
+     * | `-- Scan(T2)
+     * `-- LogicalAssertNumRows(output=(x, ...))
+     *
+     * After:
+     * Project(...)
+     * |-- bottomJoin(T1.id = T2.sid)
+     * |-- topJoin(T1.age + 1 > x)
+     * | |-- Scan(T1)
+     * | `-- LogicalAssertNumRows(output=(x, ...))
+     * `-- Scan(T2)
+     */
+    @Test
+    void testPushDownWithProjectNode() {
+        // Create a one-row relation wrapped in LogicalAssertNumRows
+        Plan oneRowRelation = new LogicalPlanBuilder(t3)
+                .limit(1)
+                .build();
+
+        AssertNumRowsElement assertElement = new AssertNumRowsElement(1, "", Assertion.EQ);
+        LogicalAssertNumRows<Plan> assertNumRows = new LogicalAssertNumRows<>(assertElement, oneRowRelation);
+
+        // Create bottom join: T1 JOIN T2 on T1.id = T2.sid
+        Expression bottomJoinCondition = new EqualTo(t1Slots.get(0), t2Slots.get(1));
+
+        LogicalPlan bottomJoin = new LogicalPlanBuilder(t1)
+                .join(t2, JoinType.INNER_JOIN, ImmutableList.of(bottomJoinCondition),
+                                ImmutableList.of())
+                .build();
+
+        // Create project with alias: T1.age + 1 as alias_col
+        Expression addExpr = new Add(t1Slots.get(1), Literal.of(1));
+        Alias aliasCol = new Alias(addExpr, "alias_col");
+
+        ImmutableList.Builder<NamedExpression> projectListBuilder = ImmutableList.builder();
+        projectListBuilder.add(t1Slots.get(0)); // T1.id
+        projectListBuilder.add(aliasCol); // T1.age + 1 as alias_col
+        projectListBuilder.addAll(t2Slots); // All T2 columns
+
+        LogicalProject<Plan> project = new LogicalProject<>(projectListBuilder.build(), bottomJoin);
+
+        // Create top join: project JOIN assertNumRows on alias_col > course.id
+        Expression topJoinCondition = new GreaterThan(aliasCol.toSlot(), t3Slots.get(0));
+
+        LogicalPlan root = new LogicalPlanBuilder(project)
+                .join(assertNumRows, JoinType.INNER_JOIN, ImmutableList.of(),
+                                ImmutableList.of(topJoinCondition))
+                .build();
+
+        // Apply the rule
+        PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+                .applyTopDown(new PushDownJoinOnAssertNumRows())
+                .matches(logicalProject(
+                                logicalJoin(
+                                                        logicalJoin(
+                                                                        logicalProject(logicalOlapScan()),
+                                                                        logicalAssertNumRows()),
+                                                        logicalOlapScan())));
+    }
+
+    /**
+     * Test with CROSS JOIN type.
+     */
+    @Test
+    void testWithCrossJoin() {
+        // Create a one-row relation wrapped in LogicalAssertNumRows
+        Plan oneRowRelation = new LogicalPlanBuilder(t3)
+                .limit(1)
+                .build();
+
+        AssertNumRowsElement assertElement = new AssertNumRowsElement(1, "", Assertion.EQ);
+        LogicalAssertNumRows<Plan> assertNumRows = new LogicalAssertNumRows<>(assertElement, oneRowRelation);
+
+        // Create bottom cross join: T1 CROSS JOIN T2
+        LogicalPlan bottomJoin = new LogicalPlanBuilder(t1)
+                .join(t2, JoinType.CROSS_JOIN, ImmutableList.of(), ImmutableList.of())
+                .build();
+
+        // Create top join with condition: T1.age > course.id
+        Expression topJoinCondition = new GreaterThan(t1Slots.get(1), t3Slots.get(0));
+
+        LogicalPlan root = new LogicalPlanBuilder(bottomJoin)
+                .join(assertNumRows, JoinType.CROSS_JOIN, ImmutableList.of(),
+                                ImmutableList.of(topJoinCondition))
+                .build();
+
+        // Apply the rule
+        PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+                .applyTopDown(new PushDownJoinOnAssertNumRows())
+                .matches(logicalJoin(
+                                logicalJoin(
+                                                        logicalOlapScan(),
+                                                        logicalAssertNumRows()),
+                                logicalOlapScan()));
+    }
+
+    /**
+     * Test that rule doesn't apply when bottom join already has assertNumRows.
+     */
+    @Test
+    void testNoApplyWhenBottomJoinHasAssertNumRows() {
+        // Create two one-row relations
+        Plan oneRowRelation1 = new LogicalPlanBuilder(t2)
+                .limit(1)
+                .build();
+        Plan oneRowRelation2 = new LogicalPlanBuilder(t3)
+                .limit(1)
+                .build();
+
+        AssertNumRowsElement assertElement1 = new AssertNumRowsElement(1, "", Assertion.EQ);
+        AssertNumRowsElement assertElement2 = new AssertNumRowsElement(1, "", Assertion.EQ);
+        LogicalAssertNumRows<Plan> assertNumRows1 = new LogicalAssertNumRows<>(assertElement1, oneRowRelation1);
+        LogicalAssertNumRows<Plan> assertNumRows2 = new LogicalAssertNumRows<>(assertElement2, oneRowRelation2);
+
+        // Create bottom join that already has assertNumRows: T1 JOIN assertNumRows1
+        Expression bottomJoinCondition = new GreaterThan(t1Slots.get(1), t2Slots.get(0));
+
+        LogicalPlan bottomJoin = new LogicalPlanBuilder(t1)
+                .join(assertNumRows1, JoinType.INNER_JOIN, ImmutableList.of(),
+                                ImmutableList.of(bottomJoinCondition))
+                .build();
+
+        // Create top join with another assertNumRows
+        Expression topJoinCondition = new GreaterThan(t1Slots.get(1), t3Slots.get(0));
+
+        LogicalPlan root = new LogicalPlanBuilder(bottomJoin)
+                .join(assertNumRows2, JoinType.INNER_JOIN, ImmutableList.of(),
+                                ImmutableList.of(topJoinCondition))
+                .build();
+
+        // Apply the rule - should not transform because bottom join already has
+        // assertNumRows
+        PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+                .applyTopDown(new PushDownJoinOnAssertNumRows())
+                .matches(logicalJoin(
+                                logicalJoin(
+                                                        logicalOlapScan(),
+                                                        logicalAssertNumRows()),
+                                logicalAssertNumRows()));
+    }
+
+    /**
+     * Test that rule doesn't apply when there are multiple join conditions.
+     */
+    @Test
+    void testNoApplyWithMultipleJoinConditions() {
+        // Create a one-row relation wrapped in LogicalAssertNumRows
+        Plan oneRowRelation = new LogicalPlanBuilder(t3)
+                .limit(1)
+                .build();
+
+        AssertNumRowsElement assertElement = new AssertNumRowsElement(1, "", Assertion.EQ);
+        LogicalAssertNumRows<Plan> assertNumRows = new LogicalAssertNumRows<>(assertElement, oneRowRelation);
+
+        // Create bottom join: T1 JOIN T2 on T1.id = T2.sid
+        Expression bottomJoinCondition = new EqualTo(t1Slots.get(0), t2Slots.get(1));
+
+        LogicalPlan bottomJoin = new LogicalPlanBuilder(t1)
+                .join(t2, JoinType.INNER_JOIN, ImmutableList.of(bottomJoinCondition),
+                                ImmutableList.of())
+                .build();
+
+        // Create top join with multiple conditions
+        Expression topJoinCondition1 = new GreaterThan(t1Slots.get(1), t3Slots.get(0));
+        Expression topJoinCondition2 = new GreaterThan(t2Slots.get(0), t3Slots.get(0));
+
+        LogicalPlan root = new LogicalPlanBuilder(bottomJoin)
+                .join(assertNumRows, JoinType.INNER_JOIN, ImmutableList.of(),
+                                ImmutableList.of(topJoinCondition1, topJoinCondition2))
+                .build();
+
+        // Apply the rule - should not transform because there are multiple join
+        // conditions
+        PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+                .applyTopDown(new PushDownJoinOnAssertNumRows())
+                .matches(logicalJoin(
+                                logicalJoin(
+                                                        logicalOlapScan(),
+                                                        logicalOlapScan()),
+                                logicalAssertNumRows()));
+    }
+
+    /**
+     * Test that rule doesn't apply with outer join types.
+     */
+    @Test
+    void testNoApplyWithOuterJoin() {
+        // Create a one-row relation wrapped in LogicalAssertNumRows
+        Plan oneRowRelation = new LogicalPlanBuilder(t3)
+                .limit(1)
+                .build();
+
+        AssertNumRowsElement assertElement = new AssertNumRowsElement(1, "", Assertion.EQ);
+        LogicalAssertNumRows<Plan> assertNumRows = new LogicalAssertNumRows<>(assertElement, oneRowRelation);
+
+        // Create bottom left outer join: T1 LEFT JOIN T2 on T1.id = T2.sid
+        Expression bottomJoinCondition = new EqualTo(t1Slots.get(0), t2Slots.get(1));
+
+        LogicalPlan bottomJoin = new LogicalPlanBuilder(t1)
+                .join(t2, JoinType.LEFT_OUTER_JOIN, ImmutableList.of(bottomJoinCondition),
+                                ImmutableList.of())
+                .build();
+
+        // Create top join with condition
+        Expression topJoinCondition = new GreaterThan(t1Slots.get(1), t3Slots.get(0));
+
+        LogicalPlan root = new LogicalPlanBuilder(bottomJoin)
+                .join(assertNumRows, JoinType.INNER_JOIN, ImmutableList.of(),
+                                ImmutableList.of(topJoinCondition))
+                .build();
+
+        // Apply the rule - should not transform because bottom join is outer join
+        PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+                .applyTopDown(new PushDownJoinOnAssertNumRows())
+                .matches(logicalJoin(
+                                logicalJoin(
+                                                        logicalOlapScan(),
+                                                        logicalOlapScan()),
+                                logicalAssertNumRows()));
+    }
+
+    /**
+     * Test with assertNumRows wrapped in project.
+     */
+    @Test
+    void testWithAssertNumRowsInProject() {
+        // Create a one-row relation wrapped in LogicalAssertNumRows and then Project
+        Plan oneRowRelation = new LogicalPlanBuilder(t3)
+                .limit(1)
+                .build();
+
+        AssertNumRowsElement assertElement = new AssertNumRowsElement(1, "", Assertion.EQ);
+        LogicalAssertNumRows<Plan> assertNumRows = new LogicalAssertNumRows<>(assertElement, oneRowRelation);
+
+        // Wrap assertNumRows in a project
+        LogicalProject<Plan> assertProject = new LogicalProject<>(
+                ImmutableList.copyOf(assertNumRows.getOutput()), assertNumRows);
+
+        // Create bottom join: T1 JOIN T2 on T1.id = T2.sid
+        Expression bottomJoinCondition = new EqualTo(t1Slots.get(0), t2Slots.get(1));
+
+        LogicalPlan bottomJoin = new LogicalPlanBuilder(t1)
+                .join(t2, JoinType.INNER_JOIN, ImmutableList.of(bottomJoinCondition),
+                                ImmutableList.of())
+                .build();
+
+        // Create top join with project-wrapped assertNumRows
+        Expression topJoinCondition = new GreaterThan(t1Slots.get(1), t3Slots.get(0));
+
+        LogicalPlan root = new LogicalPlanBuilder(bottomJoin)
+                .join(assertProject, JoinType.INNER_JOIN, ImmutableList.of(),
+                                ImmutableList.of(topJoinCondition))
+                .build();
+
+        // Apply the rule
+        PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+                .applyTopDown(new PushDownJoinOnAssertNumRows())
+                .matches(logicalJoin(
+                                logicalJoin(
+                                                        logicalOlapScan(),
+                                                        logicalProject(
+                                                                        logicalAssertNumRows())),
+                                logicalOlapScan()));
+    }
+}
diff --git a/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query54.out b/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query54.out
index 46ff85eba23a08..84b5754b67aedb 100644
--- a/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query54.out
+++ b/regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query54.out
@@ -11,48 +11,56 @@ PhysicalResultSink
 ----------------PhysicalProject
 ------------------hashAgg[GLOBAL]
 --------------------PhysicalProject
-----------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=()
 ------------------------PhysicalProject
---------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=()
 ----------------------------PhysicalProject
-------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=()
+------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=()
 --------------------------------PhysicalProject
-----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=()
+----------------------------------hashJoin[INNER_JOIN shuffleBucket] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF3 c_customer_sk->[ss_customer_sk]
 ------------------------------------PhysicalProject
---------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=()
+--------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF3
+------------------------------------PhysicalProject
+--------------------------------------hashAgg[GLOBAL]
 ----------------------------------------PhysicalProject
-------------------------------------------hashJoin[INNER_JOIN shuffleBucket] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF3 c_customer_sk->[ss_customer_sk]
---------------------------------------------PhysicalProject
-----------------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF3
+------------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=()
 --------------------------------------------PhysicalProject
-----------------------------------------------hashAgg[GLOBAL]
+----------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
 ------------------------------------------------PhysicalProject
---------------------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=()
-----------------------------------------------------PhysicalProject
-------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
+--------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
+----------------------------------------------------PhysicalUnion
+------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 --------------------------------------------------------PhysicalProject
-----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
-------------------------------------------------------------PhysicalUnion
---------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
---------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------filter((item.i_category = 'Women') and (item.i_class = 'maternity'))
-----------------------------------------------------------------PhysicalOlapScan[item]
+----------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
+------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 --------------------------------------------------------PhysicalProject
-----------------------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
-------------------------------------------------------------PhysicalOlapScan[date_dim]
+----------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
 ----------------------------------------------------PhysicalProject
-------------------------------------------------------PhysicalOlapScan[customer]
-----------------------------------------PhysicalProject
-------------------------------------------PhysicalOlapScan[customer_address]
-------------------------------------PhysicalProject
---------------------------------------PhysicalOlapScan[store]
+------------------------------------------------------filter((item.i_category = 'Women') and (item.i_class = 'maternity'))
+--------------------------------------------------------PhysicalOlapScan[item]
+------------------------------------------------PhysicalProject
+--------------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
+----------------------------------------------------PhysicalOlapScan[date_dim]
+--------------------------------------------PhysicalProject
+----------------------------------------------PhysicalOlapScan[customer]
+--------------------------------PhysicalProject
+----------------------------------PhysicalOlapScan[customer_address]
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[store]
+------------------------PhysicalProject
+--------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+----------------------------PhysicalProject
+------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
 --------------------------------PhysicalProject
 ----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalAssertNumRows
+----------------------------------PhysicalDistribute[DistributionSpecGather]
+------------------------------------hashAgg[GLOBAL]
+--------------------------------------PhysicalDistribute[DistributionSpecHash]
+----------------------------------------hashAgg[LOCAL]
+------------------------------------------PhysicalProject
+--------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
+----------------------------------------------PhysicalOlapScan[date_dim]
 ----------------------------PhysicalAssertNumRows
 ------------------------------PhysicalDistribute[DistributionSpecGather]
 --------------------------------hashAgg[GLOBAL]
@@ -61,12 +69,4 @@ PhysicalResultSink
 --------------------------------------PhysicalProject
 ----------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
 ------------------------------------------PhysicalOlapScan[date_dim]
-------------------------PhysicalAssertNumRows
---------------------------PhysicalDistribute[DistributionSpecGather]
-----------------------------hashAgg[GLOBAL]
-------------------------------PhysicalDistribute[DistributionSpecHash]
---------------------------------hashAgg[LOCAL]
-----------------------------------PhysicalProject
-------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
---------------------------------------PhysicalOlapScan[date_dim]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query54.out b/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query54.out
index 0be42521255ae6..a2a326baf8f39a 100644
--- a/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query54.out
+++ b/regression-test/data/shape_check/tpcds_sf100/no_stats_shape/query54.out
@@ -11,48 +11,56 @@ PhysicalResultSink
 ----------------PhysicalProject
 ------------------hashAgg[GLOBAL]
 --------------------PhysicalProject
-----------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
 ------------------------PhysicalProject
---------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF5 s_county->[ca_county];RF6 s_state->[ca_state]
 ----------------------------PhysicalProject
-------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
+------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF4 ca_address_sk->[c_current_addr_sk]
 --------------------------------PhysicalProject
-----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF5 s_county->[ca_county];RF6 s_state->[ca_state]
+----------------------------------hashJoin[INNER_JOIN shuffleBucket] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF3 c_customer_sk->[ss_customer_sk]
 ------------------------------------PhysicalProject
---------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF4 ca_address_sk->[c_current_addr_sk]
+--------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF3 RF7
+------------------------------------PhysicalProject
+--------------------------------------hashAgg[GLOBAL]
 ----------------------------------------PhysicalProject
-------------------------------------------hashJoin[INNER_JOIN shuffleBucket] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF3 c_customer_sk->[ss_customer_sk]
---------------------------------------------PhysicalProject
-----------------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF3 RF7
+------------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 c_customer_sk->[cs_bill_customer_sk,ws_bill_customer_sk]
 --------------------------------------------PhysicalProject
-----------------------------------------------hashAgg[GLOBAL]
+----------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
 ------------------------------------------------PhysicalProject
---------------------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 c_customer_sk->[cs_bill_customer_sk,ws_bill_customer_sk]
-----------------------------------------------------PhysicalProject
-------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
+--------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
+----------------------------------------------------PhysicalUnion
+------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 --------------------------------------------------------PhysicalProject
-----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
-------------------------------------------------------------PhysicalUnion
---------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
---------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------filter((item.i_category = 'Women') and (item.i_class = 'maternity'))
-----------------------------------------------------------------PhysicalOlapScan[item]
+----------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
+------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 --------------------------------------------------------PhysicalProject
-----------------------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
-------------------------------------------------------------PhysicalOlapScan[date_dim]
+----------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
 ----------------------------------------------------PhysicalProject
-------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF4
-----------------------------------------PhysicalProject
-------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF5 RF6
-------------------------------------PhysicalProject
---------------------------------------PhysicalOlapScan[store]
+------------------------------------------------------filter((item.i_category = 'Women') and (item.i_class = 'maternity'))
+--------------------------------------------------------PhysicalOlapScan[item]
+------------------------------------------------PhysicalProject
+--------------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
+----------------------------------------------------PhysicalOlapScan[date_dim]
+--------------------------------------------PhysicalProject
+----------------------------------------------PhysicalOlapScan[customer] apply RFs: RF4
+--------------------------------PhysicalProject
+----------------------------------PhysicalOlapScan[customer_address] apply RFs: RF5 RF6
+----------------------------PhysicalProject
+------------------------------PhysicalOlapScan[store]
+------------------------PhysicalProject
+--------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+----------------------------PhysicalProject
+------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
 --------------------------------PhysicalProject
 ----------------------------------PhysicalOlapScan[date_dim]
+--------------------------------PhysicalAssertNumRows
+----------------------------------PhysicalDistribute[DistributionSpecGather]
+------------------------------------hashAgg[GLOBAL]
+--------------------------------------PhysicalDistribute[DistributionSpecHash]
+----------------------------------------hashAgg[LOCAL]
+------------------------------------------PhysicalProject
+--------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
+----------------------------------------------PhysicalOlapScan[date_dim]
 ----------------------------PhysicalAssertNumRows
 ------------------------------PhysicalDistribute[DistributionSpecGather]
 --------------------------------hashAgg[GLOBAL]
@@ -61,12 +69,4 @@ PhysicalResultSink
 --------------------------------------PhysicalProject
 ----------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
 ------------------------------------------PhysicalOlapScan[date_dim]
-------------------------PhysicalAssertNumRows
---------------------------PhysicalDistribute[DistributionSpecGather]
-----------------------------hashAgg[GLOBAL]
-------------------------------PhysicalDistribute[DistributionSpecHash]
---------------------------------hashAgg[LOCAL]
-----------------------------------PhysicalProject
-------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
---------------------------------------PhysicalOlapScan[date_dim]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/rf_prune/query54.out b/regression-test/data/shape_check/tpcds_sf100/rf_prune/query54.out
index a1637bb5a33f34..ad8a262d79400b 100644
--- a/regression-test/data/shape_check/tpcds_sf100/rf_prune/query54.out
+++ b/regression-test/data/shape_check/tpcds_sf100/rf_prune/query54.out
@@ -13,50 +13,58 @@ PhysicalResultSink
 --------------------PhysicalDistribute[DistributionSpecHash]
 ----------------------hashAgg[LOCAL]
 ------------------------PhysicalProject
---------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
 ----------------------------PhysicalProject
-------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
+------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF6 c_customer_sk->[ss_customer_sk]
 --------------------------------PhysicalProject
-----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=()
+----------------------------------PhysicalOlapScan[store_sales] apply RFs: RF6 RF7
+--------------------------------PhysicalProject
+----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF4 s_county->[ca_county];RF5 s_state->[ca_state]
 ------------------------------------PhysicalProject
---------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF6 c_customer_sk->[ss_customer_sk]
+--------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 c_current_addr_sk->[ca_address_sk]
 ----------------------------------------PhysicalProject
-------------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF6
+------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF3 RF4 RF5
 ----------------------------------------PhysicalProject
-------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF4 s_county->[ca_county];RF5 s_state->[ca_state]
---------------------------------------------PhysicalProject
-----------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 c_current_addr_sk->[ca_address_sk]
-------------------------------------------------PhysicalProject
---------------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF3 RF4 RF5
+------------------------------------------hashAgg[GLOBAL]
+--------------------------------------------PhysicalDistribute[DistributionSpecHash]
+----------------------------------------------hashAgg[LOCAL]
 ------------------------------------------------PhysicalProject
---------------------------------------------------hashAgg[GLOBAL]
-----------------------------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------------------------hashAgg[LOCAL]
+--------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 customer_sk->[c_customer_sk]
+----------------------------------------------------PhysicalProject
+------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF2
+----------------------------------------------------PhysicalProject
+------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
 --------------------------------------------------------PhysicalProject
-----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 customer_sk->[c_customer_sk]
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF2
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
+----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
+------------------------------------------------------------PhysicalUnion
+--------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
---------------------------------------------------------------------PhysicalUnion
-----------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-------------------------------------------------------------------------PhysicalProject
---------------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
-----------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-------------------------------------------------------------------------PhysicalProject
---------------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
---------------------------------------------------------------------PhysicalProject
-----------------------------------------------------------------------filter((item.i_category = 'Women') and (item.i_class = 'maternity'))
-------------------------------------------------------------------------PhysicalOlapScan[item]
+------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
+--------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
---------------------------------------------------------------------PhysicalOlapScan[date_dim]
---------------------------------------------PhysicalProject
-----------------------------------------------PhysicalOlapScan[store]
+------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
+------------------------------------------------------------PhysicalProject
+--------------------------------------------------------------filter((item.i_category = 'Women') and (item.i_class = 'maternity'))
+----------------------------------------------------------------PhysicalOlapScan[item]
+--------------------------------------------------------PhysicalProject
+----------------------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
+------------------------------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalProject
+--------------------------------------PhysicalOlapScan[store]
+----------------------------PhysicalProject
+------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------------PhysicalProject
+----------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
 ------------------------------------PhysicalProject
 --------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalAssertNumRows
+--------------------------------------PhysicalDistribute[DistributionSpecGather]
+----------------------------------------hashAgg[GLOBAL]
+------------------------------------------PhysicalDistribute[DistributionSpecHash]
+--------------------------------------------hashAgg[LOCAL]
+----------------------------------------------PhysicalProject
+------------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
+--------------------------------------------------PhysicalOlapScan[date_dim]
 --------------------------------PhysicalAssertNumRows
 ----------------------------------PhysicalDistribute[DistributionSpecGather]
 ------------------------------------hashAgg[GLOBAL]
@@ -65,12 +73,4 @@ PhysicalResultSink
 ------------------------------------------PhysicalProject
 --------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
 ----------------------------------------------PhysicalOlapScan[date_dim]
-----------------------------PhysicalAssertNumRows
-------------------------------PhysicalDistribute[DistributionSpecGather]
---------------------------------hashAgg[GLOBAL]
-----------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------hashAgg[LOCAL]
---------------------------------------PhysicalProject
-----------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
-------------------------------------------PhysicalOlapScan[date_dim]
 
diff --git a/regression-test/data/shape_check/tpcds_sf100/shape/query54.out b/regression-test/data/shape_check/tpcds_sf100/shape/query54.out
index 0f6f28c79ce7ce..ad8a262d79400b 100644
--- a/regression-test/data/shape_check/tpcds_sf100/shape/query54.out
+++ b/regression-test/data/shape_check/tpcds_sf100/shape/query54.out
@@ -13,50 +13,58 @@ PhysicalResultSink
 --------------------PhysicalDistribute[DistributionSpecHash]
 ----------------------hashAgg[LOCAL]
 ------------------------PhysicalProject
---------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
 ----------------------------PhysicalProject
-------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
+------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF6 c_customer_sk->[ss_customer_sk]
 --------------------------------PhysicalProject
-----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
+----------------------------------PhysicalOlapScan[store_sales] apply RFs: RF6 RF7
+--------------------------------PhysicalProject
+----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF4 s_county->[ca_county];RF5 s_state->[ca_state]
 ------------------------------------PhysicalProject
---------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF6 c_customer_sk->[ss_customer_sk]
+--------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 c_current_addr_sk->[ca_address_sk]
 ----------------------------------------PhysicalProject
-------------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF6 RF7
+------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF3 RF4 RF5
 ----------------------------------------PhysicalProject
-------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF4 s_county->[ca_county];RF5 s_state->[ca_state]
---------------------------------------------PhysicalProject
-----------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 c_current_addr_sk->[ca_address_sk]
-------------------------------------------------PhysicalProject
---------------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF3 RF4 RF5
+------------------------------------------hashAgg[GLOBAL]
+--------------------------------------------PhysicalDistribute[DistributionSpecHash]
+----------------------------------------------hashAgg[LOCAL]
 ------------------------------------------------PhysicalProject
---------------------------------------------------hashAgg[GLOBAL]
-----------------------------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------------------------hashAgg[LOCAL]
+--------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 customer_sk->[c_customer_sk]
+----------------------------------------------------PhysicalProject
+------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF2
+----------------------------------------------------PhysicalProject
+------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
 --------------------------------------------------------PhysicalProject
-----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 customer_sk->[c_customer_sk]
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF2
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
+----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
+------------------------------------------------------------PhysicalUnion
+--------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
---------------------------------------------------------------------PhysicalUnion
-----------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-------------------------------------------------------------------------PhysicalProject
---------------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
-----------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-------------------------------------------------------------------------PhysicalProject
---------------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
---------------------------------------------------------------------PhysicalProject
-----------------------------------------------------------------------filter((item.i_category = 'Women') and (item.i_class = 'maternity'))
-------------------------------------------------------------------------PhysicalOlapScan[item]
+------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
+--------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
---------------------------------------------------------------------PhysicalOlapScan[date_dim]
---------------------------------------------PhysicalProject
-----------------------------------------------PhysicalOlapScan[store]
+------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
+------------------------------------------------------------PhysicalProject
+--------------------------------------------------------------filter((item.i_category = 'Women') and (item.i_class = 'maternity'))
+----------------------------------------------------------------PhysicalOlapScan[item]
+--------------------------------------------------------PhysicalProject
+----------------------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
+------------------------------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalProject
+--------------------------------------PhysicalOlapScan[store]
+----------------------------PhysicalProject
+------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------------PhysicalProject
+----------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
 ------------------------------------PhysicalProject
 --------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalAssertNumRows
+--------------------------------------PhysicalDistribute[DistributionSpecGather]
+----------------------------------------hashAgg[GLOBAL]
+------------------------------------------PhysicalDistribute[DistributionSpecHash]
+--------------------------------------------hashAgg[LOCAL]
+----------------------------------------------PhysicalProject
+------------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
+--------------------------------------------------PhysicalOlapScan[date_dim]
 --------------------------------PhysicalAssertNumRows
 ----------------------------------PhysicalDistribute[DistributionSpecGather]
 ------------------------------------hashAgg[GLOBAL]
@@ -65,12 +73,4 @@ PhysicalResultSink
 ------------------------------------------PhysicalProject
 --------------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
 ----------------------------------------------PhysicalOlapScan[date_dim]
-----------------------------PhysicalAssertNumRows
-------------------------------PhysicalDistribute[DistributionSpecGather]
---------------------------------hashAgg[GLOBAL]
-----------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------hashAgg[LOCAL]
---------------------------------------PhysicalProject
-----------------------------------------filter((date_dim.d_moy = 5) and (date_dim.d_year = 1998))
-------------------------------------------PhysicalOlapScan[date_dim]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query54.out b/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query54.out
index ba570488dc0f94..b7c2fe2adcd5b6 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query54.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/bs_downgrade_shape/query54.out
@@ -13,50 +13,58 @@ PhysicalResultSink
 --------------------PhysicalDistribute[DistributionSpecHash]
 ----------------------hashAgg[LOCAL]
 ------------------------PhysicalProject
---------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
 ----------------------------PhysicalProject
-------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
+------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF6 c_customer_sk->[ss_customer_sk]
 --------------------------------PhysicalProject
-----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
+----------------------------------PhysicalOlapScan[store_sales] apply RFs: RF6 RF7
+--------------------------------PhysicalProject
+----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF4 s_county->[ca_county];RF5 s_state->[ca_state]
 ------------------------------------PhysicalProject
---------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF6 c_customer_sk->[ss_customer_sk]
+--------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 c_current_addr_sk->[ca_address_sk]
 ----------------------------------------PhysicalProject
-------------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF6 RF7
+------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF3 RF4 RF5
 ----------------------------------------PhysicalProject
-------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF4 s_county->[ca_county];RF5 s_state->[ca_state]
---------------------------------------------PhysicalProject
-----------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 c_current_addr_sk->[ca_address_sk]
-------------------------------------------------PhysicalProject
---------------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF3 RF4 RF5
+------------------------------------------hashAgg[GLOBAL]
+--------------------------------------------PhysicalDistribute[DistributionSpecHash]
+----------------------------------------------hashAgg[LOCAL]
 ------------------------------------------------PhysicalProject
---------------------------------------------------hashAgg[GLOBAL]
-----------------------------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------------------------hashAgg[LOCAL]
+--------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 customer_sk->[c_customer_sk]
+----------------------------------------------------PhysicalProject
+------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF2
+----------------------------------------------------PhysicalProject
+------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
 --------------------------------------------------------PhysicalProject
-----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 customer_sk->[c_customer_sk]
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF2
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
+----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
+------------------------------------------------------------PhysicalUnion
+--------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
---------------------------------------------------------------------PhysicalUnion
-----------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-------------------------------------------------------------------------PhysicalProject
---------------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
-----------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-------------------------------------------------------------------------PhysicalProject
---------------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
---------------------------------------------------------------------PhysicalProject
-----------------------------------------------------------------------filter((item.i_category = 'Music') and (item.i_class = 'country'))
-------------------------------------------------------------------------PhysicalOlapScan[item]
+------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
+--------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
---------------------------------------------------------------------PhysicalOlapScan[date_dim]
---------------------------------------------PhysicalProject
-----------------------------------------------PhysicalOlapScan[store]
+------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
+------------------------------------------------------------PhysicalProject
+--------------------------------------------------------------filter((item.i_category = 'Music') and (item.i_class = 'country'))
+----------------------------------------------------------------PhysicalOlapScan[item]
+--------------------------------------------------------PhysicalProject
+----------------------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
+------------------------------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalProject
+--------------------------------------PhysicalOlapScan[store]
+----------------------------PhysicalProject
+------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------------PhysicalProject
+----------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
 ------------------------------------PhysicalProject
 --------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalAssertNumRows
+--------------------------------------PhysicalDistribute[DistributionSpecGather]
+----------------------------------------hashAgg[GLOBAL]
+------------------------------------------PhysicalDistribute[DistributionSpecHash]
+--------------------------------------------hashAgg[LOCAL]
+----------------------------------------------PhysicalProject
+------------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
+--------------------------------------------------PhysicalOlapScan[date_dim]
 --------------------------------PhysicalAssertNumRows
 ----------------------------------PhysicalDistribute[DistributionSpecGather]
 ------------------------------------hashAgg[GLOBAL]
@@ -65,12 +73,4 @@ PhysicalResultSink
 ------------------------------------------PhysicalProject
 --------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
 ----------------------------------------------PhysicalOlapScan[date_dim]
-----------------------------PhysicalAssertNumRows
-------------------------------PhysicalDistribute[DistributionSpecGather]
---------------------------------hashAgg[GLOBAL]
-----------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------hashAgg[LOCAL]
---------------------------------------PhysicalProject
-----------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
-------------------------------------------PhysicalOlapScan[date_dim]
 
diff --git a/regression-test/data/shape_check/tpcds_sf1000/hint/query54.out b/regression-test/data/shape_check/tpcds_sf1000/hint/query54.out
index 381fcc6712eeb5..ddc3b69f35f222 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/hint/query54.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/hint/query54.out
@@ -13,50 +13,58 @@ PhysicalResultSink
 --------------------PhysicalDistribute[DistributionSpecHash]
 ----------------------hashAgg[LOCAL]
 ------------------------PhysicalProject
---------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
 ----------------------------PhysicalProject
-------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
+------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF6 c_customer_sk->[ss_customer_sk]
 --------------------------------PhysicalProject
-----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
+----------------------------------PhysicalOlapScan[store_sales] apply RFs: RF6 RF7
+--------------------------------PhysicalProject
+----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF4 s_county->[ca_county];RF5 s_state->[ca_state]
 ------------------------------------PhysicalProject
---------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF6 c_customer_sk->[ss_customer_sk]
+--------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 c_current_addr_sk->[ca_address_sk]
 ----------------------------------------PhysicalProject
-------------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF6 RF7
+------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF3 RF4 RF5
 ----------------------------------------PhysicalProject
-------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF4 s_county->[ca_county];RF5 s_state->[ca_state]
---------------------------------------------PhysicalProject
-----------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 c_current_addr_sk->[ca_address_sk]
-------------------------------------------------PhysicalProject
---------------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF3 RF4 RF5
+------------------------------------------hashAgg[GLOBAL]
+--------------------------------------------PhysicalDistribute[DistributionSpecHash]
+----------------------------------------------hashAgg[LOCAL]
 ------------------------------------------------PhysicalProject
---------------------------------------------------hashAgg[GLOBAL]
-----------------------------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------------------------hashAgg[LOCAL]
+--------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 customer_sk->[c_customer_sk]
+----------------------------------------------------PhysicalProject
+------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF2
+----------------------------------------------------PhysicalProject
+------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
 --------------------------------------------------------PhysicalProject
-----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 customer_sk->[c_customer_sk]
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF2
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
+----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
+------------------------------------------------------------PhysicalUnion
+--------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
---------------------------------------------------------------------PhysicalUnion
-----------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-------------------------------------------------------------------------PhysicalProject
---------------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
-----------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-------------------------------------------------------------------------PhysicalProject
---------------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
---------------------------------------------------------------------PhysicalProject
-----------------------------------------------------------------------filter((item.i_category = 'Music') and (item.i_class = 'country'))
-------------------------------------------------------------------------PhysicalOlapScan[item]
+------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
+--------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
---------------------------------------------------------------------PhysicalOlapScan[date_dim]
---------------------------------------------PhysicalProject
-----------------------------------------------PhysicalOlapScan[store]
+------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
+------------------------------------------------------------PhysicalProject
+--------------------------------------------------------------filter((item.i_category = 'Music') and (item.i_class = 'country'))
+----------------------------------------------------------------PhysicalOlapScan[item]
+--------------------------------------------------------PhysicalProject
+----------------------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
+------------------------------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalProject
+--------------------------------------PhysicalOlapScan[store]
+----------------------------PhysicalProject
+------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------------PhysicalProject
+----------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
 ------------------------------------PhysicalProject
 --------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalAssertNumRows
+--------------------------------------PhysicalDistribute[DistributionSpecGather]
+----------------------------------------hashAgg[GLOBAL]
+------------------------------------------PhysicalDistribute[DistributionSpecHash]
+--------------------------------------------hashAgg[LOCAL]
+----------------------------------------------PhysicalProject
+------------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
+--------------------------------------------------PhysicalOlapScan[date_dim]
 --------------------------------PhysicalAssertNumRows
 ----------------------------------PhysicalDistribute[DistributionSpecGather]
 ------------------------------------hashAgg[GLOBAL]
@@ -65,14 +73,6 @@ PhysicalResultSink
 ------------------------------------------PhysicalProject
 --------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
 ----------------------------------------------PhysicalOlapScan[date_dim]
-----------------------------PhysicalAssertNumRows
-------------------------------PhysicalDistribute[DistributionSpecGather]
---------------------------------hashAgg[GLOBAL]
-----------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------hashAgg[LOCAL]
---------------------------------------PhysicalProject
-----------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
-------------------------------------------PhysicalOlapScan[date_dim]
 
 Hint log:
 Used: leading(customer { cs_or_ws_sales item date_dim } ) leading(store_sales { customer_address { my_customers store } } date_dim )
diff --git a/regression-test/data/shape_check/tpcds_sf1000/shape/query54.out b/regression-test/data/shape_check/tpcds_sf1000/shape/query54.out
index ba570488dc0f94..b7c2fe2adcd5b6 100644
--- a/regression-test/data/shape_check/tpcds_sf1000/shape/query54.out
+++ b/regression-test/data/shape_check/tpcds_sf1000/shape/query54.out
@@ -13,50 +13,58 @@ PhysicalResultSink
 --------------------PhysicalDistribute[DistributionSpecHash]
 ----------------------hashAgg[LOCAL]
 ------------------------PhysicalProject
---------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
 ----------------------------PhysicalProject
-------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
+------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF6 c_customer_sk->[ss_customer_sk]
 --------------------------------PhysicalProject
-----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
+----------------------------------PhysicalOlapScan[store_sales] apply RFs: RF6 RF7
+--------------------------------PhysicalProject
+----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF4 s_county->[ca_county];RF5 s_state->[ca_state]
 ------------------------------------PhysicalProject
---------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF6 c_customer_sk->[ss_customer_sk]
+--------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 c_current_addr_sk->[ca_address_sk]
 ----------------------------------------PhysicalProject
-------------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF6 RF7
+------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF3 RF4 RF5
 ----------------------------------------PhysicalProject
-------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF4 s_county->[ca_county];RF5 s_state->[ca_state]
---------------------------------------------PhysicalProject
-----------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF3 c_current_addr_sk->[ca_address_sk]
-------------------------------------------------PhysicalProject
---------------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF3 RF4 RF5
+------------------------------------------hashAgg[GLOBAL]
+--------------------------------------------PhysicalDistribute[DistributionSpecHash]
+----------------------------------------------hashAgg[LOCAL]
 ------------------------------------------------PhysicalProject
---------------------------------------------------hashAgg[GLOBAL]
-----------------------------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------------------------hashAgg[LOCAL]
+--------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 customer_sk->[c_customer_sk]
+----------------------------------------------------PhysicalProject
+------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF2
+----------------------------------------------------PhysicalProject
+------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
 --------------------------------------------------------PhysicalProject
-----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF2 customer_sk->[c_customer_sk]
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF2
-------------------------------------------------------------PhysicalProject
---------------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
+----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
+------------------------------------------------------------PhysicalUnion
+--------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF0 i_item_sk->[cs_item_sk,ws_item_sk]
---------------------------------------------------------------------PhysicalUnion
-----------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-------------------------------------------------------------------------PhysicalProject
---------------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
-----------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
-------------------------------------------------------------------------PhysicalProject
---------------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
---------------------------------------------------------------------PhysicalProject
-----------------------------------------------------------------------filter((item.i_category = 'Music') and (item.i_class = 'country'))
-------------------------------------------------------------------------PhysicalOlapScan[item]
+------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1
+--------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
---------------------------------------------------------------------PhysicalOlapScan[date_dim]
---------------------------------------------PhysicalProject
-----------------------------------------------PhysicalOlapScan[store]
+------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1
+------------------------------------------------------------PhysicalProject
+--------------------------------------------------------------filter((item.i_category = 'Music') and (item.i_class = 'country'))
+----------------------------------------------------------------PhysicalOlapScan[item]
+--------------------------------------------------------PhysicalProject
+----------------------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
+------------------------------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalProject
+--------------------------------------PhysicalOlapScan[store]
+----------------------------PhysicalProject
+------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------------PhysicalProject
+----------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
 ------------------------------------PhysicalProject
 --------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalAssertNumRows
+--------------------------------------PhysicalDistribute[DistributionSpecGather]
+----------------------------------------hashAgg[GLOBAL]
+------------------------------------------PhysicalDistribute[DistributionSpecHash]
+--------------------------------------------hashAgg[LOCAL]
+----------------------------------------------PhysicalProject
+------------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
+--------------------------------------------------PhysicalOlapScan[date_dim]
 --------------------------------PhysicalAssertNumRows
 ----------------------------------PhysicalDistribute[DistributionSpecGather]
 ------------------------------------hashAgg[GLOBAL]
@@ -65,12 +73,4 @@ PhysicalResultSink
 ------------------------------------------PhysicalProject
 --------------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
 ----------------------------------------------PhysicalOlapScan[date_dim]
-----------------------------PhysicalAssertNumRows
-------------------------------PhysicalDistribute[DistributionSpecGather]
---------------------------------hashAgg[GLOBAL]
-----------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------hashAgg[LOCAL]
---------------------------------------PhysicalProject
-----------------------------------------filter((date_dim.d_moy = 1) and (date_dim.d_year = 1999))
-------------------------------------------PhysicalOlapScan[date_dim]
 
diff --git a/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query54.out b/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query54.out
index 3bf22e267718b4..6d88dce347de99 100644
--- a/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query54.out
+++ b/regression-test/data/shape_check/tpcds_sf10t_orc/shape/query54.out
@@ -13,48 +13,56 @@ PhysicalResultSink
 --------------------PhysicalDistribute[DistributionSpecHash]
 ----------------------hashAgg[LOCAL]
 ------------------------PhysicalProject
---------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
 ----------------------------PhysicalProject
-------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
+------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF5 s_county->[ca_county];RF6 s_state->[ca_state]
 --------------------------------PhysicalProject
-----------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((store_sales.ss_sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF7 d_date_sk->[ss_sold_date_sk]
+----------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF4 ca_address_sk->[c_current_addr_sk]
 ------------------------------------PhysicalProject
---------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((customer_address.ca_county = store.s_county) and (customer_address.ca_state = store.s_state)) otherCondition=() build RFs:RF5 s_county->[ca_county];RF6 s_state->[ca_state]
+--------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF3 c_customer_sk->[ss_customer_sk]
 ----------------------------------------PhysicalProject
-------------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((my_customers.c_current_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF4 ca_address_sk->[c_current_addr_sk]
+------------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF3 RF7
+----------------------------------------PhysicalProject
+------------------------------------------hashAgg[GLOBAL]
 --------------------------------------------PhysicalProject
-----------------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((my_customers.c_customer_sk = store_sales.ss_customer_sk)) otherCondition=() build RFs:RF3 c_customer_sk->[ss_customer_sk]
-------------------------------------------------PhysicalProject
---------------------------------------------------PhysicalOlapScan[store_sales] apply RFs: RF3 RF7
+----------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
 ------------------------------------------------PhysicalProject
---------------------------------------------------hashAgg[GLOBAL]
+--------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF1 i_item_sk->[cs_item_sk,ws_item_sk]
 ----------------------------------------------------PhysicalProject
-------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.sold_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF2 d_date_sk->[cs_sold_date_sk,ws_sold_date_sk]
---------------------------------------------------------PhysicalProject
-----------------------------------------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs_or_ws_sales.item_sk = item.i_item_sk)) otherCondition=() build RFs:RF1 i_item_sk->[cs_item_sk,ws_item_sk]
+------------------------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF0 c_customer_sk->[cs_bill_customer_sk,ws_bill_customer_sk]
+--------------------------------------------------------PhysicalUnion
+----------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ------------------------------------------------------------PhysicalProject
---------------------------------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((customer.c_customer_sk = cs_or_ws_sales.customer_sk)) otherCondition=() build RFs:RF0 c_customer_sk->[cs_bill_customer_sk,ws_bill_customer_sk]
-----------------------------------------------------------------PhysicalUnion
-------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
---------------------------------------------------------------------PhysicalProject
-----------------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
-------------------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
---------------------------------------------------------------------PhysicalProject
-----------------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
-----------------------------------------------------------------PhysicalProject
-------------------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF4
+--------------------------------------------------------------PhysicalOlapScan[catalog_sales] apply RFs: RF0 RF1 RF2
+----------------------------------------------------------PhysicalDistribute[DistributionSpecExecutionAny]
 ------------------------------------------------------------PhysicalProject
---------------------------------------------------------------filter((item.i_category = 'Books') and (item.i_class = 'business'))
-----------------------------------------------------------------PhysicalOlapScan[item]
+--------------------------------------------------------------PhysicalOlapScan[web_sales] apply RFs: RF0 RF1 RF2
 --------------------------------------------------------PhysicalProject
-----------------------------------------------------------filter((date_dim.d_moy = 2) and (date_dim.d_year = 2000))
-------------------------------------------------------------PhysicalOlapScan[date_dim]
---------------------------------------------PhysicalProject
-----------------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF5 RF6
-----------------------------------------PhysicalProject
-------------------------------------------PhysicalOlapScan[store]
+----------------------------------------------------------PhysicalOlapScan[customer] apply RFs: RF4
+----------------------------------------------------PhysicalProject
+------------------------------------------------------filter((item.i_category = 'Books') and (item.i_class = 'business'))
+--------------------------------------------------------PhysicalOlapScan[item]
+------------------------------------------------PhysicalProject
+--------------------------------------------------filter((date_dim.d_moy = 2) and (date_dim.d_year = 2000))
+----------------------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalProject
+--------------------------------------PhysicalOlapScan[customer_address] apply RFs: RF5 RF6
+--------------------------------PhysicalProject
+----------------------------------PhysicalOlapScan[store]
+----------------------------PhysicalProject
+------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) <= d_month_seq+3)
+--------------------------------PhysicalProject
+----------------------------------NestedLoopJoin[INNER_JOIN](cast(d_month_seq as BIGINT) >= d_month_seq+1)
 ------------------------------------PhysicalProject
 --------------------------------------PhysicalOlapScan[date_dim]
+------------------------------------PhysicalAssertNumRows
+--------------------------------------PhysicalDistribute[DistributionSpecGather]
+----------------------------------------hashAgg[GLOBAL]
+------------------------------------------PhysicalDistribute[DistributionSpecHash]
+--------------------------------------------hashAgg[LOCAL]
+----------------------------------------------PhysicalProject
+------------------------------------------------filter((date_dim.d_moy = 2) and (date_dim.d_year = 2000))
+--------------------------------------------------PhysicalOlapScan[date_dim]
 --------------------------------PhysicalAssertNumRows
 ----------------------------------PhysicalDistribute[DistributionSpecGather]
 ------------------------------------hashAgg[GLOBAL]
@@ -63,12 +71,4 @@ PhysicalResultSink
 ------------------------------------------PhysicalProject
 --------------------------------------------filter((date_dim.d_moy = 2) and (date_dim.d_year = 2000))
 ----------------------------------------------PhysicalOlapScan[date_dim]
-----------------------------PhysicalAssertNumRows
-------------------------------PhysicalDistribute[DistributionSpecGather]
---------------------------------hashAgg[GLOBAL]
-----------------------------------PhysicalDistribute[DistributionSpecHash]
-------------------------------------hashAgg[LOCAL]
---------------------------------------PhysicalProject
-----------------------------------------filter((date_dim.d_moy = 2) and (date_dim.d_year = 2000))
-------------------------------------------PhysicalOlapScan[date_dim]