[improvement](fe) Reuse external scan tasks within a statement

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

@@ -0,0 +1,29 @@
+// 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.datasource;
+
+/**
+ * A statement-scoped cache key for connector-native scan tasks.
+ *
+ * <p>Implementations must include every scan property that can change the planned task list,
+ * such as the table generation, snapshot, projected schema, predicates, and connector options.
+ * The type parameter ties a connector key to its native task type without exposing that type
+ * in StatementContext.
+ */
+public interface ExternalScanTaskCacheKey<T> {
+}

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

@@ -41,6 +41,7 @@
 import org.apache.doris.datasource.mvcc.MvccSnapshot;
 import org.apache.doris.datasource.mvcc.MvccTable;
 import org.apache.doris.datasource.mvcc.MvccUtil;
+import org.apache.doris.nereids.StatementContext;
 import org.apache.doris.planner.PlanNodeId;
 import org.apache.doris.planner.ScanContext;
 import org.apache.doris.qe.ConnectContext;
@@ -85,6 +86,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
@@ -97,6 +99,7 @@ public abstract class FileQueryScanNode extends FileScanNode {
 
     protected Map<String, SlotDescriptor> destSlotDescByName;
     protected TFileScanRangeParams params;
+    private final StatementContext.ExternalScanTaskCache externalScanTaskCache;
 
     @Getter
     protected TableSample tableSample;
@@ -136,6 +139,18 @@ public FileQueryScanNode(PlanNodeId id, TupleDescriptor desc, String planNodeNam
             StatisticalType statisticalType, ScanContext scanContext, boolean needCheckColumnPriv, SessionVariable sv) {
         super(id, desc, planNodeName, statisticalType, scanContext, needCheckColumnPriv);
         this.sessionVariable = sv;
+        ConnectContext context = ConnectContext.get();
+        StatementContext statementContext = context == null ? null : context.getStatementContext();
+        this.externalScanTaskCache = statementContext == null
+                ? null : statementContext.getExternalScanTaskCache();
+    }
+
+    protected <T> List<T> getOrLoadExternalScanTasks(
+            ExternalScanTaskCacheKey<T> key, Callable<List<T>> loader) throws Exception {
+        if (externalScanTaskCache == null) {
+            return loader.call();
+        }
+        return externalScanTaskCache.getOrLoad(key, loader);
     }
 
     /**

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

@@ -29,6 +29,7 @@
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.common.util.Util;
+import org.apache.doris.datasource.ExternalScanTaskCacheKey;
 import org.apache.doris.datasource.FileQueryScanNode;
 import org.apache.doris.datasource.FileSplit;
 import org.apache.doris.datasource.FileSplitter;
@@ -78,6 +79,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
@@ -320,8 +322,17 @@ private void getFileSplitByPartitions(HiveExternalMetaCache cache, List<HivePart
             }
         } else {
             boolean withCache = Config.max_external_file_cache_num > 0;
-            fileCaches = cache.getFilesByPartitions(partitions, withCache, partitions.size() > 1,
-                    directoryLister, hmsTable);
+            HiveFileScanTaskCacheKey cacheKey = new HiveFileScanTaskCacheKey(
+                    hmsTable.getCatalog().getId(), hmsTable.getId(), partitions);
+            try {
+                fileCaches = getOrLoadExternalScanTasks(cacheKey,
+                        () -> cache.getFilesByPartitions(partitions, withCache, partitions.size() > 1,
+                                directoryLister, hmsTable));
+            } catch (IOException | UserException e) {
+                throw e;
+            } catch (Exception e) {
+                throw new IOException("Failed to list Hive files", e);
+            }
         }
         if (!isBatchMode && getSummaryProfile() != null) {
             getSummaryProfile().addExternalTableGetPartitionFilesTime(System.currentTimeMillis() - startTime);
@@ -428,11 +439,10 @@ private List<HiveExternalMetaCache.HiveFileStatus> selectFiles(List<FileCacheVal
         long totalSize = 0;
         for (FileCacheValue value : inputCacheValue) {
             for (HiveExternalMetaCache.HiveFileStatus file : value.getFiles()) {
-                file.setSplittable(value.isSplittable());
-                file.setPartitionValues(value.getPartitionValues());
-                file.setAcidInfo(value.getAcidInfo());
-                fileList.add(file);
-                totalSize += file.getLength();
+                HiveExternalMetaCache.HiveFileStatus sampledFile =
+                        copyFileStatus(file, value);
+                fileList.add(sampledFile);
+                totalSize += sampledFile.getLength();
             }
         }
         long sampleSize = 0;
@@ -458,6 +468,86 @@ private List<HiveExternalMetaCache.HiveFileStatus> selectFiles(List<FileCacheVal
         return fileList.subList(0, index);
     }
 
+    private HiveExternalMetaCache.HiveFileStatus copyFileStatus(
+            HiveExternalMetaCache.HiveFileStatus file, FileCacheValue cacheValue) {
+        HiveExternalMetaCache.HiveFileStatus copy = new HiveExternalMetaCache.HiveFileStatus();
+        copy.setBlockLocations(file.getBlockLocations());
+        copy.setPath(file.getPath());
+        copy.setLength(file.getLength());
+        copy.setBlockSize(file.getBlockSize());
+        copy.setModificationTime(file.getModificationTime());
+        copy.setSplittable(cacheValue.isSplittable());
+        copy.setPartitionValues(cacheValue.getPartitionValues());
+        copy.setAcidInfo(cacheValue.getAcidInfo());
+        return copy;
+    }
+
+    private static final class HiveFileScanTaskCacheKey
+            implements ExternalScanTaskCacheKey<FileCacheValue> {
+        private final long catalogId;
+        private final long tableId;
+        private final List<HivePartitionCacheKey> partitions;
+
+        private HiveFileScanTaskCacheKey(long catalogId, long tableId, List<HivePartition> partitions) {
+            this.catalogId = catalogId;
+            this.tableId = tableId;
+            this.partitions = partitions.stream()
+                    .map(HivePartitionCacheKey::new)
+                    .collect(Collectors.toList());
+        }
+
+        @Override
+        public boolean equals(Object object) {
+            if (this == object) {
+                return true;
+            }
+            if (!(object instanceof HiveFileScanTaskCacheKey)) {
+                return false;
+            }
+            HiveFileScanTaskCacheKey that = (HiveFileScanTaskCacheKey) object;
+            return catalogId == that.catalogId
+                    && tableId == that.tableId
+                    && partitions.equals(that.partitions);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(catalogId, tableId, partitions);
+        }
+    }
+
+    private static final class HivePartitionCacheKey {
+        private final String inputFormat;
+        private final String path;
+        private final List<String> partitionValues;
+
+        private HivePartitionCacheKey(HivePartition partition) {
+            this.inputFormat = partition.getInputFormat();
+            this.path = partition.getPath();
+            this.partitionValues = partition.getPartitionValues() == null
+                    ? null : Collections.unmodifiableList(new ArrayList<>(partition.getPartitionValues()));
+        }
+
+        @Override
+        public boolean equals(Object object) {
+            if (this == object) {
+                return true;
+            }
+            if (!(object instanceof HivePartitionCacheKey)) {
+                return false;
+            }
+            HivePartitionCacheKey that = (HivePartitionCacheKey) object;
+            return Objects.equals(inputFormat, that.inputFormat)
+                    && Objects.equals(path, that.path)
+                    && Objects.equals(partitionValues, that.partitionValues);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(inputFormat, path, partitionValues);
+        }
+    }
+
     private List<FileCacheValue> getFileSplitByTransaction(HiveExternalMetaCache cache, List<HivePartition> partitions,
                                                            String bindBrokerName) {
         for (HivePartition partition : partitions) {