diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
index 6980765049f37b..c37573c4c2227b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
@@ -152,7 +152,7 @@ private CascadesContext(Optional<CascadesContext> parent, Optional<CTEId> curren
         this.ruleSet = new RuleSet();
         this.jobPool = new JobStack();
         this.jobScheduler = new SimpleJobScheduler();
-        this.currentJobContext = new JobContext(this, requireProperties, Double.MAX_VALUE);
+        this.currentJobContext = new JobContext(this, requireProperties);
         this.subqueryExprIsAnalyzed = new HashMap<>();
         IdGenerator<RuntimeFilterId> runtimeFilterIdGen = RuntimeFilterId.createGenerator();
         this.runtimeFilterContext = new RuntimeFilterContext(getConnectContext().getSessionVariable(),
@@ -341,7 +341,7 @@ public void setCurrentJobContext(JobContext currentJobContext) {
     }
 
     public CascadesContext setJobContext(PhysicalProperties physicalProperties) {
-        this.currentJobContext = new JobContext(this, physicalProperties, Double.MAX_VALUE);
+        this.currentJobContext = new JobContext(this, physicalProperties);
         return this;
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/PlanContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/PlanContext.java
index 3ab95423e24a5e..09285a638dee35 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/PlanContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/PlanContext.java
@@ -18,12 +18,13 @@
 package org.apache.doris.nereids;
 
 import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.DistributionSpecReplicated;
+import org.apache.doris.nereids.properties.PhysicalProperties;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.statistics.Statistics;
 
-import java.util.ArrayList;
 import java.util.List;
 
 /**
@@ -34,23 +35,21 @@
  */
 public class PlanContext {
     private final ConnectContext connectContext;
-    private final List<Statistics> childrenStats;
-    private final Statistics planStats;
-    private final int arity;
-    private boolean isBroadcastJoin = false;
-    private final boolean isStatsReliable;
+    private final GroupExpression groupExpression;
+    private final boolean isBroadcastJoin;
 
     /**
      * Constructor for PlanContext.
      */
-    public PlanContext(ConnectContext connectContext, GroupExpression groupExpression) {
+    public PlanContext(ConnectContext connectContext, GroupExpression groupExpression,
+            List<PhysicalProperties> childrenProperties) {
         this.connectContext = connectContext;
-        this.arity = groupExpression.arity();
-        this.planStats = groupExpression.getOwnerGroup().getStatistics();
-        this.isStatsReliable = groupExpression.getOwnerGroup().isStatsReliable();
-        this.childrenStats = new ArrayList<>(groupExpression.arity());
-        for (int i = 0; i < groupExpression.arity(); i++) {
-            childrenStats.add(groupExpression.childStatistics(i));
+        this.groupExpression = groupExpression;
+        if (childrenProperties.size() >= 2
+                && childrenProperties.get(1).getDistributionSpec() instanceof DistributionSpecReplicated) {
+            isBroadcastJoin = true;
+        } else {
+            isBroadcastJoin = false;
         }
     }
 
@@ -58,35 +57,27 @@ public SessionVariable getSessionVariable() {
         return connectContext.getSessionVariable();
     }
 
-    public void setBroadcastJoin() {
-        isBroadcastJoin = true;
-    }
-
     public boolean isBroadcastJoin() {
         return isBroadcastJoin;
     }
 
     public int arity() {
-        return arity;
+        return groupExpression.arity();
     }
 
     public Statistics getStatisticsWithCheck() {
-        return planStats;
+        return groupExpression.getOwnerGroup().getStatistics();
     }
 
     public boolean isStatsReliable() {
-        return isStatsReliable;
+        return groupExpression.getOwnerGroup().isStatsReliable();
     }
 
     /**
      * Get child statistics.
      */
     public Statistics getChildStatistics(int index) {
-        return childrenStats.get(index);
-    }
-
-    public List<Statistics> getChildrenStatistics() {
-        return childrenStats;
+        return groupExpression.childStatistics(index);
     }
 
     public StatementContext getStatementContext() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostCalculator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostCalculator.java
index 7661e1aec512b1..8ec69391691d5e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostCalculator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostCalculator.java
@@ -19,7 +19,6 @@
 
 import org.apache.doris.nereids.PlanContext;
 import org.apache.doris.nereids.memo.GroupExpression;
-import org.apache.doris.nereids.properties.DistributionSpecReplicated;
 import org.apache.doris.nereids.properties.PhysicalProperties;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.qe.ConnectContext;
@@ -38,12 +37,7 @@ public class CostCalculator {
      */
     public static Cost calculateCost(ConnectContext connectContext, GroupExpression groupExpression,
             List<PhysicalProperties> childrenProperties) {
-        PlanContext planContext = new PlanContext(connectContext, groupExpression);
-        if (childrenProperties.size() >= 2
-                && childrenProperties.get(1).getDistributionSpec() instanceof DistributionSpecReplicated) {
-            planContext.setBroadcastJoin();
-        }
-
+        PlanContext planContext = new PlanContext(connectContext, groupExpression, childrenProperties);
         CostModel costModelV1 = new CostModel(connectContext);
         return groupExpression.getPlan().accept(costModelV1, planContext);
     }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/JobContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/JobContext.java
index 88ca252e537fe8..24c3beadf6b5db 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/JobContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/JobContext.java
@@ -37,7 +37,6 @@ public class JobContext {
     // use for optimizer
     protected final ScheduleContext scheduleContext;
     protected final PhysicalProperties requiredProperties;
-    protected double costUpperBound;
 
     // use for rewriter
     protected boolean rewritten = false;
@@ -46,10 +45,9 @@ public class JobContext {
     // user for trace
     protected Map<RuleType, Integer> ruleInvokeTimes = Maps.newLinkedHashMap();
 
-    public JobContext(ScheduleContext scheduleContext, PhysicalProperties requiredProperties, double costUpperBound) {
+    public JobContext(ScheduleContext scheduleContext, PhysicalProperties requiredProperties) {
         this.scheduleContext = scheduleContext;
         this.requiredProperties = requiredProperties;
-        this.costUpperBound = costUpperBound;
     }
 
     public ScheduleContext getScheduleContext() {
@@ -64,14 +62,6 @@ public PhysicalProperties getRequiredProperties() {
         return requiredProperties;
     }
 
-    public double getCostUpperBound() {
-        return costUpperBound;
-    }
-
-    public void setCostUpperBound(double costUpperBound) {
-        this.costUpperBound = costUpperBound;
-    }
-
     public boolean isRewritten() {
         return rewritten;
     }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/cascades/CostAndEnforcerJob.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/cascades/CostAndEnforcerJob.java
index 76a659870d09fd..31206e85e38ad5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/cascades/CostAndEnforcerJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/cascades/CostAndEnforcerJob.java
@@ -174,9 +174,7 @@ public void execute() {
                     // Meaning that optimize recursively by derive job.
                     prevChildIndex = curChildIndex;
                     pushJob(clone());
-                    double newCostUpperBound = context.getCostUpperBound() - curTotalCost.getValue();
-                    JobContext jobContext = new JobContext(context.getCascadesContext(),
-                            requestChildProperty, newCostUpperBound);
+                    JobContext jobContext = new JobContext(context.getCascadesContext(), requestChildProperty);
                     pushJob(new OptimizeGroupJob(childGroup, jobContext));
                     return;
                 }
@@ -215,13 +213,7 @@ public void execute() {
             // This mean that we successfully optimize all child groups.
             // if break when running the loop above, the condition must be false.
             if (curChildIndex == groupExpression.arity()) {
-                if (!calculateEnforce(requestChildrenProperties, outputChildrenProperties)) {
-                    clear();
-                    continue; // if error exists, return
-                }
-                if (curTotalCost.getValue() < context.getCostUpperBound()) {
-                    context.setCostUpperBound(curTotalCost.getValue());
-                }
+                calculateEnforce(requestChildrenProperties, outputChildrenProperties);
             }
             clear();
         }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/joinorder/hypergraph/receiver/PlanReceiver.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/joinorder/hypergraph/receiver/PlanReceiver.java
index b4629ef9de6fd7..9047a1763220f4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/joinorder/hypergraph/receiver/PlanReceiver.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/joinorder/hypergraph/receiver/PlanReceiver.java
@@ -188,7 +188,7 @@ private void processMissedEdges(long left, long right, List<JoinEdge> edges) {
 
     private void proposeAllDistributedPlans(GroupExpression groupExpression) {
         jobContext.getCascadesContext().pushJob(new OptimizeGroupExpressionJob(groupExpression,
-                new JobContext(jobContext.getCascadesContext(), PhysicalProperties.ANY, Double.MAX_VALUE)));
+                new JobContext(jobContext.getCascadesContext(), PhysicalProperties.ANY)));
         if (!groupExpression.isStatDerived()) {
             jobContext.getCascadesContext().pushJob(new DeriveStatsJob(groupExpression,
                     jobContext.getCascadesContext().getCurrentJobContext()));
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/memo/Group.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/memo/Group.java
index b3d45e6c80a50a..67bdaf56050608 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/memo/Group.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/memo/Group.java
@@ -19,6 +19,7 @@
 
 import org.apache.doris.common.Pair;
 import org.apache.doris.nereids.cost.Cost;
+import org.apache.doris.nereids.properties.DistributionSpec;
 import org.apache.doris.nereids.properties.LogicalProperties;
 import org.apache.doris.nereids.properties.PhysicalProperties;
 import org.apache.doris.nereids.trees.expressions.literal.Literal;
@@ -27,6 +28,7 @@
 import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
 import org.apache.doris.nereids.util.TreeStringUtils;
 import org.apache.doris.nereids.util.Utils;
@@ -60,6 +62,7 @@ public class Group {
     private final List<GroupExpression> logicalExpressions = Lists.newArrayList();
     private final List<GroupExpression> physicalExpressions = Lists.newArrayList();
     private final Map<GroupExpression, GroupExpression> enforcers = Maps.newHashMap();
+    private final Map<DistributionSpec, GroupExpression> enforcerSpecs = Maps.newHashMap();
     private boolean isStatsReliable = true;
     private LogicalProperties logicalProperties;
 
@@ -243,8 +246,17 @@ public GroupExpression getBestPlan(PhysicalProperties properties) {
         return null;
     }
 
+    /**
+     * add a new enforcer to this group.
+     */
     public void addEnforcer(GroupExpression enforcer) {
         enforcer.setOwnerGroup(this);
+        if (enforcer.getPlan() instanceof PhysicalDistribute) {
+            DistributionSpec distributionSpec = ((PhysicalDistribute) enforcer.getPlan()).getDistributionSpec();
+            if (null != enforcerSpecs.put(distributionSpec, enforcer)) {
+                return;
+            }
+        }
         enforcers.put(enforcer, enforcer);
     }
 
@@ -252,6 +264,10 @@ public Map<GroupExpression, GroupExpression> getEnforcers() {
         return enforcers;
     }
 
+    public Map<DistributionSpec, GroupExpression> getEnforcerSpecs() {
+        return enforcerSpecs;
+    }
+
     /**
      * Set or update lowestCostPlans: properties --> Pair.of(cost, expression)
      */
@@ -356,6 +372,7 @@ public void mergeTo(Group target) {
         // TODO: dedup?
         enforcers.forEach((k, v) -> target.addEnforcer(k));
         enforcers.clear();
+        enforcerSpecs.clear();
 
         // move LogicalExpression PhysicalExpression Ownership
         Map<GroupExpression, GroupExpression> logicalSet = target.getLogicalExpressions().stream()
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/memo/GroupExpression.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/memo/GroupExpression.java
index d6a9e29dd9ef46..96b4180db15c66 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/memo/GroupExpression.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/memo/GroupExpression.java
@@ -54,7 +54,7 @@ public class GroupExpression {
     private static final EventProducer COST_STATE_TRACER = new EventProducer(CostStateUpdateEvent.class,
             EventChannel.getDefaultChannel().addConsumers(new LogConsumer(CostStateUpdateEvent.class,
                     EventChannel.LOG)));
-    private Cost cost;
+    private Cost cost = null;
     private Group ownerGroup;
     private final List<Group> children;
     private final Plan plan;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
index 5a1c232bc38742..0632c2ac54572b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
@@ -73,6 +73,7 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -98,7 +99,8 @@ public ChildOutputPropertyDeriver(List<PhysicalProperties> childrenOutputPropert
     }
 
     public PhysicalProperties getOutputProperties(ConnectContext connectContext, GroupExpression groupExpression) {
-        return groupExpression.getPlan().accept(this, new PlanContext(connectContext, groupExpression));
+        return groupExpression.getPlan().accept(this,
+                new PlanContext(connectContext, groupExpression, Collections.emptyList()));
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
index 41a1bfc37a97e1..d95cdb5db961d1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
@@ -867,6 +867,9 @@ private void updateChildEnforceAndCost(GroupExpression child, PhysicalProperties
             currentCost = newChildAndCost.first;
         }
 
+        if (child.getOwnerGroup().getEnforcerSpecs().containsKey(target)) {
+            return;
+        }
         PhysicalProperties newOutputProperty = new PhysicalProperties(target);
         GroupExpression enforcer = target.addEnforcer(child.getOwnerGroup());
         child.getOwnerGroup().addEnforcer(enforcer);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHiveTableSinkHashPartitioned.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHiveTableSinkHashPartitioned.java
index 4275bb53dd85ca..87a5eafc46f5e4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHiveTableSinkHashPartitioned.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHiveTableSinkHashPartitioned.java
@@ -20,6 +20,7 @@
 import org.apache.doris.nereids.trees.expressions.ExprId;
 
 import java.util.List;
+import java.util.Objects;
 
 /**
  * use for shuffle data by partition keys before sink.
@@ -44,4 +45,21 @@ public void setOutputColExprIds(List<ExprId> outputColExprIds) {
     public boolean satisfy(DistributionSpec other) {
         return other instanceof DistributionSpecHiveTableSinkHashPartitioned;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        if (!super.equals(o)) {
+            return false;
+        }
+        DistributionSpecHiveTableSinkHashPartitioned that = (DistributionSpecHiveTableSinkHashPartitioned) o;
+        return Objects.equals(outputColExprIds, that.outputColExprIds);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), outputColExprIds);
+    }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/EnforceMissingPropertiesHelper.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/EnforceMissingPropertiesHelper.java
index 6dc6fe565a21a6..81ce94e5f2f3fa 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/EnforceMissingPropertiesHelper.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/EnforceMissingPropertiesHelper.java
@@ -123,7 +123,11 @@ private PhysicalProperties enforceDistribution(PhysicalProperties oldOutputPrope
         }
 
         PhysicalProperties newOutputProperty = new PhysicalProperties(outputDistributionSpec);
-        GroupExpression enforcer = outputDistributionSpec.addEnforcer(groupExpression.getOwnerGroup());
+        GroupExpression enforcer = groupExpression.getOwnerGroup().getEnforcerSpecs().get(outputDistributionSpec);
+
+        if (enforcer == null) {
+            enforcer = outputDistributionSpec.addEnforcer(groupExpression.getOwnerGroup());
+        }
         addEnforcerUpdateCost(enforcer, oldOutputProperty, newOutputProperty);
         return newOutputProperty;
     }
@@ -160,10 +164,13 @@ private void addEnforcerUpdateCost(GroupExpression enforcer,
                 oldOutputProperty, newOutputProperty);
         ENFORCER_TRACER.log(EnforcerEvent.of(groupExpression, ((PhysicalPlan) enforcer.getPlan()),
                 oldOutputProperty, newOutputProperty));
-        enforcer.setEstOutputRowCount(enforcer.getOwnerGroup().getStatistics().getRowCount());
-        Cost enforcerCost = CostCalculator.calculateCost(connectContext, enforcer,
-                Lists.newArrayList(oldOutputProperty));
-        enforcer.setCost(enforcerCost);
+        Cost enforcerCost = enforcer.getCost();
+        if (enforcerCost == null) {
+            enforcer.setEstOutputRowCount(enforcer.getOwnerGroup().getStatistics().getRowCount());
+            enforcerCost = CostCalculator.calculateCost(connectContext, enforcer,
+                    Lists.newArrayList(oldOutputProperty));
+            enforcer.setCost(enforcerCost);
+        }
         curTotalCost = CostCalculator.addChildCost(
                 connectContext,
                 enforcer.getPlan(),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
index 7c31e11531f47b..3584b58dae426c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
@@ -73,6 +73,7 @@
 import com.google.common.collect.Sets;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -113,7 +114,8 @@ public RequestPropertyDeriver(ConnectContext connectContext, PhysicalProperties
      */
     public List<List<PhysicalProperties>> getRequestChildrenPropertyList(GroupExpression groupExpression) {
         requestPropertyToChildren = Lists.newArrayList();
-        groupExpression.getPlan().accept(this, new PlanContext(connectContext, groupExpression));
+        groupExpression.getPlan().accept(this,
+                new PlanContext(connectContext, groupExpression, Collections.emptyList()));
         return requestPropertyToChildren;
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/join/OuterJoinLAsscomProject.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/join/OuterJoinLAsscomProject.java
index ac9787bed48dc1..24fa772297080a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/join/OuterJoinLAsscomProject.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/join/OuterJoinLAsscomProject.java
@@ -23,7 +23,6 @@
 import org.apache.doris.nereids.rules.exploration.CBOUtils;
 import org.apache.doris.nereids.rules.exploration.OneExplorationRuleFactory;
 import org.apache.doris.nereids.trees.expressions.ExprId;
-import org.apache.doris.nereids.trees.expressions.SlotReference;
 import org.apache.doris.nereids.trees.plans.GroupPlan;
 import org.apache.doris.nereids.trees.plans.JoinType;
 import org.apache.doris.nereids.trees.plans.Plan;
@@ -36,7 +35,6 @@
 
 import java.util.HashSet;
 import java.util.Set;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 /**
@@ -116,11 +114,7 @@ public static boolean checkCondition(LogicalJoin<? extends Plan, GroupPlan> topJ
                         topJoin.getHashJoinConjuncts().stream(),
                         topJoin.getOtherJoinConjuncts().stream())
                 .allMatch(expr -> {
-                    Set<ExprId> usedExprIdSet = expr.<SlotReference>collect(SlotReference.class::isInstance)
-                            .stream()
-                            .map(SlotReference::getExprId)
-                            .collect(Collectors.toSet());
-                    return !Utils.isIntersecting(usedExprIdSet, bOutputExprIdSet);
+                    return !Utils.isIntersecting(expr.getInputSlotExprIds(), bOutputExprIdSet);
                 });
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Expression.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Expression.java
index d6b091b8718d0d..3f5dbca246828a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Expression.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Expression.java
@@ -71,6 +71,16 @@ public abstract class Expression extends AbstractTreeNode<Expression> implements
     private final boolean hasUnbound;
     private final Supplier<Set<Slot>> inputSlots = LazyCompute.of(
             () -> collect(e -> e instanceof Slot && !(e instanceof ArrayItemSlot)));
+    private final Supplier<Set<ExprId>> inputExprIds = LazyCompute.of(
+            () -> {
+                Set<Slot> inputSlots = getInputSlots();
+                Builder<ExprId> exprIds = ImmutableSet.builderWithExpectedSize(inputSlots.size());
+                for (Slot inputSlot : inputSlots) {
+                    exprIds.add(inputSlot.getExprId());
+                }
+                return exprIds.build();
+            }
+    );
     private final int fastChildrenHashCode;
     private final Supplier<String> toSqlCache = LazyCompute.of(this::computeToSql);
     private final Supplier<Integer> hashCodeCache = LazyCompute.of(this::computeHashCode);
@@ -394,12 +404,7 @@ public final Set<Slot> getInputSlots() {
      * Note that the input slots of subquery's inner plan is not included.
      */
     public final Set<ExprId> getInputSlotExprIds() {
-        Set<Slot> inputSlots = getInputSlots();
-        Builder<ExprId> exprIds = ImmutableSet.builderWithExpectedSize(inputSlots.size());
-        for (Slot inputSlot : inputSlots) {
-            exprIds.add(inputSlot.getExprId());
-        }
-        return exprIds.build();
+        return inputExprIds.get();
     }
 
     public boolean isLiteral() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalHashJoin.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalHashJoin.java
index ca84e1afc2fd1a..d90a74f4199aad 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalHashJoin.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalHashJoin.java
@@ -131,13 +131,11 @@ public Pair<List<ExprId>, List<ExprId>> getHashConjunctsExprIds() {
         //  mark join with non-empty hash join conjuncts allow shuffle join by hash join conjuncts
         Preconditions.checkState(!(isMarkJoin() && hashJoinConjuncts.isEmpty()),
                 "shouldn't call mark join's getHashConjunctsExprIds method for standalone mark join");
-        int size = hashJoinConjuncts.size();
-
-        List<ExprId> exprIds1 = new ArrayList<>(size);
-        List<ExprId> exprIds2 = new ArrayList<>(size);
 
         Set<ExprId> leftExprIds = left().getOutputExprIdSet();
         Set<ExprId> rightExprIds = right().getOutputExprIdSet();
+        List<ExprId> exprIds1 = new ArrayList<>(leftExprIds.size());
+        List<ExprId> exprIds2 = new ArrayList<>(rightExprIds.size());
 
         for (Expression expr : hashJoinConjuncts) {
             for (ExprId exprId : expr.getInputSlotExprIds()) {
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/jobs/cascades/DeriveStatsJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/jobs/cascades/DeriveStatsJobTest.java
index 181ab5ba08cd02..aa5a21adcdc259 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/jobs/cascades/DeriveStatsJobTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/jobs/cascades/DeriveStatsJobTest.java
@@ -68,7 +68,7 @@ public void testExecute() throws Exception {
         LogicalAggregate agg = constructAgg(olapScan);
         CascadesContext cascadesContext = MemoTestUtils.createCascadesContext(agg);
         new DeriveStatsJob(cascadesContext.getMemo().getRoot().getLogicalExpression(),
-                new JobContext(cascadesContext, null, Double.MAX_VALUE)).execute();
+                new JobContext(cascadesContext, null)).execute();
         while (!cascadesContext.getJobPool().isEmpty()) {
             cascadesContext.getJobPool().pop().execute();
         }
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/ConstantPropagationTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/ConstantPropagationTest.java
index a1ef8f8d7b414e..285e8313e6a48c 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/ConstantPropagationTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/ConstantPropagationTest.java
@@ -88,7 +88,7 @@ class ConstantPropagationTest {
     ConstantPropagationTest() {
         cascadesContext = MemoTestUtils.createCascadesContext(
                 new UnboundRelation(new RelationId(1), ImmutableList.of("tbl")));
-        jobContext = new JobContext(cascadesContext, null, Double.MAX_VALUE);
+        jobContext = new JobContext(cascadesContext, null);
 
         student = new LogicalOlapScan(PlanConstructor.getNextRelationId(), PlanConstructor.student, ImmutableList.of(""));
         studentId = (SlotReference) student.getOutput().get(0);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/util/PlanChecker.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/util/PlanChecker.java
index c23ff24c868d89..5084ef09823f5b 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/util/PlanChecker.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/util/PlanChecker.java
@@ -700,8 +700,7 @@ public PlanChecker checkMemo(Consumer<Memo> memoChecker) {
     public PlanChecker setMaxInvokeTimesPerRule(int maxInvokeTime) {
         JobContext originJobContext = cascadesContext.getCurrentJobContext();
         cascadesContext.setCurrentJobContext(
-                new JobContext(cascadesContext,
-                        originJobContext.getRequiredProperties(), originJobContext.getCostUpperBound()) {
+                new JobContext(cascadesContext, originJobContext.getRequiredProperties()) {
                     @Override
                     public void onInvokeRule(RuleType ruleType) {
                         // add invoke times