[opt](memo) reuse GroupPlan in Group to reduce memory usage

Author: morrySnow

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/joinorder/hypergraph/HyperGraph.java

@@ -400,8 +400,9 @@ private Pair<BitSet, Long> buildForMv(Plan plan) {
                 Group group = ((GroupPlan) plan).getGroup();
                 GroupExpression groupExpression = group.getLogicalExpressions().get(0);
                 return buildForMv(groupExpression.getPlan()
-                        .withChildren(
-                                groupExpression.children().stream().map(GroupPlan::new).collect(Collectors.toList())));
+                        .withChildren(groupExpression.children().stream()
+                                .map(Group::getGroupPlan)
+                                .collect(Collectors.toList())));
             }
             // process Project
             if (isValidProject(plan)) {

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/joinorder/hypergraph/receiver/PlanReceiver.java

@@ -102,8 +102,8 @@ public boolean emitCsgCmp(long left, long right, List<JoinEdge> edges) {
         }
 
         Memo memo = jobContext.getCascadesContext().getMemo();
-        GroupPlan leftPlan = new GroupPlan(planTable.get(left));
-        GroupPlan rightPlan = new GroupPlan(planTable.get(right));
+        GroupPlan leftPlan = planTable.get(left).getGroupPlan();
+        GroupPlan rightPlan = planTable.get(right).getGroupPlan();
 
         // First, we implement all possible physical plans
         // In this step, we don't generate logical expression because they are useless in DPhyp.
@@ -205,7 +205,7 @@ private LogicalPlan proposeJoin(JoinType joinType, Plan left, Plan right, List<E
     public void addGroup(long bitmap, Group group) {
         Preconditions.checkArgument(LongBitmap.getCardinality(bitmap) == 1);
         usdEdges.put(bitmap, new BitSet());
-        Plan plan = proposeProject(new GroupPlan(group), new ArrayList<>(), bitmap, bitmap);
+        Plan plan = proposeProject(group.getGroupPlan(), new ArrayList<>(), bitmap, bitmap);
         if (!(plan instanceof GroupPlan)) {
             CopyInResult copyInResult = jobContext.getCascadesContext().getMemo().copyIn(plan, null, false, planTable);
             group = copyInResult.correspondingExpression.getOwnerGroup();

fe/fe-core/src/main/java/org/apache/doris/nereids/memo/Group.java

@@ -22,6 +22,7 @@
 import org.apache.doris.nereids.properties.LogicalProperties;
 import org.apache.doris.nereids.properties.PhysicalProperties;
 import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
 import org.apache.doris.nereids.trees.plans.JoinType;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
@@ -60,6 +61,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 GroupPlan groupPlan;
     private boolean isStatsReliable = true;
     private LogicalProperties logicalProperties;
 
@@ -89,6 +91,7 @@ public Group(GroupId groupId, GroupExpression groupExpression, LogicalProperties
         this.groupId = groupId;
         addGroupExpression(groupExpression);
         this.logicalProperties = logicalProperties;
+        this.groupPlan = new GroupPlan(this);
     }
 
     /**
@@ -99,6 +102,7 @@ public Group(GroupId groupId, GroupExpression groupExpression, LogicalProperties
     public Group(GroupId groupId, LogicalProperties logicalProperties) {
         this.groupId = groupId;
         this.logicalProperties = logicalProperties;
+        this.groupPlan = new GroupPlan(this);
     }
 
     public GroupId getGroupId() {
@@ -173,6 +177,10 @@ public List<GroupExpression> getPhysicalExpressions() {
         return physicalExpressions;
     }
 
+    public GroupPlan getGroupPlan() {
+        return groupPlan;
+    }
+
     /**
      * Remove groupExpression from this group.
      *

fe/fe-core/src/main/java/org/apache/doris/nereids/memo/Memo.java

@@ -746,7 +746,7 @@ private Plan replaceChildrenToGroupPlan(Plan plan, List<Group> childrenGroups) {
 
         ImmutableList.Builder<Plan> groupPlanChildren = ImmutableList.builderWithExpectedSize(childrenGroups.size());
         for (Group childrenGroup : childrenGroups) {
-            groupPlanChildren.add(new GroupPlan(childrenGroup));
+            groupPlanChildren.add(childrenGroup.getGroupPlan());
         }
         return plan.withChildren(groupPlanChildren.build());
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/pattern/GroupExpressionMatching.java

@@ -20,7 +20,6 @@
 import org.apache.doris.nereids.memo.Group;
 import org.apache.doris.nereids.memo.GroupExpression;
 import org.apache.doris.nereids.properties.LogicalProperties;
-import org.apache.doris.nereids.trees.plans.GroupPlan;
 import org.apache.doris.nereids.trees.plans.Plan;
 
 import com.google.common.collect.ImmutableList;
@@ -111,7 +110,7 @@ public GroupExpressionIterator(Pattern<Plan> pattern, GroupExpression groupExpre
 
                     if (childrenPlan.isEmpty()) {
                         if (pattern instanceof SubTreePattern) {
-                            childrenPlan = ImmutableList.of(new GroupPlan(childGroup));
+                            childrenPlan = ImmutableList.of(childGroup.getGroupPlan());
                         } else {
                             // current pattern is match but children patterns not match
                             return;

fe/fe-core/src/main/java/org/apache/doris/nereids/pattern/GroupMatching.java

@@ -35,7 +35,7 @@ public class GroupMatching {
     public static List<Plan> getAllMatchingPlans(Pattern pattern, Group group) {
         List<Plan> matchingPlans = new ArrayList<>();
         if (pattern.isGroup() || pattern.isMultiGroup()) {
-            GroupPlan groupPlan = new GroupPlan(group);
+            GroupPlan groupPlan = group.getGroupPlan();
             if (((Pattern<Plan>) pattern).matchPredicates(groupPlan)) {
                 matchingPlans.add(groupPlan);
             }

fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpec.java

@@ -44,7 +44,7 @@ public GroupExpression addEnforcer(Group child) {
         // If we don't set LogicalProperties explicitly, node will compute a applicable LogicalProperties for itself.
         PhysicalDistribute<GroupPlan> distribution = new PhysicalDistribute<>(
                 this,
-                new GroupPlan(child));
+                child.getGroupPlan());
         return new GroupExpression(distribution, Lists.newArrayList(child));
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/properties/OrderSpec.java

@@ -19,7 +19,6 @@
 
 import org.apache.doris.nereids.memo.Group;
 import org.apache.doris.nereids.memo.GroupExpression;
-import org.apache.doris.nereids.trees.plans.GroupPlan;
 import org.apache.doris.nereids.trees.plans.SortPhase;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalQuickSort;
 
@@ -67,7 +66,7 @@ public boolean satisfy(OrderSpec other) {
     public GroupExpression addLocalQuickSortEnforcer(Group child) {
         return new GroupExpression(
                 new PhysicalQuickSort<>(orderKeys, SortPhase.LOCAL_SORT, child.getLogicalProperties(),
-                        new GroupPlan(child)),
+                        child.getGroupPlan()),
                 Lists.newArrayList(child)
         );
     }
@@ -78,7 +77,7 @@ public GroupExpression addLocalQuickSortEnforcer(Group child) {
     public GroupExpression addGlobalQuickSortEnforcer(Group child) {
         return new GroupExpression(
                 new PhysicalQuickSort<>(orderKeys, SortPhase.MERGE_SORT, child.getLogicalProperties(),
-                        new GroupPlan(child)),
+                        child.getGroupPlan()),
                 Lists.newArrayList(child)
         );
     }

fe/fe-core/src/test/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulatorTest.java

@@ -100,6 +100,8 @@ public void testMustShuffleProject(Class<? extends Plan> childClazz,
             List<GroupExpression> children;
             Group childGroup = Mockito.mock(Group.class);
             Mockito.when(childGroup.getLogicalProperties()).thenReturn(Mockito.mock(LogicalProperties.class));
+            GroupPlan childGroupPlan = new GroupPlan(childGroup);
+            Mockito.when(childGroup.getGroupPlan()).thenReturn(childGroupPlan);
             GroupExpression child = Mockito.mock(GroupExpression.class);
             Mockito.when(child.getOutputProperties(Mockito.any())).thenReturn(PhysicalProperties.MUST_SHUFFLE);
             Mockito.when(child.getOwnerGroup()).thenReturn(childGroup);
@@ -159,6 +161,8 @@ private void testMustShuffleFilter(Class<? extends Plan> childClazz) {
             List<GroupExpression> children;
             Group childGroup = Mockito.mock(Group.class);
             Mockito.when(childGroup.getLogicalProperties()).thenReturn(Mockito.mock(LogicalProperties.class));
+            GroupPlan childGroupPlan = new GroupPlan(childGroup);
+            Mockito.when(childGroup.getGroupPlan()).thenReturn(childGroupPlan);
             GroupExpression child = Mockito.mock(GroupExpression.class);
             Mockito.when(child.getOutputProperties(Mockito.any())).thenReturn(PhysicalProperties.MUST_SHUFFLE);
             Mockito.when(child.getOwnerGroup()).thenReturn(childGroup);