From ffe5c08995c649d5d9bb2632a23b8745a6ff16f0 Mon Sep 17 00:00:00 2001 From: bobhan1 Date: Fri, 12 Jun 2026 21:40:55 +0800 Subject: [PATCH 1/4] [fix](delta writer) Fix shared delta writer state lifetime (#64349) Issue Number: None Problem Summary: Shared `DeltaWriterV2` instances can be reused by multiple local sinks from the same load. Before this change, the shared writer stored the `RuntimeState*` from the sink that first created it. If that creator sink finished and its `RuntimeState` was destroyed while another local sink continued to reuse the shared writer, `DeltaWriterV2::write()` could access the destroyed state in the memtable flush-limit cancellation path, causing a BE crash or ASAN use-after-free. This PR adds a BE unit test that reproduces the lifetime boundary: - one `VTabletWriterV2` creates the shared `DeltaWriterV2`; - the creator writer and its `RuntimeState` are destroyed without cancelling the shared writer; - a second writer reuses the shared writer and is forced into the `DeltaWriterV2::write()` flush-limit wait path; - the old code reads the destroyed creator state, while the fixed code observes the current writer's cancel state and exits cleanly. The fix removes the stored `RuntimeState*` from `DeltaWriterV2`. The shared writer now keeps only the stable `WorkloadGroup` shared pointer needed by `MemTableWriter` initialization, and `VTabletWriterV2` passes a per-call cancel checker into `DeltaWriterV2::write()` so cancellation is evaluated against the current sink. Fix a possible BE crash when shared delta writers are reused by multiple local sinks. --- be/src/olap/delta_writer_v2.cpp | 29 +-- be/src/olap/delta_writer_v2.h | 10 +- be/src/vec/sink/writer/vtablet_writer_v2.cpp | 18 +- .../vec/exec/delta_writer_v2_pool_test.cpp | 13 +- be/test/vec/sink/vtablet_writer_v2_test.cpp | 237 +++++++++++++++++- 5 files changed, 274 insertions(+), 33 deletions(-) diff --git a/be/src/olap/delta_writer_v2.cpp b/be/src/olap/delta_writer_v2.cpp index 8cf815aa7c0ffd..2af2070295e5d4 100644 --- a/be/src/olap/delta_writer_v2.cpp +++ b/be/src/olap/delta_writer_v2.cpp @@ -66,9 +66,9 @@ using namespace ErrorCode; DeltaWriterV2::DeltaWriterV2(WriteRequest* req, const std::vector>& streams, - RuntimeState* state) - : _state(state), - _req(*req), + std::shared_ptr workload_group) + : _req(*req), + _workload_group(std::move(workload_group)), _tablet_schema(new TabletSchema), _memtable_writer(new MemTableWriter(*req)), _streams(streams) {} @@ -127,19 +127,17 @@ Status DeltaWriterV2::init() { _rowset_writer = std::make_shared(_streams); RETURN_IF_ERROR(_rowset_writer->init(context)); - std::shared_ptr wg_sptr = nullptr; - if (_state->get_query_ctx()) { - wg_sptr = _state->get_query_ctx()->workload_group(); - } RETURN_IF_ERROR(_memtable_writer->init(_rowset_writer, _tablet_schema, _partial_update_info, - wg_sptr, _streams[0]->enable_unique_mow(_req.index_id))); + _workload_group, + _streams[0]->enable_unique_mow(_req.index_id))); ExecEnv::GetInstance()->memtable_memory_limiter()->register_writer(_memtable_writer); _is_init = true; _streams.clear(); return Status::OK(); } -Status DeltaWriterV2::write(const vectorized::Block* block, const DorisVector& row_idxs) { +Status DeltaWriterV2::write(const vectorized::Block* block, const DorisVector& row_idxs, + const std::function& cancel_check) { if (UNLIKELY(row_idxs.empty())) { return Status::OK(); } @@ -155,9 +153,8 @@ Status DeltaWriterV2::write(const vectorized::Block* block, const DorisVectorflush_running_count() >= memtable_flush_running_count_limit) { - if (_state->is_cancelled()) { - return _state->cancel_reason(); - } + DBUG_EXECUTE_IF("DeltaWriterV2.write.flush_limit_wait", DBUG_RUN_CALLBACK()); + RETURN_IF_ERROR(cancel_check()); std::this_thread::sleep_for(std::chrono::milliseconds(10)); } } @@ -186,14 +183,10 @@ Status DeltaWriterV2::close_wait(int32_t& num_segments, RuntimeProfile* profile) DCHECK(_is_init) << "delta writer is supposed be to initialized before close_wait() being called"; - if (_state->profile_level() >= 2 && profile != nullptr) { + if (profile != nullptr) { _update_profile(profile); } - if (_state->profile_level() >= 2) { - RETURN_IF_ERROR(_memtable_writer->close_wait(profile)); - } else { - RETURN_IF_ERROR(_memtable_writer->close_wait()); - } + RETURN_IF_ERROR(_memtable_writer->close_wait(profile)); num_segments = _rowset_writer->next_segment_id(); _delta_written_success = true; diff --git a/be/src/olap/delta_writer_v2.h b/be/src/olap/delta_writer_v2.h index 550c4b72f873bd..71458a4b709ab3 100644 --- a/be/src/olap/delta_writer_v2.h +++ b/be/src/olap/delta_writer_v2.h @@ -24,6 +24,7 @@ #include #include +#include #include #include #include @@ -52,6 +53,7 @@ class SlotDescriptor; class OlapTableSchemaParam; class BetaRowsetWriterV2; class LoadStreamStub; +class WorkloadGroup; namespace vectorized { class Block; @@ -64,13 +66,14 @@ class DeltaWriterV2 { public: DeltaWriterV2(WriteRequest* req, const std::vector>& streams, - RuntimeState* state); + std::shared_ptr workload_group); ~DeltaWriterV2(); Status init(); - Status write(const vectorized::Block* block, const DorisVector& row_idxs); + Status write(const vectorized::Block* block, const DorisVector& row_idxs, + const std::function& cancel_check); // flush the last memtable to flush queue, must call it before close_wait() Status close(); @@ -90,11 +93,10 @@ class DeltaWriterV2 { void _update_profile(RuntimeProfile* profile); - RuntimeState* _state = nullptr; - bool _is_init = false; bool _is_cancelled = false; WriteRequest _req; + std::shared_ptr _workload_group; std::shared_ptr _rowset_writer; TabletSchemaSPtr _tablet_schema; bool _delta_written_success = false; diff --git a/be/src/vec/sink/writer/vtablet_writer_v2.cpp b/be/src/vec/sink/writer/vtablet_writer_v2.cpp index bce52bead14ee4..93577b69eaf27f 100644 --- a/be/src/vec/sink/writer/vtablet_writer_v2.cpp +++ b/be/src/vec/sink/writer/vtablet_writer_v2.cpp @@ -578,7 +578,11 @@ Status VTabletWriterV2::_write_memtable(std::shared_ptr block << " not found in schema, load_id=" << print_id(_load_id); return std::unique_ptr(nullptr); } - return DeltaWriterV2::create_unique(&req, streams, _state); + std::shared_ptr workload_group = nullptr; + if (_state->get_query_ctx()) { + workload_group = _state->workload_group(); + } + return DeltaWriterV2::create_unique(&req, streams, workload_group); }); if (delta_writer == nullptr) { LOG(WARNING) << "failed to open DeltaWriter for tablet " << tablet_id @@ -594,7 +598,12 @@ Status VTabletWriterV2::_write_memtable(std::shared_ptr block } } SCOPED_TIMER(_write_memtable_timer); - st = delta_writer->write(block.get(), rows.row_idxes); + st = delta_writer->write(block.get(), rows.row_idxes, [state = _state]() { + if (state->is_cancelled()) { + return state->cancel_reason(); + } + return Status::OK(); + }); return st; } @@ -677,7 +686,10 @@ Status VTabletWriterV2::close(Status exec_status) { std::unordered_map segments_for_tablet; SCOPED_TIMER(_close_writer_timer); // close all delta writers if this is the last user - auto st = _delta_writer_for_tablet->close(segments_for_tablet, _operator_profile); + RuntimeProfile* delta_writer_profile = + _state->enable_profile() && _state->profile_level() >= 2 ? _operator_profile + : nullptr; + auto st = _delta_writer_for_tablet->close(segments_for_tablet, delta_writer_profile); _delta_writer_for_tablet.reset(); if (!st.ok()) { _cancel(st); diff --git a/be/test/vec/exec/delta_writer_v2_pool_test.cpp b/be/test/vec/exec/delta_writer_v2_pool_test.cpp index dc86ce8c3a28aa..92f84fd31b1b46 100644 --- a/be/test/vec/exec/delta_writer_v2_pool_test.cpp +++ b/be/test/vec/exec/delta_writer_v2_pool_test.cpp @@ -57,18 +57,17 @@ TEST_F(DeltaWriterV2PoolTest, test_map) { auto map = pool.get_or_create(load_id); EXPECT_EQ(1, pool.size()); WriteRequest req; - RuntimeState state; - auto writer = map->get_or_create(100, [&req, &state]() { + auto writer = map->get_or_create(100, [&req]() { return std::make_unique( - &req, std::vector> {}, &state); + &req, std::vector> {}, nullptr); }); - auto writer2 = map->get_or_create(101, [&req, &state]() { + auto writer2 = map->get_or_create(101, [&req]() { return std::make_unique( - &req, std::vector> {}, &state); + &req, std::vector> {}, nullptr); }); - auto writer3 = map->get_or_create(100, [&req, &state]() { + auto writer3 = map->get_or_create(100, [&req]() { return std::make_unique( - &req, std::vector> {}, &state); + &req, std::vector> {}, nullptr); }); EXPECT_EQ(2, map->size()); EXPECT_EQ(writer, writer3); diff --git a/be/test/vec/sink/vtablet_writer_v2_test.cpp b/be/test/vec/sink/vtablet_writer_v2_test.cpp index ce467fb1d45f69..840eb8ab58d8bc 100644 --- a/be/test/vec/sink/vtablet_writer_v2_test.cpp +++ b/be/test/vec/sink/vtablet_writer_v2_test.cpp @@ -18,12 +18,40 @@ #include "vec/sink/writer/vtablet_writer_v2.h" #include - +#include + +#include +#include +#include + +#include "common/config.h" +#include "io/fs/local_file_system.h" +#include "olap/memtable_memory_limiter.h" +#include "olap/storage_engine.h" +#include "olap/tablet_schema.h" +#include "pipeline/operator/operator_helper.h" +#include "runtime/exec_env.h" +#include "testutil/column_helper.h" +#include "util/debug_points.h" +#include "util/defer_op.h" +#include "vec/data_types/data_type_number.h" +#include "vec/sink/delta_writer_v2_pool.h" #include "vec/sink/load_stream_map_pool.h" #include "vec/sink/load_stream_stub.h" +#include "vec/sink/sink_test_utils.h" namespace doris { +using pipeline::OperatorContext; +using vectorized::Block; +using vectorized::ColumnHelper; +using vectorized::DataTypeInt32; +using vectorized::DeltaWriterV2Pool; +using vectorized::Rows; +using vectorized::VExprContextSPtrs; +using vectorized::VTabletWriterV2; +namespace sink_test_utils = vectorized::sink_test_utils; + class TestVTabletWriterV2 : public ::testing::Test { public: TestVTabletWriterV2() = default; @@ -63,6 +91,213 @@ static std::unique_ptr create_vtablet_writer(int nu return writer; } +static TColumn create_int_column_desc(bool is_nullable) { + TColumn column; + column.__set_column_name("c1"); + column.column_type.type = TPrimitiveType::INT; + column.__set_is_key(true); + column.__set_is_allow_null(is_nullable); + column.__set_col_unique_id(1); + return column; +} + +static void prepare_load_runtime_state(MockRuntimeState& state, int sender_id) { + state.set_backend_id(1); + state.set_per_fragment_instance_idx(sender_id); + state.set_num_per_fragment_instances(2); + state.set_load_stream_per_node(1); + state.set_total_load_streams(2); + state.set_num_local_sink(2); +} + +static void prepare_open_streams(std::shared_ptr load_stream_map, int64_t node_id, + int64_t index_id, const TabletSchemaSPtr& tablet_schema) { + auto streams = load_stream_map->get_or_create(node_id); + streams->mark_open(); + for (auto& stream : streams->streams()) { + stream->_is_open.store(true); + stream->_status = Status::OK(); + stream->_tablet_schema_for_index->emplace(index_id, tablet_schema); + stream->_enable_unique_mow_for_index->emplace(index_id, false); + } +} + +static TabletSchemaSPtr create_int_tablet_schema() { + TabletSchemaPB tablet_schema_pb; + tablet_schema_pb.set_keys_type(DUP_KEYS); + tablet_schema_pb.set_num_short_key_columns(1); + tablet_schema_pb.set_num_rows_per_row_block(1024); + tablet_schema_pb.set_compress_kind(COMPRESS_NONE); + tablet_schema_pb.set_next_column_unique_id(2); + ColumnPB* column = tablet_schema_pb.add_column(); + column->set_unique_id(1); + column->set_name("c1"); + column->set_type("INT"); + column->set_is_key(true); + column->set_is_nullable(false); + + auto tablet_schema = std::make_shared(); + tablet_schema->init_from_pb(tablet_schema_pb); + return tablet_schema; +} + +static TDataSink create_vtablet_writer_sink(const TOlapTableSchemaParam& schema, + const TOlapTablePartitionParam& partition, + const TOlapTableLocationParam& location, + TTupleId tuple_id, const TUniqueId& load_id) { + TDataSink t_sink; + t_sink.__isset.olap_table_sink = true; + auto& olap_sink = t_sink.olap_table_sink; + olap_sink.__set_load_id(load_id); + olap_sink.__set_txn_id(1); + olap_sink.__set_db_id(schema.db_id); + olap_sink.__set_table_id(schema.table_id); + olap_sink.__set_tuple_id(tuple_id); + olap_sink.__set_num_replicas(1); + olap_sink.__set_need_gen_rollup(false); + olap_sink.__set_schema(schema); + olap_sink.__set_partition(partition); + olap_sink.__set_location(location); + + TNodeInfo node; + node.__set_id(1); + node.__set_option(0); + node.__set_host("127.0.0.1"); + node.__set_async_internal_port(8060); + TPaloNodesInfo nodes; + nodes.nodes.push_back(node); + olap_sink.__set_nodes_info(nodes); + return t_sink; +} + +TEST_F(TestVTabletWriterV2, shared_delta_writer_should_not_access_destroyed_creator_runtime_state) { + const bool old_share_delta_writers = config::share_delta_writers; + const int32_t old_flush_running_count_limit = config::memtable_flush_running_count_limit; + const bool old_enable_debug_points = config::enable_debug_points; + config::share_delta_writers = true; + Defer restore_configs([&] { + config::share_delta_writers = old_share_delta_writers; + config::memtable_flush_running_count_limit = old_flush_running_count_limit; + config::enable_debug_points = old_enable_debug_points; + DebugPoints::instance()->remove("DeltaWriterV2.write.flush_limit_wait"); + }); + + ExecEnv* exec_env = ExecEnv::GetInstance(); + auto old_load_stream_map_pool = std::move(exec_env->_load_stream_map_pool); + auto old_delta_writer_v2_pool = std::move(exec_env->_delta_writer_v2_pool); + auto old_memtable_memory_limiter = std::move(exec_env->_memtable_memory_limiter); + auto old_storage_engine = std::move(exec_env->_storage_engine); + const std::string old_storage_root_path = config::storage_root_path; + char cwd_buffer[1024]; + ASSERT_NE(nullptr, getcwd(cwd_buffer, sizeof(cwd_buffer))); + const std::string test_data_dir = + std::string(cwd_buffer) + "/vtablet_writer_v2_shared_delta_writer_test"; + Defer restore_exec_env([&]() mutable { + exec_env->_delta_writer_v2_pool.reset(); + exec_env->_load_stream_map_pool.reset(); + exec_env->_storage_engine.reset(); + exec_env->_memtable_memory_limiter.reset(); + exec_env->_storage_engine = std::move(old_storage_engine); + exec_env->_memtable_memory_limiter = std::move(old_memtable_memory_limiter); + exec_env->_delta_writer_v2_pool = std::move(old_delta_writer_v2_pool); + exec_env->_load_stream_map_pool = std::move(old_load_stream_map_pool); + config::storage_root_path = old_storage_root_path; + static_cast(io::global_local_filesystem()->delete_directory(test_data_dir)); + }); + + config::storage_root_path = test_data_dir; + ASSERT_TRUE(io::global_local_filesystem()->delete_directory(test_data_dir).ok()); + ASSERT_TRUE(io::global_local_filesystem()->create_directory(test_data_dir).ok()); + EngineOptions options; + options.store_paths.emplace_back(test_data_dir, -1); + auto engine = std::make_unique(options); + ASSERT_TRUE(engine->open().ok()); + exec_env->_storage_engine = std::move(engine); + auto memtable_memory_limiter = std::make_unique(); + ASSERT_TRUE(memtable_memory_limiter->init(1024 * 1024 * 1024).ok()); + exec_env->_memtable_memory_limiter = std::move(memtable_memory_limiter); + exec_env->_load_stream_map_pool = std::make_unique(); + exec_env->_delta_writer_v2_pool = std::make_unique(); + + auto creator_ctx = std::make_unique(); + OperatorContext current_ctx; + prepare_load_runtime_state(creator_ctx->state, 0); + prepare_load_runtime_state(current_ctx.state, 1); + + TOlapTableSchemaParam schema; + TTupleId tuple_id = 0; + int64_t index_id = 0; + sink_test_utils::build_desc_tbl_and_schema(*creator_ctx, schema, tuple_id, index_id, false); + sink_test_utils::build_desc_tbl_and_schema(current_ctx, schema, tuple_id, index_id, false); + schema.indexes[0].__set_columns_desc({create_int_column_desc(false)}); + + TUniqueId load_id; + load_id.hi = 380; + load_id.lo = 1; + const auto partition = sink_test_utils::build_partition_param(index_id); + const auto location = sink_test_utils::build_location_param(); + const auto t_sink = create_vtablet_writer_sink(schema, partition, location, tuple_id, load_id); + + VExprContextSPtrs output_exprs; + auto creator_writer = std::make_unique(t_sink, output_exprs, nullptr, nullptr); + auto current_writer = std::make_unique(t_sink, output_exprs, nullptr, nullptr); + ASSERT_TRUE(creator_writer->_init(&creator_ctx->state, &creator_ctx->profile).ok()); + ASSERT_TRUE(current_writer->_init(¤t_ctx.state, ¤t_ctx.profile).ok()); + ASSERT_EQ(creator_writer->_delta_writer_for_tablet, current_writer->_delta_writer_for_tablet); + + const auto tablet_schema = create_int_tablet_schema(); + prepare_open_streams(creator_writer->_load_stream_map, 1, index_id, tablet_schema); + + auto block = std::make_shared(ColumnHelper::create_block({1})); + Rows rows; + rows.partition_id = 1; + rows.index_id = index_id; + rows.row_idxes.push_back(0); + + const auto first_write_status = creator_writer->_write_memtable(block, 100, rows); + ASSERT_TRUE(first_write_status.ok()) << first_write_status; + ASSERT_EQ(1, creator_writer->_delta_writer_for_tablet->size()); + + // The first write above creates the shared DeltaWriterV2 and stores + // creator_ctx->state in DeltaWriterV2::_state. Destroy the creator sink and + // its RuntimeState to reproduce the original lifetime boundary: another + // local sink can still reuse the shared writer after the creator state is + // gone. Do not call creator_writer->_cancel() here because it cancels the + // shared writer and would hide the dangling RuntimeState path. + creator_writer.reset(); + creator_ctx.reset(); + + // Force the current sink into DeltaWriterV2::write()'s flush-limit wait + // path, then cancel the current RuntimeState. Fixed code should observe the + // current sink's cancel state and exit cleanly; current broken code reads + // the destroyed creator RuntimeState from the shared DeltaWriterV2 and ASAN + // reports heap-use-after-free in the child process. + auto debug_point = std::make_shared(); + debug_point->execute_limit = 1; + debug_point->callback = std::function( + [&] { current_ctx.state.cancel(Status::Cancelled("current state cancelled")); }); + config::enable_debug_points = true; + DebugPoints::instance()->add("DeltaWriterV2.write.flush_limit_wait", debug_point); + config::memtable_flush_running_count_limit = 0; + + EXPECT_EXIT( + { + alarm(10); + auto status = current_writer->_write_memtable(block, 100, rows); + if (!status.ok() && + status.msg().find("current state cancelled") != std::string::npos) { + _exit(0); + } + _exit(1); + }, + ::testing::ExitedWithCode(0), ""); + + config::memtable_flush_running_count_limit = old_flush_running_count_limit; + DebugPoints::instance()->remove("DeltaWriterV2.write.flush_limit_wait"); + + current_writer->_cancel(Status::Cancelled("test cleanup")); +} + TEST_F(TestVTabletWriterV2, one_replica) { UniqueId load_id; std::vector tablet_commit_infos; From e730f5622046d744a7042c78cb4562322c8abe76 Mon Sep 17 00:00:00 2001 From: bobhan1 Date: Wed, 17 Jun 2026 19:02:57 +0800 Subject: [PATCH 2/4] [test](be) Clean up WalManager in stream load test StreamLoadTest.TestHeader installs a WalManager into the global ExecEnv but did not clear it. In full BE UT coverage runs, later tests can change global metric registry state, and the leaked WalManager's ThreadPool may deregister metrics during process exit after DorisMetrics has already been destroyed, causing an ASAN heap-use-after-free. Stop and clear the test WalManager in StreamLoadTest::TearDown so the global ExecEnv is restored before the next test and before process shutdown. Validation:\n- ./run-be-ut.sh --run --coverage --filter=StreamLoadTest.TestHeader:TestVTabletWriterV2.shared_delta_writer_should_not_access_destroyed_creator_runtime_state -j100\n- ./run-be-ut.sh --run --coverage --filter=StreamLoadTest.TestHeader:TestVTabletWriterV2.*:DeltaWriterV2PoolTest.* -j100 --- be/test/http/stream_load_test.cpp | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/be/test/http/stream_load_test.cpp b/be/test/http/stream_load_test.cpp index faa582704d11cc..9d566572585095 100644 --- a/be/test/http/stream_load_test.cpp +++ b/be/test/http/stream_load_test.cpp @@ -45,7 +45,13 @@ class StreamLoadTest : public testing::Test { StreamLoadTest() = default; virtual ~StreamLoadTest() = default; void SetUp() override {} - void TearDown() override {} + void TearDown() override { + auto* exec_env = ExecEnv::GetInstance(); + if (auto* wal_mgr = exec_env->wal_mgr(); wal_mgr != nullptr) { + wal_mgr->stop(); + } + exec_env->clear_wal_mgr(); + } }; void http_request_done_cb(struct evhttp_request* req, void* arg) { @@ -123,4 +129,4 @@ TEST_F(StreamLoadTest, TestHeader) { evhttp_request_free(evhttp_req); } } -} // namespace doris \ No newline at end of file +} // namespace doris From 7cfde72f87f691bbee38d08b3857eddfea714673 Mon Sep 17 00:00:00 2001 From: bobhan1 Date: Wed, 24 Jun 2026 19:42:48 +0800 Subject: [PATCH 3/4] [test](be) Clean up global state in workload tests --- be/test/pipeline/pipeline_task_test.cpp | 2 ++ .../workload_group_manager_test.cpp | 31 ++++++++++++++++--- 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/be/test/pipeline/pipeline_task_test.cpp b/be/test/pipeline/pipeline_task_test.cpp index a629344363a134..e9c6c0f62dd2a3 100644 --- a/be/test/pipeline/pipeline_task_test.cpp +++ b/be/test/pipeline/pipeline_task_test.cpp @@ -959,6 +959,7 @@ TEST_F(PipelineTaskTest, TEST_RESERVE_MEMORY_FAIL) { ((MockWorkloadGroupMgr*)ExecEnv::GetInstance()->_workload_group_manager)->_paused); } delete ExecEnv::GetInstance()->_workload_group_manager; + ExecEnv::GetInstance()->_workload_group_manager = nullptr; } // Test reserve memory fail for spillable pipeline task @@ -1135,6 +1136,7 @@ TEST_F(PipelineTaskTest, TEST_RESERVE_MEMORY_FAIL_SPILLABLE) { ((MockWorkloadGroupMgr*)ExecEnv::GetInstance()->_workload_group_manager)->_paused); } delete ExecEnv::GetInstance()->_workload_group_manager; + ExecEnv::GetInstance()->_workload_group_manager = nullptr; } TEST_F(PipelineTaskTest, TEST_INJECT_SHARED_STATE) { diff --git a/be/test/runtime/workload_group/workload_group_manager_test.cpp b/be/test/runtime/workload_group/workload_group_manager_test.cpp index f81488928c9b80..5808128bd93510 100644 --- a/be/test/runtime/workload_group/workload_group_manager_test.cpp +++ b/be/test/runtime/workload_group/workload_group_manager_test.cpp @@ -22,10 +22,13 @@ #include #include #include +#include +#include #include -#include +#include #include +#include #include "common/config.h" #include "common/status.h" @@ -44,10 +47,21 @@ class WorkloadGroupManagerTest : public testing::Test { protected: void SetUp() override { _wg_manager = std::make_unique(); - EXPECT_EQ(system("rm -rf ./wg_test_run && mkdir -p ./wg_test_run"), 0); + std::ostringstream oss; + oss << "./wg_test_run_" << std::chrono::system_clock::now().time_since_epoch().count() + << "_" << getpid(); + _test_dir = oss.str(); + + std::error_code ec; + std::filesystem::remove_all(_test_dir, ec); + if (ec) { + FAIL() << "Failed to remove " << _test_dir << ": " << ec.message(); + } + std::filesystem::create_directories(_test_dir, ec); + ASSERT_FALSE(ec) << "Failed to create " << _test_dir << ": " << ec.message(); std::vector paths; - std::string path = std::filesystem::absolute("./wg_test_run").string(); + std::string path = std::filesystem::absolute(_test_dir).string(); auto olap_res = doris::parse_conf_store_paths(path, &paths); EXPECT_TRUE(olap_res.ok()) << olap_res.to_string(); @@ -78,7 +92,15 @@ class WorkloadGroupManagerTest : public testing::Test { ExecEnv::GetInstance()->_runtime_query_statistics_mgr->stop_report_thread(); SAFE_DELETE(ExecEnv::GetInstance()->_runtime_query_statistics_mgr); - EXPECT_EQ(system("rm -rf ./wg_test_run"), 0); + if (ExecEnv::GetInstance()->_spill_stream_mgr != nullptr) { + ExecEnv::GetInstance()->_spill_stream_mgr->stop(); + } + SAFE_DELETE(ExecEnv::GetInstance()->_spill_stream_mgr); + ExecEnv::GetInstance()->_pipeline_tracer_ctx.reset(); + + std::error_code ec; + std::filesystem::remove_all(_test_dir, ec); + EXPECT_FALSE(ec) << "Failed to remove " << _test_dir << ": " << ec.message(); config::spill_in_paused_queue_timeout_ms = _spill_in_paused_queue_timeout_ms; doris::ExecEnv::GetInstance()->set_memtable_memory_limiter(nullptr); } @@ -117,6 +139,7 @@ class WorkloadGroupManagerTest : public testing::Test { } std::unique_ptr _wg_manager; + std::string _test_dir; const int64_t _spill_in_paused_queue_timeout_ms = config::spill_in_paused_queue_timeout_ms; }; From 7f11ebf515d89736979a27afebc2855b40de345e Mon Sep 17 00:00:00 2001 From: bobhan1 Date: Thu, 25 Jun 2026 11:50:05 +0800 Subject: [PATCH 4/4] [test](be) Avoid death test for shared delta writer lifetime --- be/test/http/stream_load_test.cpp | 10 ++---- be/test/pipeline/pipeline_task_test.cpp | 2 -- .../workload_group_manager_test.cpp | 31 +++---------------- be/test/vec/sink/vtablet_writer_v2_test.cpp | 24 +++++++------- 4 files changed, 18 insertions(+), 49 deletions(-) diff --git a/be/test/http/stream_load_test.cpp b/be/test/http/stream_load_test.cpp index 9d566572585095..faa582704d11cc 100644 --- a/be/test/http/stream_load_test.cpp +++ b/be/test/http/stream_load_test.cpp @@ -45,13 +45,7 @@ class StreamLoadTest : public testing::Test { StreamLoadTest() = default; virtual ~StreamLoadTest() = default; void SetUp() override {} - void TearDown() override { - auto* exec_env = ExecEnv::GetInstance(); - if (auto* wal_mgr = exec_env->wal_mgr(); wal_mgr != nullptr) { - wal_mgr->stop(); - } - exec_env->clear_wal_mgr(); - } + void TearDown() override {} }; void http_request_done_cb(struct evhttp_request* req, void* arg) { @@ -129,4 +123,4 @@ TEST_F(StreamLoadTest, TestHeader) { evhttp_request_free(evhttp_req); } } -} // namespace doris +} // namespace doris \ No newline at end of file diff --git a/be/test/pipeline/pipeline_task_test.cpp b/be/test/pipeline/pipeline_task_test.cpp index e9c6c0f62dd2a3..a629344363a134 100644 --- a/be/test/pipeline/pipeline_task_test.cpp +++ b/be/test/pipeline/pipeline_task_test.cpp @@ -959,7 +959,6 @@ TEST_F(PipelineTaskTest, TEST_RESERVE_MEMORY_FAIL) { ((MockWorkloadGroupMgr*)ExecEnv::GetInstance()->_workload_group_manager)->_paused); } delete ExecEnv::GetInstance()->_workload_group_manager; - ExecEnv::GetInstance()->_workload_group_manager = nullptr; } // Test reserve memory fail for spillable pipeline task @@ -1136,7 +1135,6 @@ TEST_F(PipelineTaskTest, TEST_RESERVE_MEMORY_FAIL_SPILLABLE) { ((MockWorkloadGroupMgr*)ExecEnv::GetInstance()->_workload_group_manager)->_paused); } delete ExecEnv::GetInstance()->_workload_group_manager; - ExecEnv::GetInstance()->_workload_group_manager = nullptr; } TEST_F(PipelineTaskTest, TEST_INJECT_SHARED_STATE) { diff --git a/be/test/runtime/workload_group/workload_group_manager_test.cpp b/be/test/runtime/workload_group/workload_group_manager_test.cpp index 5808128bd93510..f81488928c9b80 100644 --- a/be/test/runtime/workload_group/workload_group_manager_test.cpp +++ b/be/test/runtime/workload_group/workload_group_manager_test.cpp @@ -22,13 +22,10 @@ #include #include #include -#include -#include #include -#include +#include #include -#include #include "common/config.h" #include "common/status.h" @@ -47,21 +44,10 @@ class WorkloadGroupManagerTest : public testing::Test { protected: void SetUp() override { _wg_manager = std::make_unique(); - std::ostringstream oss; - oss << "./wg_test_run_" << std::chrono::system_clock::now().time_since_epoch().count() - << "_" << getpid(); - _test_dir = oss.str(); - - std::error_code ec; - std::filesystem::remove_all(_test_dir, ec); - if (ec) { - FAIL() << "Failed to remove " << _test_dir << ": " << ec.message(); - } - std::filesystem::create_directories(_test_dir, ec); - ASSERT_FALSE(ec) << "Failed to create " << _test_dir << ": " << ec.message(); + EXPECT_EQ(system("rm -rf ./wg_test_run && mkdir -p ./wg_test_run"), 0); std::vector paths; - std::string path = std::filesystem::absolute(_test_dir).string(); + std::string path = std::filesystem::absolute("./wg_test_run").string(); auto olap_res = doris::parse_conf_store_paths(path, &paths); EXPECT_TRUE(olap_res.ok()) << olap_res.to_string(); @@ -92,15 +78,7 @@ class WorkloadGroupManagerTest : public testing::Test { ExecEnv::GetInstance()->_runtime_query_statistics_mgr->stop_report_thread(); SAFE_DELETE(ExecEnv::GetInstance()->_runtime_query_statistics_mgr); - if (ExecEnv::GetInstance()->_spill_stream_mgr != nullptr) { - ExecEnv::GetInstance()->_spill_stream_mgr->stop(); - } - SAFE_DELETE(ExecEnv::GetInstance()->_spill_stream_mgr); - ExecEnv::GetInstance()->_pipeline_tracer_ctx.reset(); - - std::error_code ec; - std::filesystem::remove_all(_test_dir, ec); - EXPECT_FALSE(ec) << "Failed to remove " << _test_dir << ": " << ec.message(); + EXPECT_EQ(system("rm -rf ./wg_test_run"), 0); config::spill_in_paused_queue_timeout_ms = _spill_in_paused_queue_timeout_ms; doris::ExecEnv::GetInstance()->set_memtable_memory_limiter(nullptr); } @@ -139,7 +117,6 @@ class WorkloadGroupManagerTest : public testing::Test { } std::unique_ptr _wg_manager; - std::string _test_dir; const int64_t _spill_in_paused_queue_timeout_ms = config::spill_in_paused_queue_timeout_ms; }; diff --git a/be/test/vec/sink/vtablet_writer_v2_test.cpp b/be/test/vec/sink/vtablet_writer_v2_test.cpp index 840eb8ab58d8bc..f8168efd2f29b1 100644 --- a/be/test/vec/sink/vtablet_writer_v2_test.cpp +++ b/be/test/vec/sink/vtablet_writer_v2_test.cpp @@ -271,7 +271,7 @@ TEST_F(TestVTabletWriterV2, shared_delta_writer_should_not_access_destroyed_crea // path, then cancel the current RuntimeState. Fixed code should observe the // current sink's cancel state and exit cleanly; current broken code reads // the destroyed creator RuntimeState from the shared DeltaWriterV2 and ASAN - // reports heap-use-after-free in the child process. + // reports heap-use-after-free. auto debug_point = std::make_shared(); debug_point->execute_limit = 1; debug_point->callback = std::function( @@ -280,17 +280,17 @@ TEST_F(TestVTabletWriterV2, shared_delta_writer_should_not_access_destroyed_crea DebugPoints::instance()->add("DeltaWriterV2.write.flush_limit_wait", debug_point); config::memtable_flush_running_count_limit = 0; - EXPECT_EXIT( - { - alarm(10); - auto status = current_writer->_write_memtable(block, 100, rows); - if (!status.ok() && - status.msg().find("current state cancelled") != std::string::npos) { - _exit(0); - } - _exit(1); - }, - ::testing::ExitedWithCode(0), ""); + // Master still uses a death test for this regression, but branch-4.0 BE UT + // reaches this case after many other suites have started process-wide + // worker threads. Forking from that state is not part of the regression and + // can abort in CI before the real assertion runs. Keep the same lifetime + // setup in process: after creator_ctx has been destroyed, the write must + // return the cancellation from current_ctx. If the shared DeltaWriterV2 + // still reads the creator RuntimeState, this path exercises the original + // dangling-state bug instead of returning "current state cancelled". + const auto status = current_writer->_write_memtable(block, 100, rows); + ASSERT_FALSE(status.ok()) << status; + EXPECT_NE(status.msg().find("current state cancelled"), std::string::npos) << status; config::memtable_flush_running_count_limit = old_flush_running_count_limit; DebugPoints::instance()->remove("DeltaWriterV2.write.flush_limit_wait");