[enhance](mtmv)MTMV allow paimon table has multi partition keys

docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/paimon/run01.sql

@@ -22,4 +22,37 @@ insert into test_tb_mix_format values (1,1,'b'),(2,1,'b'),(3,1,'b'),(4,1,'b'),(5
 -- update some data, these splits will be readed by jni
 insert into test_tb_mix_format values (1,2,'b'),(2,2,'b'),(3,2,'b'),(4,2,'b'),(5,2,'b');
 -- delete foramt in table properties, doris should get format by file name
-alter table test_tb_mix_format unset TBLPROPERTIES ('file.format');
+alter table test_tb_mix_format unset TBLPROPERTIES ('file.format');
+
+drop table if exists two_partition;
+CREATE TABLE two_partition (
+   id BIGINT,
+   create_date STRING,
+   region STRING
+) PARTITIONED BY (create_date,region) TBLPROPERTIES (
+    'primary-key' = 'create_date,region,id',
+    'bucket'=10,
+    'file.format'='orc'
+);
+
+insert into two_partition values(1,'2020-01-01','bj');
+insert into two_partition values(2,'2020-01-01','sh');
+insert into two_partition values(3,'2038-01-01','bj');
+insert into two_partition values(4,'2038-01-01','sh');
+insert into two_partition values(5,'2038-01-02','bj');
+
+drop table if exists null_partition;
+CREATE TABLE null_partition (
+   id BIGINT,
+   region STRING
+) PARTITIONED BY (region) TBLPROPERTIES (
+    'primary-key' = 'region,id',
+    'bucket'=10,
+    'file.format'='orc'
+);
+-- null NULL "null" all will be in partition [null]
+insert into null_partition values(1,'bj');
+insert into null_partition values(2,null);
+insert into null_partition values(3,NULL);
+insert into null_partition values(4,'null');
+insert into null_partition values(5,'NULL');

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

@@ -25,8 +25,8 @@
 import org.apache.doris.analysis.FunctionParams;
 import org.apache.doris.analysis.SlotRef;
 import org.apache.doris.analysis.StringLiteral;
+import org.apache.doris.catalog.PartitionType;
 import org.apache.doris.common.DdlException;
-import org.apache.doris.datasource.hive.HMSExternalTable;
 import org.apache.doris.mtmv.MTMVPartitionExprFactory;
 import org.apache.doris.mtmv.MTMVPartitionInfo;
 import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
@@ -136,9 +136,9 @@ private RelatedTableInfo getRelatedTableInfo(NereidsPlanner planner, String part
         if (!partitionColumnNames.contains(relatedTableInfo.getColumn())) {
             throw new AnalysisException("error related column: " + relatedTableInfo.getColumn());
         }
-        if (!(mtmvBaseRelatedTable instanceof HMSExternalTable)
+        if (!(mtmvBaseRelatedTable.getPartitionType(Optional.empty()).equals(PartitionType.LIST))
                 && partitionColumnNames.size() != 1) {
-            throw new AnalysisException("only hms table support multi column partition.");
+            throw new AnalysisException("only List PartitionType support multi column partition.");
         }
         return relatedTableInfo;
     }

regression-test/data/mtmv_p0/test_paimon_mtmv.out

@@ -111,3 +111,32 @@ false
 -- !not_partition_after --
 true
 
+-- !join_one_partition --
+1	2	a	1	2
+10	1	a	\N	\N
+2	2	a	\N	\N
+3	2	a	\N	\N
+4	2	a	\N	\N
+5	2	a	\N	\N
+6	1	a	\N	\N
+7	1	a	\N	\N
+8	1	a	\N	\N
+9	1	a	\N	\N
+
+-- !two_partition --
+1	2020-01-01	bj
+2	2020-01-01	sh
+3	2038-01-01	bj
+4	2038-01-01	sh
+5	2038-01-02	bj
+
+-- !limit_partition --
+3	2038-01-01	bj
+4	2038-01-01	sh
+5	2038-01-02	bj
+
+-- !null_partition --
+1	bj
+4	null
+5	NULL
+

regression-test/data/mtmv_p0/test_paimon_olap_rewrite_mtmv.out

@@ -0,0 +1,79 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !refresh_one_partition --
+1	2	a	1	2
+10	1	a	\N	\N
+2	2	a	\N	\N
+3	2	a	\N	\N
+4	2	a	\N	\N
+5	2	a	\N	\N
+6	1	a	\N	\N
+7	1	a	\N	\N
+8	1	a	\N	\N
+9	1	a	\N	\N
+
+-- !refresh_one_partition_rewrite --
+1	2	a	1	2
+1	2	b	1	2
+10	1	a	\N	\N
+10	1	b	\N	\N
+2	2	a	\N	\N
+2	2	b	\N	\N
+3	2	a	\N	\N
+3	2	b	\N	\N
+4	2	a	\N	\N
+4	2	b	\N	\N
+5	2	a	\N	\N
+5	2	b	\N	\N
+6	1	a	\N	\N
+6	1	b	\N	\N
+7	1	a	\N	\N
+7	1	b	\N	\N
+8	1	a	\N	\N
+8	1	b	\N	\N
+9	1	a	\N	\N
+9	1	b	\N	\N
+
+-- !refresh_auto --
+1	2	a	1	2
+1	2	b	1	2
+10	1	a	\N	\N
+10	1	b	\N	\N
+2	2	a	\N	\N
+2	2	b	\N	\N
+3	2	a	\N	\N
+3	2	b	\N	\N
+4	2	a	\N	\N
+4	2	b	\N	\N
+5	2	a	\N	\N
+5	2	b	\N	\N
+6	1	a	\N	\N
+6	1	b	\N	\N
+7	1	a	\N	\N
+7	1	b	\N	\N
+8	1	a	\N	\N
+8	1	b	\N	\N
+9	1	a	\N	\N
+9	1	b	\N	\N
+
+-- !refresh_all_partition_rewrite --
+1	2	a	1	2
+1	2	b	1	2
+10	1	a	\N	\N
+10	1	b	\N	\N
+2	2	a	\N	\N
+2	2	b	\N	\N
+3	2	a	\N	\N
+3	2	b	\N	\N
+4	2	a	\N	\N
+4	2	b	\N	\N
+5	2	a	\N	\N
+5	2	b	\N	\N
+6	1	a	\N	\N
+6	1	b	\N	\N
+7	1	a	\N	\N
+7	1	b	\N	\N
+8	1	a	\N	\N
+8	1	b	\N	\N
+9	1	a	\N	\N
+9	1	b	\N	\N
+

regression-test/suites/mtmv_p0/test_paimon_mtmv.groovy

@@ -25,6 +25,24 @@ suite("test_paimon_mtmv", "p0,external,mtmv,external_docker,external_docker_dori
     String catalogName = "${suiteName}_catalog"
     String mvName = "${suiteName}_mv"
     String dbName = context.config.getDbNameByFile(context.file)
+    String otherDbName = "${suiteName}_otherdb"
+    String tableName = "${suiteName}_table"
+
+    sql """drop database if exists ${otherDbName}"""
+    sql """create database ${otherDbName}"""
+     sql """
+        CREATE TABLE  ${otherDbName}.${tableName} (
+          `user_id` INT,
+          `num` INT
+        ) ENGINE=OLAP
+        DUPLICATE KEY(`user_id`)
+        DISTRIBUTED BY HASH(`user_id`) BUCKETS 2
+        PROPERTIES ('replication_num' = '1') ;
+       """
+
+    sql """
+        insert into ${otherDbName}.${tableName} values(1,2);
+        """
 
     String minio_port = context.config.otherConfigs.get("iceberg_minio_port")
     String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
@@ -99,8 +117,10 @@ suite("test_paimon_mtmv", "p0,external,mtmv,external_docker,external_docker_dori
      sql """
         CREATE MATERIALIZED VIEW ${mvName}
             BUILD DEFERRED REFRESH AUTO ON MANUAL
-            DISTRIBUTED BY RANDOM BUCKETS 2
-            PROPERTIES ('replication_num' = '1')
+            KEY(`id`)
+            COMMENT "comment1"
+            DISTRIBUTED BY HASH(`id`) BUCKETS 2
+            PROPERTIES ('replication_num' = '1',"grace_period"="333")
             AS
             SELECT * FROM ${catalogName}.`test_paimon_spark`.test_tb_mix_format;
         """
@@ -113,6 +133,137 @@ suite("test_paimon_mtmv", "p0,external,mtmv,external_docker,external_docker_dori
     order_qt_not_partition "SELECT * FROM ${mvName} "
     order_qt_not_partition_after "select SyncWithBaseTables from mv_infos('database'='${dbName}') where Name='${mvName}'"
     sql """drop materialized view if exists ${mvName};"""
+
+    // refresh on schedule
+     sql """
+     CREATE MATERIALIZED VIEW ${mvName}
+        BUILD IMMEDIATE REFRESH COMPLETE ON SCHEDULE EVERY 10 SECOND STARTS "9999-12-13 21:07:09"
+        KEY(`id`)
+        COMMENT "comment1"
+        DISTRIBUTED BY HASH(`id`) BUCKETS 2
+        PROPERTIES ('replication_num' = '1',"grace_period"="333")
+        AS
+        SELECT * FROM ${catalogName}.`test_paimon_spark`.test_tb_mix_format;
+    """
+     waitingMTMVTaskFinishedByMvName(mvName)
+     sql """drop materialized view if exists ${mvName};"""
+
+    // refresh on schedule
+     sql """
+     CREATE MATERIALIZED VIEW ${mvName}
+        BUILD IMMEDIATE REFRESH AUTO ON commit
+        KEY(`id`)
+        COMMENT "comment1"
+        DISTRIBUTED BY HASH(`id`) BUCKETS 2
+        PROPERTIES ('replication_num' = '1',"grace_period"="333")
+        AS
+        SELECT * FROM ${catalogName}.`test_paimon_spark`.test_tb_mix_format;
+    """
+    waitingMTMVTaskFinishedByMvName(mvName)
+     sql """drop materialized view if exists ${mvName};"""
+
+    // cross db and join internal table
+    sql """
+        CREATE MATERIALIZED VIEW ${mvName}
+            BUILD DEFERRED REFRESH AUTO ON MANUAL
+            partition by(`par`)
+            DISTRIBUTED BY RANDOM BUCKETS 2
+            PROPERTIES ('replication_num' = '1')
+            AS
+            SELECT * FROM ${catalogName}.`test_paimon_spark`.test_tb_mix_format a left join internal.${otherDbName}.${tableName} b on a.id=b.user_id;
+        """
+    def showJoinPartitionsResult = sql """show partitions from ${mvName}"""
+    logger.info("showJoinPartitionsResult: " + showJoinPartitionsResult.toString())
+    assertTrue(showJoinPartitionsResult.toString().contains("p_a"))
+    assertTrue(showJoinPartitionsResult.toString().contains("p_b"))
+
+    sql """
+            REFRESH MATERIALIZED VIEW ${mvName} partitions(p_a);
+        """
+    waitingMTMVTaskFinishedByMvName(mvName)
+    order_qt_join_one_partition "SELECT * FROM ${mvName} "
+    sql """drop materialized view if exists ${mvName};"""
+
+    sql """
+        CREATE MATERIALIZED VIEW ${mvName}
+            BUILD DEFERRED REFRESH AUTO ON MANUAL
+            partition by(`create_date`)
+            DISTRIBUTED BY RANDOM BUCKETS 2
+            PROPERTIES ('replication_num' = '1')
+            AS
+            SELECT * FROM ${catalogName}.`test_paimon_spark`.two_partition;
+        """
+    def showTwoPartitionsResult = sql """show partitions from ${mvName}"""
+    logger.info("showTwoPartitionsResult: " + showTwoPartitionsResult.toString())
+    assertTrue(showTwoPartitionsResult.toString().contains("p_20200101"))
+    assertTrue(showTwoPartitionsResult.toString().contains("p_20380101"))
+    assertTrue(showTwoPartitionsResult.toString().contains("p_20380102"))
+    sql """
+            REFRESH MATERIALIZED VIEW ${mvName} auto;
+        """
+    waitingMTMVTaskFinishedByMvName(mvName)
+    order_qt_two_partition "SELECT * FROM ${mvName} "
+    sql """drop materialized view if exists ${mvName};"""
+
+    sql """
+        CREATE MATERIALIZED VIEW ${mvName}
+            BUILD DEFERRED REFRESH AUTO ON MANUAL
+            partition by(`create_date`)
+            DISTRIBUTED BY RANDOM BUCKETS 2
+            PROPERTIES ('replication_num' = '1','partition_sync_limit'='2','partition_date_format'='%Y-%m-%d',
+                        'partition_sync_time_unit'='MONTH')
+            AS
+            SELECT * FROM ${catalogName}.`test_paimon_spark`.two_partition;
+        """
+    def showLimitPartitionsResult = sql """show partitions from ${mvName}"""
+    logger.info("showLimitPartitionsResult: " + showLimitPartitionsResult.toString())
+    assertFalse(showLimitPartitionsResult.toString().contains("p_20200101"))
+    assertTrue(showLimitPartitionsResult.toString().contains("p_20380101"))
+    assertTrue(showLimitPartitionsResult.toString().contains("p_20380102"))
+    sql """
+            REFRESH MATERIALIZED VIEW ${mvName} auto;
+        """
+    waitingMTMVTaskFinishedByMvName(mvName)
+    order_qt_limit_partition "SELECT * FROM ${mvName} "
+    sql """drop materialized view if exists ${mvName};"""
+
+    // not allow date trunc
+    test {
+         sql """
+            CREATE MATERIALIZED VIEW ${mvName}
+                BUILD DEFERRED REFRESH AUTO ON MANUAL
+                partition by (date_trunc(`create_date`,'month'))
+                DISTRIBUTED BY RANDOM BUCKETS 2
+                PROPERTIES ('replication_num' = '1','partition_sync_limit'='2','partition_date_format'='%Y-%m-%d',
+                            'partition_sync_time_unit'='MONTH')
+                AS
+                SELECT * FROM ${catalogName}.`test_paimon_spark`.two_partition;
+            """
+          exception "only support"
+      }
+
+    sql """
+        CREATE MATERIALIZED VIEW ${mvName}
+            BUILD DEFERRED REFRESH AUTO ON MANUAL
+            partition by(`region`)
+            DISTRIBUTED BY RANDOM BUCKETS 2
+            PROPERTIES ('replication_num' = '1')
+            AS
+            SELECT * FROM ${catalogName}.`test_paimon_spark`.null_partition;
+        """
+    def showNullPartitionsResult = sql """show partitions from ${mvName}"""
+    logger.info("showNullPartitionsResult: " + showNullPartitionsResult.toString())
+    assertTrue(showNullPartitionsResult.toString().contains("p_null"))
+    assertTrue(showNullPartitionsResult.toString().contains("p_NULL"))
+    assertTrue(showNullPartitionsResult.toString().contains("p_bj"))
+    sql """
+            REFRESH MATERIALIZED VIEW ${mvName} auto;
+        """
+    waitingMTMVTaskFinishedByMvName(mvName)
+    // Will lose null data
+    order_qt_null_partition "SELECT * FROM ${mvName} "
+    sql """drop materialized view if exists ${mvName};"""
+
     sql """drop catalog if exists ${catalogName}"""
 
 }