github-actions[bot] commented on code in PR #68610:
URL: https://github.com/apache/doris/pull/68610#discussion_r4280785812
##########
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:
[P2] Park exhausted scanners while outstanding work drains. With parallel
scanning enabled, another local context can exhaust this operator-wide counter
while context X has returned fewer than its own LIMIT and still has scanner B
blocked in I/O. Scanner A now completes empty with `eos == false`;
`get_block_from_queue()` sees B in flight, fails its finish check, and
resubmits A. A can repeat this empty completion/submission cycle for all of B's
I/O latency, consuming scanner and pipeline CPU although the shared LIMIT is
already satisfied. Previously the shared-limit EOF path retired A after one
completion. Preserve reader EOS semantics, but give exhausted tasks a quiescent
context/operator path instead of resubmitting them, and cover this interleaving
with a scheduler-level test.
##########
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:
[P2] Join the RPC worker before destroying its captured locals. If the
30-second wait times out while the callback is still running, the fatal
assertion returns from the test and destroys `finished` and `response` before
this earlier-declared `stop_scheduler` guard runs. The worker can then write
the destroyed protobuf or call `count_down()` on the destroyed latch while
`scheduler.stop()` waits for it, turning a timeout into a use-after-free/crash.
Declare the response and latch before the stop guard so every exit joins active
work before destroying either captured object.
--
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]
