[enhance](mtmv)Mtmv rollup

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -101,7 +101,7 @@ statementBase
         (REFRESH refreshMethod? refreshTrigger?)?
         ((DUPLICATE)? KEY keys=identifierList)?
         (COMMENT STRING_LITERAL)?
-        (PARTITION BY LEFT_PAREN partitionKey = identifier RIGHT_PAREN)?
+        (PARTITION BY mvPartition)?
         (DISTRIBUTED BY (HASH hashKeys=identifierList | RANDOM) (BUCKETS (INTEGER_VALUE | AUTO))?)?
         propertyClause?
         AS query                                                        #createMTMV
@@ -225,6 +225,11 @@ refreshMethod
     : COMPLETE | AUTO
     ;
 
+mvPartition
+    : LEFT_PAREN partitionKey = identifier RIGHT_PAREN
+    | partitionExpr = functionCallExpression
+    ;
+
 identifierOrStringLiteral
     : identifier
     | STRING_LITERAL

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

@@ -30,6 +30,7 @@
 import org.apache.doris.mtmv.MTMVJobManager;
 import org.apache.doris.mtmv.MTMVPartitionInfo;
 import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
+import org.apache.doris.mtmv.MTMVPartitionUtil;
 import org.apache.doris.mtmv.MTMVPlanUtil;
 import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVRefreshState;
 import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVState;
@@ -38,7 +39,6 @@
 import org.apache.doris.mtmv.MTMVRefreshSnapshot;
 import org.apache.doris.mtmv.MTMVRelation;
 import org.apache.doris.mtmv.MTMVStatus;
-import org.apache.doris.mtmv.MTMVUtil;
 import org.apache.doris.persist.gson.GsonUtils;
 
 import com.google.common.collect.Maps;
@@ -51,7 +51,6 @@
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
@@ -313,35 +312,6 @@ public Map<Long, PartitionKeyDesc> generateMvPartitionDescs() {
         return result;
     }
 
-    /**
-     * generateRelatedPartitionDescs
-     * <p>
-     * Different partitions may generate the same PartitionKeyDesc through logical calculations
-     * (such as selecting only one column, or rolling up partitions), so it is a one to many relationship
-     *
-     * @return related PartitionKeyDesc ==> relatedPartitionIds
-     * @throws AnalysisException
-     */
-    public Map<PartitionKeyDesc, Set<Long>> generateRelatedPartitionDescs() throws AnalysisException {
-        if (mvPartitionInfo.getPartitionType() == MTMVPartitionType.SELF_MANAGE) {
-            return Maps.newHashMap();
-        }
-        Map<PartitionKeyDesc, Set<Long>> res = new HashMap<>();
-        int relatedColPos = mvPartitionInfo.getRelatedColPos();
-        Map<Long, PartitionItem> relatedPartitionItems = mvPartitionInfo.getRelatedTable()
-                .getPartitionItemsByTimeFilter(relatedColPos,
-                        MTMVUtil.generateMTMVPartitionSyncConfigByProperties(mvProperties));
-        for (Entry<Long, PartitionItem> entry : relatedPartitionItems.entrySet()) {
-            PartitionKeyDesc partitionKeyDesc = entry.getValue().toPartitionKeyDesc(relatedColPos);
-            if (res.containsKey(partitionKeyDesc)) {
-                res.get(partitionKeyDesc).add(entry.getKey());
-            } else {
-                res.put(partitionKeyDesc, Sets.newHashSet(entry.getKey()));
-            }
-        }
-        return res;
-    }
-
     /**
      * 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,
@@ -354,13 +324,19 @@ public Map<Long, Set<Long>> calculatePartitionMappings() throws AnalysisExceptio
         if (mvPartitionInfo.getPartitionType() == MTMVPartitionType.SELF_MANAGE) {
             return Maps.newHashMap();
         }
+        long start = System.currentTimeMillis();
         Map<Long, Set<Long>> res = Maps.newHashMap();
-        Map<PartitionKeyDesc, Set<Long>> relatedPartitionDescs = generateRelatedPartitionDescs();
+        Map<PartitionKeyDesc, Set<Long>> relatedPartitionDescs = MTMVPartitionUtil
+                .generateRelatedPartitionDescs(mvPartitionInfo, mvProperties);
         Map<Long, PartitionItem> mvPartitionItems = getAndCopyPartitionItems();
         for (Entry<Long, PartitionItem> entry : mvPartitionItems.entrySet()) {
             res.put(entry.getKey(),
                     relatedPartitionDescs.getOrDefault(entry.getValue().toPartitionKeyDesc(), Sets.newHashSet()));
         }
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("calculatePartitionMappings use [{}] mills, mvName is [{}]",
+                    System.currentTimeMillis() - start, name);
+        }
         return res;
     }
 

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

@@ -169,7 +169,7 @@ public void run() throws JobException {
             // Now, the MTMV first ensures consistency with the data in the cache.
             // To be completely consistent with hive, you need to manually refresh the cache
             // refreshHmsTable();
-            if (mtmv.getMvPartitionInfo().getPartitionType() == MTMVPartitionType.FOLLOW_BASE_TABLE) {
+            if (mtmv.getMvPartitionInfo().getPartitionType() != MTMVPartitionType.SELF_MANAGE) {
                 MTMVPartitionUtil.alignMvPartition(mtmv);
             }
             Map<Long, Set<Long>> partitionMappings = mtmv.calculatePartitionMappings();
@@ -217,7 +217,7 @@ private void exec(ConnectContext ctx, Set<Long> refreshPartitionIds,
         lastQueryId = DebugUtil.printId(queryId);
         // if SELF_MANAGE mv, only have default partition,  will not have partitionItem, so we give empty set
         UpdateMvByPartitionCommand command = UpdateMvByPartitionCommand
-                .from(mtmv, mtmv.getMvPartitionInfo().getPartitionType() == MTMVPartitionType.FOLLOW_BASE_TABLE
+                .from(mtmv, mtmv.getMvPartitionInfo().getPartitionType() != MTMVPartitionType.SELF_MANAGE
                         ? refreshPartitionIds : Sets.newHashSet(), tableWithPartKey);
         executor = new StmtExecutor(ctx, new LogicalPlanAdapter(command, ctx.getStatementContext()));
         ctx.setExecutor(executor);
@@ -380,7 +380,7 @@ private void after() {
 
     private Map<TableIf, String> getIncrementalTableMap() throws AnalysisException {
         Map<TableIf, String> tableWithPartKey = Maps.newHashMap();
-        if (mtmv.getMvPartitionInfo().getPartitionType() == MTMVPartitionType.FOLLOW_BASE_TABLE) {
+        if (mtmv.getMvPartitionInfo().getPartitionType() != MTMVPartitionType.SELF_MANAGE) {
             tableWithPartKey
                     .put(mtmv.getMvPartitionInfo().getRelatedTable(), mtmv.getMvPartitionInfo().getRelatedCol());
         }

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

@@ -0,0 +1,224 @@
+// 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 org.apache.doris.analysis.Expr;
+import org.apache.doris.analysis.FunctionCallExpr;
+import org.apache.doris.analysis.PartitionExprUtil;
+import org.apache.doris.analysis.PartitionKeyDesc;
+import org.apache.doris.analysis.PartitionValue;
+import org.apache.doris.analysis.StringLiteral;
+import org.apache.doris.catalog.PartitionType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.util.PropertyAnalyzer;
+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;
+import org.apache.doris.nereids.trees.expressions.literal.DateTimeV2Literal;
+import org.apache.doris.nereids.trees.expressions.literal.DateV2Literal;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.VarcharLiteral;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+public class MTMVPartitionExprDateTrunc implements MTMVPartitionExprService {
+    private static Set<String> timeUnits = ImmutableSet.of("year", "month", "day");
+    private String timeUnit;
+
+    public MTMVPartitionExprDateTrunc(FunctionCallExpr functionCallExpr) throws AnalysisException {
+        List<Expr> paramsExprs = functionCallExpr.getParams().exprs();
+        if (paramsExprs.size() != 2) {
+            throw new AnalysisException("date_trunc params exprs size should be 2.");
+        }
+        Expr param = paramsExprs.get(1);
+        if (!(param instanceof StringLiteral)) {
+            throw new AnalysisException("date_trunc param of time unit is not string literal.");
+        }
+        this.timeUnit = param.getStringValue().toLowerCase();
+    }
+
+    @Override
+    public void analyze(MTMVPartitionInfo mvPartitionInfo) throws AnalysisException {
+        if (!timeUnits.contains(this.timeUnit)) {
+            throw new AnalysisException(
+                    String.format("timeUnit not support: %s, only support: %s", this.timeUnit, timeUnits));
+        }
+        MTMVRelatedTableIf relatedTable = mvPartitionInfo.getRelatedTable();
+        PartitionType partitionType = relatedTable.getPartitionType();
+        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");
+            }
+        }
+    }
+
+    @Override
+    public String getRollUpIdentity(PartitionKeyDesc partitionKeyDesc, Map<String, String> mvProperties)
+            throws AnalysisException {
+        String res = null;
+        Optional<String> dateFormat = getDateFormat(mvProperties);
+        List<List<PartitionValue>> inValues = partitionKeyDesc.getInValues();
+        for (int i = 0; i < inValues.size(); i++) {
+            // mtmv only support one partition column
+            PartitionValue partitionValue = inValues.get(i).get(0);
+            if (partitionValue.isNullPartition()) {
+                throw new AnalysisException("date trunc not support null partition value");
+            }
+            String identity = dateTrunc(partitionValue.getStringValue(), dateFormat, false).toString();
+            if (i == 0) {
+                res = identity;
+            } else {
+                if (!Objects.equals(res, identity)) {
+                    throw new AnalysisException(
+                            String.format("partition values not equal, res: %s, identity: %s", res,
+                                    identity));
+                }
+            }
+        }
+        return res;
+    }
+
+    private Optional<String> getDateFormat(Map<String, String> mvProperties) {
+        Optional<String> dateFormat =
+                StringUtils.isEmpty(mvProperties.get(PropertyAnalyzer.PROPERTIES_PARTITION_DATE_FORMAT))
+                        ? Optional.empty()
+                        : Optional.of(mvProperties.get(PropertyAnalyzer.PROPERTIES_PARTITION_DATE_FORMAT));
+        return dateFormat;
+    }
+
+    @Override
+    public PartitionKeyDesc generateRollUpPartitionKeyDesc(PartitionKeyDesc partitionKeyDesc,
+            MTMVPartitionInfo mvPartitionInfo) throws AnalysisException {
+        Type partitionColumnType = MTMVPartitionUtil
+                .getPartitionColumnType(mvPartitionInfo.getRelatedTable(), mvPartitionInfo.getRelatedCol());
+        // mtmv only support one partition column
+        DateTimeV2Literal beginTime = dateTrunc(
+                partitionKeyDesc.getLowerValues().get(0).getStringValue(),
+                Optional.empty(), false);
+
+        PartitionValue lowerValue = new PartitionValue(timeToStr(beginTime, partitionColumnType));
+        return PartitionKeyDesc.createFixed(
+                Collections.singletonList(lowerValue),
+                Collections.singletonList(
+                        getUpperValue(partitionKeyDesc.getUpperValues().get(0), beginTime, partitionColumnType)));
+    }
+
+    private PartitionValue getUpperValue(PartitionValue upperValue, DateTimeV2Literal beginTruncTime,
+            Type partitionColumnType) throws AnalysisException {
+        if (upperValue.isMax()) {
+            throw new AnalysisException("date trunc not support MAXVALUE partition");
+        }
+        // begin time and end time dateTrunc should has same result
+        DateTimeV2Literal endTruncTime = dateTrunc(upperValue.getStringValue(), Optional.empty(), true);
+        if (!Objects.equals(beginTruncTime, endTruncTime)) {
+            throw new AnalysisException(
+                    String.format("partition values not equal, beginTruncTime: %s, endTruncTime: %s", beginTruncTime,
+                            endTruncTime));
+        }
+        DateTimeV2Literal endTime = dateAdd(beginTruncTime);
+        return new PartitionValue(timeToStr(endTime, partitionColumnType));
+    }
+
+    private DateTimeV2Literal dateTrunc(String value,
+            Optional<String> dateFormat, boolean isUpper) throws AnalysisException {
+        DateTimeV2Literal dateTimeLiteral = strToDate(value, dateFormat);
+        // for (2020-01-31,2020-02-01),if not -1, lower value and upper value will not same after rollup
+        if (isUpper) {
+            dateTimeLiteral = (DateTimeV2Literal) DateTimeArithmetic.secondsSub(dateTimeLiteral, new IntegerLiteral(1));
+        }
+        Expression expression = DateTimeExtractAndTransform.dateTrunc(dateTimeLiteral, new VarcharLiteral(timeUnit));
+        if (!(expression instanceof DateTimeV2Literal)) {
+            throw new AnalysisException("dateTrunc() should return DateLiteral, expression: " + expression);
+        }
+        return (DateTimeV2Literal) expression;
+    }
+
+    private DateTimeV2Literal strToDate(String value,
+            Optional<String> dateFormat) throws AnalysisException {
+        try {
+            return new DateTimeV2Literal(value);
+        } catch (Exception e) {
+            if (!dateFormat.isPresent()) {
+                throw e;
+            }
+            Expression strToDate = DateTimeExtractAndTransform
+                    .strToDate(new VarcharLiteral(value),
+                            new VarcharLiteral(dateFormat.get()));
+            if (strToDate instanceof DateV2Literal) {
+                DateV2Literal dateV2Literal = (DateV2Literal) strToDate;
+                return new DateTimeV2Literal(dateV2Literal.getYear(), dateV2Literal.getMonth(), dateV2Literal.getDay(),
+                        0, 0, 0);
+            } else if (strToDate instanceof DateTimeV2Literal) {
+                return (DateTimeV2Literal) strToDate;
+            } else {
+                throw new AnalysisException(
+                        String.format("strToDate failed, stringValue: %s, dateFormat: %s", value,
+                                dateFormat));
+            }
+        }
+    }
+
+    private DateTimeV2Literal dateAdd(DateTimeV2Literal value) throws AnalysisException {
+        Expression result;
+        switch (timeUnit) {
+            case "year":
+                result = value.plusYears(1L);
+                break;
+            case "month":
+                result = value.plusMonths(1L);
+                break;
+            case "day":
+                result = value.plusDays(1L);
+                break;
+            default:
+                throw new AnalysisException("MTMV partition roll up not support timeUnit: " + timeUnit);
+        }
+        if (!(result instanceof DateTimeV2Literal)) {
+            throw new AnalysisException("sub() should return  DateTimeLiteral, result: " + result);
+        }
+        return (DateTimeV2Literal) result;
+    }
+
+    private String timeToStr(DateTimeV2Literal literal,
+            Type partitionColumnType) throws AnalysisException {
+        if (partitionColumnType.isDate() || partitionColumnType.isDateV2()) {
+            return String.format(PartitionExprUtil.DATE_FORMATTER, literal.getYear(), literal.getMonth(),
+                    literal.getDay());
+        } else if (partitionColumnType.isDatetime() || partitionColumnType.isDatetimeV2()) {
+            return String.format(PartitionExprUtil.DATETIME_FORMATTER,
+                    literal.getYear(), literal.getMonth(), literal.getDay(),
+                    literal.getHour(), literal.getMinute(), literal.getSecond());
+        } else {
+            throw new AnalysisException(
+                    "MTMV swnot support partition with column type : " + partitionColumnType.toString());
+        }
+    }
+}

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

@@ -0,0 +1,39 @@
+// 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 org.apache.doris.analysis.Expr;
+import org.apache.doris.analysis.FunctionCallExpr;
+import org.apache.doris.common.AnalysisException;
+
+/**
+ * MTMV Partition Expr Factory
+ */
+public class MTMVPartitionExprFactory {
+    public static MTMVPartitionExprService getExprSerice(Expr expr) throws AnalysisException {
+        if (!(expr instanceof FunctionCallExpr)) {
+            throw new AnalysisException("now mtmv partition only support FunctionCallExpr");
+        }
+        FunctionCallExpr functionCallExpr = (FunctionCallExpr) expr;
+        String fnName = functionCallExpr.getFnName().getFunction().toLowerCase();
+        if ("date_trunc".equals(fnName)) {
+            return new MTMVPartitionExprDateTrunc(functionCallExpr);
+        }
+        throw new AnalysisException("MTMV partition not support function name: " + fnName);
+    }
+}