This is an automated email from the ASF dual-hosted git repository.
mrhhsg 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 e0bf4874070 [improvement](be) Use thread pool scan scheduler by
default (#68610)
e0bf4874070 is described below
commit e0bf4874070ba0b6837ad93f695ea63532b77a47
Author: Jerry Hu <[email protected]>
AuthorDate: Mon Oct 19 15:53:55 2026 +0800
[improvement](be) Use thread pool scan scheduler by default (#68610)
### What problem does this PR solve?
Issue Number: None
Related PR: #51690, #58144, #58273, #65814, #67070
Problem Summary: Workload groups pick the scan scheduler for internal
and
external table scans from `enable_task_executor_in_internal_table` and
`enable_task_executor_in_external_table`. Both configs currently default
to
`true` on master, so every workload group creates
`TaskExecutorSimplifiedScanScheduler` (the time-sharing task executor)
for
local (`ls_*`) and remote (`rs_*`) scans. The time-sharing executor is
still
maturing (for example the self-deadlock fixed in #58273; the
internal-table
default was already reverted once in #58144), so the default should be
the
`ThreadPoolSimplifiedScanScheduler` for both internal and external
tables.
This change sets both configs' default values to `false`. The task
executor
can still be enabled explicitly through `be.conf`.
Making the thread pool path the default exposed four gaps.
1. Pending scanners after an exhausted shared LIMIT. For a `SELECT ...
LIMIT n`
over many tablets, the thread pool admission
(`ScannerContext::can_admit_scan_task`,
rewritten in #67070) keeps admitting the remaining pending scanners
after peer
scanners have exhausted the shared scan LIMIT. Admission is kept that
way on
purpose: holding pending scanners back based on the in-flight or
completed
task counters, as #65814 does on the TaskExecutor path, ties the
schedule to
bookkeeping that is easy to get wrong and can leave a context with
nothing to
wake the operator. Instead, `Scanner::get_block_after_projects()` now
checks
the shared LIMIT before it reads: once the counter is exhausted, the
scanner
returns an empty block, so an admitted scanner never reads its range for
rows
the operator will cut anyway. It does not report EOS for that: EOS means
the
underlying read is exhausted, while the LIMIT is applied by the scan
operator,
which stops the scanners after `reached_limit()`;
`ScannerContext::get_block_from_queue()` also finishes the context once
the
counter is exhausted with nothing in flight. The two pre-existing checks
in
`Scanner::get_block()` that set `eof` from the shared counter are
removed for
the same reason, so the shared LIMIT is handled in one place.
2. Rows held in scanner padding under a shared LIMIT.
`Scanner::get_block()`
charges the shared LIMIT counter as soon as rows pass the filters, but
with
projections `get_block_after_projects()` may keep a small block in
`_padding_block` and emit a later block first. Once peer scanners
exhaust the
counter, the context reaches EOS without running that scanner again, and
the
charged rows it holds are lost; for example, a parallel projected
`LIMIT 4096` could return 4086 rows. The counter is now decremented only
by
the rows a call actually returns, so held padding is never charged and
the
counter equals "LIMIT minus returned rows" between calls. The padding
batch
target is capped by the remaining LIMIT when a call starts, so a scanner
does
not read on just to fill half a batch when fewer rows are needed, and
the
read loop stops as soon as peers exhaust the LIMIT, returning the rows
it
already holds. Projected scans with selective filters keep coalescing
small
blocks. The scan does not cap its output at the LIMIT (the operator cuts
extra rows), so no further accounting is needed.
3. Exceptions in the row-ID fetch RPC callback. `multiget_data_v2`
submits
its callback to the remote scan scheduler; on a `ThreadPool` worker
there is
no exception boundary, and `RowIdStorageReader::read_by_rowids()`
constructs
`SlotDescriptor(pslot)` before its local `try`, so an unknown
`PTypeNode.type` would throw `doris::Exception` and terminate the BE.
The
callback now calls `RowIdStorageReader::read_by_rowids_for_rpc()`, which
converts a thrown exception into an error status on the response before
the
closure completes the RPC. `ExecEnv::set_id_manager()` is added under
`BE_TEST` for the unit test.
4. The shared LIMIT pointer of a scanner initialized while the counter
is
negative. `ScanLocalState::shared_scan_limit_ptr()` returned null
whenever
the live counter was negative, and `Scanner::init()` caches that pointer
once. Concurrent scanners can return more than the LIMIT before the
operator
makes its final cut, leaving the counter negative, so a scanner whose
local
state opened at that moment never checked or charged the shared LIMIT
and
could read a long selective tail after peers had satisfied the query.
The pointer is now returned whenever the operator has a SQL LIMIT, which
is
how `ScannerContext` already decides it.
### Release note
The BE configs `enable_task_executor_in_internal_table` and
`enable_task_executor_in_external_table` now default to `false`;
internal
and external table scans use the thread pool scan scheduler by default.
### Check List (For Author)
- Test:
- Unit Test: added
`WorkloadGroupManagerTest.DefaultScanSchedulersUseThreadPool`,
`ScannerContextTest.thread_pool_admission_ignores_shared_limit`,
`ScannerProjectionTest.shared_limit_exhausted_before_call_returns_empty_block_without_eos`,
`ScannerProjectionTest.shared_limit_counts_only_emitted_rows`,
`ScannerProjectionTest.shared_limit_drops_padding_after_peer_satisfies_limit`,
`ScannerProjectionTest.shared_limit_flushes_padding_before_reading_more`,
`ScannerProjectionTest.shared_limit_flushes_merged_padding_at_limit`,
`ScannerProjectionTest.shared_limit_counts_returned_rows_without_projection`,
`ScannerProjectionTest.shared_limit_counts_returned_rows_without_padding`,
`ScannerProjectionTest.shared_limit_keeps_padding_small_blocks`,
`ScannerProjectionTest.shared_limit_exhausted_by_own_rows_stops_reading_without_eos`,
`ScannerProjectionTest.shared_limit_exhausted_by_peer_mid_read_returns_held_rows_without_eos`,
`ScannerProjectionTest.shared_limit_observed_by_scanner_initialized_under_negative_counter`,
`ScannerProjectionTest.no_shared_limit_without_sql_limit`
and
`RowIdStorageReaderTest.RpcHelperReportsThrownExceptionOnThreadPoolPath`
(runs the RPC helper on a real `ThreadPoolSimplifiedScanScheduler`;
the test process aborts without the fix);
ran `ScannerProjectionTest.*`, `ScannerContextTest.*`,
`ScannerLateArrivalRfTest.*`, `WorkloadGroupManagerTest.*`,
`RowIdStorageReaderTest.*`, `SubmitExternalScanTasksTest.*`
- Regression test: ran on a local cluster with the new
default: query_p0/limit (19 suites) and
correctness_p0/test_shared_scan_limit_pending_tasks
- Behavior changed: Yes (default scan scheduler for internal and
external
table scans changes from the task executor to the thread pool; it can be
switched back via the two BE configs)
- Does this need documentation: No
---
be/src/common/config.cpp | 4 +-
be/src/exec/operator/mock_scan_operator.h | 6 +
be/src/exec/operator/scan_operator.cpp | 10 +-
be/src/exec/rowid_fetcher.cpp | 7 +
be/src/exec/rowid_fetcher.h | 8 +
be/src/exec/scan/scanner.cpp | 106 +++--
be/src/exec/scan/scanner.h | 9 +
be/src/exec/scan/scanner_context.cpp | 9 +-
be/src/runtime/exec_env.h | 1 +
be/src/service/internal_service.cpp | 4 +-
be/test/exec/rowid_fetcher_test.cpp | 72 +++
be/test/exec/scan/scanner_context_test.cpp | 134 ++++++
be/test/exec/scan/scanner_late_arrival_rf_test.cpp | 485 +++++++++++++++++++++
.../workload_group/workload_group_manager_test.cpp | 57 +++
14 files changed, 856 insertions(+), 56 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 1361137bae7..cffac653cf1 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -339,9 +339,9 @@ DEFINE_Int32(task_executor_max_concurrency_per_task, "-1");
DEFINE_Int32(task_executor_initial_max_concurrency_per_task, "-1");
// Enable task executor in internal table scan.
-DEFINE_Bool(enable_task_executor_in_internal_table, "true");
+DEFINE_Bool(enable_task_executor_in_internal_table, "false");
// Enable task executor in external table scan.
-DEFINE_Bool(enable_task_executor_in_external_table, "true");
+DEFINE_Bool(enable_task_executor_in_external_table, "false");
// number of scanner thread pool size for olap table
// and the min thread num of remote scanner thread pool
diff --git a/be/src/exec/operator/mock_scan_operator.h
b/be/src/exec/operator/mock_scan_operator.h
index 609e631ce8c..27f1f8ef53d 100644
--- a/be/src/exec/operator/mock_scan_operator.h
+++ b/be/src/exec/operator/mock_scan_operator.h
@@ -84,6 +84,12 @@ public:
_output_blocks.push_back(std::move(block));
}
+ // Mirrors the constructor of ScanOperatorX: the shared counter starts at
the SQL LIMIT.
+ void set_limit_for_test(int64_t limit) {
+ _limit = limit;
+ _shared_scan_limit.store(limit, std::memory_order_relaxed);
+ }
+
Status get_block_impl(RuntimeState* state, Block* block, bool* eos)
override {
if (_output_blocks.empty()) {
*eos = true;
diff --git a/be/src/exec/operator/scan_operator.cpp
b/be/src/exec/operator/scan_operator.cpp
index 7a0c4042883..e01fdc33bf0 100644
--- a/be/src/exec/operator/scan_operator.cpp
+++ b/be/src/exec/operator/scan_operator.cpp
@@ -1072,10 +1072,12 @@ int64_t ScanLocalState<Derived>::limit_per_scanner() {
template <typename Derived>
std::atomic<int64_t>* ScanLocalState<Derived>::shared_scan_limit_ptr() {
- auto* p = &_parent->cast<typename Derived::Parent>()._shared_scan_limit;
- // -1 means "no SQL LIMIT" — return nullptr so callers naturally skip
- // all limit logic.
- return p->load(std::memory_order_relaxed) < 0 ? nullptr : p;
+ auto& parent = _parent->cast<typename Derived::Parent>();
+ // Decide from the SQL LIMIT, not from the counter. -1 means "no SQL
LIMIT" — return nullptr
+ // so callers naturally skip all limit logic. Concurrent scanners can
return more than the
+ // LIMIT before the operator makes its final cut, leaving this counter
negative. A scanner
+ // initialized then must still observe that the shared LIMIT is exhausted.
+ return parent.limit() < 0 ? nullptr : &parent._shared_scan_limit;
}
template <typename Derived>
diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp
index 3595a735b67..3f9c0efc82e 100644
--- a/be/src/exec/rowid_fetcher.cpp
+++ b/be/src/exec/rowid_fetcher.cpp
@@ -196,6 +196,13 @@ static void scatter_scan_blocks_to_result_block(
}
}
+void RowIdStorageReader::read_by_rowids_for_rpc(const PMultiGetRequestV2&
request,
+ PMultiGetResponseV2* response)
{
+ Status st = Status::OK();
+ ASSIGN_STATUS_IF_CATCH_EXCEPTION(st = read_by_rowids(request, response),
st);
+ st.to_protobuf(response->mutable_status());
+}
+
Status RowIdStorageReader::read_by_rowids(const PMultiGetRequestV2& request,
PMultiGetResponseV2* response) {
if (request.request_block_descs_size()) {
diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h
index 60b18fb8e82..916b77ff214 100644
--- a/be/src/exec/rowid_fetcher.h
+++ b/be/src/exec/rowid_fetcher.h
@@ -82,6 +82,14 @@ public:
static Status read_by_rowids(const PMultiGetRequestV2& request,
PMultiGetResponseV2* response);
+ // Runs read_by_rowids() for the multiget_data_v2 RPC and serializes the
outcome into the
+ // response status. The remote scan scheduler runs the RPC callback on a
plain ThreadPool
+ // worker, which has no exception boundary of its own, so a
doris::Exception thrown while
+ // reading (for example by a slot of the request whose PTypeNode this BE
does not know) is
+ // converted into an error status here instead of terminating the process.
+ static void read_by_rowids_for_rpc(const PMultiGetRequestV2& request,
+ PMultiGetResponseV2* response);
+
private:
static bool should_use_file_scanner_v2(const TQueryOptions& query_options,
const TFileScanRangeParams&
scan_params,
diff --git a/be/src/exec/scan/scanner.cpp b/be/src/exec/scan/scanner.cpp
index 836dacf7125..c4eec1e8f67 100644
--- a/be/src/exec/scan/scanner.cpp
+++ b/be/src/exec/scan/scanner.cpp
@@ -19,6 +19,8 @@
#include <glog/logging.h>
+#include <algorithm>
+
#include "common/config.h"
#include "common/status.h"
#include "core/block/column_with_type_and_name.h"
@@ -87,8 +89,18 @@ Status Scanner::init(RuntimeState* state, const
VExprContextSPtrs& conjuncts) {
return Status::OK();
}
+// Keep the single-use padding loop inline with its projection and LIMIT
accounting.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity)
Status Scanner::get_block_after_projects(RuntimeState* state, Block* block,
bool* eos) {
SCOPED_CONCURRENCY_COUNT(ConcurrencyStatsManager::instance().vscanner_get_block);
+ // Peer scanners may have exhausted the shared LIMIT while this scanner
waited to be
+ // scheduled. Admission does not look at the LIMIT, so the scanner returns
an empty block
+ // instead of reading on. It does not report EOS: the operator already
holds enough rows and
+ // stops the scanners after cutting the LIMIT itself, and ScannerContext
parks the scanner
+ // rather than resubmitting it while the outstanding work drains.
+ if (_shared_scan_limit_exhausted()) {
+ return Status::OK();
+ }
const auto& row_descriptor =
_local_state->_parent->operator_row_desc_before_projection();
if (_has_projection) {
_origin_block.clear_column_data(row_descriptor.num_slots());
@@ -99,58 +111,68 @@ Status Scanner::get_block_after_projects(RuntimeState*
state, Block* block, bool
// before the next split is read. Padding those blocks first would
make correctness
// depend on whether two file tails happen to share one output
batch.
RETURN_IF_ERROR(get_block(state, &_origin_block, eos));
- return _do_projections(&_origin_block, block);
- }
- const auto min_batch_size = std::max(state->batch_size() / 2, 1);
- const auto block_max_bytes = state->preferred_block_size_bytes();
- while (_padding_block.rows() < min_batch_size &&
_padding_block.bytes() < block_max_bytes &&
- !*eos) {
- RETURN_IF_ERROR(get_block(state, &_origin_block, eos));
- if (*eos) {
- // For the final block, merge any padding directly and return
eos in this call.
- // The merged tail can be larger than the target batch, but
each source block is
- // already bounded by the lower scanner.
- RETURN_IF_ERROR(_merge_padding_block());
- _origin_block.clear_column_data(row_descriptor.num_slots());
- break;
- }
- if (_origin_block.rows() >= min_batch_size) {
- break;
+ } else {
+ auto min_batch_size = std::max(state->batch_size() / 2, 1);
+ if (_shared_scan_limit) {
+ // Pending padding is not charged until it is returned. Bound
this call's batch
+ // target by the shared remaining rows to avoid reading on
just to fill a batch.
+ min_batch_size = cast_set<int>(std::clamp<int64_t>(
+ _shared_scan_limit->load(std::memory_order_acquire),
0, min_batch_size));
}
+ const auto block_max_bytes = state->preferred_block_size_bytes();
+ // Stop reading as soon as peers exhaust the shared LIMIT; the
rows already held are
+ // still returned below.
+ while (_padding_block.rows() < min_batch_size &&
+ _padding_block.bytes() < block_max_bytes && !*eos &&
+ !_shared_scan_limit_exhausted()) {
+ RETURN_IF_ERROR(get_block(state, &_origin_block, eos));
+ if (*eos) {
+ // For the final block, merge any padding directly and
return eos in this call.
+ // The merged tail can be larger than the target batch,
but each source block
+ // is already bounded by the lower scanner.
+ RETURN_IF_ERROR(_merge_padding_block());
+
_origin_block.clear_column_data(row_descriptor.num_slots());
+ break;
+ }
+ if (_origin_block.rows() >= min_batch_size) {
+ break;
+ }
- if (_origin_block.rows() + _padding_block.rows() <=
state->batch_size() &&
- _origin_block.bytes() + _padding_block.bytes() <=
block_max_bytes) {
- RETURN_IF_ERROR(_merge_padding_block());
- _origin_block.clear_column_data(row_descriptor.num_slots());
- } else {
- if (_origin_block.rows() < _padding_block.rows()) {
- _padding_block.swap(_origin_block);
+ if (_origin_block.rows() + _padding_block.rows() <=
state->batch_size() &&
+ _origin_block.bytes() + _padding_block.bytes() <=
block_max_bytes) {
+ RETURN_IF_ERROR(_merge_padding_block());
+
_origin_block.clear_column_data(row_descriptor.num_slots());
+ } else {
+ if (_origin_block.rows() < _padding_block.rows()) {
+ _padding_block.swap(_origin_block);
+ }
+ break;
}
- break;
}
- }
- if (_origin_block.empty() && !_padding_block.empty()) {
- _padding_block.swap(_origin_block);
+ if (_origin_block.empty() && !_padding_block.empty()) {
+ _padding_block.swap(_origin_block);
+ }
}
- return _do_projections(&_origin_block, block);
+ RETURN_IF_ERROR(_do_projections(&_origin_block, block));
} else {
- return get_block(state, block, eos);
+ RETURN_IF_ERROR(get_block(state, block, eos));
+ }
+
+ // Only rows actually produced by this call consume the shared LIMIT. In
particular, padding
+ // kept for a later call must not make the context finish before that call
can return it.
+ // Exhausting the counter is not EOS either: the next call returns an
empty block, and the
+ // scan operator stops this scanner once it has cut the LIMIT.
+ if (_shared_scan_limit != nullptr && block->rows() > 0) {
+ _shared_scan_limit->fetch_sub(cast_set<int64_t>(block->rows()),
std::memory_order_acq_rel);
}
+ return Status::OK();
}
Status Scanner::get_block(RuntimeState* state, Block* block, bool* eof) {
// only empty block should be here
DCHECK(block->rows() == 0);
- // Stop early if other scanners have already collected enough rows
- // for the SQL LIMIT. Skipped when _shared_scan_limit is null (topn
- // path or no LIMIT).
- if (_shared_scan_limit &&
_shared_scan_limit->load(std::memory_order_acquire) <= 0) {
- *eof = true;
- return Status::OK();
- }
-
// scanner running time
SCOPED_RAW_TIMER(&_per_scanner_timer);
int64_t rows_read_threshold = _num_rows_read +
config::doris_scanner_row_num;
@@ -183,13 +205,6 @@ Status Scanner::get_block(RuntimeState* state, Block*
block, bool* eof) {
}
// record rows return (after filter) for _limit check
_num_rows_return += block->rows();
- // Publish progress to the shared counter so peer scanners can
- // observe it. The counter may go negative when several scanners
- // subtract concurrently; that is harmless because the operator's
- // reached_limit() makes the final cut.
- if (_shared_scan_limit && block->rows() > 0) {
- _shared_scan_limit->fetch_sub(block->rows(),
std::memory_order_acq_rel);
- }
} while (!_should_stop && !state->is_cancelled() && block->rows() == 0
&& !(*eof) &&
_num_rows_read < rows_read_threshold);
}
@@ -202,7 +217,6 @@ Status Scanner::get_block(RuntimeState* state, Block*
block, bool* eof) {
// set eof to true if per scanner limit is reached
// currently for query: ORDER BY key LIMIT n
*eof = *eof || (_limit > 0 && _num_rows_return >= _limit);
- *eof = *eof || (_shared_scan_limit &&
_shared_scan_limit->load(std::memory_order_acquire) <= 0);
return Status::OK();
}
diff --git a/be/src/exec/scan/scanner.h b/be/src/exec/scan/scanner.h
index 927472b2fa2..dfed572406e 100644
--- a/be/src/exec/scan/scanner.h
+++ b/be/src/exec/scan/scanner.h
@@ -140,6 +140,15 @@ protected:
return Status::OK();
}
+ // True once scanners of this operator have produced the whole shared
LIMIT. The scanner then
+ // stops reading but does not report EOS for it: EOS means the underlying
read is exhausted,
+ // while the LIMIT is applied by the scan operator, which stops the
scanners once it holds
+ // enough rows.
+ bool _shared_scan_limit_exhausted() const {
+ return _shared_scan_limit != nullptr &&
+ _shared_scan_limit->load(std::memory_order_acquire) <= 0;
+ }
+
// Update the counters before closing this scanner
virtual void _collect_profile_before_close();
diff --git a/be/src/exec/scan/scanner_context.cpp
b/be/src/exec/scan/scanner_context.cpp
index c8727893f49..4c4bb8ae939 100644
--- a/be/src/exec/scan/scanner_context.cpp
+++ b/be/src/exec/scan/scanner_context.cpp
@@ -418,6 +418,12 @@ Status ScannerContext::get_block_from_queue(RuntimeState*
state, Block* block, b
(_is_shared_scan_limit_exhausted() && _in_flight_tasks_num == 0))) {
_set_scanner_done();
_is_finished = true;
+ } else if (scan_task != nullptr && !scan_task->is_eos() &&
_is_shared_scan_limit_exhausted()) {
+ // Scanners of this operator have already produced the whole shared
LIMIT, so this scanner
+ // would only return empty blocks until the outstanding work drains.
Park it instead of
+ // resubmitting it. The finish check above did not fire, so a
completed or in-flight task
+ // remains to wake the operator, and the context finishes once nothing
is in flight.
+ push_pending_scan_task(std::move(scan_task), l);
} else if (scan_task != nullptr) {
// Check the terminal state before scheduling more work. In
particular, after the shared
// LIMIT is exhausted, submitting another Context runnable can turn a
completed query into
@@ -867,7 +873,8 @@ std::shared_ptr<ScanTask>
ScannerContext::_pull_next_scan_task(
if (!_pending_tasks.empty()) {
// 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.
+ // scanners to be submitted so they can complete without reading and
wake the pipeline
+ // task, which finishes the context once nothing is in flight.
if (_is_shared_scan_limit_exhausted() &&
(_in_flight_tasks_num != 0 || !_completed_tasks.empty())) {
return nullptr;
diff --git a/be/src/runtime/exec_env.h b/be/src/runtime/exec_env.h
index 8d30f60258a..be7d92c4fff 100644
--- a/be/src/runtime/exec_env.h
+++ b/be/src/runtime/exec_env.h
@@ -348,6 +348,7 @@ public:
_inverted_index_query_cache = inverted_index_query_cache;
}
void set_process_profile(ProcessProfile* pp) { this->_process_profile =
pp; }
+ void set_id_manager(IdManager* id_manager) { this->_id_manager =
id_manager; }
void set_delete_bitmap_agg_cache(DeleteBitmapAggCache* c) {
_delete_bitmap_agg_cache = c; }
void set_storage_page_cache(StoragePageCache* c) {
this->_storage_page_cache = c; }
void set_ann_index_ivf_list_cache(AnnIndexIVFListCache* c) {
diff --git a/be/src/service/internal_service.cpp
b/be/src/service/internal_service.cpp
index a2014b7bb3d..d64349a6d29 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -2063,9 +2063,7 @@ void
PInternalService::multiget_data_v2(google::protobuf::RpcController* control
MonotonicStopWatch watch;
watch.start();
brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- Status st =
RowIdStorageReader::read_by_rowids(*request, response);
- st.to_protobuf(response->mutable_status());
+ RowIdStorageReader::read_by_rowids_for_rpc(*request,
response);
LOG(INFO) << "multiget_data finished, cost(us):"
<< watch.elapsed_time() / 1000;
return true;
diff --git a/be/test/exec/rowid_fetcher_test.cpp
b/be/test/exec/rowid_fetcher_test.cpp
index 77bf26a98d1..b0478cfc4ab 100644
--- a/be/test/exec/rowid_fetcher_test.cpp
+++ b/be/test/exec/rowid_fetcher_test.cpp
@@ -22,13 +22,22 @@
#include <string>
#include <vector>
+#include "agent/cgroup_cpu_ctl.h"
+#include "common/exception.h"
#include "exec/operator/file_scan_operator.h"
#include "exec/scan/file_scanner_v2.h"
+#include "exec/scan/scanner_scheduler.h"
#include "format_v2/column_mapper.h"
#include "format_v2/table/hive_reader.h"
+#include "gen_cpp/internal_service.pb.h"
#include "runtime/descriptor_helper.h"
#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
+#include "storage/id_manager.h"
+#include "storage/rowset_id.h"
+#include "util/countdown_latch.h"
+#include "util/defer_op.h"
namespace doris {
@@ -429,4 +438,67 @@ TEST_F(SubmitExternalScanTasksTest,
ThrownExceptionReachesTheCaller) {
EXPECT_NE(result.to_string().find("scanner threw"), std::string::npos);
}
+// multiget_data_v2 runs read_by_rowids() on the remote scan scheduler, which
by default is a
+// ThreadPool whose worker has no exception boundary. A request whose file
mapping exists but
+// whose slot carries a PTypeNode this BE cannot decode makes SlotDescriptor's
constructor throw;
+// the RPC helper must turn that into an error status on the response instead
of letting the
+// exception leave the worker and terminate the process.
+TEST_F(RowIdStorageReaderTest,
RpcHelperReportsThrownExceptionOnThreadPoolPath) {
+ IdManager id_manager;
+ ExecEnv* env = ExecEnv::GetInstance();
+ IdManager* saved_id_manager = env->get_id_manager();
+ env->set_id_manager(&id_manager);
+ Defer restore_id_manager {[&] { env->set_id_manager(saved_id_manager); }};
+
+ PMultiGetRequestV2 request;
+ request.mutable_query_id()->set_hi(68610);
+ request.mutable_query_id()->set_lo(1);
+ request.set_wg_id(1);
+ auto id_file_map = id_manager.add_id_file_map(request.query_id(), 1024);
+ const uint32_t file_id = id_file_map->get_file_mapping_id(
+ std::make_shared<FileMapping>(/*tablet_id=*/1, RowsetId(),
/*segment_id=*/0));
+
+ PRequestBlockDesc* block_desc = request.add_request_block_descs();
+ block_desc->add_file_id(file_id);
+ block_desc->add_row_id(0);
+ PSlotDescriptor* slot = block_desc->add_slots();
+ slot->set_id(0);
+ slot->set_parent(0);
+ slot->set_column_pos(0);
+ slot->set_byte_offset(0);
+ slot->set_null_indicator_byte(0);
+ slot->set_null_indicator_bit(-1);
+ slot->set_col_name("c");
+ slot->set_slot_idx(0);
+ // A type node kind unknown to this BE, as a newer FE could send.
+ slot->mutable_slot_type()->add_types()->set_type(99);
+
+ // The worker captures the response and the latch by reference. Declare
them before the
+ // scheduler so every exit, including a failed wait below, joins the
worker in stop_scheduler
+ // before either captured object is destroyed.
+ PMultiGetResponseV2 response;
+ CountDownLatch finished(1);
+ std::shared_ptr<CgroupCpuCtl> cgroup_cpu_ctl =
std::make_shared<CgroupV2CpuCtl>(1);
+ ThreadPoolSimplifiedScanScheduler scheduler("rowid_rpc_test",
cgroup_cpu_ctl);
+ ASSERT_TRUE(scheduler.start(1, 1, 4, 1).ok());
+ Defer stop_scheduler {[&] { scheduler.stop(); }};
+
+ ASSERT_TRUE(scheduler
+ .submit_scan_task(SimplifiedScanTask(
+ [&]() {
+
RowIdStorageReader::read_by_rowids_for_rpc(
+ request,
&response);
+ finished.count_down();
+ return true;
+ },
+ nullptr, nullptr),
+ "rowid_rpc_test_task")
+ .ok());
+ ASSERT_TRUE(finished.wait_for(std::chrono::seconds(30)));
+
+ EXPECT_NE(response.status().status_code(), 0);
+ ASSERT_EQ(response.status().error_msgs_size(), 1);
+ EXPECT_NE(response.status().error_msgs(0).find("invalid TTypeNodeType"),
std::string::npos);
+}
+
} // namespace doris
diff --git a/be/test/exec/scan/scanner_context_test.cpp
b/be/test/exec/scan/scanner_context_test.cpp
index d60f4e77d62..a9a3eeac3b9 100644
--- a/be/test/exec/scan/scanner_context_test.cpp
+++ b/be/test/exec/scan/scanner_context_test.cpp
@@ -705,6 +705,140 @@ TEST_F(ScannerContextTest, thread_pool_admission_state) {
EXPECT_EQ(scanner_context->try_get_next_scan_task(context_transfer_lock),
nullptr);
}
+TEST_F(ScannerContextTest, thread_pool_admission_ignores_shared_limit) {
+ const int parallel_tasks = 1;
+ const int64_t limit = 100;
+ auto scan_operator = std::make_unique<OlapScanOperatorX>(obj_pool.get(),
tnode, 0, *descs,
+ parallel_tasks,
TQueryCacheParam {});
+ auto olap_scan_local_state =
+ OlapScanLocalState::create_unique(state.get(),
scan_operator.get());
+
+ OlapScanner::Params scanner_params;
+ scanner_params.state = state.get();
+ scanner_params.profile = profile.get();
+ scanner_params.limit = limit;
+ scanner_params.key_ranges = std::vector<OlapScanRange*>();
+ std::shared_ptr<Scanner> scanner =
+ OlapScanner::create_shared(olap_scan_local_state.get(),
std::move(scanner_params));
+ std::list<std::shared_ptr<ScannerDelegate>> scanners {
+ std::make_shared<ScannerDelegate>(scanner)};
+ auto scanner_context = ScannerContext::create_shared(
+ state.get(), olap_scan_local_state.get(), output_tuple_desc,
false, scanners, limit,
+ scan_dependency, &shared_limit, nullptr, nullptr, 0, false,
parallel_tasks);
+
+ std::unique_ptr<MockSimplifiedScanScheduler> scheduler =
+ std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl);
+ EXPECT_CALL(*scheduler,
get_active_threads()).WillRepeatedly(testing::Return(0));
+ EXPECT_CALL(*scheduler,
get_queue_size()).WillRepeatedly(testing::Return(0));
+ scanner_context->_scanner_scheduler = scheduler.get();
+ scanner_context->_min_scan_concurrency_of_scan_scheduler = 20;
+ scanner_context->_max_scan_concurrency = 4;
+
+ std::unique_lock<std::mutex>
transfer_lock(scanner_context->transfer_lock());
+ scanner_context->_pending_tasks = std::stack<std::shared_ptr<ScanTask>>();
+ scanner_context->_completed_tasks.clear();
+ scanner_context->push_pending_scan_task(
+
std::make_shared<ScanTask>(std::make_shared<ScannerDelegate>(scanner)),
transfer_lock);
+
+ // The shared LIMIT is exhausted while a task is in flight and a completed
result waits for the
+ // operator. Admission only applies the concurrency limit: the pending
scanner is still
+ // admitted, and Scanner::get_block_after_projects() returns an empty
block without reading
+ // (and without EOS, which the scan operator decides from the LIMIT).
Holding it back based on
+ // the in-flight or completed tasks is what can leave a context with
nothing to wake the
+ // operator.
+ shared_limit.store(0);
+ scanner_context->_in_flight_tasks_num = 1;
+ auto completed_task =
std::make_shared<ScanTask>(std::make_shared<ScannerDelegate>(scanner));
+ completed_task->set_state(ScanTask::State::IN_FLIGHT);
+ completed_task->set_state(ScanTask::State::EOS);
+ scanner_context->_completed_tasks.push_back(completed_task);
+ EXPECT_TRUE(scanner_context->can_admit_scan_task(transfer_lock));
+ auto admitted_task =
scanner_context->try_get_next_scan_task(transfer_lock);
+ ASSERT_NE(admitted_task, nullptr);
+ EXPECT_EQ(scanner_context->_in_flight_tasks_num, 2);
+ EXPECT_TRUE(scanner_context->_pending_tasks.empty());
+}
+
+TEST_F(ScannerContextTest,
exhausted_shared_limit_parks_empty_task_until_in_flight_drains) {
+ const int parallel_tasks = 2;
+ const int64_t limit = 10;
+ auto scan_operator = std::make_unique<OlapScanOperatorX>(obj_pool.get(),
tnode, 0, *descs,
+ parallel_tasks,
TQueryCacheParam {});
+ auto olap_scan_local_state =
+ OlapScanLocalState::create_unique(state.get(),
scan_operator.get());
+
+ OlapScanner::Params scanner_params;
+ scanner_params.state = state.get();
+ scanner_params.profile = profile.get();
+ scanner_params.limit = limit;
+ scanner_params.key_ranges = std::vector<OlapScanRange*>();
+ std::shared_ptr<Scanner> scanner =
+ OlapScanner::create_shared(olap_scan_local_state.get(),
std::move(scanner_params));
+ std::list<std::shared_ptr<ScannerDelegate>> scanners {
+ std::make_shared<ScannerDelegate>(scanner),
std::make_shared<ScannerDelegate>(scanner)};
+ auto scanner_context = ScannerContext::create_shared(
+ state.get(), olap_scan_local_state.get(), output_tuple_desc,
false, scanners, limit,
+ scan_dependency, &shared_limit, nullptr, nullptr, 0, false,
parallel_tasks);
+ scanner_context->_newly_create_free_blocks_num =
newly_create_free_blocks_num.get();
+ scanner_context->_scanner_memory_used_counter =
scanner_memory_used_counter.get();
+
+ // Both schedulers resubmit through schedule_scan_task(); an exhausted
LIMIT must not reach it.
+ std::unique_ptr<MockSimplifiedScanScheduler> scheduler =
+ std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl);
+ EXPECT_CALL(*scheduler, schedule_scan_task(testing::_, testing::_,
testing::_)).Times(0);
+ scanner_context->_scanner_scheduler = scheduler.get();
+
+ // Scanners of a parallel peer instance produced the whole operator-wide
LIMIT while this
+ // context has scanner B blocked in I/O. Scanner A returned fewer rows
than this instance's
+ // LIMIT and now completes with an empty block but without EOS, as the
LIMIT is not EOS.
+ auto task_a = std::make_shared<ScanTask>(scanners.front());
+ auto task_b = std::make_shared<ScanTask>(scanners.back());
+ {
+ std::unique_lock<std::mutex>
transfer_lock(scanner_context->transfer_lock());
+ scanner_context->_pending_tasks =
std::stack<std::shared_ptr<ScanTask>>();
+ shared_limit.store(0);
+ task_a->set_state(ScanTask::State::IN_FLIGHT);
+ task_b->set_state(ScanTask::State::IN_FLIGHT);
+ scanner_context->_in_flight_tasks_num = 2;
+ }
+ task_a->cached_block = Block::create_unique();
+ task_a->set_state(ScanTask::State::COMPLETED);
+ scanner_context->push_completed_scan_task(task_a);
+
+ // Consuming the empty result parks A instead of resubmitting it: A would
only complete empty
+ // again for as long as B's I/O lasts. B's completion is what wakes the
operator next.
+ Block block;
+ bool eos = false;
+ Status st = scanner_context->get_block_from_queue(state.get(), &block,
&eos, 0);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(block.rows(), 0);
+ {
+ std::unique_lock<std::mutex>
transfer_lock(scanner_context->transfer_lock());
+ EXPECT_FALSE(scanner_context->done());
+ EXPECT_EQ(scanner_context->_in_flight_tasks_num, 1);
+ EXPECT_EQ(task_a->_state, ScanTask::State::PENDING);
+ ASSERT_EQ(scanner_context->_pending_tasks.size(), 1);
+ EXPECT_EQ(scanner_context->_pending_tasks.top(), task_a);
+ }
+
+ // B drains with no more rows to add. Nothing is in flight anymore, so the
context finishes
+ // without ever running A again.
+ task_b->cached_block = Block::create_unique();
+ task_b->set_state(ScanTask::State::COMPLETED);
+ scanner_context->push_completed_scan_task(task_b);
+ st = scanner_context->get_block_from_queue(state.get(), &block, &eos, 0);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ EXPECT_TRUE(eos);
+ {
+ std::unique_lock<std::mutex>
transfer_lock(scanner_context->transfer_lock());
+ EXPECT_TRUE(scanner_context->_is_finished);
+ EXPECT_EQ(scanner_context->_in_flight_tasks_num, 0);
+ ASSERT_EQ(scanner_context->_pending_tasks.size(), 1);
+ EXPECT_EQ(scanner_context->_pending_tasks.top(), task_a);
+ }
+}
+
TEST_F(ScannerContextTest, thread_pool_admission_refreshes_adaptive_limit) {
const int parallel_tasks = 2;
auto scan_operator = std::make_unique<OlapScanOperatorX>(obj_pool.get(),
tnode, 0, *descs,
diff --git a/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
b/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
index a21e7f47bda..5989ce1e7f0 100644
--- a/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
+++ b/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
@@ -19,6 +19,7 @@
#include <gtest/gtest.h>
#include <atomic>
+#include <functional>
#include <latch>
#include <list>
#include <thread>
@@ -62,6 +63,9 @@ public:
: Scanner(state, local_state, limit, profile),
_allow_padding(allow_padding) {}
void add_block(Block block) { _blocks.push_back(std::move(block)); }
+ size_t unread_blocks() const { return _blocks.size(); }
+ // Runs after each source block is handed out, to model a peer scanner
acting mid-read.
+ void set_after_read_hook(std::function<void()> hook) { _after_read_hook =
std::move(hook); }
protected:
bool _can_merge_padding_blocks(const Block& /*left*/, const Block&
/*right*/) const override {
@@ -76,12 +80,16 @@ protected:
*eof = false;
block->swap(_blocks.front());
_blocks.pop_front();
+ if (_after_read_hook) {
+ _after_read_hook();
+ }
return Status::OK();
}
private:
bool _allow_padding = true;
std::list<Block> _blocks;
+ std::function<void()> _after_read_hook;
};
class FakeTablet final : public BaseTablet {
@@ -739,6 +747,45 @@ TEST_F(ScannerLateArrivalRfTest,
olap_scanner_releases_resources_before_open) {
prepared_scanner->update_realtime_counters();
}
+// Admission does not look at the shared LIMIT, so a pending scanner is still
scheduled after its
+// peers have exhausted it. The scanner returns an empty block on that call
instead of reading its
+// range, and does not report EOS: the scan operator decides that from the
LIMIT.
+TEST(ScannerProjectionTest,
shared_limit_exhausted_before_call_returns_empty_block_without_eos) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ state._batch_size = 8;
+
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(4);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0, 1}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({2, 3, 4}));
+
+ // Peer scanners returned the whole LIMIT before this scanner ran.
+ op->_shared_scan_limit.fetch_sub(4);
+
+ Block output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &output, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ EXPECT_EQ(scanner.unread_blocks(), 2);
+ EXPECT_EQ(op->_shared_scan_limit.load(), 0);
+}
+
TEST(ScannerProjectionTest,
merges_padding_block_when_limit_eos_without_extra_flag) {
ObjectPool pool;
auto data_type = std::make_shared<DataTypeInt32>();
@@ -854,4 +901,442 @@ TEST(ScannerProjectionTest,
projects_incompatible_blocks_before_reading_the_next
EXPECT_EQ(final_output.rows(), 0);
}
+// With parallel scanners, the scan context may finish as soon as the shared
LIMIT counter is
+// exhausted. Count only the rows actually produced: after the first call, one
row is still held
+// in padding, so a peer returning four rows leaves one row for this scanner
to produce.
+TEST(ScannerProjectionTest, shared_limit_counts_only_emitted_rows) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ state._batch_size = 8;
+
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(10);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ // The first block is below the padding threshold and is held while the
second block, which
+ // reaches the threshold on its own, is emitted.
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({1, 2, 3, 4,
5}));
+
+ Block first_output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &first_output,
&eos).ok());
+ EXPECT_FALSE(eos);
+ ASSERT_EQ(first_output.rows(), 5);
+ EXPECT_EQ(first_output.get_by_position(0).column->get_int(0), 1);
+ // Only the five returned rows consume the shared LIMIT; the held row does
not.
+ EXPECT_EQ(op->_shared_scan_limit.load(), 5);
+
+ // A peer returns four more rows. Counting the held row before it is
produced would let the
+ // context finish at nine output rows instead of scheduling this scanner
again.
+ TestScanner peer(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(peer.init(&state, {}).ok());
+ peer.add_block(ColumnHelper::create_block<DataTypeInt32>({6, 7, 8, 9}));
+ Block peer_output;
+ bool peer_eos = false;
+ ASSERT_TRUE(peer.get_block_after_projects(&state, &peer_output,
&peer_eos).ok());
+ EXPECT_EQ(peer_output.rows(), 4);
+ EXPECT_EQ(op->_shared_scan_limit.load(), 1);
+
+ // The held row already satisfies the one remaining row, so flush it
without reading on. The
+ // reader still has no EOS to report; the operator stops the scanner after
cutting the LIMIT.
+ Block final_output;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &final_output,
&eos).ok());
+ EXPECT_FALSE(eos);
+ ASSERT_EQ(final_output.rows(), 1);
+ EXPECT_EQ(final_output.get_by_position(0).column->get_int(0), 0);
+ EXPECT_EQ(op->_shared_scan_limit.load(), 0);
+}
+
+// Padding stays unreturned when a peer has already produced the whole LIMIT.
It never consumed
+// shared credit, so skipping it cannot leave the operator short of rows, and
the scanner leaves
+// EOS to the operator's LIMIT cut.
+TEST(ScannerProjectionTest,
shared_limit_drops_padding_after_peer_satisfies_limit) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ state._batch_size = 8;
+
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(10);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({1, 2, 3, 4,
5}));
+ Block first_output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &first_output,
&eos).ok());
+ ASSERT_EQ(first_output.rows(), 5);
+ ASSERT_EQ(op->_shared_scan_limit.load(), 5);
+
+ TestScanner peer(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(peer.init(&state, {}).ok());
+ peer.add_block(ColumnHelper::create_block<DataTypeInt32>({6, 7, 8, 9,
10}));
+ Block peer_output;
+ bool peer_eos = false;
+ ASSERT_TRUE(peer.get_block_after_projects(&state, &peer_output,
&peer_eos).ok());
+ ASSERT_EQ(peer_output.rows(), 5);
+ ASSERT_EQ(op->_shared_scan_limit.load(), 0);
+
+ Block skipped_output;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &skipped_output,
&eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(skipped_output.rows(), 0);
+ EXPECT_EQ(op->_shared_scan_limit.load(), 0);
+}
+
+// LIMIT 1 must not turn padding into a long scan through a selective tail
just to reach half a
+// batch. The first surviving row is returned and charged before the next
source block is read.
+TEST(ScannerProjectionTest, shared_limit_flushes_padding_before_reading_more) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ state._batch_size = 8;
+
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(1);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({1, 2, 3, 4,
5}));
+
+ Block output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &output, &eos).ok());
+ EXPECT_FALSE(eos);
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_EQ(output.get_by_position(0).column->get_int(0), 0);
+ EXPECT_EQ(scanner.unread_blocks(), 1);
+ EXPECT_EQ(op->_shared_scan_limit.load(), 0);
+}
+
+// Several small source blocks can satisfy the LIMIT together without giving
up the normal
+// small-block merge or scanning one more source block to reach the half-batch
target.
+TEST(ScannerProjectionTest, shared_limit_flushes_merged_padding_at_limit) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ state._batch_size = 8;
+
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(2);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({1}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({2, 3, 4, 5}));
+
+ Block output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &output, &eos).ok());
+ EXPECT_FALSE(eos);
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(output.get_by_position(0).column->get_int(0), 0);
+ EXPECT_EQ(output.get_by_position(0).column->get_int(1), 1);
+ EXPECT_EQ(scanner.unread_blocks(), 1);
+ EXPECT_EQ(op->_shared_scan_limit.load(), 0);
+}
+
+// The no-projection path must charge the returned block once as well; it does
not use padding.
+TEST(ScannerProjectionTest,
shared_limit_counts_returned_rows_without_projection) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ op->_output_tuple_desc = row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(2);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0, 1, 2}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({3}));
+
+ Block output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &output, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(output.rows(), 3);
+ EXPECT_EQ(scanner.unread_blocks(), 1);
+ EXPECT_EQ(op->_shared_scan_limit.load(), -1);
+}
+
+// File-local projections that cannot pad still use the same output-only
shared accounting.
+TEST(ScannerProjectionTest, shared_limit_counts_returned_rows_without_padding)
{
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(2);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile, false);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0, 1, 2}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({3}));
+
+ Block output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &output, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(output.rows(), 3);
+ EXPECT_EQ(scanner.unread_blocks(), 1);
+ EXPECT_EQ(op->_shared_scan_limit.load(), -1);
+}
+
+// A selective scan under a large shared LIMIT still coalesces small blocks:
one call reads several
+// source blocks and returns them as one block, instead of handing every
surviving row to the
+// operator alone.
+TEST(ScannerProjectionTest, shared_limit_keeps_padding_small_blocks) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ state._batch_size = 8;
+
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(100);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ for (int i = 0; i < 6; ++i) {
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({i}));
+ }
+
+ // Four single-row blocks fill half a batch and are returned together.
+ Block output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &output, &eos).ok());
+ EXPECT_FALSE(eos);
+ ASSERT_EQ(output.rows(), 4);
+ for (int i = 0; i < 4; ++i) {
+ EXPECT_EQ(output.get_by_position(0).column->get_int(i), i);
+ }
+ EXPECT_EQ(op->_shared_scan_limit.load(), 96);
+
+ // The tail is returned with eos; nothing is held across the calls.
+ Block tail;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &tail, &eos).ok());
+ EXPECT_TRUE(eos);
+ ASSERT_EQ(tail.rows(), 2);
+ EXPECT_EQ(op->_shared_scan_limit.load(), 94);
+}
+
+// Concurrent scanners may produce more than the shared LIMIT before the
operator cuts their
+// output. A scanner initialized after that must still retain the pointer and
stop without reading.
+TEST(ScannerProjectionTest,
shared_limit_observed_by_scanner_initialized_under_negative_counter) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ state._batch_size = 8;
+
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(8);
+ // Peers produced nine rows for LIMIT 8 before the operator made its final
cut.
+ op->_shared_scan_limit.store(-1);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ ASSERT_EQ(scanner._shared_scan_limit, &op->_shared_scan_limit);
+
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0, 1, 2}));
+
+ Block output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &output, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ EXPECT_EQ(op->_shared_scan_limit.load(), -1);
+ EXPECT_EQ(scanner.unread_blocks(), 1);
+}
+
+// Without a SQL LIMIT the scanner gets no counter, whatever the counter holds.
+TEST(ScannerProjectionTest, no_shared_limit_without_sql_limit) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ ASSERT_EQ(op->limit(), -1);
+ op->_shared_scan_limit.store(5);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ EXPECT_EQ(scanner._shared_scan_limit, nullptr);
+}
+
+// Once its own rows exhaust the shared LIMIT the scanner stops reading, even
though its reader
+// still has data. It does not report EOS for that: the next call returns an
empty block and the
+// scan operator stops the scanner after cutting the LIMIT.
+TEST(ScannerProjectionTest,
shared_limit_exhausted_by_own_rows_stops_reading_without_eos) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ state._batch_size = 8;
+
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(3);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0, 1, 2, 3,
4}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({5, 6, 7, 8,
9}));
+
+ Block output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &output, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(output.rows(), 5);
+ EXPECT_EQ(scanner.unread_blocks(), 1);
+ EXPECT_EQ(op->_shared_scan_limit.load(), -2);
+
+ Block next_output;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &next_output,
&eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(next_output.rows(), 0);
+ EXPECT_EQ(scanner.unread_blocks(), 1);
+ EXPECT_EQ(op->_shared_scan_limit.load(), -2);
+}
+
+// A peer may exhaust the shared LIMIT while this scanner is still padding
small blocks. The read
+// loop stops at once, the rows already held are returned, and EOS is again
left to the operator.
+TEST(ScannerProjectionTest,
shared_limit_exhausted_by_peer_mid_read_returns_held_rows_without_eos) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+
+ MockRuntimeState state;
+ state._batch_size = 8;
+
+ auto op = std::make_shared<MockScanOperatorX>();
+ op->_row_descriptor = row_descriptor;
+ auto& projection = op->set_projection_for_test(
+ MockRowDescriptor(std::vector<DataTypePtr> {data_type}, &pool));
+ op->_output_tuple_desc =
projection.output_row_descriptor.tuple_descriptors()[0];
+ op->set_limit_for_test(10);
+
+ auto local_state = std::make_shared<MockScanLocalState>(&state, op.get());
+ local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ for (int i = 0; i < 5; ++i) {
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({i}));
+ }
+ // A peer produces the whole LIMIT right after this scanner reads its
first block.
+ bool peer_done = false;
+ scanner.set_after_read_hook([&] {
+ if (!peer_done) {
+ peer_done = true;
+ op->_shared_scan_limit.fetch_sub(10);
+ }
+ });
+
+ Block output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &output, &eos).ok());
+ EXPECT_FALSE(eos);
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_EQ(output.get_by_position(0).column->get_int(0), 0);
+ EXPECT_EQ(scanner.unread_blocks(), 4);
+ EXPECT_EQ(op->_shared_scan_limit.load(), -1);
+}
+
} // namespace doris
diff --git a/be/test/runtime/workload_group/workload_group_manager_test.cpp
b/be/test/runtime/workload_group/workload_group_manager_test.cpp
index db047257af5..a7f4543dc81 100644
--- a/be/test/runtime/workload_group/workload_group_manager_test.cpp
+++ b/be/test/runtime/workload_group/workload_group_manager_test.cpp
@@ -41,6 +41,7 @@
#include "common/status.h"
#include "cpp/sync_point.h"
#include "exec/pipeline/dependency.h"
+#include "exec/scan/scanner_scheduler.h"
#include "exec/spill/spill_file_manager.h"
#include "load/memtable/memtable_memory_limiter.h"
#include "runtime/exec_env.h"
@@ -1489,6 +1490,62 @@ TEST_F(WorkloadGroupManagerTest,
FailedInternalGroupSetupCancelsAdaptiveFlush) {
controller->stop();
}
+TEST_F(WorkloadGroupManagerTest, DefaultScanSchedulersUseThreadPool) {
+ // Check the registered defaults, which do not depend on the be.conf
loaded by the UT runner.
+ const auto& fields = *config::Register::_s_field_map;
+ EXPECT_STREQ("false",
fields.at("enable_task_executor_in_internal_table").defval);
+ EXPECT_STREQ("false",
fields.at("enable_task_executor_in_external_table").defval);
+
+ const auto saved_config = std::make_tuple(
+ config::enable_adaptive_flush_threads,
config::enable_task_executor_in_internal_table,
+ config::enable_task_executor_in_external_table,
config::pipeline_executor_size,
+ config::blocking_pipeline_executor_size,
config::doris_scanner_thread_pool_thread_num,
+ config::doris_max_remote_scanner_thread_pool_thread_num,
+ config::doris_scanner_min_thread_pool_thread_num,
config::min_active_scan_threads,
+ config::min_active_file_scan_threads,
config::flush_thread_num_per_store);
+ Defer restore {[&] {
+ std::tie(config::enable_adaptive_flush_threads,
+ config::enable_task_executor_in_internal_table,
+ config::enable_task_executor_in_external_table,
config::pipeline_executor_size,
+ config::blocking_pipeline_executor_size,
+ config::doris_scanner_thread_pool_thread_num,
+ config::doris_max_remote_scanner_thread_pool_thread_num,
+ config::doris_scanner_min_thread_pool_thread_num,
config::min_active_scan_threads,
+ config::min_active_file_scan_threads,
config::flush_thread_num_per_store) =
+ saved_config;
+ }};
+ config::enable_adaptive_flush_threads = false;
+ config::enable_task_executor_in_internal_table = false;
+ config::enable_task_executor_in_external_table = false;
+ config::pipeline_executor_size = 1;
+ config::blocking_pipeline_executor_size = 1;
+ config::doris_scanner_thread_pool_thread_num = 1;
+ config::doris_max_remote_scanner_thread_pool_thread_num = 1;
+ config::doris_scanner_min_thread_pool_thread_num = 1;
+ config::min_active_scan_threads = 1;
+ config::min_active_file_scan_threads = 1;
+ config::flush_thread_num_per_store = 1;
+
+ ASSERT_TRUE(_wg_manager->create_internal_wg().ok());
+ Defer stop_schedulers {[&] {
+ for (auto& entry : _wg_manager->_workload_groups) {
+ entry.second->try_stop_schedulers();
+ entry.second->destroy_schedulers();
+ }
+ }};
+ ASSERT_EQ(_wg_manager->_workload_groups.size(), 1);
+ auto wg = _wg_manager->_workload_groups.begin()->second;
+
+ TaskScheduler* exec_sched = nullptr;
+ ScannerScheduler* scan_sched = nullptr;
+ ScannerScheduler* remote_scan_sched = nullptr;
+ wg->get_query_scheduler(&exec_sched, &scan_sched, &remote_scan_sched);
+ ASSERT_NE(scan_sched, nullptr);
+ ASSERT_NE(remote_scan_sched, nullptr);
+ EXPECT_NE(dynamic_cast<ThreadPoolSimplifiedScanScheduler*>(scan_sched),
nullptr);
+
EXPECT_NE(dynamic_cast<ThreadPoolSimplifiedScanScheduler*>(remote_scan_sched),
nullptr);
+}
+
// Exercise the actual registration/cancellation paths without starting query
schedulers.
TEST_F(WorkloadGroupManagerTest,
AdaptiveFlushRegistrationSurvivesIdChangeAndReuse) {
auto* env = ExecEnv::GetInstance();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
