branch-2.1:[enhance](mtmv)cache table snapshot in refresh context (#50855) (#51493)

pick: #50855

Author: zddr

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

@@ -487,11 +487,34 @@ private static boolean isSyncWithBaseTable(MTMVRefreshContext context, String mt
         if (!baseTable.needAutoRefresh()) {
             return true;
         }
-        MTMVSnapshotIf baseTableCurrentSnapshot = baseTable.getTableSnapshot(context, Optional.empty());
+        MTMVSnapshotIf baseTableCurrentSnapshot = getTableSnapshotFromContext(baseTable, context);
         return mtmv.getRefreshSnapshot()
                 .equalsWithBaseTable(mtmvPartitionName, new BaseTableInfo(baseTable), baseTableCurrentSnapshot);
     }
 
+    /**
+     * Try context first, then load via getTableSnapshot and cache
+     *
+     * @param mtmvRelatedTableIf Base table of materialized views
+     * @param context The context data persists for the duration of either a refresh task
+     *         or a transparent rewrite operation
+     * @return The snapshot information of the MTMV
+     * @throws AnalysisException
+     */
+    public static MTMVSnapshotIf getTableSnapshotFromContext(MTMVRelatedTableIf mtmvRelatedTableIf,
+            MTMVRefreshContext context)
+            throws AnalysisException {
+        BaseTableInfo baseTableInfo = new BaseTableInfo(mtmvRelatedTableIf);
+        Map<BaseTableInfo, MTMVSnapshotIf> baseTableSnapshotCache = context.getBaseTableSnapshotCache();
+        if (baseTableSnapshotCache.containsKey(baseTableInfo)) {
+            return baseTableSnapshotCache.get(baseTableInfo);
+        }
+        MTMVSnapshotIf baseTableCurrentSnapshot = mtmvRelatedTableIf.getTableSnapshot(context,
+                Optional.empty());
+        baseTableSnapshotCache.put(baseTableInfo, baseTableCurrentSnapshot);
+        return baseTableCurrentSnapshot;
+    }
+
     /**
      * Generate updated snapshots of partitions to determine if they are synchronized
      *
@@ -538,7 +561,7 @@ private static MTMVRefreshPartitionSnapshot generatePartitionSnapshot(MTMVRefres
                 continue;
             }
             refreshPartitionSnapshot.addTableSnapshot(baseTableInfo,
-                    ((MTMVRelatedTableIf) table).getTableSnapshot(context, Optional.empty()));
+                    getTableSnapshotFromContext((MTMVRelatedTableIf) table, context));
         }
         return refreshPartitionSnapshot;
     }

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

@@ -20,13 +20,19 @@
 import org.apache.doris.catalog.MTMV;
 import org.apache.doris.common.AnalysisException;
 
+import com.google.common.collect.Maps;
+
 import java.util.Map;
 import java.util.Set;
 
 public class MTMVRefreshContext {
     private MTMV mtmv;
     private Map<String, Set<String>> partitionMappings;
     private MTMVBaseVersions baseVersions;
+    // Within the same context, repeated fetches of the same table's snapshot must return consistent values.
+    // Hence, the results are cached at this stage.
+    // The value is loaded/cached on the first fetch
+    private Map<BaseTableInfo, MTMVSnapshotIf> baseTableSnapshotCache = Maps.newHashMap();
 
     public MTMVRefreshContext(MTMV mtmv) {
         this.mtmv = mtmv;
@@ -44,6 +50,10 @@ public MTMVBaseVersions getBaseVersions() {
         return baseVersions;
     }
 
+    public Map<BaseTableInfo, MTMVSnapshotIf> getBaseTableSnapshotCache() {
+        return baseTableSnapshotCache;
+    }
+
     public static MTMVRefreshContext buildContext(MTMV mtmv) throws AnalysisException {
         MTMVRefreshContext context = new MTMVRefreshContext(mtmv);
         context.partitionMappings = mtmv.calculatePartitionMappings();

fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPartitionUtilTest.java

@@ -38,6 +38,7 @@
 import org.junit.Test;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
@@ -92,6 +93,10 @@ public void setUp() throws NoSuchMethodException, SecurityException, AnalysisExc
                 minTimes = 0;
                 result = versions;
 
+                context.getBaseTableSnapshotCache();
+                minTimes = 0;
+                result = Maps.newHashMap();
+
                 mtmv.getPartitions();
                 minTimes = 0;
                 result = Lists.newArrayList(p1);
@@ -296,4 +301,20 @@ public void testIsTableNamelike() {
         Assert.assertFalse(MTMVPartitionUtil.isTableNamelike(new TableName("db1"), tableNameToCheck));
         Assert.assertFalse(MTMVPartitionUtil.isTableNamelike(new TableName("ctl1"), tableNameToCheck));
     }
+
+    @Test
+    public void testGetTableSnapshotFromContext() throws AnalysisException {
+        Map<BaseTableInfo, MTMVSnapshotIf> cache = Maps.newHashMap();
+        new Expectations() {
+            {
+                context.getBaseTableSnapshotCache();
+                minTimes = 0;
+                result = cache;
+            }
+        };
+        Assert.assertTrue(cache.isEmpty());
+        MTMVPartitionUtil.getTableSnapshotFromContext(baseOlapTable, context);
+        Assert.assertEquals(1, cache.size());
+        Assert.assertEquals(baseSnapshotIf, cache.values().iterator().next());
+    }
 }