yiguolei commented on code in PR #47783: URL: https://github.com/apache/doris/pull/47783#discussion_r1956207742
########## be/test/scan/scanner_context_test.cpp: ########## @@ -0,0 +1,603 @@ +// 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. + +#include "vec/exec/scan/scanner_context.h" + +#include <gen_cpp/Descriptors_types.h> +#include <gen_cpp/Metrics_types.h> +#include <gen_cpp/PaloInternalService_types.h> +#include <gen_cpp/Types_types.h> +#include <gtest/gtest.h> + +#include <list> +#include <memory> +#include <mutex> +#include <tuple> + +#include "common/object_pool.h" +#include "mock_scanner_scheduler.h" +#include "mock_simplified_scan_scheduler.h" +#include "pipeline/dependency.h" +#include "pipeline/exec/olap_scan_operator.h" +#include "runtime/descriptors.h" +#include "vec/core/block.h" +#include "vec/exec/scan/new_olap_scanner.h" +#include "vec/exec/scan/scanner_scheduler.h" +#include "vec/exec/scan/vscan_node.h" + +namespace doris::vectorized { +class ScannerContextTest : public testing::Test { +public: + void SetUp() override { + obj_pool = std::make_unique<ObjectPool>(); + // This ScanNode has two tuples. + // First one is input tuple, second one is output tuple. + tnode.row_tuples.push_back(TTupleId(0)); + tnode.row_tuples.push_back(TTupleId(1)); + tbl_desc.tableType = TTableType::OLAP_TABLE; + + tuple_desc.id = 0; + tuple_descs.push_back(tuple_desc); + tuple_desc.id = 1; + tuple_descs.push_back(tuple_desc); + + type_node.type = TTypeNodeType::SCALAR; + + scalar_type.__set_type(TPrimitiveType::STRING); + type_node.__set_scalar_type(scalar_type); + slot_desc.slotType.types.push_back(type_node); + slot_desc.id = 0; + slot_desc.parent = 0; + slot_descs.push_back(slot_desc); + slot_desc.id = 1; + slot_desc.parent = 1; + slot_descs.push_back(slot_desc); + std::vector<bool> null_map {false, false}; + tnode.nullable_tuples = null_map; + thrift_tbl.tableDescriptors.push_back(tbl_desc); + thrift_tbl.tupleDescriptors = tuple_descs; + thrift_tbl.slotDescriptors = slot_descs; + std::ignore = DescriptorTbl::create(obj_pool.get(), thrift_tbl, &descs); + auto task_exec_ctx = std::make_shared<TaskExecutionContext>(); + state->set_task_execution_context(task_exec_ctx); + output_tuple_desc = descs->get_tuple_descriptor(0); + } + +private: + std::unique_ptr<ObjectPool> obj_pool; + TPlanNode tnode; + TTableDescriptor tbl_desc; + std::vector<TTupleDescriptor> tuple_descs; + TTupleDescriptor tuple_desc; + std::vector<TSlotDescriptor> slot_descs; + TSlotDescriptor slot_desc; + TTypeNode type_node; + TScalarType scalar_type; + TDescriptorTable thrift_tbl; + DescriptorTbl* descs = nullptr; + std::unique_ptr<RuntimeState> state = std::make_unique<RuntimeState>(); + std::unique_ptr<RuntimeProfile> profile = std::make_unique<RuntimeProfile>("TestProfile"); + std::unique_ptr<RuntimeProfile::Counter> max_concurrency_counter = + std::make_unique<RuntimeProfile::Counter>(TUnit::UNIT, 1, 3); + std::unique_ptr<RuntimeProfile::Counter> min_concurrency_counter = + std::make_unique<RuntimeProfile::Counter>(TUnit::UNIT, 1, 3); + TupleDescriptor* output_tuple_desc = nullptr; + RowDescriptor* output_row_descriptor = nullptr; + std::shared_ptr<pipeline::Dependency> scan_dependency = + pipeline::Dependency::create_shared(0, 0, "TestScanDependency"); + std::shared_ptr<CgroupCpuCtl> cgroup_cpu_ctl = std::make_shared<CgroupV2CpuCtl>(1); + std::unique_ptr<SimplifiedScanScheduler> scan_scheduler = + std::make_unique<SimplifiedScanScheduler>("ForTest", cgroup_cpu_ctl); +}; + +TEST_F(ScannerContextTest, test_init) { + const int parallel_tasks = 1; + auto scan_operator = std::make_unique<pipeline::OlapScanOperatorX>( + obj_pool.get(), tnode, 0, *descs, parallel_tasks, TQueryCacheParam {}); + + auto olap_scan_local_state = + pipeline::OlapScanLocalState::create_unique(state.get(), scan_operator.get()); + + const int64_t limit = 100; + + NewOlapScanner::Params scanner_params; + scanner_params.state = state.get(); + scanner_params.profile = profile.get(); + scanner_params.limit = limit; + scanner_params.key_ranges = std::vector<OlapScanRange*>(); // empty + + std::shared_ptr<VScanner> scanner = + NewOlapScanner::create_shared(olap_scan_local_state.get(), std::move(scanner_params)); + + std::list<std::shared_ptr<ScannerDelegate>> scanners; + for (int i = 0; i < 11; ++i) { + scanners.push_back(std::make_shared<ScannerDelegate>(scanner)); + } + + std::shared_ptr<ScannerContext> scanner_context = ScannerContext::create_shared( + state.get(), olap_scan_local_state.get(), output_tuple_desc, output_row_descriptor, + scanners, limit, scan_dependency, parallel_tasks); + + scan_operator->_should_run_serial = false; + + olap_scan_local_state->_max_scan_concurrency = max_concurrency_counter.get(); + olap_scan_local_state->_min_scan_concurrency = min_concurrency_counter.get(); + + olap_scan_local_state->_parent = scan_operator.get(); + + TQueryOptions query_options; + query_options.__set_num_scanner_threads(2); + query_options.__set_max_column_reader_num(0); + state->set_query_options(query_options); + std::unique_ptr<MockSimplifiedScanScheduler> scheduler = + std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + EXPECT_CALL(*scheduler, schedule_scan_task(testing::_, testing::_, testing::_)) + .WillRepeatedly(testing::Return(Status::OK())); + scanner_context->_scanner_scheduler = scheduler.get(); + + scanner_context->_min_scan_concurrency_of_scan_scheduler = 10; + Status st = scanner_context->init(); + ASSERT_TRUE(st.ok()); + ASSERT_EQ(scanner_context->_max_scan_concurrency, 2); + + query_options.__set_num_scanner_threads(0); + state->set_query_options(query_options); + + st = scanner_context->init(); + ASSERT_TRUE(st.ok()); + + ASSERT_EQ(scanner_context->_max_scan_concurrency, + scanner_context->_min_scan_concurrency_of_scan_scheduler / parallel_tasks); +} + +TEST_F(ScannerContextTest, test_serial_run) { + const int parallel_tasks = 1; + auto scan_operator = std::make_unique<pipeline::OlapScanOperatorX>( + obj_pool.get(), tnode, 0, *descs, parallel_tasks, TQueryCacheParam {}); + + auto olap_scan_local_state = + pipeline::OlapScanLocalState::create_unique(state.get(), scan_operator.get()); + + const int64_t limit = 100; + + NewOlapScanner::Params scanner_params; + scanner_params.state = state.get(); + scanner_params.profile = profile.get(); + scanner_params.limit = limit; + scanner_params.key_ranges = std::vector<OlapScanRange*>(); // empty + + std::shared_ptr<VScanner> scanner = + NewOlapScanner::create_shared(olap_scan_local_state.get(), std::move(scanner_params)); + + std::list<std::shared_ptr<ScannerDelegate>> scanners; + for (int i = 0; i < 11; ++i) { + scanners.push_back(std::make_shared<ScannerDelegate>(scanner)); + } + + std::shared_ptr<ScannerContext> scanner_context = ScannerContext::create_shared( + state.get(), olap_scan_local_state.get(), output_tuple_desc, output_row_descriptor, + scanners, limit, scan_dependency, parallel_tasks); + + scan_operator->_should_run_serial = true; + + olap_scan_local_state->_max_scan_concurrency = max_concurrency_counter.get(); + olap_scan_local_state->_min_scan_concurrency = min_concurrency_counter.get(); + + olap_scan_local_state->_parent = scan_operator.get(); + + TQueryOptions query_options; + query_options.__set_num_scanner_threads(2); + query_options.__set_max_column_reader_num(0); + state->set_query_options(query_options); + std::unique_ptr<MockSimplifiedScanScheduler> scheduler = + std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + EXPECT_CALL(*scheduler, schedule_scan_task(testing::_, testing::_, testing::_)) + .WillRepeatedly(testing::Return(Status::OK())); + scanner_context->_scanner_scheduler = scheduler.get(); + + scanner_context->_min_scan_concurrency_of_scan_scheduler = 10; + Status st = scanner_context->init(); + ASSERT_TRUE(st.ok()); + ASSERT_EQ(scanner_context->_max_scan_concurrency, 1); + + query_options.__set_num_scanner_threads(0); + state->set_query_options(query_options); + st = scanner_context->init(); + ASSERT_TRUE(st.ok()); + + ASSERT_EQ(scanner_context->_max_scan_concurrency, 1); +} + +TEST_F(ScannerContextTest, test_max_column_reader_num) { + const int parallel_tasks = 1; + auto scan_operator = std::make_unique<pipeline::OlapScanOperatorX>( + obj_pool.get(), tnode, 0, *descs, parallel_tasks, TQueryCacheParam {}); + + auto olap_scan_local_state = + pipeline::OlapScanLocalState::create_unique(state.get(), scan_operator.get()); + + const int64_t limit = 100; + + NewOlapScanner::Params scanner_params; + scanner_params.state = state.get(); + scanner_params.profile = profile.get(); + scanner_params.limit = limit; + scanner_params.key_ranges = std::vector<OlapScanRange*>(); // empty + + std::shared_ptr<VScanner> scanner = + NewOlapScanner::create_shared(olap_scan_local_state.get(), std::move(scanner_params)); + + std::list<std::shared_ptr<ScannerDelegate>> scanners; + for (int i = 0; i < 20; ++i) { + scanners.push_back(std::make_shared<ScannerDelegate>(scanner)); + } + + std::shared_ptr<ScannerContext> scanner_context = ScannerContext::create_shared( + state.get(), olap_scan_local_state.get(), output_tuple_desc, output_row_descriptor, + scanners, limit, scan_dependency, parallel_tasks); + + scan_operator->_should_run_serial = false; + + olap_scan_local_state->_max_scan_concurrency = max_concurrency_counter.get(); + olap_scan_local_state->_min_scan_concurrency = min_concurrency_counter.get(); + + olap_scan_local_state->_parent = scan_operator.get(); + + TQueryOptions query_options; + query_options.__set_num_scanner_threads(20); + query_options.__set_max_column_reader_num(1); + state->set_query_options(query_options); + std::unique_ptr<MockSimplifiedScanScheduler> scheduler = + std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + EXPECT_CALL(*scheduler, schedule_scan_task(testing::_, testing::_, testing::_)) + .WillRepeatedly(testing::Return(Status::OK())); + scanner_context->_scanner_scheduler = scheduler.get(); + scanner_context->_min_scan_concurrency_of_scan_scheduler = 10; + Status st = scanner_context->init(); + ASSERT_TRUE(st.ok()); + ASSERT_EQ(scanner_context->_max_scan_concurrency, 1); +} + +TEST_F(ScannerContextTest, test_push_back_scan_task) { + const int parallel_tasks = 1; + auto scan_operator = std::make_unique<pipeline::OlapScanOperatorX>( + obj_pool.get(), tnode, 0, *descs, parallel_tasks, TQueryCacheParam {}); + + auto olap_scan_local_state = + pipeline::OlapScanLocalState::create_unique(state.get(), scan_operator.get()); + + const int64_t limit = 100; + + NewOlapScanner::Params scanner_params; + scanner_params.state = state.get(); + scanner_params.profile = profile.get(); + scanner_params.limit = limit; + scanner_params.key_ranges = std::vector<OlapScanRange*>(); // empty + + std::shared_ptr<VScanner> scanner = + NewOlapScanner::create_shared(olap_scan_local_state.get(), std::move(scanner_params)); + + std::list<std::shared_ptr<ScannerDelegate>> scanners; + for (int i = 0; i < 11; ++i) { + scanners.push_back(std::make_shared<ScannerDelegate>(scanner)); + } + + std::shared_ptr<ScannerContext> scanner_context = ScannerContext::create_shared( + state.get(), olap_scan_local_state.get(), output_tuple_desc, output_row_descriptor, + scanners, limit, scan_dependency, parallel_tasks); + + scanner_context->_num_scheduled_scanners = 11; + + for (int i = 0; i < 5; ++i) { + auto scan_task = std::make_shared<ScanTask>(std::make_shared<ScannerDelegate>(scanner)); + scanner_context->push_back_scan_task(scan_task); + ASSERT_EQ(scanner_context->_num_scheduled_scanners, 10 - i); + } +} + +TEST_F(ScannerContextTest, get_margin) { + const int parallel_tasks = 4; + auto scan_operator = std::make_unique<pipeline::OlapScanOperatorX>( + obj_pool.get(), tnode, 0, *descs, parallel_tasks, TQueryCacheParam {}); + + auto olap_scan_local_state = + pipeline::OlapScanLocalState::create_unique(state.get(), scan_operator.get()); + + const int64_t limit = 100; + + NewOlapScanner::Params scanner_params; + scanner_params.state = state.get(); + scanner_params.profile = profile.get(); + scanner_params.limit = limit; + scanner_params.key_ranges = std::vector<OlapScanRange*>(); // empty + + std::shared_ptr<VScanner> scanner = + NewOlapScanner::create_shared(olap_scan_local_state.get(), std::move(scanner_params)); + + std::list<std::shared_ptr<ScannerDelegate>> scanners; + for (int i = 0; i < 11; ++i) { + scanners.push_back(std::make_shared<ScannerDelegate>(scanner)); + } + + std::shared_ptr<ScannerContext> scanner_context = ScannerContext::create_shared( + state.get(), olap_scan_local_state.get(), output_tuple_desc, output_row_descriptor, + scanners, limit, scan_dependency, parallel_tasks); + + std::mutex transfer_mutex; + std::unique_lock<std::mutex> transfer_lock(transfer_mutex); + std::shared_mutex scheduler_mutex; + std::unique_lock<std::shared_mutex> scheduler_lock(scheduler_mutex); + scanner_context->_scanner_scheduler = scan_scheduler.get(); + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + // _task_queue.size is 0. + // _num_schedule_scanners is 0. + std::shared_ptr<CgroupCpuCtl> cgroup_cpu_ctl = std::make_shared<CgroupV2CpuCtl>(1); + + // Has not submit any scan tasks. + // ScanScheduler is empty too. + // So margin shuold be equal to _min_scan_concurrency_of_scan_scheduler / parallel_tasks. + // We can make full utilization of the resource. + std::unique_ptr<MockSimplifiedScanScheduler> scheduler = + std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + EXPECT_CALL(*scheduler, get_active_threads()).WillOnce(testing::Return(0)); + EXPECT_CALL(*scheduler, get_queue_size()).WillOnce(testing::Return(0)); + scanner_context->_scanner_scheduler = scheduler.get(); + int32_t margin = scanner_context->_get_margin(transfer_lock, scheduler_lock); + + ASSERT_EQ(margin, scanner_context->_min_scan_concurrency_of_scan_scheduler / parallel_tasks); + + // ScanSchedule has 5 active threads and 10 tasks in queue. + // So remaing margin(3) is less than parallel_tasks(4). + scheduler = std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + EXPECT_CALL(*scheduler, get_active_threads()).WillOnce(testing::Return(5)); + EXPECT_CALL(*scheduler, get_queue_size()).WillOnce(testing::Return(10)); + scanner_context->_scanner_scheduler = scheduler.get(); + scanner_context->_min_scan_concurrency_of_scan_scheduler = 18; + margin = scanner_context->_get_margin(transfer_lock, scheduler_lock); + ASSERT_EQ(margin, 1); + + // ScanSchedule has 10 active threads and 2 tasks in queue. + // Remaing margin(8) is greater than parallel_tasks(4). + // So margin should be equal to margin(8)/parallel_tasks(4) == 2. + scheduler = std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + EXPECT_CALL(*scheduler, get_active_threads()).WillOnce(testing::Return(10)); + EXPECT_CALL(*scheduler, get_queue_size()).WillOnce(testing::Return(2)); + scanner_context->_scanner_scheduler = scheduler.get(); + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + margin = scanner_context->_get_margin(transfer_lock, scheduler_lock); + ASSERT_EQ(margin, + (scanner_context->_min_scan_concurrency_of_scan_scheduler - 12) / parallel_tasks); + + // ScanSchedule is busy. + // Just submit _min_scan_concurrency tasks. + scheduler = std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + EXPECT_CALL(*scheduler, get_active_threads()).WillOnce(testing::Return(50)); + EXPECT_CALL(*scheduler, get_queue_size()).WillOnce(testing::Return(10)); + scanner_context->_scanner_scheduler = scheduler.get(); + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + scanner_context->_num_scheduled_scanners = 0; + margin = scanner_context->_get_margin(transfer_lock, scheduler_lock); + ASSERT_EQ(margin, scanner_context->_min_scan_concurrency); + + // ScanSchedule is busy. + // _min_scan_concurrency is already satisfied. + scheduler = std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + EXPECT_CALL(*scheduler, get_active_threads()).WillOnce(testing::Return(50)); + EXPECT_CALL(*scheduler, get_queue_size()).WillOnce(testing::Return(10)); + scanner_context->_scanner_scheduler = scheduler.get(); + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + scanner_context->_num_scheduled_scanners = 20; + margin = scanner_context->_get_margin(transfer_lock, scheduler_lock); + ASSERT_EQ(margin, 0); +} + +TEST_F(ScannerContextTest, pull_next_scan_task) { + const int parallel_tasks = 4; + auto scan_operator = std::make_unique<pipeline::OlapScanOperatorX>( + obj_pool.get(), tnode, 0, *descs, parallel_tasks, TQueryCacheParam {}); + + auto olap_scan_local_state = + pipeline::OlapScanLocalState::create_unique(state.get(), scan_operator.get()); + + const int64_t limit = 100; + + NewOlapScanner::Params scanner_params; + scanner_params.state = state.get(); + scanner_params.profile = profile.get(); + scanner_params.limit = limit; + scanner_params.key_ranges = std::vector<OlapScanRange*>(); // empty + + std::shared_ptr<VScanner> scanner = + NewOlapScanner::create_shared(olap_scan_local_state.get(), std::move(scanner_params)); + + std::list<std::shared_ptr<ScannerDelegate>> scanners; + for (int i = 0; i < 11; ++i) { + scanners.push_back(std::make_shared<ScannerDelegate>(scanner)); + } + + std::shared_ptr<ScannerContext> scanner_context = ScannerContext::create_shared( + state.get(), olap_scan_local_state.get(), output_tuple_desc, output_row_descriptor, + scanners, limit, scan_dependency, parallel_tasks); + + std::mutex transfer_mutex; + std::unique_lock<std::mutex> transfer_lock(transfer_mutex); + std::shared_mutex scheduler_mutex; + std::unique_lock<std::shared_mutex> scheduler_lock(scheduler_mutex); + scanner_context->_scanner_scheduler = scan_scheduler.get(); + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + std::shared_ptr<CgroupCpuCtl> cgroup_cpu_ctl = std::make_shared<CgroupV2CpuCtl>(1); + std::unique_ptr<MockSimplifiedScanScheduler> scheduler = + std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + + scanner_context->_scanner_scheduler = scan_scheduler.get(); + scanner_context->_max_scan_concurrency = 1; + std::shared_ptr<ScanTask> pull_scan_task = + scanner_context->_pull_next_scan_task(nullptr, scanner_context->_max_scan_concurrency); + ASSERT_EQ(pull_scan_task, nullptr); + auto scan_task = std::make_shared<ScanTask>(std::make_shared<ScannerDelegate>(scanner)); + pull_scan_task = scanner_context->_pull_next_scan_task(scan_task, + scanner_context->_max_scan_concurrency); + ASSERT_EQ(pull_scan_task, nullptr); + + scanner_context->_max_scan_concurrency = 2; + BlockUPtr cached_block = Block::create_unique(); + scan_task->cached_blocks.emplace_back(std::move(cached_block), 0); + EXPECT_ANY_THROW(scanner_context->_pull_next_scan_task( + scan_task, scanner_context->_max_scan_concurrency - 1)); + scan_task->cached_blocks.clear(); + scan_task->eos = true; + EXPECT_ANY_THROW(scanner_context->_pull_next_scan_task( + scan_task, scanner_context->_max_scan_concurrency - 1)); + + scan_task->cached_blocks.clear(); + scan_task->eos = false; + pull_scan_task = scanner_context->_pull_next_scan_task( + scan_task, scanner_context->_max_scan_concurrency - 1); + EXPECT_EQ(pull_scan_task.get(), scan_task.get()); + + scanner_context->_pending_scanners = std::stack<std::weak_ptr<ScannerDelegate>>(); + pull_scan_task = scanner_context->_pull_next_scan_task( + nullptr, scanner_context->_max_scan_concurrency - 1); + EXPECT_EQ(pull_scan_task, nullptr); + + scanner_context->_pending_scanners.push(std::make_shared<ScannerDelegate>(scanner)); + pull_scan_task = scanner_context->_pull_next_scan_task( + nullptr, scanner_context->_max_scan_concurrency - 1); + EXPECT_NE(pull_scan_task, nullptr); +} + +TEST_F(ScannerContextTest, schedule_scan_task) { + const int parallel_tasks = 4; + auto scan_operator = std::make_unique<pipeline::OlapScanOperatorX>( + obj_pool.get(), tnode, 0, *descs, parallel_tasks, TQueryCacheParam {}); + + auto olap_scan_local_state = + pipeline::OlapScanLocalState::create_unique(state.get(), scan_operator.get()); + + const int64_t limit = 100; + + NewOlapScanner::Params scanner_params; + scanner_params.state = state.get(); + scanner_params.profile = profile.get(); + scanner_params.limit = limit; + scanner_params.key_ranges = std::vector<OlapScanRange*>(); // empty + + std::shared_ptr<VScanner> scanner = + NewOlapScanner::create_shared(olap_scan_local_state.get(), std::move(scanner_params)); + + std::list<std::shared_ptr<ScannerDelegate>> scanners; + for (int i = 0; i < 15; ++i) { + scanners.push_back(std::make_shared<ScannerDelegate>(scanner)); + } + + std::shared_ptr<ScannerContext> scanner_context = ScannerContext::create_shared( + state.get(), olap_scan_local_state.get(), output_tuple_desc, output_row_descriptor, + scanners, limit, scan_dependency, parallel_tasks); + + std::mutex transfer_mutex; + std::unique_lock<std::mutex> transfer_lock(transfer_mutex); + std::shared_mutex scheduler_mutex; + std::unique_lock<std::shared_mutex> scheduler_lock(scheduler_mutex); + std::shared_ptr<CgroupCpuCtl> cgroup_cpu_ctl = std::make_shared<CgroupV2CpuCtl>(1); + + // Scan resource is enough. + 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)); + + std::unique_ptr<MockScannerScheduler> scanner_scheduler = + std::make_unique<MockScannerScheduler>(); + EXPECT_CALL(*scanner_scheduler, submit(testing::_, testing::_)) + .WillRepeatedly(testing::Return(Status::OK())); + + scanner_context->_scanner_scheduler_global = scanner_scheduler.get(); + scanner_context->_scanner_scheduler = scheduler.get(); + scanner_context->_max_scan_concurrency = 1; + scanner_context->_max_scan_concurrency = 1; + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + + Status st = scanner_context->_schedule_scan_task(nullptr, transfer_lock, scheduler_lock); + ASSERT_TRUE(st.ok()); + ASSERT_EQ(scanner_context->_num_scheduled_scanners, 1); + + scanner_context->_max_scan_concurrency = 10; + scanner_context->_max_scan_concurrency = 1; + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + st = scanner_context->_schedule_scan_task(nullptr, transfer_lock, scheduler_lock); + ASSERT_TRUE(st.ok()); + ASSERT_EQ(scanner_context->_num_scheduled_scanners, scanner_context->_max_scan_concurrency); + + scanner_context = ScannerContext::create_shared( + state.get(), olap_scan_local_state.get(), output_tuple_desc, output_row_descriptor, + scanners, limit, scan_dependency, parallel_tasks); + + scanner_context->_scanner_scheduler_global = scanner_scheduler.get(); + scanner_context->_scanner_scheduler = scheduler.get(); + + scanner_context->_max_scan_concurrency = 100; + scanner_context->_min_scan_concurrency = 1; + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + int margin = scanner_context->_get_margin(transfer_lock, scheduler_lock); + ASSERT_EQ(margin, scanner_context->_min_scan_concurrency_of_scan_scheduler / parallel_tasks); + st = scanner_context->_schedule_scan_task(nullptr, transfer_lock, scheduler_lock); + ASSERT_TRUE(st.ok()); + ASSERT_EQ(scanner_context->_num_scheduled_scanners, + scanner_context->_min_scan_concurrency_of_scan_scheduler / parallel_tasks); + + scanners = std::list<std::shared_ptr<ScannerDelegate>>(); + for (int i = 0; i < 1; ++i) { + scanners.push_back(std::make_shared<ScannerDelegate>(scanner)); + } + + scanner_context = ScannerContext::create_shared( + state.get(), olap_scan_local_state.get(), output_tuple_desc, output_row_descriptor, + scanners, limit, scan_dependency, parallel_tasks); + + scanner_context->_scanner_scheduler_global = scanner_scheduler.get(); + scanner_context->_scanner_scheduler = scheduler.get(); + + scanner_context->_max_scan_concurrency = 1; + scanner_context->_min_scan_concurrency = 1; + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + st = scanner_context->_schedule_scan_task(nullptr, transfer_lock, scheduler_lock); + auto scan_task = std::make_shared<ScanTask>(std::make_shared<ScannerDelegate>(scanner)); + st = scanner_context->_schedule_scan_task(scan_task, transfer_lock, scheduler_lock); + // current scan task is added back. + ASSERT_EQ(scanner_context->_pending_scanners.size(), 1); + ASSERT_EQ(scanner_context->_num_scheduled_scanners, 1); + + scanner_context = ScannerContext::create_shared( + state.get(), olap_scan_local_state.get(), output_tuple_desc, output_row_descriptor, + scanners, limit, scan_dependency, parallel_tasks); + + scanner_context->_scanner_scheduler_global = scanner_scheduler.get(); + scanner_context->_scanner_scheduler = scheduler.get(); + + scanner_context->_max_scan_concurrency = 1; + scanner_context->_min_scan_concurrency = 1; + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + st = scanner_context->_schedule_scan_task(nullptr, transfer_lock, scheduler_lock); + scan_task = std::make_shared<ScanTask>(std::make_shared<ScannerDelegate>(scanner)); + scan_task->cached_blocks.emplace_back(Block::create_unique(), 0); Review Comment: 增加一些 free blocks 的check -- 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: commits-unsubscr...@doris.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For additional commands, e-mail: commits-h...@doris.apache.org
