branch-4.1: [fix](fe) Stop split enumeration when queries finish or terminate - #70004

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?

branch-4.1 backport validation and performance

Validated head: 5b90e691fce12665de078bb7b944357dfdc2ac72.
Baseline: dd72933ec78a8a4daf3e1d204a6ea96d6443ada7 on branch-4.1, before this backport. Both sides use the same BE 4.1.3 package (doris-4.1.3-rc02-7126cf65d96).

Test conditions

• Hive tables through a local Docker HMS, with real OSS and a dedicated HDFS fixture. Each partition contains 100 small Parquet files. The 100-, 1,000-, and 10,000-partition tables contain 10,000, 100,000, and 1,000,000 files respectively.
• One serial query at a time. Both versions use one file-listing worker and one scheduling worker. Remaining relevant settings use branch-4.1 defaults, including the 1,024-partition threshold; the candidate's producer-parallelism setting stays at its default.
• Same machine, CPU allocation, 4 GiB FE heap, and 6 GiB BE limit on both sides. An initial 3 GiB BE series was discarded after a baseline OOM; the reported comparison was rerun with 6 GiB on both sides.
• Main comparison order: candidate / baseline / baseline / candidate, with two fresh FE processes per version. Each scenario has 20 measured samples per version, following warm-up. All reported times are median SQL wall time in milliseconds.
• "Fresh catalog" means a newly created FE catalog. It does not imply cold OSS service caches or cold OS caches. "Warm" means repeated queries after warm-up in the same catalog. Native listing workers and BE readers were checked for idleness around queries.
• Performance timing runs use normal binaries. Resource/SDK observation is performed separately from timing.

Here t100, t1000, and t10000 are shorthand for the three table sizes. LIMIT SQL is SELECT * FROM tN LIMIT 10; full-scan SQL is SELECT SUM(id) FROM t100; filtered full-scan SQL is SELECT SUM(id) FROM t10000 WHERE p < 100. The two SUM scenarios read the same 100 partition directories / 10,000 physical files, verified through HMS.

Main performance comparison

Change = (after / before - 1) * 100%; a negative value means lower latency.

The 10,000-partition LIMIT improvement reproduced in both execution orders for both storage systems. Smaller LIMIT scenarios did not show a stable improvement or regression across orders. These sizes remain below the existing 1,024-partition threshold, so the results should not be extrapolated as a uniform speedup for every table size.

Additional confirmation of the slower filtered-SUM cases

The main series showed roughly 4% higher medians for filtered SUM. These cases were repeated with 60 measured samples per version per scenario, baseline / candidate / candidate / baseline process order, and alternating full-SUM / filtered-SUM query order. These are separate samples from the main table.

The initial ~4% filtered-SUM slowdown was not consistently reproduced: the sign changed between process orders in the confirmation runs. The intervals include zero. This does not prove zero overhead; the machine is shared, and there are only two FE processes per version in each series. The bootstrap intervals describe the collected query samples and do not capture all process-level or host-level variability.

Across the main and confirmation series, 1,120 timed queries and 368 warm-up queries completed successfully. Pilot queries and the discarded 3 GiB series are outside these totals. A confirmation-run stop-controller failure occurred after that round's queries had succeeded; process exit was checked before resuming the remaining baseline round with the same settings.

Functional and resource acceptance

Acceptance fixture: Hive, local HMS + real OSS, 10,000 partitions x 100 files, common BE 4.1.3.

Example first-Split timeout error:

Timed out waiting for the first split after 1000 ms: partitions_processed=0/10000, files_enumerated=0, splits_generated=0, active_producers=1

The first-Split timer starts when waiting for the first Split; total SQL wall time also includes earlier planning. The two first-wait cases inject a 5-second SDK delay to exercise cleanup while an SDK call is in progress. Their background termination times after the timeout signal were 4,221 ms and 2,783 ms, both below 10 seconds. An in-flight SDK call can finish after cancellation; this result does not imply instantaneous interruption at the stop signal.

The request observer covers the AWS SDK 2.29.52 UrlConnectionHttpClient.RequestCallable dispatch to HttpURLConnection.connect; internal JDK retries are not separately counted. Hive end-to-end acceptance passed 11/11. Iceberg was covered by code review and unit tests; equivalent Iceberg cluster acceptance was not performed in this round.

Other verification

• FE unit tests: 476 passed, zero failures/errors/skips.
• FE build through build.sh --fe: passed.
• Checkstyle and git diff --check: passed.

Benchmark scripts, data, runtime configuration, and build workarounds are excluded from this PR's source changes. No associated prerequisite PRs were separately backported.

run buildall