This is an automated email from the ASF dual-hosted git repository.
panxiaolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 35742c5db6d [Bug](runtime-filter) remove
RuntimeFilterProducer::collect_realtime_profile (#50712)
35742c5db6d is described below
commit 35742c5db6deb8a3683391d5e90c2c78a7e1c966
Author: Pxl <[email protected]>
AuthorDate: Fri May 9 16:26:00 2025 +0800
[Bug](runtime-filter) remove
RuntimeFilterProducer::collect_realtime_profile (#50712)
### What problem does this PR solve?
```cpp
start BE in local mode
*** Query id: 287c1ab5b4394453-a0b065764ab98882 ***
*** is nereids: 1 ***
*** tablet id: 0 ***
*** Aborted at 1746692099 (unix time) try "date -d @1746692099" if you are
using GNU date ***
*** Current BE git commitID: ba085f6419 ***
*** SIGSEGV invalid permissions for mapped object (@0xf9200627cf) received
by PID 2839536 (TID 2845246 OR 0x7fa869ffb700) from PID 537274319; stack trace:
***
0# doris::signal::(anonymous namespace)::FailureSignalHandler(int,
siginfo_t*, void*) at /root/doris/be/src/common/signal_handler.h:421
1# PosixSignals::chained_handler(int, siginfo_t*, void*) [clone .part.0]
in /usr/lib/jvm/java-17-openjdk-amd64/lib/server/libjvm.so
2# JVM_handle_linux_signal in
/usr/lib/jvm/java-17-openjdk-amd64/lib/server/libjvm.so
3# 0x00007FABD2923090 in /lib/x86_64-linux-gnu/libc.so.6
4# doris::RuntimeFilterWrapper::debug_string[abi:cxx11]() const at
/root/doris/be/src/runtime_filter/runtime_filter_wrapper.cpp:592
5# doris::RuntimeFilter::_debug_string[abi:cxx11]() const in
/mnt/ssd01/pipline/OpenSourceDoris/clusterEnv/P0/Cluster0/be/lib/doris_be
6# doris::RuntimeFilterProducer::debug_string[abi:cxx11]() in
/mnt/ssd01/pipline/OpenSourceDoris/clusterEnv/P0/Cluster0/be/lib/doris_be
7#
doris::RuntimeFilterProducer::collect_realtime_profile(doris::RuntimeProfile*)
in /mnt/ssd01/pipline/OpenSourceDoris/clusterEnv/P0/Cluster0/be/lib/doris_be
8#
doris::RuntimeFilterProducerHelper::collect_realtime_profile(doris::RuntimeProfile*)
at /root/doris/be/src/runtime_filter/runtime_filter_producer_helper.cpp:179
9#
doris::pipeline::HashJoinBuildSinkLocalState::close(doris::RuntimeState*,
doris::Status) in
/mnt/ssd01/pipline/OpenSourceDoris/clusterEnv/P0/Cluster0/be/lib/doris_be
10# doris::pipeline::DataSinkOperatorXBase::close(doris::RuntimeState*,
doris::Status) in
/mnt/ssd01/pipline/OpenSourceDoris/clusterEnv/P0/Cluster0/be/lib/doris_be
11#
doris::pipeline::PartitionedHashJoinSinkLocalState::close(doris::RuntimeState*,
doris::Status) at
/root/doris/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:94
12# doris::pipeline::DataSinkOperatorXBase::close(doris::RuntimeState*,
doris::Status) in
/mnt/ssd01/pipline/OpenSourceDoris/clusterEnv/P0/Cluster0/be/lib/doris_be
13# doris::pipeline::PipelineTask::close(doris::Status, bool) in
/mnt/ssd01/pipline/OpenSourceDoris/clusterEnv/P0/Cluster0/be/lib/doris_be
14# doris::pipeline::close_task(doris::pipeline::PipelineTask*,
doris::Status, doris::pipeline::PipelineFragmentContext*) at
/root/doris/be/src/pipeline/task_scheduler.cpp:88
15#
doris::Defer<doris::pipeline::TaskScheduler::_do_work(int)::$_0>::~Defer() at
/root/doris/be/src/util/defer_op.h:37
16# doris::pipeline::TaskScheduler::_do_work(int) in
/mnt/ssd01/pipline/OpenSourceDoris/clusterEnv/P0/Cluster0/be/lib/doris_be
17# doris::ThreadPool::dispatch_thread() in
/mnt/ssd01/pipline/OpenSourceDoris/clusterEnv/P0/Cluster0/be/lib/doris_be
18# doris::Thread::supervise_thread(void*) at
/root/doris/be/src/util/thread.cpp:496
19# start_thread at /build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:478
20# __clone at ../sysdeps/unix/sysv/linux/x86_64/clone.S:97
```
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/runtime_filter/runtime_filter.cpp | 7 ++++++-
be/src/runtime_filter/runtime_filter.h | 2 +-
be/src/runtime_filter/runtime_filter_merger.h | 2 --
be/src/runtime_filter/runtime_filter_producer.cpp | 2 ++
be/src/runtime_filter/runtime_filter_producer.h | 14 --------------
.../runtime_filter_producer_helper.cpp | 4 ----
be/test/io/fs/s3_obj_storage_client_role_test.cpp | 3 +++
be/test/io/fs/s3_obj_storage_client_test.cpp | 2 ++
be/test/olap/rowset/beta_rowset_test.cpp | 2 ++
be/test/pipeline/pipeline_test.cpp | 20 +++++++-------------
.../runtime_filter/runtime_filter_producer_test.cpp | 16 ++++++++++++++++
11 files changed, 39 insertions(+), 35 deletions(-)
diff --git a/be/src/runtime_filter/runtime_filter.cpp
b/be/src/runtime_filter/runtime_filter.cpp
index f1b0dd5b8aa..0b61c98f382 100644
--- a/be/src/runtime_filter/runtime_filter.cpp
+++ b/be/src/runtime_filter/runtime_filter.cpp
@@ -129,7 +129,12 @@ std::string RuntimeFilter::_debug_string() const {
_has_remote_target ? "GLOBAL" : "LOCAL");
}
-void
RuntimeFilter::_check_wrapper_state(std::vector<RuntimeFilterWrapper::State>
assumed_states) {
+void RuntimeFilter::_check_wrapper_state(
+ const std::vector<RuntimeFilterWrapper::State>& assumed_states) {
+ // _wrapper is null mean rf is published
+ if (!_wrapper) {
+ return;
+ }
try {
_wrapper->check_state(assumed_states);
} catch (const Exception& e) {
diff --git a/be/src/runtime_filter/runtime_filter.h
b/be/src/runtime_filter/runtime_filter.h
index 8b42576c7b4..7273db88ecb 100644
--- a/be/src/runtime_filter/runtime_filter.h
+++ b/be/src/runtime_filter/runtime_filter.h
@@ -106,7 +106,7 @@ protected:
std::string _debug_string() const;
- void _check_wrapper_state(std::vector<RuntimeFilterWrapper::State>
assumed_states);
+ void _check_wrapper_state(const std::vector<RuntimeFilterWrapper::State>&
assumed_states);
// _wrapper is a runtime filter function wrapper
std::shared_ptr<RuntimeFilterWrapper> _wrapper;
diff --git a/be/src/runtime_filter/runtime_filter_merger.h
b/be/src/runtime_filter/runtime_filter_merger.h
index 79ae9dc869b..d373bb8be05 100644
--- a/be/src/runtime_filter/runtime_filter_merger.h
+++ b/be/src/runtime_filter/runtime_filter_merger.h
@@ -55,8 +55,6 @@ public:
}
// If input is a disabled predicate, the final result is a disabled
predicate.
- // If input is a ignored predicate, then we will skip this predicate.
- // If all inputs are ignored predicate, the final result is a ignored
predicate.
Status merge_from(const RuntimeFilter* other) {
_received_producer_num++;
if (_expected_producer_num < _received_producer_num) {
diff --git a/be/src/runtime_filter/runtime_filter_producer.cpp
b/be/src/runtime_filter/runtime_filter_producer.cpp
index 85d55f9f5c7..9bfe3cfad5b 100644
--- a/be/src/runtime_filter/runtime_filter_producer.cpp
+++ b/be/src/runtime_filter/runtime_filter_producer.cpp
@@ -89,6 +89,8 @@ Status RuntimeFilterProducer::publish(RuntimeState* state,
bool build_hash_table
DCHECK(_is_broadcast_join);
}
+ // wrapper may moved to rf merger, release wrapper here to make sure
thread safe
+ _wrapper.reset();
set_state(State::PUBLISHED);
return Status::OK();
}
diff --git a/be/src/runtime_filter/runtime_filter_producer.h
b/be/src/runtime_filter/runtime_filter_producer.h
index ea013625462..0edf85cd1d9 100644
--- a/be/src/runtime_filter/runtime_filter_producer.h
+++ b/be/src/runtime_filter/runtime_filter_producer.h
@@ -131,20 +131,6 @@ public:
_wrapper = wrapper;
}
- void collect_realtime_profile(RuntimeProfile* parent_operator_profile) {
- std::unique_lock<std::recursive_mutex> l(_rmtx);
- DCHECK(parent_operator_profile != nullptr);
- if (parent_operator_profile == nullptr) {
- return;
- }
- /*
- RuntimeFilterInfo:
- - RF0 Info: xxxx
- */
- parent_operator_profile->add_description(fmt::format("RF{} Info",
_wrapper->filter_id()),
- debug_string(),
"RuntimeFilterInfo");
- }
-
private:
RuntimeFilterProducer(const QueryContext* query_ctx, const
TRuntimeFilterDesc* desc)
: RuntimeFilter(desc), _is_broadcast_join(desc->is_broadcast_join)
{}
diff --git a/be/src/runtime_filter/runtime_filter_producer_helper.cpp
b/be/src/runtime_filter/runtime_filter_producer_helper.cpp
index 435585da3d3..6306fab1447 100644
--- a/be/src/runtime_filter/runtime_filter_producer_helper.cpp
+++ b/be/src/runtime_filter/runtime_filter_producer_helper.cpp
@@ -175,10 +175,6 @@ void RuntimeFilterProducerHelper::collect_realtime_profile(
"SkipProcess", _skip_runtime_filters_process ? "True" : "False",
"RuntimeFilterInfo");
publish_timer->set(_publish_runtime_filter_timer->value());
build_timer->set(_runtime_filter_compute_timer->value());
-
- for (auto& producer : _producers) {
- producer->collect_realtime_profile(parent_operator_profile);
- }
}
} // namespace doris
diff --git a/be/test/io/fs/s3_obj_storage_client_role_test.cpp
b/be/test/io/fs/s3_obj_storage_client_role_test.cpp
index 5760e6e4df3..4646de690df 100644
--- a/be/test/io/fs/s3_obj_storage_client_role_test.cpp
+++ b/be/test/io/fs/s3_obj_storage_client_role_test.cpp
@@ -51,6 +51,9 @@ protected:
S3ObjStorageClientRoleTest::obj_storage_client =
S3ClientFactory::instance().create(
{.endpoint = endpoint,
.region = region,
+ .ak = "",
+ .sk = "",
+ .token = "",
.bucket = bucket,
.provider = io::ObjStorageType::AWS,
.use_virtual_addressing = false,
diff --git a/be/test/io/fs/s3_obj_storage_client_test.cpp
b/be/test/io/fs/s3_obj_storage_client_test.cpp
index 3e68adbeaa1..97b44b8fa0c 100644
--- a/be/test/io/fs/s3_obj_storage_client_test.cpp
+++ b/be/test/io/fs/s3_obj_storage_client_test.cpp
@@ -48,6 +48,8 @@ protected:
.bucket = bucket,
.provider = io::ObjStorageType::AWS,
.use_virtual_addressing = false,
+ .role_arn = "",
+ .external_id = "",
});
ASSERT_TRUE(S3ObjStorageClientTest::obj_storage_client != nullptr);
diff --git a/be/test/olap/rowset/beta_rowset_test.cpp
b/be/test/olap/rowset/beta_rowset_test.cpp
index a1e6a0ec633..730999e650a 100644
--- a/be/test/olap/rowset/beta_rowset_test.cpp
+++ b/be/test/olap/rowset/beta_rowset_test.cpp
@@ -295,6 +295,8 @@ TEST_F(BetaRowsetTest, ReadTest) {
.sk = "sk",
.token = "",
.bucket = "",
+ .role_arn = "",
+ .external_id = "",
}};
std::string resource_id = "10000";
auto res = io::S3FileSystem::create(std::move(s3_conf),
io::FileSystem::TMP_FS_ID);
diff --git a/be/test/pipeline/pipeline_test.cpp
b/be/test/pipeline/pipeline_test.cpp
index 1b54db5a00b..391d99231f3 100644
--- a/be/test/pipeline/pipeline_test.cpp
+++ b/be/test/pipeline/pipeline_test.cpp
@@ -1095,28 +1095,22 @@ TEST_F(PipelineTest, PLAN_HASH_JOIN) {
->_runtime_filter_type,
RuntimeFilterType::IN_OR_BLOOM_FILTER);
EXPECT_EQ(_pipeline_tasks[1][j]->_is_pending_finish(), false);
+ auto wrapper =
+
sink_local_state._runtime_filter_producer_helper->_producers[0]->_wrapper;
EXPECT_EQ(_pipeline_tasks[1][j]->close(Status::OK()),
Status::OK());
-
EXPECT_EQ(sink_local_state._runtime_filter_producer_helper->_producers[0]
- ->_wrapper->get_real_type(),
+ EXPECT_EQ(wrapper->get_real_type(),
j == 0 ? RuntimeFilterType::IN_FILTER :
RuntimeFilterType::BLOOM_FILTER)
<< " " << j << " "
<<
sink_local_state._runtime_filter_producer_helper->_producers[0]
->debug_string();
-
EXPECT_TRUE(sink_local_state._runtime_filter_producer_helper->_producers[0]
- ->_wrapper->_state ==
RuntimeFilterWrapper::State::READY);
+ EXPECT_TRUE(wrapper->_state == RuntimeFilterWrapper::State::READY);
if (j == 0) {
-
EXPECT_EQ(sink_local_state._runtime_filter_producer_helper->_producers[0]
- ->_wrapper->_hybrid_set->size(),
- 1);
+ EXPECT_EQ(wrapper->_hybrid_set->size(), 1);
} else {
-
EXPECT_EQ(sink_local_state._runtime_filter_producer_helper->_producers[0]
-
->_wrapper->_bloom_filter_func->build_bf_by_runtime_size(),
- false);
+
EXPECT_EQ(wrapper->_bloom_filter_func->build_bf_by_runtime_size(), false);
-
EXPECT_EQ(sink_local_state._runtime_filter_producer_helper->_producers[0]
-
->_wrapper->_bloom_filter_func->_bloom_filter_length,
- 1048576);
+ EXPECT_EQ(wrapper->_bloom_filter_func->_bloom_filter_length,
1048576);
}
}
}
diff --git a/be/test/runtime_filter/runtime_filter_producer_test.cpp
b/be/test/runtime_filter/runtime_filter_producer_test.cpp
index b075247759a..50dee8f1903 100644
--- a/be/test/runtime_filter/runtime_filter_producer_test.cpp
+++ b/be/test/runtime_filter/runtime_filter_producer_test.cpp
@@ -175,4 +175,20 @@ TEST_F(RuntimeFilterProducerTest, set_disable) {
ASSERT_EQ(consumer->_wrapper->_state,
RuntimeFilterWrapper::State::DISABLED);
}
+TEST_F(RuntimeFilterProducerTest, publish_release_wrapper) {
+ auto desc = TRuntimeFilterDescBuilder()
+ .set_build_bf_by_runtime_size(false)
+ .set_is_broadcast_join(false)
+ .add_planId_to_target_expr(0)
+ .build();
+
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ FAIL_IF_ERROR_OR_CATCH_EXCEPTION(
+ _runtime_states[0]->register_producer_runtime_filter(desc,
&producer));
+
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::DISABLED);
+
FAIL_IF_ERROR_OR_CATCH_EXCEPTION(producer->publish(_runtime_states[0].get(),
true));
+ ASSERT_EQ(producer->_wrapper, nullptr);
+}
+
} // namespace doris
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
