[feature](mtmv)MTMV refresh support multi pct tables

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

@@ -21,12 +21,12 @@
 import org.apache.doris.catalog.OlapTableFactory.MTMVParams;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
-import org.apache.doris.common.Pair;
 import org.apache.doris.common.util.PropertyAnalyzer;
 import org.apache.doris.datasource.CatalogMgr;
 import org.apache.doris.info.TableNameInfo;
 import org.apache.doris.job.common.TaskStatus;
 import org.apache.doris.job.extensions.mtmv.MTMVTask;
+import org.apache.doris.mtmv.BaseTableInfo;
 import org.apache.doris.mtmv.EnvInfo;
 import org.apache.doris.mtmv.MTMVCache;
 import org.apache.doris.mtmv.MTMVJobInfo;
@@ -40,17 +40,21 @@
 import org.apache.doris.mtmv.MTMVRefreshInfo;
 import org.apache.doris.mtmv.MTMVRefreshPartitionSnapshot;
 import org.apache.doris.mtmv.MTMVRefreshSnapshot;
+import org.apache.doris.mtmv.MTMVRelatedTableIf;
 import org.apache.doris.mtmv.MTMVRelation;
+import org.apache.doris.mtmv.MTMVSnapshotIf;
 import org.apache.doris.mtmv.MTMVStatus;
 import org.apache.doris.qe.ConnectContext;
 
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import com.google.gson.annotations.SerializedName;
+import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.io.IOException;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
@@ -242,6 +246,7 @@ public boolean addTaskResult(MTMVTask task, MTMVRelation relation,
                 this.status.setRefreshState(MTMVRefreshState.FAIL);
             }
             this.jobInfo.addHistoryTask(task);
+            compatiblePctSnapshot(partitionSnapshots);
             this.refreshSnapshot.updateSnapshots(partitionSnapshots, getPartitionNames());
             Env.getCurrentEnv().getMtmvService()
                     .refreshComplete(this, relation, task);
@@ -397,64 +402,26 @@ public Map<String, PartitionKeyDesc> generateMvPartitionDescs() {
         return result;
     }
 
-    /**
-     * Calculate the partition and associated partition mapping relationship of the MTMV
-     * It is the result of real-time comparison calculation, so there may be some costs,
-     * so it should be called with caution.
-     * The reason for not directly calling `calculatePartitionMappings` and
-     * generating a reverse index is to directly generate two maps here,
-     * without the need to traverse them again
-     *
-     * @return mvPartitionName ==> relationPartitionNames and relationPartitionName ==> mvPartitionName
-     * @throws AnalysisException
-     */
-    public Pair<Map<String, Set<String>>, Map<String, String>> calculateDoublyPartitionMappings()
-            throws AnalysisException {
-        if (mvPartitionInfo.getPartitionType() == MTMVPartitionType.SELF_MANAGE) {
-            return Pair.of(Maps.newHashMap(), Maps.newHashMap());
-        }
-        long start = System.currentTimeMillis();
-        Map<String, Set<String>> mvToBase = Maps.newHashMap();
-        Map<String, String> baseToMv = Maps.newHashMap();
-        Map<PartitionKeyDesc, Set<String>> relatedPartitionDescs = MTMVPartitionUtil
-                .generateRelatedPartitionDescs(mvPartitionInfo, mvProperties);
-        Map<String, PartitionItem> mvPartitionItems = getAndCopyPartitionItems();
-        for (Entry<String, PartitionItem> entry : mvPartitionItems.entrySet()) {
-            Set<String> basePartitionNames = relatedPartitionDescs.getOrDefault(entry.getValue().toPartitionKeyDesc(),
-                    Sets.newHashSet());
-            String mvPartitionName = entry.getKey();
-            mvToBase.put(mvPartitionName, basePartitionNames);
-            for (String basePartitionName : basePartitionNames) {
-                baseToMv.put(basePartitionName, mvPartitionName);
-            }
-        }
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("calculateDoublyPartitionMappings use [{}] mills, mvName is [{}]",
-                    System.currentTimeMillis() - start, name);
-        }
-        return Pair.of(mvToBase, baseToMv);
-    }
-
     /**
      * Calculate the partition and associated partition mapping relationship of the MTMV
      * It is the result of real-time comparison calculation, so there may be some costs,
      * so it should be called with caution
      *
-     * @return mvPartitionName ==> relationPartitionNames
+     * @return mvPartitionName ==> pctTable ==> pctPartitionName
      * @throws AnalysisException
      */
-    public Map<String, Set<String>> calculatePartitionMappings() throws AnalysisException {
+    public Map<String, Map<MTMVRelatedTableIf, Set<String>>> calculatePartitionMappings() throws AnalysisException {
         if (mvPartitionInfo.getPartitionType() == MTMVPartitionType.SELF_MANAGE) {
             return Maps.newHashMap();
         }
         long start = System.currentTimeMillis();
-        Map<String, Set<String>> res = Maps.newHashMap();
-        Map<PartitionKeyDesc, Set<String>> relatedPartitionDescs = MTMVPartitionUtil
-                .generateRelatedPartitionDescs(mvPartitionInfo, mvProperties);
+        Map<String, Map<MTMVRelatedTableIf, Set<String>>> res = Maps.newHashMap();
+        Map<PartitionKeyDesc, Map<MTMVRelatedTableIf, Set<String>>> pctPartitionDescs = MTMVPartitionUtil
+                .generateRelatedPartitionDescs(mvPartitionInfo, mvProperties, getPartitionColumns());
         Map<String, PartitionItem> mvPartitionItems = getAndCopyPartitionItems();
         for (Entry<String, PartitionItem> entry : mvPartitionItems.entrySet()) {
             res.put(entry.getKey(),
-                    relatedPartitionDescs.getOrDefault(entry.getValue().toPartitionKeyDesc(), Sets.newHashSet()));
+                    pctPartitionDescs.getOrDefault(entry.getValue().toPartitionKeyDesc(), Maps.newHashMap()));
         }
         if (LOG.isDebugEnabled()) {
             LOG.debug("calculatePartitionMappings use [{}] mills, mvName is [{}]",
@@ -580,4 +547,28 @@ private void compatibleInternal(CatalogMgr catalogMgr) throws Exception {
             refreshSnapshot.compatible(this);
         }
     }
+
+    @Override
+    public void gsonPostProcess() throws IOException {
+        super.gsonPostProcess();
+        Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots = refreshSnapshot.getPartitionSnapshots();
+        compatiblePctSnapshot(partitionSnapshots);
+    }
+
+    private void compatiblePctSnapshot(Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots) {
+        BaseTableInfo relatedTableInfo = mvPartitionInfo.getRelatedTableInfo();
+        if (relatedTableInfo == null) {
+            return;
+        }
+        if (MapUtils.isEmpty(partitionSnapshots)) {
+            return;
+        }
+        for (MTMVRefreshPartitionSnapshot partitionSnapshot : partitionSnapshots.values()) {
+            Map<String, MTMVSnapshotIf> partitions = partitionSnapshot.getPartitions();
+            Map<BaseTableInfo, Map<String, MTMVSnapshotIf>> pcts = partitionSnapshot.getPcts();
+            if (!MapUtils.isEmpty(partitions) && MapUtils.isEmpty(pcts)) {
+                pcts.put(relatedTableInfo, partitions);
+            }
+        }
+    }
 }

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

@@ -3418,7 +3418,7 @@ public List<Column> getPartitionColumns() {
     public MTMVSnapshotIf getPartitionSnapshot(String partitionName, MTMVRefreshContext context,
             Optional<MvccSnapshot> snapshot)
             throws AnalysisException {
-        Map<String, Long> partitionVersions = context.getBaseVersions().getPartitionVersions();
+        Map<String, Long> partitionVersions = context.getBaseVersions().getPartitionVersions(this);
         long partitionId = getPartitionOrAnalysisException(partitionName).getId();
         long visibleVersion = partitionVersions.containsKey(partitionName) ? partitionVersions.get(partitionName)
                 : getPartitionOrAnalysisException(partitionName).getVisibleVersion();

fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java

@@ -44,6 +44,7 @@
 import org.apache.doris.job.exception.JobException;
 import org.apache.doris.job.task.AbstractTask;
 import org.apache.doris.metric.MetricRepo;
+import org.apache.doris.mtmv.BaseColInfo;
 import org.apache.doris.mtmv.BaseTableInfo;
 import org.apache.doris.mtmv.MTMVBaseTableIf;
 import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
@@ -209,12 +210,14 @@ public void run() throws JobException {
                     MTMVPlanUtil.ensureMTMVQueryUsable(mtmv, ctx);
                 }
                 if (mtmv.getMvPartitionInfo().getPartitionType() != MTMVPartitionType.SELF_MANAGE) {
-                    MTMVRelatedTableIf relatedTable = mtmv.getMvPartitionInfo().getRelatedTable();
-                    if (!relatedTable.isValidRelatedTable()) {
-                        throw new JobException("MTMV " + mtmv.getName() + "'s related table " + relatedTable.getName()
-                                + " is not a valid related table anymore, stop refreshing."
-                                + " e.g. Table has multiple partition columns"
-                                + " or including not supported transform functions.");
+                    Set<MTMVRelatedTableIf> pctTables = mtmv.getMvPartitionInfo().getPctTables();
+                    for (MTMVRelatedTableIf pctTable : pctTables) {
+                        if (!pctTable.isValidRelatedTable()) {
+                            throw new JobException("MTMV " + mtmv.getName() + "'s pct table " + pctTable.getName()
+                                    + " is not a valid pct table anymore, stop refreshing."
+                                    + " e.g. Table has multiple partition columns"
+                                    + " or including not supported transform functions.");
+                        }
                     }
                     syncPartitions = MTMVPartitionUtil.alignMvPartition(mtmv);
                 }
@@ -578,8 +581,11 @@ protected void closeOrReleaseResources() {
     private Map<TableIf, String> getIncrementalTableMap() throws AnalysisException {
         Map<TableIf, String> tableWithPartKey = Maps.newHashMap();
         if (mtmv.getMvPartitionInfo().getPartitionType() != MTMVPartitionType.SELF_MANAGE) {
-            tableWithPartKey
-                    .put(mtmv.getMvPartitionInfo().getRelatedTable(), mtmv.getMvPartitionInfo().getRelatedCol());
+            List<BaseColInfo> pctInfos = mtmv.getMvPartitionInfo().getPctInfos();
+            for (BaseColInfo pctInfo : pctInfos) {
+                tableWithPartKey
+                        .put(MTMVUtil.getTable(pctInfo.getTableInfo()), pctInfo.getColName());
+            }
         }
         return tableWithPartKey;
     }

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

@@ -0,0 +1,73 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.mtmv;
+
+import com.google.gson.annotations.SerializedName;
+
+import java.util.Objects;
+
+public class BaseColInfo {
+    @SerializedName("ti")
+    private BaseTableInfo tableInfo;
+    @SerializedName("rn")
+    private String colName;
+
+    public BaseColInfo(String colName, BaseTableInfo tableInfo) {
+        this.colName = colName;
+        this.tableInfo = tableInfo;
+    }
+
+    public String getColName() {
+        return colName;
+    }
+
+    public void setColName(String colName) {
+        this.colName = colName;
+    }
+
+    public BaseTableInfo getTableInfo() {
+        return tableInfo;
+    }
+
+    public void setTableInfo(BaseTableInfo tableInfo) {
+        this.tableInfo = tableInfo;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        BaseColInfo that = (BaseColInfo) o;
+        return Objects.equals(tableInfo, that.tableInfo) && Objects.equals(colName, that.colName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(tableInfo, colName);
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder("BaseColInfo{");
+        sb.append("colName='").append(colName).append('\'');
+        sb.append(", tableInfo=").append(tableInfo);
+        sb.append('}');
+        return sb.toString();
+    }
+}

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

@@ -17,13 +17,16 @@
 
 package org.apache.doris.mtmv;
 
+import com.google.common.collect.Maps;
+
 import java.util.Map;
 
 public class MTMVBaseVersions {
     private final Map<Long, Long> tableVersions;
-    private final Map<String, Long> partitionVersions;
+    private final Map<MTMVRelatedTableIf, Map<String, Long>> partitionVersions;
 
-    public MTMVBaseVersions(Map<Long, Long> tableVersions, Map<String, Long> partitionVersions) {
+    public MTMVBaseVersions(Map<Long, Long> tableVersions,
+            Map<MTMVRelatedTableIf, Map<String, Long>> partitionVersions) {
         this.tableVersions = tableVersions;
         this.partitionVersions = partitionVersions;
     }
@@ -32,7 +35,7 @@ public Map<Long, Long> getTableVersions() {
         return tableVersions;
     }
 
-    public Map<String, Long> getPartitionVersions() {
-        return partitionVersions;
+    public Map<String, Long> getPartitionVersions(MTMVRelatedTableIf mtmvRelatedTableIf) {
+        return partitionVersions.getOrDefault(mtmvRelatedTableIf, Maps.newHashMap());
     }
 }

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

@@ -120,7 +120,7 @@ public static Pair<Boolean, String> compareDynamicPartition(OlapTable originalTa
     @VisibleForTesting
     public static Pair<Boolean, String> compareAutoPartition(OlapTable originalTable,
             OlapTable relatedTable) throws AnalysisException {
-        if (!isDynamicPartition(relatedTable)) {
+        if (!isAutoPartition(relatedTable)) {
             return Pair.of(false, "relatedTable is not dynamic partition.");
         }
         FunctionIntervalInfo originalFunctionIntervalInfo = PartitionExprUtil.getFunctionIntervalInfo(

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

@@ -69,18 +69,21 @@ public void analyze(MTMVPartitionInfo mvPartitionInfo) throws AnalysisException
             throw new AnalysisException(
                     String.format("timeUnit not support: %s, only support: %s", this.timeUnit, timeUnits));
         }
-        MTMVRelatedTableIf relatedTable = mvPartitionInfo.getRelatedTable();
-        PartitionType partitionType = relatedTable.getPartitionType(MvccUtil.getSnapshotFromContext(relatedTable));
-        if (partitionType == PartitionType.RANGE) {
-            Type partitionColumnType = MTMVPartitionUtil
-                    .getPartitionColumnType(mvPartitionInfo.getRelatedTable(), mvPartitionInfo.getRelatedCol());
-            if (!partitionColumnType.isDateType()) {
-                throw new AnalysisException(
-                        "partitionColumnType should be date/datetime "
-                                + "when PartitionType is range and expr is date_trunc");
+        List<BaseColInfo> pctInfos = mvPartitionInfo.getPctInfos();
+        for (BaseColInfo pctInfo : pctInfos) {
+            MTMVRelatedTableIf pctTable = MTMVUtil.getRelatedTable(pctInfo.getTableInfo());
+            PartitionType partitionType = pctTable.getPartitionType(MvccUtil.getSnapshotFromContext(pctTable));
+            if (partitionType == PartitionType.RANGE) {
+                Type partitionColumnType = MTMVPartitionUtil
+                        .getPartitionColumnType(pctTable, pctInfo.getColName());
+                if (!partitionColumnType.isDateType()) {
+                    throw new AnalysisException(
+                            "partitionColumnType should be date/datetime "
+                                    + "when PartitionType is range and expr is date_trunc");
+                }
+            } else {
+                throw new AnalysisException("date_trunc only support range partition");
             }
-        } else {
-            throw new AnalysisException("date_trunc only support range partition");
         }
     }
 
@@ -125,9 +128,9 @@ private Optional<String> getDateFormat(Map<String, String> mvProperties) {
 
     @Override
     public PartitionKeyDesc generateRollUpPartitionKeyDesc(PartitionKeyDesc partitionKeyDesc,
-            MTMVPartitionInfo mvPartitionInfo) throws AnalysisException {
+            MTMVPartitionInfo mvPartitionInfo, MTMVRelatedTableIf pctTable) throws AnalysisException {
         Type partitionColumnType = MTMVPartitionUtil
-                .getPartitionColumnType(mvPartitionInfo.getRelatedTable(), mvPartitionInfo.getRelatedCol());
+                .getPartitionColumnType(pctTable, mvPartitionInfo.getPartitionColByPctTable(pctTable));
         // mtmv only support one partition column
         Preconditions.checkState(partitionKeyDesc.getLowerValues().size() == 1,
                 "only support one partition column");
@@ -243,4 +246,18 @@ private String dateTimeToStr(DateTimeV2Literal literal,
                     "MTMV not support partition with column type : " + partitionColumnType);
         }
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        MTMVPartitionExprDateTrunc that = (MTMVPartitionExprDateTrunc) o;
+        return Objects.equals(timeUnit, that.timeUnit);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(timeUnit);
+    }
 }

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

@@ -47,7 +47,7 @@ String getRollUpIdentity(PartitionKeyDesc partitionKeyDesc, Map<String, String>
      * @throws AnalysisException
      */
     PartitionKeyDesc generateRollUpPartitionKeyDesc(
-            PartitionKeyDesc partitionKeyDesc, MTMVPartitionInfo mvPartitionInfo)
+            PartitionKeyDesc partitionKeyDesc, MTMVPartitionInfo mvPartitionInfo, MTMVRelatedTableIf pctTable)
             throws AnalysisException;
 
     /**