[enhance](catalog)Unified external partition prune interface

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

@@ -20,6 +20,7 @@
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.PartitionItem;
 import org.apache.doris.catalog.TableAttributes;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.TableIndexes;
@@ -30,6 +31,7 @@
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.common.util.PropertyAnalyzer;
 import org.apache.doris.common.util.Util;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions;
 import org.apache.doris.persist.gson.GsonPostProcessable;
 import org.apache.doris.persist.gson.GsonUtils;
 import org.apache.doris.statistics.AnalysisInfo;
@@ -41,16 +43,19 @@
 import com.google.common.collect.Sets;
 import com.google.gson.annotations.SerializedName;
 import lombok.Getter;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.NotImplementedException;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.Set;
 
 /**
@@ -364,4 +369,52 @@ protected Optional<SchemaCacheValue> getSchemaCacheValue() {
     public TableIndexes getTableIndexes() {
         return new TableIndexes();
     }
+
+    /**
+     * Retrieve all partitions and initialize SelectedPartitions
+     *
+     * @param snapshotId if not support mvcc, ignore this
+     * @return
+     */
+    public SelectedPartitions initSelectedPartitions(OptionalLong snapshotId) {
+        if (!supportPartitionPruned()) {
+            return SelectedPartitions.NOT_PRUNED;
+        }
+        if (CollectionUtils.isEmpty(this.getPartitionColumns(snapshotId))) {
+            return SelectedPartitions.NOT_PRUNED;
+        }
+        Map<String, PartitionItem> nameToPartitionItems = getNameToPartitionItems(snapshotId);
+        return new SelectedPartitions(nameToPartitionItems.size(), nameToPartitionItems, false);
+    }
+
+    /**
+     * get partition map
+     * If partition related operations are supported, this method needs to be implemented in the subclass
+     *
+     * @param snapshotId if not support mvcc, ignore this
+     * @return partitionName ==> PartitionItem
+     */
+    public Map<String, PartitionItem> getNameToPartitionItems(OptionalLong snapshotId) {
+        return Collections.emptyMap();
+    }
+
+    /**
+     * get partition column list
+     * If partition related operations are supported, this method needs to be implemented in the subclass
+     *
+     * @param snapshotId if not support mvcc, ignore this
+     * @return
+     */
+    public List<Column> getPartitionColumns(OptionalLong snapshotId) {
+        return Collections.emptyList();
+    }
+
+    /**
+     * Does it support partition cpruned， If so, this method needs to be overridden in subclasses
+     *
+     * @return
+     */
+    public boolean supportPartitionPruned() {
+        return false;
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java

@@ -40,7 +40,6 @@
 import org.apache.doris.mtmv.MTMVSnapshotIf;
 import org.apache.doris.mtmv.MTMVTimestampSnapshot;
 import org.apache.doris.nereids.exceptions.NotSupportedException;
-import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions;
 import org.apache.doris.qe.GlobalVariable;
 import org.apache.doris.statistics.AnalysisInfo;
 import org.apache.doris.statistics.BaseAnalysisTask;
@@ -297,11 +296,25 @@ public List<Column> getPartitionColumns() {
                 .orElse(Collections.emptyList());
     }
 
-    public SelectedPartitions getAllPartitions() {
+    @Override
+    public List<Column> getPartitionColumns(OptionalLong snapshotId) {
+        return getPartitionColumns();
+    }
+
+    @Override
+    public boolean supportPartitionPruned() {
+        return getDlaType() == DLAType.HIVE;
+    }
+
+    @Override
+    public Map<String, PartitionItem> getNameToPartitionItems(OptionalLong snapshotId) {
+        return getNameToPartitionItems();
+    }
+
+    public Map<String, PartitionItem> getNameToPartitionItems() {
         if (CollectionUtils.isEmpty(this.getPartitionColumns())) {
-            return SelectedPartitions.NOT_PRUNED;
+            return Collections.emptyMap();
         }
-
         HiveMetaStoreCache cache = Env.getCurrentEnv().getExtMetaCacheMgr()
                 .getMetaStoreCache((HMSExternalCatalog) this.getCatalog());
         List<Type> partitionColumnTypes = this.getPartitionColumnTypes();
@@ -314,7 +327,7 @@ public SelectedPartitions getAllPartitions() {
         for (Entry<Long, PartitionItem> entry : idToPartitionItem.entrySet()) {
             nameToPartitionItem.put(idToName.get(entry.getKey()), entry.getValue());
         }
-        return new SelectedPartitions(idToPartitionItem.size(), nameToPartitionItem, false);
+        return nameToPartitionItem;
     }
 
     public boolean isHiveTransactionalTable() {

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java

@@ -435,7 +435,6 @@ private LogicalPlan getLogicalPlan(TableIf table, UnboundRelation unboundRelatio
                     } else {
                         return new LogicalFileScan(unboundRelation.getRelationId(), (HMSExternalTable) table,
                                 qualifierWithoutTableName,
-                                ((HMSExternalTable) table).getAllPartitions(),
                                 unboundRelation.getTableSample(),
                                 unboundRelation.getTableSnapshot());
                     }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneFileScanPartition.java

@@ -19,8 +19,6 @@
 
 import org.apache.doris.catalog.PartitionItem;
 import org.apache.doris.datasource.ExternalTable;
-import org.apache.doris.datasource.hive.HMSExternalTable;
-import org.apache.doris.datasource.hive.HMSExternalTable.DLAType;
 import org.apache.doris.nereids.CascadesContext;
 import org.apache.doris.nereids.rules.Rule;
 import org.apache.doris.nereids.rules.RuleType;
@@ -38,6 +36,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.OptionalLong;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -60,10 +59,8 @@ public Rule build() {
                     ExternalTable tbl = scan.getTable();
 
                     SelectedPartitions selectedPartitions;
-                    // TODO(cmy): support other external table
-                    if (tbl instanceof HMSExternalTable && ((HMSExternalTable) tbl).getDlaType() == DLAType.HIVE) {
-                        HMSExternalTable hiveTbl = (HMSExternalTable) tbl;
-                        selectedPartitions = pruneHivePartitions(hiveTbl, filter, scan, ctx.cascadesContext);
+                    if (tbl.supportPartitionPruned()) {
+                        selectedPartitions = pruneExternalPartitions(tbl, filter, scan, ctx.cascadesContext);
                     } else {
                         // set isPruned so that it won't go pass the partition prune again
                         selectedPartitions = new SelectedPartitions(0, ImmutableMap.of(), true);
@@ -74,19 +71,20 @@ public Rule build() {
                 }).toRule(RuleType.FILE_SCAN_PARTITION_PRUNE);
     }
 
-    private SelectedPartitions pruneHivePartitions(HMSExternalTable hiveTbl,
+    private SelectedPartitions pruneExternalPartitions(ExternalTable externalTable,
             LogicalFilter<LogicalFileScan> filter, LogicalFileScan scan, CascadesContext ctx) {
         Map<String, PartitionItem> selectedPartitionItems = Maps.newHashMap();
-        if (CollectionUtils.isEmpty(hiveTbl.getPartitionColumns())) {
+        // todo: real snapshotId
+        if (CollectionUtils.isEmpty(externalTable.getPartitionColumns(OptionalLong.empty()))) {
             // non partitioned table, return NOT_PRUNED.
             // non partition table will be handled in HiveScanNode.
             return SelectedPartitions.NOT_PRUNED;
         }
         Map<String, Slot> scanOutput = scan.getOutput()
                 .stream()
                 .collect(Collectors.toMap(slot -> slot.getName().toLowerCase(), Function.identity()));
-
-        List<Slot> partitionSlots = hiveTbl.getPartitionColumns()
+        // todo: real snapshotId
+        List<Slot> partitionSlots = externalTable.getPartitionColumns(OptionalLong.empty())
                 .stream()
                 .map(column -> scanOutput.get(column.getName().toLowerCase()))
                 .collect(Collectors.toList());

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java

@@ -36,6 +36,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.OptionalLong;
 
 /**
  * Logical file scan for external catalog.
@@ -59,17 +60,11 @@ protected LogicalFileScan(RelationId id, ExternalTable table, List<String> quali
         this.tableSnapshot = tableSnapshot;
     }
 
-    public LogicalFileScan(RelationId id, ExternalTable table, List<String> qualifier,
-            SelectedPartitions selectedPartitions,
-            Optional<TableSample> tableSample, Optional<TableSnapshot> tableSnapshot) {
-        this(id, table, qualifier, Optional.empty(), Optional.empty(),
-                selectedPartitions, tableSample, tableSnapshot);
-    }
-
     public LogicalFileScan(RelationId id, ExternalTable table, List<String> qualifier,
                            Optional<TableSample> tableSample, Optional<TableSnapshot> tableSnapshot) {
+        // todo: real snapshotId
         this(id, table, qualifier, Optional.empty(), Optional.empty(),
-                SelectedPartitions.NOT_PRUNED, tableSample, tableSnapshot);
+                table.initSelectedPartitions(OptionalLong.empty()), tableSample, tableSnapshot);
     }
 
     public SelectedPartitions getSelectedPartitions() {