[enhance](mtmv)Enable the MTMVRelatedTableIf interface to support mvcc 2

fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java

@@ -1048,7 +1048,6 @@ public PartitionInfo getPartitionInfo() {
         return partitionInfo;
     }
 
-    @Override
     public Set<String> getPartitionColumnNames() throws DdlException {
         Set<String> partitionColumnNames = Sets.newHashSet();
         if (partitionInfo instanceof SinglePartitionInfo) {
@@ -1066,6 +1065,11 @@ public Set<String> getPartitionColumnNames() throws DdlException {
         }
     }
 
+    @Override
+    public Set<String> getPartitionColumnNames(OptionalLong snapshotId) throws DdlException {
+        return getPartitionColumnNames();
+    }
+
     public DistributionInfo getDefaultDistributionInfo() {
         return defaultDistributionInfo;
     }
@@ -3251,6 +3255,10 @@ public long getVisibleVersionTime() {
     }
 
     @Override
+    public PartitionType getPartitionType(OptionalLong snapshotId) {
+        return getPartitionType();
+    }
+
     public PartitionType getPartitionType() {
         return partitionInfo.getType();
     }
@@ -3275,6 +3283,10 @@ public Map<String, PartitionItem> getAndCopyPartitionItems(OptionalLong snapshot
     }
 
     @Override
+    public List<Column> getPartitionColumns(OptionalLong snapshotId) {
+        return getPartitionColumns();
+    }
+
     public List<Column> getPartitionColumns() {
         return getPartitionInfo().getPartitionColumns();
     }

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java

@@ -288,7 +288,6 @@ public List<Type> getPartitionColumnTypes() {
                 .orElse(Collections.emptyList());
     }
 
-    @Override
     public List<Column> getPartitionColumns() {
         makeSureInitialized();
         Optional<SchemaCacheValue> schemaCacheValue = getSchemaCacheValue();
@@ -756,16 +755,24 @@ public Set<String> getDistributionColumnNames() {
     }
 
     @Override
+    public PartitionType getPartitionType(OptionalLong snapshotId) {
+        return getPartitionType();
+    }
+
     public PartitionType getPartitionType() {
         return getPartitionColumns().size() > 0 ? PartitionType.LIST : PartitionType.UNPARTITIONED;
     }
 
-    @Override
     public Set<String> getPartitionColumnNames() {
         return getPartitionColumns().stream()
                 .map(c -> c.getName().toLowerCase()).collect(Collectors.toSet());
     }
 
+    @Override
+    public Set<String> getPartitionColumnNames(OptionalLong snapshotId) {
+        return getPartitionColumnNames();
+    }
+
     @Override
     public Map<String, PartitionItem> getAndCopyPartitionItems(OptionalLong snapshotId) {
         HiveMetaStoreCache cache = Env.getCurrentEnv().getExtMetaCacheMgr()

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java

@@ -318,18 +318,18 @@ public Map<String, PartitionItem> getAndCopyPartitionItems(OptionalLong snapshot
     }
 
     @Override
-    public PartitionType getPartitionType() {
+    public PartitionType getPartitionType(OptionalLong snapshotId) {
         return getPartitionColumnsFromCache().size() > 0 ? PartitionType.LIST : PartitionType.UNPARTITIONED;
     }
 
     @Override
-    public Set<String> getPartitionColumnNames() {
+    public Set<String> getPartitionColumnNames(OptionalLong snapshotId) {
         return getPartitionColumnsFromCache().stream()
                 .map(c -> c.getName().toLowerCase()).collect(Collectors.toSet());
     }
 
     @Override
-    public List<Column> getPartitionColumns() {
+    public List<Column> getPartitionColumns(OptionalLong snapshotId) {
         return getPartitionColumnsFromCache();
     }
 

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionExprDateTrunc.java

@@ -44,6 +44,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.Set;
 
 public class MTMVPartitionExprDateTrunc implements MTMVPartitionExprService {
@@ -69,7 +70,7 @@ public void analyze(MTMVPartitionInfo mvPartitionInfo) throws AnalysisException
                     String.format("timeUnit not support: %s, only support: %s", this.timeUnit, timeUnits));
         }
         MTMVRelatedTableIf relatedTable = mvPartitionInfo.getRelatedTable();
-        PartitionType partitionType = relatedTable.getPartitionType();
+        PartitionType partitionType = relatedTable.getPartitionType(OptionalLong.empty());
         if (partitionType == PartitionType.RANGE) {
             Type partitionColumnType = MTMVPartitionUtil
                     .getPartitionColumnType(mvPartitionInfo.getRelatedTable(), mvPartitionInfo.getRelatedCol());

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionInfo.java

@@ -25,6 +25,7 @@
 import com.google.gson.annotations.SerializedName;
 
 import java.util.List;
+import java.util.OptionalLong;
 
 /**
  * MTMVPartitionInfo
@@ -115,7 +116,7 @@ public int getRelatedColPos() throws AnalysisException {
         if (partitionType == MTMVPartitionType.SELF_MANAGE) {
             throw new AnalysisException("partitionType is: " + partitionType);
         }
-        List<Column> partitionColumns = getRelatedTable().getPartitionColumns();
+        List<Column> partitionColumns = getRelatedTable().getPartitionColumns(OptionalLong.empty());
         for (int i = 0; i < partitionColumns.size(); i++) {
             if (partitionColumns.get(i).getName().equalsIgnoreCase(relatedCol)) {
                 return i;

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java

@@ -504,7 +504,7 @@ private static MTMVRefreshPartitionSnapshot generatePartitionSnapshot(MTMVRefres
     }
 
     public static Type getPartitionColumnType(MTMVRelatedTableIf relatedTable, String col) throws AnalysisException {
-        List<Column> partitionColumns = relatedTable.getPartitionColumns();
+        List<Column> partitionColumns = relatedTable.getPartitionColumns(OptionalLong.empty());
         for (Column column : partitionColumns) {
             if (column.getName().equals(col)) {
                 return column.getType();

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedPartitionDescRollUpGenerator.java

@@ -31,6 +31,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.OptionalLong;
 import java.util.Set;
 
 /**
@@ -45,7 +46,7 @@ public void apply(MTMVPartitionInfo mvPartitionInfo, Map<String, String> mvPrope
             return;
         }
         MTMVRelatedTableIf relatedTable = mvPartitionInfo.getRelatedTable();
-        PartitionType partitionType = relatedTable.getPartitionType();
+        PartitionType partitionType = relatedTable.getPartitionType(OptionalLong.empty());
         if (partitionType == PartitionType.RANGE) {
             lastResult.setDescs(rollUpRange(lastResult.getDescs(), mvPartitionInfo));
         } else if (partitionType == PartitionType.LIST) {

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedTableIf.java

@@ -47,24 +47,27 @@ public interface MTMVRelatedTableIf extends TableIf {
     /**
      * getPartitionType LIST/RANGE/UNPARTITIONED
      *
+     * @param snapshotId
      * @return
      */
-    PartitionType getPartitionType();
+    PartitionType getPartitionType(OptionalLong snapshotId);
 
     /**
      * getPartitionColumnNames
      *
+     * @param snapshotId
      * @return
      * @throws DdlException
      */
-    Set<String> getPartitionColumnNames() throws DdlException;
+    Set<String> getPartitionColumnNames(OptionalLong snapshotId) throws DdlException;
 
     /**
      * getPartitionColumns
      *
+     * @param snapshotId
      * @return
      */
-    List<Column> getPartitionColumns();
+    List<Column> getPartitionColumns(OptionalLong snapshotId);
 
     /**
      * getPartitionSnapshot

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewUtils.java

@@ -90,6 +90,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -481,13 +482,13 @@ public Void visitLogicalRelation(LogicalRelation relation, IncrementCheckerConte
                 return null;
             }
             MTMVRelatedTableIf relatedTable = (MTMVRelatedTableIf) table;
-            PartitionType type = relatedTable.getPartitionType();
+            PartitionType type = relatedTable.getPartitionType(OptionalLong.empty());
             if (PartitionType.UNPARTITIONED.equals(type)) {
                 context.addFailReason(String.format("related base table is not partition table, the table is %s",
                         table.getName()));
                 return null;
             }
-            Set<Column> partitionColumnSet = new HashSet<>(relatedTable.getPartitionColumns());
+            Set<Column> partitionColumnSet = new HashSet<>(relatedTable.getPartitionColumns(OptionalLong.empty()));
             Column mvReferenceColumn = contextPartitionColumn.getColumn().get();
             Expr definExpr = mvReferenceColumn.getDefineExpr();
             if (definExpr instanceof SlotRef) {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateMTMVInfo.java

@@ -94,6 +94,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
@@ -351,7 +352,7 @@ private PartitionDesc generatePartitionDesc(ConnectContext ctx) {
                     allPartitionDescs.size(), ctx.getSessionVariable().getCreateTablePartitionMaxNum()));
         }
         try {
-            PartitionType type = relatedTable.getPartitionType();
+            PartitionType type = relatedTable.getPartitionType(OptionalLong.empty());
             if (type == PartitionType.RANGE) {
                 return new RangePartitionDesc(Lists.newArrayList(mvPartitionInfo.getPartitionCol()),
                         allPartitionDescs);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/MTMVPartitionDefinition.java

@@ -54,6 +54,7 @@
 import com.google.common.collect.Sets;
 
 import java.util.List;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -147,7 +148,7 @@ private RelatedTableInfo getRelatedTableInfo(NereidsPlanner planner, ConnectCont
             MTMVRelatedTableIf mtmvBaseRealtedTable = MTMVUtil.getRelatedTable(relatedTableInfo.getTableInfo());
             Set<String> partitionColumnNames = Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER);
             try {
-                partitionColumnNames.addAll(mtmvBaseRealtedTable.getPartitionColumnNames());
+                partitionColumnNames.addAll(mtmvBaseRealtedTable.getPartitionColumnNames(OptionalLong.empty()));
             } catch (DdlException e) {
                 throw new AnalysisException(e.getMessage(), e);
             }