[opt](memo) reduce memo size

Author: morrySnow

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;
@@ -37,6 +39,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 
 import java.text.DecimalFormat;
 import java.util.ArrayList;
@@ -46,6 +49,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -60,6 +64,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,15 +248,28 @@ 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);
     }
 
     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 +374,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()

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;

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);

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);
+    }
 }

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(),

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalDistribute.java

@@ -35,6 +35,7 @@
 import org.json.JSONObject;
 
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 
 /**