[improvement](datasource) Bound and fairly schedule external scan planning

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2209,6 +2209,15 @@ public class Config extends ConfigBase {
     @ConfField(mutable = false, masterOnly = false)
     public static int max_external_cache_loader_thread_pool_size = 64;
 
+    @ConfField(description = "Workers for fair split production")
+    public static int external_scan_split_threads = 64;
+
+    @ConfField(description = "Workers for independent metadata tasks")
+    public static int external_scan_metadata_threads = 64;
+
+    @ConfField(description = "Maximum queued tasks per query in each external scan scheduler")
+    public static int external_scan_task_queue_capacity = 128;
+
     /**
      * Max cache num of external catalog's file
      * Decrease this value if FE's memory is small

fe/fe-connector/README.md

@@ -120,6 +120,9 @@ tests and embedded use.
 
 ## Reading the API
 
+For the scan task scheduler SPI, admission, cancellation and configuration, see
+[Cooperative external scan scheduling](docs/scan-task-scheduling.md).
+
 This document never lists SPI methods. The truth lives in code, behind four
 mechanisms:
 

fe/fe-connector/docs/scan-task-scheduling.md

@@ -0,0 +1,125 @@
+# Cooperative external scan scheduling
+
+A statement obtains its stage schedulers through
+`ConnectorSession.getTaskScheduler`. The engine owns the task groups and closes
+them with the statement. Connectors submit neutral `Runnable` tasks through this
+SPI; they do not import FE executor classes.
+
+The two stages have independent workers:
+
+- `SPLIT` advances scan producers and assigns ranges to the BE split queues.
+- `METADATA` performs independent operations such as cold Manifest reads.
+
+A producer may need metadata tasks to finish. Keeping those tasks out of the
+producer pool avoids a pool full of parents waiting for children queued behind
+them. A task must not block waiting for children submitted to its own executor.
+
+## Admission and continuation
+
+Each executor has a concurrent FIFO ready queue containing task groups. A
+worker claims a task from the next group, republishes that group if it still
+has pending work, and then executes the task outside queue synchronization.
+This allows several workers to serve one query and gives other ready queries
+an opportunity at subsequent task boundaries. It does not preempt a running
+task or guarantee equal CPU time.
+
+`tryExecute` never waits for space. A full group queue registers the supplied
+`onSpace` callback and returns `false`; the caller retains the task until it
+can retry. Reuse the same callback identity and remove it when the task is
+cancelled or no longer needs admission. Failures are delivered through the
+failure callback rather than silently dropping work.
+
+A serial producer can use `tryExecuteContinuation` for bounded further work.
+The group still returns behind other ready groups, but a worker need not wake
+another worker just to execute its own serial continuation. Initial tasks,
+external admission and recovery from backpressure retain the normal wakeup
+protocol. Split production yields after at most 16 source batches or after
+observing 1 ms elapsed between batches. One blocking batch can exceed that
+interval; the bound is cooperative.
+
+## Iceberg Manifest reads
+
+Cold Manifest prefetch uses a bounded window and a statement-wide read budget.
+Within that window, completed reads can be consumed before a slower earlier
+Manifest. First-split production does not wait for speculative prefetch when
+one Manifest already supplies output; subsequent BE demand enables further
+prefetch. Filtered or rewrite-scoped scans must still advance while no output
+has been found.
+
+Closing a scan stops new submissions and wakes completion waiters. Load failures
+propagate through the corresponding future. A running shared cache load is not
+interrupted just because one waiting scan stops. Authentication and plugin
+thread-context class loading remain scoped around each actual metadata read.
+A scheduled reader that finds another query already loading the same Manifest
+observes that shared completion and returns its metadata worker immediately.
+The completion callback only publishes the bounded window entry; it does not
+read another Manifest or run split production on the loading query's thread.
+Returned asynchronous tasks remain owned by their task group until terminal
+completion, so closing the group still wakes the waiting producer. Closing a
+scan detaches its observer without cancelling the shared cache load. Cache
+publication continues to use the original scope/key generation fence.
+Cached partition values stay immutable: adapt a file task to the pinned scan
+schema at the scan boundary. Select the format from each file range rather
+than planning files merely to infer a table-wide format.
+
+The captured scheduler exposes its closed state even after query cleanup removes
+the statement registry. A Manifest reader waiting for another scan's read permit
+checks this state at most 100 ms apart and rechecks after admission. Cancellation
+therefore stops that waiter without releasing another scan's permit or starting
+another remote read. This does not interrupt an already running shared cache load.
+Hive checks the same captured state between directory requests, including when
+empty partitions are skipped within one iterator call.
+
+## Configuration and limits
+
+| Setting | Default | Meaning |
+|---|---:|---|
+| FE `external_scan_split_threads` | 64 | Worker threads in the producer scheduler |
+| FE `external_scan_metadata_threads` | 64 | Worker threads in the metadata scheduler |
+| FE `external_scan_task_queue_capacity` | 128 | Pending tasks per query in each stage |
+| Session `iceberg_manifest_load_concurrency` | 8 | Manifest read budget per query and window bound per scan; 1 uses sequential loading |
+
+These task queues are distinct from `split_queue_capacity`, which bounds the
+produced Split buffer. Queue capacity does not reserve that many running
+threads. Worker counts limit the respective pools, and the query read budget
+also covers caller-assisted Manifest loading.
+
+Workers are ordinary threads. Synchronous remote calls occupy workers until
+they return; FIFO admission cannot recover a worker blocked inside an SDK.
+Effective remote timeouts and dependency fault isolation are separate concerns.
+
+Planning-only attempts also run their query-finish callbacks when statement
+execution returns. This releases batch Split producers and metadata leases for
+EXPLAIN and planning failures even when no coordinator was registered. Normal
+coordinator unregistration remains idempotent. Deferred Arrow Flight queries keep
+those callbacks until their retained executor finalizes the query after consumption.
+
+Backend assignment has its own policy-state lock. First-split readiness and stop
+notifications use a separate short-held lock, so an already converted range can
+wake the planner while another lane is still assigning its batch. The policy's
+mutable load-balancing state remains serialized.
+
+The engine captures task-group cancellation on the statement executor before
+asynchronous work starts. KILL and timeout close those groups even before a
+coordinator exists; a group captured after cancellation is immediately closed.
+Cancellation listeners run outside their registry operation and are claimed once.
+This stops queued work and wakes future waiters without interrupting a running
+shared cache loader. Normal completion retains the separate read-lease cleanup.
+
+### Consumer completion notification
+
+Each backend has a bounded split queue and its own data-ready wait channel. A consumer registers before checking queue contents and terminal state, then waits only when neither is available. Successful enqueue, producer completion, failure and cancellation signal that channel. The empty queue is not periodically polled to discover EOF: otherwise finishing just after a poll begins adds 100 ms despite completed production. Producers avoid the channel monitor when no consumer waits; callbacks that resume producers run after releasing the consumer channel monitor. Cancellation signals consumers before closing SDK resources.
+
+### Initial delivery and producer lookahead
+
+`split_initial_batch_size` (default 64) bounds the first Split fetch per backend, also capped by the BE request size. Subsequent fetches use the BE request size. The bound is applied before compatible file ranges are compacted into RPC envelopes. It is independent of `split_batch_size` (producer batch size) and `split_queue_capacity` (buffered batches per backend). Producers continue listing files and preparing splits before BE consumption, yielding only for task fairness and configured queue backpressure. A small initial RPC therefore does not suspend useful directory prefetch while BE starts.
+
+Actual Manifest readers retain a lease on the exact read-table generation until the callable exits. Query cleanup releases the statement owner, while an already running shared reader keeps its FileIO alive. Future cancellation is not evidence that the callable has finished.
+
+## Batch mode switch
+
+`enable_external_table_batch_mode` governs both partition-batch and file-streaming
+planning. The engine checks it before calling provider capability or file-count
+estimation methods. Setting it to `false` selects synchronous split planning for
+connectors that otherwise support batch mode. Connectors without a batch or
+streaming implementation keep their synchronous path regardless of this switch.

fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java

@@ -19,6 +19,7 @@
 
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.CompletionStage;
 import java.util.function.BiConsumer;
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
@@ -51,6 +52,16 @@ public V get(K key, Function<K, V> loader) {
         return delegate.get(nonNullKey, definition.scope(nonNullKey), loader);
     }
 
+    /**
+     * Loads on the caller when elected, or observes an existing load without blocking this caller.
+     * Cancelling or completing a future obtained from the stage does not affect the shared load.
+     * Callbacks run on the completing thread and must not perform blocking work.
+     */
+    public CompletionStage<V> getOrLoadFuture(K key, Function<K, V> loader) {
+        K nonNullKey = Objects.requireNonNull(key, "key can not be null");
+        return delegate.getOrLoadFuture(nonNullKey, definition.scope(nonNullKey), loader);
+    }
+
     /**
      * Loads a missing value while allowing an adapter to publish an auxiliary side effect immediately before the
      * cache value at the same guarded commit point. The coordinator must invoke the supplied commit callback once;

fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java

@@ -43,6 +43,7 @@
 import java.util.TreeMap;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executor;
@@ -181,6 +182,23 @@ public V get(K key, ScopePath path, Function<K, V> loader) {
 
     public V getWithPublicationAction(K key, ScopePath path, Function<K, V> loader,
             BiConsumer<V, Consumer<Runnable>> publicationCoordinator) {
+        return getWithPublicationAction(key, path, loader, publicationCoordinator,
+                Function.identity(), this::awaitLoad);
+    }
+
+    /**
+     * Runs an elected loader on the caller, but returns without waiting when another caller owns the load.
+     * The returned stage cannot complete or cancel the shared load. Completion callbacks must stay short:
+     * they may execute on the elected loader before its publication lease is released.
+     */
+    public CompletionStage<V> getOrLoadFuture(K key, ScopePath path, Function<K, V> loader) {
+        return getWithPublicationAction(key, path, loader, (loaded, commit) -> commit.accept(NO_OP),
+                CompletableFuture::completedFuture, CompletableFuture::minimalCompletionStage);
+    }
+
+    private <R> R getWithPublicationAction(K key, ScopePath path, Function<K, V> loader,
+            BiConsumer<V, Consumer<Runnable>> publicationCoordinator, Function<V, R> loadedResult,
+            Function<CompletableFuture<V>, R> sharedResult) {
         Objects.requireNonNull(key, "key can not be null");
         Objects.requireNonNull(path, "path can not be null");
         Function<K, V> loadFunction = Objects.requireNonNull(loader, "loader can not be null");
@@ -206,22 +224,22 @@ public V getWithPublicationAction(K key, ScopePath path, Function<K, V> loader,
                     notifyDiscarded(key, loaded);
                 }
             }
-            return loaded;
+            return loadedResult.apply(loaded);
         }
 
         VersionedValue<K, V> presentVersioned = currentVersionedValue(key, path);
         if (presentVersioned != null) {
             recordAccess(true);
             scheduleRefresh(key, path, loader, presentVersioned);
-            return presentVersioned.value;
+            return loadedResult.apply(presentVersioned.value);
         }
         recordAccess(false);
         try (PublicationLease<K, V> lease = acquirePublicationLease(key, path, true)) {
             LoadAddress<K> loadAddress = new LoadAddress<>(key, path, lease);
             CompletableFuture<V> ownLoad = new CompletableFuture<>();
             CompletableFuture<V> existingLoad = inFlightLoads.putIfAbsent(loadAddress, ownLoad);
             if (existingLoad != null) {
-                return awaitLoad(existingLoad);
+                return sharedResult.apply(existingLoad);
             }
             try {
                 afterLoadElection.run();
@@ -237,7 +255,7 @@ public V getWithPublicationAction(K key, ScopePath path, Function<K, V> loader,
                 if (electedValuePresent) {
                     VersionedValue<K, V> present = electedValue.get();
                     ownLoad.complete(present.value);
-                    return present.value;
+                    return loadedResult.apply(present.value);
                 }
                 V loaded = loadAndRecord(key, loadFunction);
                 if (loaded != null) {
@@ -261,7 +279,7 @@ public V getWithPublicationAction(K key, ScopePath path, Function<K, V> loader,
                     }
                 }
                 ownLoad.complete(loaded);
-                return loaded;
+                return loadedResult.apply(loaded);
             } catch (RuntimeException | Error throwable) {
                 ownLoad.completeExceptionally(throwable);
                 throw throwable;