[fix](nereids)LogicalPlanDeepCopier will lost some info when coping logical relation

fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundOneRowRelation.java

@@ -86,6 +86,11 @@ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpr
         return new UnboundOneRowRelation(relationId, projects, groupExpression, logicalProperties);
     }
 
+    @Override
+    public UnboundOneRowRelation withRelationId(RelationId relationId) {
+        throw new UnboundException("should not call UnboundOneRowRelation's withRelationId method");
+    }
+
     @Override
     public List<Slot> computeOutput() {
         throw new UnboundException("output");

fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundRelation.java

@@ -143,6 +143,11 @@ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpr
                 isTempPart, tabletIds, hints, tableSample, indexName, null, indexInSqlString);
     }
 
+    @Override
+    public UnboundRelation withRelationId(RelationId relationId) {
+        throw new UnboundException("should not call UnboundRelation's withRelationId method");
+    }
+
     @Override
     public List<Slot> computeOutput() {
         throw new UnboundException("output");

fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTVFRelation.java

@@ -101,6 +101,11 @@ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpr
         return new UnboundTVFRelation(relationId, functionName, properties, groupExpression, logicalProperties);
     }
 
+    @Override
+    public UnboundTVFRelation withRelationId(RelationId relationId) {
+        throw new UnboundException("should not call UnboundTVFRelation's withRelationId method");
+    }
+
     @Override
     public String toString() {
         return Utils.toSqlString("UnboundTVFRelation",

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/copier/LogicalPlanDeepCopier.java

@@ -55,11 +55,10 @@
 import org.apache.doris.nereids.trees.plans.logical.LogicalPartitionTopN;
 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.logical.LogicalRelation;
 import org.apache.doris.nereids.trees.plans.logical.LogicalRepeat;
-import org.apache.doris.nereids.trees.plans.logical.LogicalSchemaScan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalSink;
 import org.apache.doris.nereids.trees.plans.logical.LogicalSort;
-import org.apache.doris.nereids.trees.plans.logical.LogicalTVFRelation;
 import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
 import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
 import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
@@ -85,20 +84,44 @@ public LogicalPlan deepCopy(LogicalPlan plan, DeepCopierContext context) {
         return (LogicalPlan) plan.accept(this, context);
     }
 
+    @Override
+    public Plan visitLogicalRelation(LogicalRelation logicalRelation, DeepCopierContext context) {
+        if (context.getRelationReplaceMap().containsKey(logicalRelation.getRelationId())) {
+            return context.getRelationReplaceMap().get(logicalRelation.getRelationId());
+        }
+        LogicalRelation newRelation =
+                logicalRelation.withRelationId(StatementScopeIdGenerator.newRelationId());
+        updateReplaceMapWithOutput(logicalRelation, newRelation, context.exprIdReplaceMap);
+        context.putRelation(logicalRelation.getRelationId(), newRelation);
+        return newRelation;
+    }
+
     @Override
     public Plan visitLogicalEmptyRelation(LogicalEmptyRelation emptyRelation, DeepCopierContext context) {
+        if (context.getRelationReplaceMap().containsKey(emptyRelation.getRelationId())) {
+            return context.getRelationReplaceMap().get(emptyRelation.getRelationId());
+        }
         List<NamedExpression> newProjects = emptyRelation.getProjects().stream()
                 .map(p -> (NamedExpression) ExpressionDeepCopier.INSTANCE.deepCopy(p, context))
                 .collect(ImmutableList.toImmutableList());
-        return new LogicalEmptyRelation(StatementScopeIdGenerator.newRelationId(), newProjects);
+        LogicalEmptyRelation newEmptyRelation =
+                new LogicalEmptyRelation(StatementScopeIdGenerator.newRelationId(), newProjects);
+        context.putRelation(emptyRelation.getRelationId(), newEmptyRelation);
+        return newEmptyRelation;
     }
 
     @Override
     public Plan visitLogicalOneRowRelation(LogicalOneRowRelation oneRowRelation, DeepCopierContext context) {
+        if (context.getRelationReplaceMap().containsKey(oneRowRelation.getRelationId())) {
+            return context.getRelationReplaceMap().get(oneRowRelation.getRelationId());
+        }
         List<NamedExpression> newProjects = oneRowRelation.getProjects().stream()
                 .map(p -> (NamedExpression) ExpressionDeepCopier.INSTANCE.deepCopy(p, context))
                 .collect(ImmutableList.toImmutableList());
-        return new LogicalOneRowRelation(StatementScopeIdGenerator.newRelationId(), newProjects);
+        LogicalOneRowRelation newOneRowRelation =
+                new LogicalOneRowRelation(StatementScopeIdGenerator.newRelationId(), newProjects);
+        context.putRelation(oneRowRelation.getRelationId(), newOneRowRelation);
+        return newOneRowRelation;
     }
 
     @Override
@@ -154,28 +177,6 @@ public Plan visitLogicalFilter(LogicalFilter<? extends Plan> filter, DeepCopierC
         return new LogicalFilter<>(conjuncts, child);
     }
 
-    @Override
-    public Plan visitLogicalOlapScan(LogicalOlapScan olapScan, DeepCopierContext context) {
-        if (context.getRelationReplaceMap().containsKey(olapScan.getRelationId())) {
-            return context.getRelationReplaceMap().get(olapScan.getRelationId());
-        }
-        LogicalOlapScan newOlapScan;
-        if (olapScan.getManuallySpecifiedPartitions().isEmpty()) {
-            newOlapScan = new LogicalOlapScan(StatementScopeIdGenerator.newRelationId(),
-                    olapScan.getTable(), olapScan.getQualifier(), olapScan.getSelectedTabletIds(),
-                    olapScan.getHints(), olapScan.getTableSample());
-        } else {
-            newOlapScan = new LogicalOlapScan(StatementScopeIdGenerator.newRelationId(),
-                    olapScan.getTable(), olapScan.getQualifier(),
-                    olapScan.getManuallySpecifiedPartitions(), olapScan.getSelectedTabletIds(),
-                    olapScan.getHints(), olapScan.getTableSample());
-        }
-        newOlapScan.getOutput();
-        context.putRelation(olapScan.getRelationId(), newOlapScan);
-        updateReplaceMapWithOutput(olapScan, newOlapScan, context.exprIdReplaceMap);
-        return newOlapScan;
-    }
-
     @Override
     public Plan visitLogicalDeferMaterializeOlapScan(LogicalDeferMaterializeOlapScan deferMaterializeOlapScan,
             DeepCopierContext context) {
@@ -186,55 +187,36 @@ public Plan visitLogicalDeferMaterializeOlapScan(LogicalDeferMaterializeOlapScan
                 .collect(ImmutableSet.toImmutableSet());
         SlotReference newRowId = (SlotReference) ExpressionDeepCopier.INSTANCE
                 .deepCopy(deferMaterializeOlapScan.getColumnIdSlot(), context);
-        return new LogicalDeferMaterializeOlapScan(newScan, newSlotIds, newRowId);
-    }
-
-    @Override
-    public Plan visitLogicalSchemaScan(LogicalSchemaScan schemaScan, DeepCopierContext context) {
-        if (context.getRelationReplaceMap().containsKey(schemaScan.getRelationId())) {
-            return context.getRelationReplaceMap().get(schemaScan.getRelationId());
-        }
-        LogicalSchemaScan newSchemaScan = new LogicalSchemaScan(StatementScopeIdGenerator.newRelationId(),
-                schemaScan.getTable(), schemaScan.getQualifier());
-        updateReplaceMapWithOutput(schemaScan, newSchemaScan, context.exprIdReplaceMap);
-        context.putRelation(schemaScan.getRelationId(), newSchemaScan);
-        return newSchemaScan;
+        LogicalDeferMaterializeOlapScan newMaterializeOlapScan =
+                new LogicalDeferMaterializeOlapScan(newScan, newSlotIds, newRowId);
+        return newMaterializeOlapScan;
     }
 
     @Override
     public Plan visitLogicalFileScan(LogicalFileScan fileScan, DeepCopierContext context) {
         if (context.getRelationReplaceMap().containsKey(fileScan.getRelationId())) {
             return context.getRelationReplaceMap().get(fileScan.getRelationId());
         }
-        LogicalFileScan newFileScan = new LogicalFileScan(StatementScopeIdGenerator.newRelationId(),
-                fileScan.getTable(), fileScan.getQualifier(), fileScan.getTableSample());
-        updateReplaceMapWithOutput(fileScan, newFileScan, context.exprIdReplaceMap);
-        context.putRelation(fileScan.getRelationId(), newFileScan);
         Set<Expression> conjuncts = fileScan.getConjuncts().stream()
                 .map(p -> ExpressionDeepCopier.INSTANCE.deepCopy(p, context))
                 .collect(ImmutableSet.toImmutableSet());
-        return newFileScan.withConjuncts(conjuncts);
-    }
-
-    @Override
-    public Plan visitLogicalTVFRelation(LogicalTVFRelation tvfRelation, DeepCopierContext context) {
-        if (context.getRelationReplaceMap().containsKey(tvfRelation.getRelationId())) {
-            return context.getRelationReplaceMap().get(tvfRelation.getRelationId());
-        }
-        LogicalTVFRelation newTVFRelation = new LogicalTVFRelation(StatementScopeIdGenerator.newRelationId(),
-                tvfRelation.getFunction());
-        updateReplaceMapWithOutput(tvfRelation, newTVFRelation, context.exprIdReplaceMap);
-        context.putRelation(tvfRelation.getRelationId(), newTVFRelation);
-        return newTVFRelation;
+        LogicalFileScan newFileScan = fileScan.withConjuncts(conjuncts)
+                .withRelationId(StatementScopeIdGenerator.newRelationId());
+        updateReplaceMapWithOutput(fileScan, newFileScan, context.exprIdReplaceMap);
+        context.putRelation(fileScan.getRelationId(), newFileScan);
+        return newFileScan;
     }
 
     @Override
     public Plan visitLogicalJdbcScan(LogicalJdbcScan jdbcScan, DeepCopierContext context) {
         if (context.getRelationReplaceMap().containsKey(jdbcScan.getRelationId())) {
             return context.getRelationReplaceMap().get(jdbcScan.getRelationId());
         }
-        LogicalJdbcScan newJdbcScan = new LogicalJdbcScan(StatementScopeIdGenerator.newRelationId(),
-                jdbcScan.getTable(), jdbcScan.getQualifier());
+        Set<Expression> conjuncts = jdbcScan.getConjuncts().stream()
+                .map(p -> ExpressionDeepCopier.INSTANCE.deepCopy(p, context))
+                .collect(ImmutableSet.toImmutableSet());
+        LogicalJdbcScan newJdbcScan = jdbcScan.withConjuncts(conjuncts)
+                .withRelationId(StatementScopeIdGenerator.newRelationId());
         updateReplaceMapWithOutput(jdbcScan, newJdbcScan, context.exprIdReplaceMap);
         context.putRelation(jdbcScan.getRelationId(), newJdbcScan);
         return newJdbcScan;
@@ -245,8 +227,11 @@ public Plan visitLogicalOdbcScan(LogicalOdbcScan odbcScan, DeepCopierContext con
         if (context.getRelationReplaceMap().containsKey(odbcScan.getRelationId())) {
             return context.getRelationReplaceMap().get(odbcScan.getRelationId());
         }
-        LogicalOdbcScan newOdbcScan = new LogicalOdbcScan(StatementScopeIdGenerator.newRelationId(),
-                odbcScan.getTable(), odbcScan.getQualifier());
+        Set<Expression> conjuncts = odbcScan.getConjuncts().stream()
+                .map(p -> ExpressionDeepCopier.INSTANCE.deepCopy(p, context))
+                .collect(ImmutableSet.toImmutableSet());
+        LogicalOdbcScan newOdbcScan = odbcScan.withConjuncts(conjuncts)
+                .withRelationId(StatementScopeIdGenerator.newRelationId());
         updateReplaceMapWithOutput(odbcScan, newOdbcScan, context.exprIdReplaceMap);
         context.putRelation(odbcScan.getRelationId(), newOdbcScan);
         return newOdbcScan;
@@ -257,8 +242,11 @@ public Plan visitLogicalEsScan(LogicalEsScan esScan, DeepCopierContext context)
         if (context.getRelationReplaceMap().containsKey(esScan.getRelationId())) {
             return context.getRelationReplaceMap().get(esScan.getRelationId());
         }
-        LogicalEsScan newEsScan = new LogicalEsScan(StatementScopeIdGenerator.newRelationId(),
-                esScan.getTable(), esScan.getQualifier());
+        Set<Expression> conjuncts = esScan.getConjuncts().stream()
+                .map(p -> ExpressionDeepCopier.INSTANCE.deepCopy(p, context))
+                .collect(ImmutableSet.toImmutableSet());
+        LogicalEsScan newEsScan = esScan.withConjuncts(conjuncts)
+                .withRelationId(StatementScopeIdGenerator.newRelationId());
         updateReplaceMapWithOutput(esScan, newEsScan, context.exprIdReplaceMap);
         context.putRelation(esScan.getRelationId(), newEsScan);
         return newEsScan;

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEConsumer.java

@@ -142,6 +142,11 @@ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpr
                 groupExpression, logicalProperties);
     }
 
+    @Override
+    public LogicalCTEConsumer withRelationId(RelationId relationId) {
+        throw new RuntimeException("should not call LogicalCTEConsumer's withRelationId method");
+    }
+
     @Override
     public List<Slot> computeOutput() {
         return ImmutableList.copyOf(producerToConsumerOutputMap.values());

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalDeferMaterializeOlapScan.java

@@ -24,6 +24,7 @@
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.expressions.SlotReference;
 import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.RelationId;
 import org.apache.doris.nereids.trees.plans.algebra.OlapScan;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.nereids.util.Utils;
@@ -130,6 +131,11 @@ public Plan withChildren(List<Plan> children) {
         return this;
     }
 
+    @Override
+    public LogicalDeferMaterializeOlapScan withRelationId(RelationId relationId) {
+        throw new RuntimeException("should not call LogicalDeferMaterializeOlapScan's withRelationId method");
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitLogicalDeferMaterializeOlapScan(this, context);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalEmptyRelation.java

@@ -81,6 +81,11 @@ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpr
         return new LogicalEmptyRelation(relationId, projects, groupExpression, logicalProperties);
     }
 
+    @Override
+    public LogicalEmptyRelation withRelationId(RelationId relationId) {
+        throw new RuntimeException("should not call LogicalEmptyRelation's withRelationId method");
+    }
+
     @Override
     public List<Slot> computeOutput() {
         return projects.stream()

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalEsScan.java

@@ -84,10 +84,16 @@ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpr
     }
 
     public LogicalEsScan withConjuncts(Set<Expression> conjuncts) {
-        return new LogicalEsScan(relationId, (ExternalTable) table, qualifier, groupExpression,
+        return new LogicalEsScan(relationId, (ExternalTable) table, qualifier, Optional.empty(),
                 Optional.of(getLogicalProperties()), conjuncts);
     }
 
+    @Override
+    public LogicalEsScan withRelationId(RelationId relationId) {
+        return new LogicalEsScan(relationId, (ExternalTable) table, qualifier, Optional.empty(),
+                Optional.empty(), conjuncts);
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitLogicalEsScan(this, context);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java

@@ -109,15 +109,21 @@ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpr
     }
 
     public LogicalFileScan withConjuncts(Set<Expression> conjuncts) {
-        return new LogicalFileScan(relationId, (ExternalTable) table, qualifier, groupExpression,
+        return new LogicalFileScan(relationId, (ExternalTable) table, qualifier, Optional.empty(),
                 Optional.of(getLogicalProperties()), conjuncts, selectedPartitions, tableSample);
     }
 
     public LogicalFileScan withSelectedPartitions(SelectedPartitions selectedPartitions) {
-        return new LogicalFileScan(relationId, (ExternalTable) table, qualifier, groupExpression,
+        return new LogicalFileScan(relationId, (ExternalTable) table, qualifier, Optional.empty(),
                 Optional.of(getLogicalProperties()), conjuncts, selectedPartitions, tableSample);
     }
 
+    @Override
+    public LogicalFileScan withRelationId(RelationId relationId) {
+        return new LogicalFileScan(relationId, (ExternalTable) table, qualifier, Optional.empty(),
+                Optional.empty(), conjuncts, selectedPartitions, tableSample);
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitLogicalFileScan(this, context);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalJdbcScan.java

@@ -82,7 +82,7 @@ public LogicalJdbcScan withGroupExpression(Optional<GroupExpression> groupExpres
     }
 
     public LogicalJdbcScan withConjuncts(Set<Expression> conjuncts) {
-        return new LogicalJdbcScan(relationId, table, qualifier, groupExpression,
+        return new LogicalJdbcScan(relationId, table, qualifier, Optional.empty(),
             Optional.of(getLogicalProperties()), conjuncts);
     }
 
@@ -92,6 +92,11 @@ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpr
         return new LogicalJdbcScan(relationId, table, qualifier, groupExpression, logicalProperties, conjuncts);
     }
 
+    @Override
+    public LogicalJdbcScan withRelationId(RelationId relationId) {
+        return new LogicalJdbcScan(relationId, table, qualifier, Optional.empty(), Optional.empty(), conjuncts);
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitLogicalJdbcScan(this, context);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOdbcScan.java

@@ -78,7 +78,7 @@ public LogicalOdbcScan withGroupExpression(Optional<GroupExpression> groupExpres
     }
 
     public LogicalOdbcScan withConjuncts(Set<Expression> conjuncts) {
-        return new LogicalOdbcScan(relationId, table, qualifier, groupExpression,
+        return new LogicalOdbcScan(relationId, table, qualifier, Optional.empty(),
                 Optional.of(getLogicalProperties()), conjuncts);
     }
 
@@ -88,6 +88,11 @@ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpr
         return new LogicalOdbcScan(relationId, table, qualifier, groupExpression, logicalProperties, conjuncts);
     }
 
+    @Override
+    public LogicalOdbcScan withRelationId(RelationId relationId) {
+        return new LogicalOdbcScan(relationId, table, qualifier, Optional.empty(), Optional.empty(), conjuncts);
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitLogicalOdbcScan(this, context);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java

@@ -311,6 +311,16 @@ public LogicalOlapScan withPreAggStatus(PreAggStatus preAggStatus) {
                 hints, cacheSlotWithSlotName, tableSample, directMvScan, projectPulledUp);
     }
 
+    @Override
+    public LogicalOlapScan withRelationId(RelationId relationId) {
+        // we have to set partitionPruned to false, so that mtmv rewrite can prevent deadlock when rewriting union
+        return new LogicalOlapScan(relationId, (Table) table, qualifier,
+                Optional.empty(), Optional.empty(),
+                selectedPartitionIds, false, selectedTabletIds,
+                selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions,
+                hints, Maps.newHashMap(), tableSample, directMvScan, projectPulledUp);
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitLogicalOlapScan(this, context);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOneRowRelation.java

@@ -89,6 +89,11 @@ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpr
         return new LogicalOneRowRelation(relationId, projects, groupExpression, logicalProperties);
     }
 
+    @Override
+    public LogicalOneRowRelation withRelationId(RelationId relationId) {
+        throw new RuntimeException("should not call LogicalOneRowRelation's withRelationId method");
+    }
+
     @Override
     public List<Slot> computeOutput() {
         return projects.stream()

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalRelation.java

@@ -83,6 +83,8 @@ public RelationId getRelationId() {
         return relationId;
     }
 
+    public abstract LogicalRelation withRelationId(RelationId relationId);
+
     @Override
     public JSONObject toJson() {
         JSONObject logicalRelation = super.toJson();