github-actions[bot] commented on code in PR #67762:
URL: https://github.com/apache/doris/pull/67762#discussion_r4022565968
##########
be/src/exec/rowid_fetcher.cpp:
##########
@@ -317,31 +323,47 @@ Status RowIdStorageReader::read_by_rowids(const
PMultiGetRequestV2& request,
return Status::OK();
}
+Status RowIdStorageReader::read_internal_segment_groups(
+ size_t group_count, int batch_groups, int concurrency, bool
fetch_row_store,
+ const std::function<Status(size_t, size_t)>& read_groups) {
+ if (group_count == 0) {
+ return Status::OK();
+ }
+ auto read_range = [&](size_t begin, size_t end) -> Status {
+ Status status;
+ // A thrown exception must not escape bthread_fork_join: it would skip
completion
+ // accounting and leave the RPC waiting for a task that can never
finish.
+ try {
+ ASSIGN_STATUS_IF_CATCH_EXCEPTION(status = read_groups(begin, end),
status);
+ } catch (const std::exception& e) {
+ status = Status::InternalError("Row id fetch failed because {}",
e.what());
+ }
+ return status;
+ };
+ if (fetch_row_store || batch_groups <= 0 || concurrency <= 1 ||
+ std::cmp_less_equal(group_count, batch_groups)) {
+ return read_range(0, group_count);
+ }
+ const auto groups_per_task = static_cast<size_t>(batch_groups);
+ std::vector<std::function<Status()>> tasks;
+ tasks.reserve(1 + (group_count - 1) / groups_per_task);
+ for (size_t begin = 0; begin < group_count; begin += groups_per_task) {
+ tasks.emplace_back([&, begin] {
Review Comment:
[P1] Make this task handoff exception-safe. `bthread_fork_join` starts
workers that capture its stack and elements of this `tasks` vector, but
allocates each heap callback after earlier workers may already be live. If a
later allocation throws, the helper unwinds before its final wait, then this
vector and the `read_groups` captures are destroyed while a worker can still
dereference them, causing a UAF. Please make the helper join every started
worker on dispatch exceptions, or preallocate all potentially throwing wrapper
state before starting the first worker.
##########
regression-test/suites/topn_optimize/lazy_materialize/parallel_rowid_fetch_v2.groovy:
##########
@@ -0,0 +1,80 @@
+// 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("parallel_rowid_fetch_v2") {
+ sql "DROP TABLE IF EXISTS parallel_rowid_fetch_v2"
+ sql """
+ CREATE TABLE parallel_rowid_fetch_v2 (
+ id INT NOT NULL,
+ sort_key INT NOT NULL,
+ payload STRING NULL,
+ values_array ARRAY<INT> NULL
+ ) DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 8
+ PROPERTIES ("replication_num" = "1", "disable_auto_compaction" =
"true")
+ """
+ // Separate rowsets and buckets give each request multiple segment groups.
+ for (int batch = 0; batch < 3; ++batch) {
+ sql """
+ INSERT INTO parallel_rowid_fetch_v2
+ SELECT number + ${batch * 200}, (number * 37) % 101,
+ IF(number % 7 = 0, NULL, CONCAT('payload-', CAST(number AS
STRING))),
+ ARRAY(CAST(number AS INT), NULL, CAST(number + 1 AS INT))
+ FROM numbers("number" = "200")
+ """
+ }
+ sql "DROP TABLE IF EXISTS parallel_rowid_fetch_v2_row_store"
+ sql """
+ CREATE TABLE parallel_rowid_fetch_v2_row_store (
+ id INT NOT NULL, sort_key INT NOT NULL, payload STRING NULL
+ ) DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 8
+ PROPERTIES ("replication_num" = "1", "store_row_column" = "true")
+ """
+ sql """INSERT INTO parallel_rowid_fetch_v2_row_store
+ SELECT id, sort_key, payload FROM parallel_rowid_fetch_v2"""
+ sql "SYNC"
+ sql "SET topn_opt_limit_threshold = 1024"
+
+ def queries = [
+ """SELECT id, payload, payload, values_array FROM
parallel_rowid_fetch_v2
+ ORDER BY sort_key, id LIMIT 137 OFFSET 3""",
+ // The join repeats source RowIDs and interleaves rows from different
segments.
+ """SELECT t.id, t.payload, t.values_array, d.x
+ FROM parallel_rowid_fetch_v2 t CROSS JOIN (SELECT 1 x UNION ALL
SELECT 2 x) d
+ ORDER BY t.sort_key, t.id, d.x LIMIT 137""",
+ """SELECT id, payload FROM parallel_rowid_fetch_v2_row_store
+ ORDER BY sort_key, id LIMIT 137""",
+ """SELECT id, payload FROM parallel_rowid_fetch_v2
+ WHERE id = -1 ORDER BY sort_key, id LIMIT 10"""
+ ]
+ sql "SET enable_two_phase_read_opt = false"
+ sql "SET topn_lazy_materialization_threshold = 0"
+ def expected = queries.collect { sql(it) }
+ sql "SET enable_two_phase_read_opt = true"
+ sql "SET topn_lazy_materialization_threshold = 1024"
+ explain {
+ sql queries[0]
+ contains "VMaterializeNode"
+ }
+ for (int batch : [0, -1, 1, 2, 100000]) {
+ sql "SET rowid_fetch_parallel_batch_rows = ${batch}"
Review Comment:
[P2] Make this test prove the parallel path is active. Every assertion below
still passes if `rowid_fetch_parallel_batch_rows` is dropped between
`TQueryOptions` and `PMultiGetRequestV2`, or if BE stays serial; the explain
check only proves `VMaterializeNode`. The FE test stops at `TQueryOptions` and
the BE tests invoke the grouping helper directly, so no test covers the new
request wiring. Please assert the generated protobuf field and add an
observable task-count, injection, or profile check for a positive value.
##########
be/src/exec/rowid_fetcher.cpp:
##########
@@ -317,31 +323,47 @@ Status RowIdStorageReader::read_by_rowids(const
PMultiGetRequestV2& request,
return Status::OK();
}
+Status RowIdStorageReader::read_internal_segment_groups(
+ size_t group_count, int batch_groups, int concurrency, bool
fetch_row_store,
+ const std::function<Status(size_t, size_t)>& read_groups) {
+ if (group_count == 0) {
+ return Status::OK();
+ }
+ auto read_range = [&](size_t begin, size_t end) -> Status {
+ Status status;
+ // A thrown exception must not escape bthread_fork_join: it would skip
completion
+ // accounting and leave the RPC waiting for a task that can never
finish.
+ try {
+ ASSIGN_STATUS_IF_CATCH_EXCEPTION(status = read_groups(begin, end),
status);
+ } catch (const std::exception& e) {
+ status = Status::InternalError("Row id fetch failed because {}",
e.what());
+ }
+ return status;
+ };
+ if (fetch_row_store || batch_groups <= 0 || concurrency <= 1 ||
+ std::cmp_less_equal(group_count, batch_groups)) {
+ return read_range(0, group_count);
+ }
+ const auto groups_per_task = static_cast<size_t>(batch_groups);
+ std::vector<std::function<Status()>> tasks;
+ tasks.reserve(1 + (group_count - 1) / groups_per_task);
+ for (size_t begin = 0; begin < group_count; begin += groups_per_task) {
+ tasks.emplace_back([&, begin] {
+ return read_range(begin, std::min(begin + groups_per_task,
group_count));
+ });
+ }
+ return cloud::bthread_fork_join(tasks, concurrency);
Review Comment:
[P1] Keep these storage reads off bthreads. A positive batch setting sends
`read_doris_format_row` through this helper, and an uncached page reaches
`FileReader::read_at`, whose entry `DCHECK`s that `bthread_self() == 0`. This
aborts DCHECK-enabled builds and runs unsupported blocking file/network I/O in
release builds. The worker also mutates pthread-local bad-allocation and
signal-query state; migration can decrement the former on a different worker,
while the latter is left on whichever worker executed the set. Please schedule
the whole read on the existing scanner pthread pool (or add a supported offload
boundary) rather than invoking this synchronous storage stack from bthreads.
--
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]
