mrhhsg commented on code in PR #68610:
URL: https://github.com/apache/doris/pull/68610#discussion_r4280990015
##########
be/src/exec/scan/scanner.cpp:
##########
@@ -87,8 +89,17 @@ 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.
+ if (_shared_scan_limit_exhausted()) {
Review Comment:
Fixed in 62afc6b03b7. The exhausted scanner is now parked instead of
resubmitted: `ScannerContext::get_block_from_queue()` pushes a consumed non-EOS
task back to `_pending_tasks` without calling `schedule_scan_task()` when
`_is_shared_scan_limit_exhausted()`. This covers both schedulers, since the
ThreadPool path (`push_pending_scan_task()` + Context runnable, where admission
ignores the LIMIT by design) and the TaskExecutor path
(`_pull_next_scan_task()` returns the current task unconditionally) both
resubmit from this one call.
Progress is still guaranteed from the same call: the branch is only reached
when the finish check right above it did not fire, i.e. there is a completed or
in-flight task left for this context. Its completion wakes the operator
(`push_completed_scan_task()` → `set_ready()`), and the context finishes once
the counter is exhausted with nothing in flight, exactly as before. The parked
scanner never runs again; the scanner itself still only stops reading and never
reports EOS for the LIMIT.
Test:
`ScannerContextTest.exhausted_shared_limit_parks_empty_task_until_in_flight_drains`
drives this interleaving through `get_block_from_queue()`: with the
operator-wide counter at 0, scanner B in flight and scanner A completing empty
without EOS, A ends up PENDING, the scheduler's `schedule_scan_task()` is
called zero times, and the context finishes (`eos == true`) when B's completion
drains with A still parked. The scan UT filter passed, and `query_p0/limit`
plus `correctness_p0/test_shared_scan_limit_pending_tasks` pass on a local
cluster with the ThreadPool default.
##########
be/test/exec/rowid_fetcher_test.cpp:
##########
@@ -429,4 +438,64 @@ 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);
+
+ 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(); }};
Review Comment:
Fixed in 62afc6b03b7. `response` and `finished` are now declared before the
scheduler and its `stop_scheduler` guard in
`RowIdStorageReaderTest.RpcHelperReportsThrownExceptionOnThreadPoolPath`, so on
every exit, including a failed `ASSERT_TRUE(finished.wait_for(...))`,
`scheduler.stop()` joins the worker before either captured object is destroyed.
The test still passes on the real ThreadPool scheduler.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
