[enhance](mtmv)When obtaining the partition list fails, treat the pai…

…mon table as an unpartitioned table (apache#46641)

### What problem does this PR solve?

When retrieving data of type Paimon Date in version 0.9 from the system
table, the value is an integer and cannot be converted to type Date.

This issue has been fixed in Paimon's latest code.

This PR downgrades this situation without affecting user data queries

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

@@ -55,4 +55,16 @@ 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');
+insert into null_partition values(5,'NULL');
+
+drop table if exists date_partition;
+CREATE TABLE date_partition (
+                               id BIGINT,
+                               create_date DATE
+) PARTITIONED BY (create_date) TBLPROPERTIES (
+    'primary-key' = 'create_date,id',
+    'bucket'=10,
+    'file.format'='orc'
+);
+
+insert into date_partition values(1,date '2020-01-01');

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java

@@ -165,6 +165,9 @@ public Map<String, PartitionItem> getAndCopyPartitionItems(Optional<MvccSnapshot
 
     @Override
     public PartitionType getPartitionType(Optional<MvccSnapshot> snapshot) {
+        if (isPartitionInvalid(snapshot)) {
+            return PartitionType.UNPARTITIONED;
+        }
         return getPartitionColumns(snapshot).size() > 0 ? PartitionType.LIST : PartitionType.UNPARTITIONED;
     }
 
@@ -176,9 +179,17 @@ public Set<String> getPartitionColumnNames(Optional<MvccSnapshot> snapshot) {
 
     @Override
     public List<Column> getPartitionColumns(Optional<MvccSnapshot> snapshot) {
+        if (isPartitionInvalid(snapshot)) {
+            return Collections.emptyList();
+        }
         return getPaimonSchemaCacheValue(snapshot).getPartitionColumns();
     }
 
+    private boolean isPartitionInvalid(Optional<MvccSnapshot> snapshot) {
+        PaimonSnapshotCacheValue paimonSnapshotCacheValue = getOrFetchSnapshotCacheValue(snapshot);
+        return paimonSnapshotCacheValue.getPartitionInfo().isPartitionInvalid();
+    }
+
     @Override
     public MTMVSnapshotIf getPartitionSnapshot(String partitionName, MTMVRefreshContext context,
             Optional<MvccSnapshot> snapshot)

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonPartitionInfo.java

@@ -45,4 +45,9 @@ public Map<String, PartitionItem> getNameToPartitionItem() {
     public Map<String, PaimonPartition> getNameToPartition() {
         return nameToPartition;
     }
+
+    public boolean isPartitionInvalid() {
+        // when transfer to partitionItem failed, will not equal
+        return nameToPartitionItem.size() != nameToPartition.size();
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java

@@ -117,7 +117,7 @@ public static PaimonPartition rowToPartition(InternalRow row) {
     }
 
     public static PaimonPartitionInfo generatePartitionInfo(List<Column> partitionColumns,
-            List<PaimonPartition> paimonPartitions) throws AnalysisException {
+            List<PaimonPartition> paimonPartitions) {
         Map<String, PartitionItem> nameToPartitionItem = Maps.newHashMap();
         Map<String, PaimonPartition> nameToPartition = Maps.newHashMap();
         PaimonPartitionInfo partitionInfo = new PaimonPartitionInfo(nameToPartitionItem, nameToPartition);
@@ -127,7 +127,14 @@ public static PaimonPartitionInfo generatePartitionInfo(List<Column> partitionCo
         for (PaimonPartition paimonPartition : paimonPartitions) {
             String partitionName = getPartitionName(partitionColumns, paimonPartition.getPartitionValues());
             nameToPartition.put(partitionName, paimonPartition);
-            nameToPartitionItem.put(partitionName, toListPartitionItem(partitionName, partitionColumns));
+            try {
+                // partition values return by paimon api, may have problem,
+                // to avoid affecting the query, we catch exceptions here
+                nameToPartitionItem.put(partitionName, toListPartitionItem(partitionName, partitionColumns));
+            } catch (Exception e) {
+                LOG.warn("toListPartitionItem failed, partitionColumns: {}, partitionValues: {}", partitionColumns,
+                        paimonPartition.getPartitionValues(), e);
+            }
         }
         return partitionInfo;
     }

regression-test/data/mtmv_p0/test_paimon_mtmv.out

@@ -140,3 +140,9 @@ true
 4	null
 5	NULL
 
+-- !date_partition_base_table --
+1	2020-01-01
+
+-- !date_partition --
+1	2020-01-01
+

regression-test/suites/mtmv_p0/test_paimon_mtmv.groovy

@@ -264,6 +264,36 @@ suite("test_paimon_mtmv", "p0,external,mtmv,external_docker,external_docker_dori
     order_qt_null_partition "SELECT * FROM ${mvName} "
     sql """drop materialized view if exists ${mvName};"""
 
+    // date type will has problem
+    order_qt_date_partition_base_table "SELECT * FROM ${catalogName}.`test_paimon_spark`.date_partition"
+    test {
+         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`.date_partition;
+            """
+          exception "Unable to find a suitable base table"
+      }
+
+    sql """
+        CREATE MATERIALIZED VIEW ${mvName}
+            BUILD DEFERRED REFRESH AUTO ON MANUAL
+            DISTRIBUTED BY RANDOM BUCKETS 2
+            PROPERTIES ('replication_num' = '1')
+            AS
+            SELECT * FROM ${catalogName}.`test_paimon_spark`.date_partition;
+        """
+    sql """
+         REFRESH MATERIALIZED VIEW ${mvName} auto;
+     """
+    waitingMTMVTaskFinishedByMvName(mvName)
+    order_qt_date_partition "SELECT * FROM ${mvName} "
+
+    sql """drop materialized view if exists ${mvName};"""
     sql """drop catalog if exists ${catalogName}"""
 
 }