[feat](mtmv)use real snapshot instead of optional.empty()

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

@@ -27,6 +27,7 @@
 import org.apache.doris.catalog.Type;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.util.PropertyAnalyzer;
+import org.apache.doris.datasource.mvcc.MvccUtil;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.functions.executable.DateTimeArithmetic;
 import org.apache.doris.nereids.trees.expressions.functions.executable.DateTimeExtractAndTransform;
@@ -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(Optional.empty());
+        PartitionType partitionType = relatedTable.getPartitionType(MvccUtil.getSnapshotFromContext(relatedTable));
         if (partitionType == PartitionType.RANGE) {
             Type partitionColumnType = MTMVPartitionUtil
                     .getPartitionColumnType(mvPartitionInfo.getRelatedTable(), mvPartitionInfo.getRelatedCol());

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

@@ -21,11 +21,11 @@
 import org.apache.doris.catalog.Column;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.mvcc.MvccUtil;
 
 import com.google.gson.annotations.SerializedName;
 
 import java.util.List;
-import java.util.Optional;
 
 /**
  * MTMVPartitionInfo
@@ -116,7 +116,9 @@ public int getRelatedColPos() throws AnalysisException {
         if (partitionType == MTMVPartitionType.SELF_MANAGE) {
             throw new AnalysisException("partitionType is: " + partitionType);
         }
-        List<Column> partitionColumns = getRelatedTable().getPartitionColumns(Optional.empty());
+        MTMVRelatedTableIf mtmvRelatedTableIf = getRelatedTable();
+        List<Column> partitionColumns = mtmvRelatedTableIf.getPartitionColumns(
+                MvccUtil.getSnapshotFromContext(mtmvRelatedTableIf));
         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

@@ -32,6 +32,7 @@
 import org.apache.doris.catalog.Type;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.DdlException;
+import org.apache.doris.datasource.mvcc.MvccUtil;
 import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
 import org.apache.doris.rpc.RpcException;
 
@@ -50,7 +51,6 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.Set;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -324,7 +324,7 @@ public static boolean isSyncWithPartitions(MTMVRefreshContext context, String mt
         }
         for (String relatedPartitionName : relatedPartitionNames) {
             MTMVSnapshotIf relatedPartitionCurrentSnapshot = relatedTable
-                    .getPartitionSnapshot(relatedPartitionName, context, Optional.empty());
+                    .getPartitionSnapshot(relatedPartitionName, context, MvccUtil.getSnapshotFromContext(relatedTable));
             if (!mtmv.getRefreshSnapshot()
                     .equalsWithRelatedPartition(mtmvPartitionName, relatedPartitionName,
                             relatedPartitionCurrentSnapshot)) {
@@ -441,7 +441,8 @@ private static boolean isSyncWithBaseTable(MTMVRefreshContext context, String mt
         if (!baseTable.needAutoRefresh()) {
             return true;
         }
-        MTMVSnapshotIf baseTableCurrentSnapshot = baseTable.getTableSnapshot(context, Optional.empty());
+        MTMVSnapshotIf baseTableCurrentSnapshot = baseTable.getTableSnapshot(context,
+                MvccUtil.getSnapshotFromContext(baseTable));
         return mtmv.getRefreshSnapshot()
                 .equalsWithBaseTable(mtmvPartitionName, new BaseTableInfo(baseTable), baseTableCurrentSnapshot);
     }
@@ -476,10 +477,9 @@ private static MTMVRefreshPartitionSnapshot generatePartitionSnapshot(MTMVRefres
         if (mtmv.getMvPartitionInfo().getPartitionType() != MTMVPartitionType.SELF_MANAGE) {
             MTMVRelatedTableIf relatedTable = mtmv.getMvPartitionInfo().getRelatedTable();
             for (String relatedPartitionName : relatedPartitionNames) {
-                MTMVSnapshotIf partitionSnapshot = relatedTable
-                        .getPartitionSnapshot(relatedPartitionName, context, Optional.empty());
-                refreshPartitionSnapshot.getPartitions()
-                        .put(relatedPartitionName, partitionSnapshot);
+                MTMVSnapshotIf partitionSnapshot = relatedTable.getPartitionSnapshot(relatedPartitionName, context,
+                        MvccUtil.getSnapshotFromContext(relatedTable));
+                refreshPartitionSnapshot.getPartitions().put(relatedPartitionName, partitionSnapshot);
             }
         }
         for (BaseTableInfo baseTableInfo : baseTables) {
@@ -492,13 +492,13 @@ private static MTMVRefreshPartitionSnapshot generatePartitionSnapshot(MTMVRefres
                 continue;
             }
             refreshPartitionSnapshot.addTableSnapshot(baseTableInfo,
-                    ((MTMVRelatedTableIf) table).getTableSnapshot(context, Optional.empty()));
+                    ((MTMVRelatedTableIf) table).getTableSnapshot(context, MvccUtil.getSnapshotFromContext(table)));
         }
         return refreshPartitionSnapshot;
     }
 
     public static Type getPartitionColumnType(MTMVRelatedTableIf relatedTable, String col) throws AnalysisException {
-        List<Column> partitionColumns = relatedTable.getPartitionColumns(Optional.empty());
+        List<Column> partitionColumns = relatedTable.getPartitionColumns(MvccUtil.getSnapshotFromContext(relatedTable));
         for (Column column : partitionColumns) {
             if (column.getName().equals(col)) {
                 return column.getType();

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

@@ -18,9 +18,9 @@
 package org.apache.doris.mtmv;
 
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.mvcc.MvccUtil;
 
 import java.util.Map;
-import java.util.Optional;
 
 /**
  * get all related partition descs
@@ -30,6 +30,7 @@ public class MTMVRelatedPartitionDescInitGenerator implements MTMVRelatedPartiti
     @Override
     public void apply(MTMVPartitionInfo mvPartitionInfo, Map<String, String> mvProperties,
             RelatedPartitionDescResult lastResult) throws AnalysisException {
-        lastResult.setItems(mvPartitionInfo.getRelatedTable().getAndCopyPartitionItems(Optional.empty()));
+        MTMVRelatedTableIf relatedTable = mvPartitionInfo.getRelatedTable();
+        lastResult.setItems(relatedTable.getAndCopyPartitionItems(MvccUtil.getSnapshotFromContext(relatedTable)));
     }
 }

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

@@ -21,6 +21,7 @@
 import org.apache.doris.analysis.PartitionValue;
 import org.apache.doris.catalog.PartitionType;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.mvcc.MvccUtil;
 import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
 
 import com.google.common.base.Preconditions;
@@ -31,7 +32,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Optional;
 import java.util.Set;
 
 /**
@@ -46,7 +46,7 @@ public void apply(MTMVPartitionInfo mvPartitionInfo, Map<String, String> mvPrope
             return;
         }
         MTMVRelatedTableIf relatedTable = mvPartitionInfo.getRelatedTable();
-        PartitionType partitionType = relatedTable.getPartitionType(Optional.empty());
+        PartitionType partitionType = relatedTable.getPartitionType(MvccUtil.getSnapshotFromContext(relatedTable));
         if (partitionType == PartitionType.RANGE) {
             lastResult.setDescs(rollUpRange(lastResult.getDescs(), mvPartitionInfo));
         } else if (partitionType == PartitionType.LIST) {

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

@@ -24,6 +24,7 @@
 import org.apache.doris.catalog.PartitionType;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.constraint.TableIdentifier;
+import org.apache.doris.datasource.mvcc.MvccUtil;
 import org.apache.doris.mtmv.BaseTableInfo;
 import org.apache.doris.mtmv.MTMVCache;
 import org.apache.doris.mtmv.MTMVRelatedTableIf;
@@ -497,13 +498,14 @@ public Void visitLogicalRelation(LogicalRelation relation, IncrementCheckerConte
                 return null;
             }
             MTMVRelatedTableIf relatedTable = (MTMVRelatedTableIf) table;
-            PartitionType type = relatedTable.getPartitionType(Optional.empty());
+            PartitionType type = relatedTable.getPartitionType(MvccUtil.getSnapshotFromContext(relatedTable));
             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(Optional.empty()));
+            Set<Column> partitionColumnSet = new HashSet<>(
+                    relatedTable.getPartitionColumns(MvccUtil.getSnapshotFromContext(relatedTable)));
             Column mvReferenceColumn = contextPartitionColumn.getColumn().get();
             Expr definExpr = mvReferenceColumn.getDefineExpr();
             if (definExpr instanceof SlotRef) {