github-actions[bot] commented on code in PR #69507:
URL: https://github.com/apache/doris/pull/69507#discussion_r4356447355
##########
be/src/exec/scan/scanner.cpp:
##########
@@ -143,24 +144,36 @@ Status Scanner::get_block_after_projects(RuntimeState*
state, Block* block, bool
RETURN_IF_ERROR(_merge_padding_block());
_origin_block.clear_column_data(row_descriptor.num_slots());
} else {
- if (_origin_block.rows() < _padding_block.rows()) {
+ // With LIMIT 10, a padded row followed by five rows must
not be swapped
+ // behind the new block: returning five would hide the
first row from peers.
+ if (_shared_scan_limit == nullptr &&
+ _origin_block.rows() < _padding_block.rows()) {
_padding_block.swap(_origin_block);
}
break;
}
}
+ // A bounded scan must publish every surviving row before peers
read on. For example,
+ // a one-row padding block and a five-row origin block together
return six rows now,
+ // even when the origin alone reached the soft batch target.
+ if (_shared_scan_limit != nullptr && !_padding_block.empty() &&
+ !_origin_block.empty()) {
+ RETURN_IF_ERROR(_merge_padding_block());
Review Comment:
[P2] Preserve the hard String column limit when flushing padding
With adaptive batching disabled, batch size 4096, LIMIT 10000 and sufficient
query memory, a projected Parquet scan of `length(s)` can read successive row
groups containing 20 and 4096 strings of 1,046,300 bytes each. The raw String
columns are individually valid (20,926,000 and 4,285,644,800 bytes), and the
first stays below the padding byte target. This new append combines them into
4,306,570,800 bytes, exceeding `ColumnString`'s 4,294,967,295-byte limit:
`insert_range_from()` throws `STRING_OVERFLOW_IN_VEC_ENGINE` before the
projection can produce its small integer results. Previously the 4096-row block
and pending rows were projected separately. Preserve the representable column
capacity when flushing padding, for example by projecting these inputs
separately before combining their outputs.
##########
be/test/exec/scan/scanner_late_arrival_rf_test.cpp:
##########
@@ -901,9 +930,217 @@ 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.
+// A source block can reach the soft batch target while earlier rows are in
padding. With a
+// shared LIMIT, those earlier rows must be emitted in the same call, even
past that soft target.
+TEST(ScannerProjectionTest,
shared_limit_emits_overflow_padding_without_hidden_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());
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0, 1, 2}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({3, 4, 5, 6,
7, 8, 9}));
+
+ Block output;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &output, &eos).ok());
+ EXPECT_FALSE(eos);
+ ASSERT_EQ(output.rows(), 10);
+ for (int i = 0; i < 10; ++i) {
+ EXPECT_EQ(output.get_by_position(0).column->get_int(i), i);
+ }
+ EXPECT_EQ(op->_shared_scan_limit.load(), 0);
+ EXPECT_EQ(scanner.read_calls(), 2);
+
+ Block skipped;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &skipped, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(skipped.rows(), 0);
+ EXPECT_EQ(scanner.read_calls(), 2);
+}
+
+TEST(ScannerProjectionTest,
shared_limit_stops_filtered_peer_after_emitting_padding) {
+ 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);
+ // A second parallel instance has its own local state but sees the
operator's counter.
+ auto other_local_state = std::make_shared<MockScanLocalState>(&state,
op.get());
+ other_local_state->_projections = MockSlotRef::create_mock_contexts(0,
data_type);
+ RuntimeProfile profile("scanner");
+ TestScanner scanner(&state, local_state.get(), -1, &profile);
+ TestScanner peer(&state, local_state.get(), -1, &profile);
+ TestScanner filtered_peer(&state, other_local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner.init(&state, {}).ok());
+ ASSERT_TRUE(peer.init(&state, {}).ok());
+ ASSERT_TRUE(filtered_peer.init(&state, {}).ok());
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({0}));
+ scanner.add_block(ColumnHelper::create_block<DataTypeInt32>({1, 2, 3, 4,
5}));
+ peer.add_block(ColumnHelper::create_block<DataTypeInt32>({6, 7, 8, 9}));
+ filtered_peer.set_filter_all_rows(true);
+ // More than one get_block() raw-row budget, if the shared counter stays
positive.
+ for (int i = 0; i < 40; ++i) {
+ filtered_peer.add_block(
+
ColumnHelper::create_block<DataTypeInt32>(std::vector<int32_t>(1024, i)));
+ }
+
+ Block first;
+ Block second;
+ Block tail;
+ bool eos = false;
+ ASSERT_TRUE(scanner.get_block_after_projects(&state, &first, &eos).ok());
+ ASSERT_TRUE(peer.get_block_after_projects(&state, &second, &eos).ok());
+ ASSERT_TRUE(filtered_peer.get_block_after_projects(&state, &tail,
&eos).ok());
+ EXPECT_EQ(first.rows() + second.rows() + tail.rows(), 10);
+ EXPECT_EQ(filtered_peer.read_calls(), 0);
+ EXPECT_EQ(filtered_peer.unread_blocks(), 40);
+ EXPECT_EQ(op->_shared_scan_limit.load(), 0);
+}
+
+// Keep the third worker at open until the first two have published their
blocks. This fixes the
+// scheduling interleaving without consuming either result, so neither
producer can be rescheduled
+// to flush old padding before the filtered worker checks the shared LIMIT.
+// The GTest assertion macros and bounded polling inflate the measured
cognitive complexity.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity)
+TEST(ScannerProjectionTest,
thread_pool_limit_stops_filtered_peer_before_source_read) {
+ ObjectPool pool;
+ auto data_type = std::make_shared<DataTypeInt32>();
+ auto row_descriptor = MockRowDescriptor({data_type}, &pool);
+ MockRuntimeState state;
+ state._batch_size = 8;
+ auto task_exec_ctx = std::make_shared<TaskExecutionContext>();
+ state.set_task_execution_context(task_exec_ctx);
+
+ 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("projected scan");
+ local_state->_scan_timer = ADD_TIMER(&profile, "ScannerGetBlockTime");
+ local_state->_scan_cpu_timer = ADD_TIMER(&profile, "ScannerCpuTime");
+ local_state->_filter_timer = ADD_TIMER(&profile, "ScannerFilterTime");
+ local_state->_rows_read_counter = ADD_COUNTER(&profile, "RowsRead",
TUnit::UNIT);
+ local_state->_max_scan_concurrency = ADD_COUNTER(&profile,
"MaxConcurrency", TUnit::UNIT);
+ local_state->_min_scan_concurrency = ADD_COUNTER(&profile,
"MinConcurrency", TUnit::UNIT);
+
+ auto scanner = std::make_shared<TestScanner>(&state, local_state.get(),
-1, &profile);
+ auto peer = std::make_shared<TestScanner>(&state, local_state.get(), -1,
&profile);
+ auto filtered_peer = std::make_shared<TestScanner>(&state,
local_state.get(), -1, &profile);
+ ASSERT_TRUE(scanner->init(&state, {}).ok());
+ ASSERT_TRUE(peer->init(&state, {}).ok());
+ ASSERT_TRUE(filtered_peer->init(&state, {}).ok());
+ scanner->add_block(ColumnHelper::create_block<DataTypeInt32>({0}));
+ scanner->add_block(ColumnHelper::create_block<DataTypeInt32>({1, 2, 3, 4,
5}));
+ peer->add_block(ColumnHelper::create_block<DataTypeInt32>({6, 7, 8, 9}));
+ filtered_peer->set_filter_all_rows(true);
+ for (int i = 0; i < 40; ++i) {
+ filtered_peer->add_block(
+
ColumnHelper::create_block<DataTypeInt32>(std::vector<int32_t>(1024, i)));
+ }
+
+ CountDownLatch release_filtered_peer(1);
+ std::atomic<bool> entered_open {false};
+ filtered_peer->set_open_hook([&] {
+ entered_open.store(true);
+ release_filtered_peer.wait();
+ });
+
+ // Pending scanners are LIFO: the two productive scanners run before the
gated peer.
+ std::list<std::shared_ptr<ScannerDelegate>> delegates;
+ for (const auto& source : {filtered_peer, peer, scanner}) {
+ std::shared_ptr<Scanner> base = source;
+ delegates.push_back(std::make_shared<ScannerDelegate>(base));
+ }
+ auto dependency = Dependency::create_shared(0, 0, "projected scan
dependency");
+ auto context = ScannerContext::create_shared(
+ &state, local_state.get(), row_descriptor.tuple_descriptors()[0],
true, delegates, 10,
+ dependency, &op->_shared_scan_limit, nullptr, nullptr, 0, false,
3);
+ context->_newly_create_free_blocks_num =
+ ADD_COUNTER(&profile, "NewlyCreatedFreeBlocks", TUnit::UNIT);
+ context->_scanner_memory_used_counter =
+ ADD_COUNTER(&profile, "ScannerMemoryUsed", TUnit::BYTES);
+
+ auto cgroup = std::make_shared<CgroupV2CpuCtl>(1);
+ ThreadPoolSimplifiedScanScheduler scheduler("projected_scan_limit_test",
cgroup);
+ ASSERT_TRUE(scheduler.start(3, 3, 16, 1).ok());
+ Defer cleanup = [&] {
+ release_filtered_peer.count_down();
+ scheduler.stop();
+ };
+ context->_scanner_scheduler = &scheduler;
+ context->_min_scan_concurrency_of_scan_scheduler = 20;
+ ASSERT_TRUE(context->init().ok());
+
+ const auto deadline = std::chrono::steady_clock::now() +
std::chrono::seconds(20);
+ auto wait_for_completed = [&](size_t count) {
+ while (std::chrono::steady_clock::now() < deadline) {
+ {
+ std::unique_lock<std::mutex> lock(context->transfer_lock());
+ if (context->_completed_tasks.size() >= count) {
+ return true;
+ }
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ return false;
+ };
+ ASSERT_TRUE(wait_for_completed(2)) << context->debug_string();
+ ASSERT_TRUE(entered_open.load());
Review Comment:
[P2] Wait for the gated worker before asserting it entered open
The two productive scanners can publish both completed tasks while the third
Context runnable is still queued (or preempted before its open hook).
`_run_context()` only submits that runnable; neither the LIFO pending queue nor
`wait_for_completed(2)` establishes that `entered_open` has been set. This
assertion can therefore fail on a correct implementation, making the new BE
test flaky under ordinary scheduling variation. Wait for the open hook
explicitly, using a latch or a bounded wait, before asserting/releasing the
peer.
--
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]
