This is an automated email from the ASF dual-hosted git repository.
HappenLee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new fb19e0da0c0 [fix](be) Finish pending scanners after shared limit
(#65814)
fb19e0da0c0 is described below
commit fb19e0da0c0db23c0b643caa4710b80b9055e9dd
Author: Mryange <[email protected]>
AuthorDate: Mon Jul 27 15:11:39 2026 +0800
[fix](be) Finish pending scanners after shared limit (#65814)
### What problem does this PR solve?
Queries with a pushed-down shared scan limit could hang when multiple
pipeline tasks shared the same limit. After one scanner exhausted the
limit, a `ScannerContext` could still own pending scan tasks while
having no completed or in-flight task. The scheduler returned without
submitting a pending task, so that context never received EOS and its
pipeline task remained blocked.
Root cause: `_pull_next_scan_task()` treated an exhausted shared limit
as sufficient reason to skip every pending scanner, even when no other
task could make progress or wake the pipeline dependency.
This change skips pending scanners only while a completed or in-flight
task can still make progress. If neither exists, it submits a pending
scanner so the scanner can observe the exhausted limit, return EOS, and
wake the pipeline task. The shared limit continues to be consumed and
enforced in the scanner block-processing path; this change only fixes
the scheduling decision after the limit has been exhausted. It also
centralizes the
---
be/src/exec/scan/scanner_context.cpp | 18 ++-
be/src/exec/scan/scanner_context.h | 1 +
.../test_shared_scan_limit_pending_tasks.out | 7 +
.../test_shared_scan_limit_pending_tasks.groovy | 175 +++++++++++++++++++++
4 files changed, 196 insertions(+), 5 deletions(-)
diff --git a/be/src/exec/scan/scanner_context.cpp
b/be/src/exec/scan/scanner_context.cpp
index d6c7347e4a5..dfa5569fcb5 100644
--- a/be/src/exec/scan/scanner_context.cpp
+++ b/be/src/exec/scan/scanner_context.cpp
@@ -404,7 +404,7 @@ Status ScannerContext::get_block_from_queue(RuntimeState*
state, Block* block, b
if (_completed_tasks.empty() &&
(_num_finished_scanners == _all_scanners.size() ||
- (_shared_scan_limit->load(std::memory_order_acquire) == 0 &&
_in_flight_tasks_num == 0))) {
+ (_is_shared_scan_limit_exhausted() && _in_flight_tasks_num == 0))) {
_set_scanner_done();
_is_finished = true;
}
@@ -545,6 +545,10 @@ void ScannerContext::_set_scanner_done() {
_dependency->set_always_ready();
}
+bool ScannerContext::_is_shared_scan_limit_exhausted() const {
+ return limit >= 0 && _shared_scan_limit->load(std::memory_order_acquire)
<= 0;
+}
+
void ScannerContext::update_peak_running_scanner(int num) {
#ifndef BE_TEST
_local_state->_peak_running_scanner->add(num);
@@ -661,9 +665,11 @@ Status
ScannerContext::schedule_scan_task(std::shared_ptr<ScanTask> current_scan
if (first_pull) {
task_to_run = _pull_next_scan_task(current_scan_task,
current_concurrency);
if (task_to_run == nullptr) {
- // In two situations we will get nullptr.
+ // In three situations we will get nullptr.
// 1. current_concurrency already reached
_max_scan_concurrency.
// 2. all scanners are finished.
+ // 3. The shared LIMIT is exhausted while completed or
in-flight tasks can still
+ // make progress.
if (current_scan_task) {
DCHECK(current_scan_task->cached_block == nullptr);
DCHECK(!current_scan_task->is_eos());
@@ -735,9 +741,11 @@ std::shared_ptr<ScanTask>
ScannerContext::_pull_next_scan_task(
}
if (!_pending_tasks.empty()) {
- // Skip submitting more pending scanners once the LIMIT budget is
- // exhausted; they would only open and immediately EOF.
- if (_shared_scan_limit->load(std::memory_order_acquire) == 0) {
+ // Do not submit more pending scanners after the shared LIMIT is
exhausted while
+ // completed or in-flight tasks can still make progress. If neither
exists, allow pending
+ // scanners to be submitted so they can report EOS and wake the
pipeline task.
+ if (_is_shared_scan_limit_exhausted() &&
+ (_in_flight_tasks_num != 0 || !_completed_tasks.empty())) {
return nullptr;
}
std::shared_ptr<ScanTask> next_scan_task;
diff --git a/be/src/exec/scan/scanner_context.h
b/be/src/exec/scan/scanner_context.h
index 0f97e82126d..013deef3544 100644
--- a/be/src/exec/scan/scanner_context.h
+++ b/be/src/exec/scan/scanner_context.h
@@ -267,6 +267,7 @@ protected:
/// 3. `_free_blocks_memory_usage` < `_max_bytes_in_queue`, remains enough
memory to scale up
/// 4. At most scale up `MAX_SCALE_UP_RATIO` times to `_max_thread_num`
void _set_scanner_done();
+ bool _is_shared_scan_limit_exhausted() const;
RuntimeState* _state = nullptr;
ScanLocalStateBase* _local_state = nullptr;
diff --git
a/regression-test/data/correctness_p0/test_shared_scan_limit_pending_tasks.out
b/regression-test/data/correctness_p0/test_shared_scan_limit_pending_tasks.out
new file mode 100644
index 00000000000..ae79c1d44f4
--- /dev/null
+++
b/regression-test/data/correctness_p0/test_shared_scan_limit_pending_tasks.out
@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !shared_scan_limit_parallel_2 --
+
+-- !shared_scan_limit_parallel_6 --
+
+-- !shared_scan_limit_parallel_8 --
+
diff --git
a/regression-test/suites/correctness_p0/test_shared_scan_limit_pending_tasks.groovy
b/regression-test/suites/correctness_p0/test_shared_scan_limit_pending_tasks.groovy
new file mode 100644
index 00000000000..72f47c16f66
--- /dev/null
+++
b/regression-test/suites/correctness_p0/test_shared_scan_limit_pending_tasks.groovy
@@ -0,0 +1,175 @@
+// 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.
+
+suite("test_shared_scan_limit_pending_tasks") {
+ sql "DROP TABLE IF EXISTS shared_scan_limit_t1"
+ sql "DROP TABLE IF EXISTS shared_scan_limit_t3"
+ sql "DROP TABLE IF EXISTS shared_scan_limit_t4"
+ sql "DROP TABLE IF EXISTS shared_scan_limit_t5"
+
+ sql """
+ CREATE TABLE shared_scan_limit_t1 (
+ pk INT,
+ v1 BIGINT,
+ v2 BIGINT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(pk)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql """
+ CREATE TABLE shared_scan_limit_t3 (
+ pk INT,
+ v1 BIGINT,
+ v2 BIGINT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(pk)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql """
+ CREATE TABLE shared_scan_limit_t4 (
+ pk INT,
+ v1 BIGINT,
+ v2 BIGINT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(pk)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql """
+ CREATE TABLE shared_scan_limit_t5 (
+ pk INT,
+ v1 BIGINT,
+ v2 BIGINT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(pk)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ INSERT INTO shared_scan_limit_t1 VALUES
+ (0,70,32163), (1,-71,NULL), (2,-85,-222125), (3,26,22867),
+ (4,17,NULL), (5,NULL,6236439), (6,-16907,-24720), (7,26877,1893),
+ (8,30405,9), (9,5873343,66), (10,-17,-47), (11,NULL,20987),
+ (12,NULL,-4672), (13,NULL,3551614), (14,28521,-1372),
(15,NULL,NULL),
+ (16,38,NULL), (17,NULL,-54), (18,NULL,-4686447), (19,NULL,-8420)
+ """
+ sql """
+ INSERT INTO shared_scan_limit_t3 VALUES
+ (0,28862,-25225), (1,5516858,-4609390), (2,-7815300,NULL),
+ (3,NULL,-7685824), (4,22293,26373), (5,NULL,NULL), (6,1237,-31),
+ (7,1606,3318374), (8,NULL,-21355), (9,NULL,-20127),
(10,5413995,-3),
+ (11,-4718995,NULL), (12,3179854,3733421), (13,18867,76),
+ (14,15517,-4932405), (15,-30778,NULL), (16,NULL,-7258),
+ (17,NULL,7520313), (18,6936629,NULL), (19,-45,6231596)
+ """
+ sql """
+ INSERT INTO shared_scan_limit_t4 VALUES
+ (0,5649134,4812327), (1,8249417,4670054), (2,106,6714155),
+ (3,NULL,NULL), (4,-52850,-5113499), (5,-23911,-9637), (6,-26168,9),
+ (7,-47,-28398), (8,-29518,-2365317), (9,20685,22956),
(10,97,25099),
+ (11,-32617,6143808), (12,NULL,NULL), (13,-52,NULL),
(14,7680925,NULL),
+ (15,10848,NULL), (16,64,-3394), (17,113,-12488), (18,-87,-12093),
+ (19,NULL,22)
+ """
+ sql """
+ INSERT INTO shared_scan_limit_t5 VALUES
+ (0,-3018453,-5763927), (1,-60,-6233027), (2,78,-5304), (3,63,NULL),
+ (4,23287,NULL), (5,98,7989209), (6,-61,-30493),
(7,-6781665,-22321),
+ (8,-5165806,-75), (9,NULL,31290), (10,1311226,7754904),
(11,-72,54),
+ (12,114,3741372), (13,NULL,NULL), (14,-40,2014555),
(15,69,-4555977),
+ (16,118,-8708), (17,NULL,NULL), (18,-123,NULL), (19,NULL,19843)
+ """
+ sql "sync"
+
+ sql "SET enable_sql_cache = false"
+ sql "SET experimental_enable_local_shuffle = false"
+ sql "SET query_timeout = 60"
+
+ // A shared LIMIT can be exhausted while another ScannerContext still owns
pending tasks.
+ // Every context must still reach EOS instead of leaving its pipeline task
blocked forever.
+ sql "SET parallel_pipeline_task_num = 2"
+ qt_shared_scan_limit_parallel_2 """
+ SELECT v1, pk
+ FROM shared_scan_limit_t1 AS t1
+ WHERE v2 > (
+ SELECT SUM(v1)
+ FROM shared_scan_limit_t5 AS t3
+ WHERE EXISTS (
+ SELECT pk
+ FROM shared_scan_limit_t3 AS t4
+ WHERE NOT EXISTS (
+ SELECT pk, v1, v1, v2, v1
+ FROM shared_scan_limit_t4 AS t5
+ WHERE v2 NOT IN (
+ SELECT pk FROM shared_scan_limit_t1 AS t6 ORDER BY pk
+ )
+ ) AND v1 < 7
+ )
+ )
+ ORDER BY v1, pk DESC
+ LIMIT 6
+ """
+
+ sql "SET parallel_pipeline_task_num = 6"
+ qt_shared_scan_limit_parallel_6 """
+ SELECT v1, pk
+ FROM shared_scan_limit_t1 AS t1
+ WHERE v2 > (
+ SELECT SUM(v1)
+ FROM shared_scan_limit_t5 AS t3
+ WHERE EXISTS (
+ SELECT pk
+ FROM shared_scan_limit_t3 AS t4
+ WHERE NOT EXISTS (
+ SELECT pk, v1, v1, v2, v1
+ FROM shared_scan_limit_t4 AS t5
+ WHERE v2 NOT IN (
+ SELECT pk FROM shared_scan_limit_t1 AS t6 ORDER BY pk
+ )
+ ) AND v1 < 7
+ )
+ )
+ ORDER BY v1, pk DESC
+ LIMIT 6
+ """
+
+ sql "SET parallel_pipeline_task_num = 8"
+ qt_shared_scan_limit_parallel_8 """
+ SELECT v1, pk
+ FROM shared_scan_limit_t1 AS t1
+ WHERE v2 > (
+ SELECT SUM(v1)
+ FROM shared_scan_limit_t5 AS t3
+ WHERE EXISTS (
+ SELECT pk
+ FROM shared_scan_limit_t3 AS t4
+ WHERE NOT EXISTS (
+ SELECT pk, v1, v1, v2, v1
+ FROM shared_scan_limit_t4 AS t5
+ WHERE v2 NOT IN (
+ SELECT pk FROM shared_scan_limit_t1 AS t6 ORDER BY pk
+ )
+ ) AND v1 < 7
+ )
+ )
+ ORDER BY v1, pk DESC
+ LIMIT 6
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
