This is an automated email from the ASF dual-hosted git repository.
sollhui 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 4a49d3434b0 [fix](be) Synchronize load error log access (#66250)
4a49d3434b0 is described below
commit 4a49d3434b01338ab36d196bd6f7f0a7d2a26a53
Author: Refrain <[email protected]>
AuthorDate: Tue Aug 18 15:07:08 2026 +0800
[fix](be) Synchronize load error log access (#66250)
### What problem does this PR solve?
Issue Number: None
Related PR: #41320
Load error-log writing and periodic status reporting can concurrently
access the same `std::ofstream`. The reporting path may close and upload
the stream while the writer is appending an error row, causing a data
race and a BE SIGSEGV in the stream write path.
This PR uses `_load_error_log_lock` to serialize stream creation and
writes and to protect the first-error and path state. A separate
`_s3_error_log_file_lock` serializes remote upload and URL publication,
while synchronous S3 I/O runs without holding the load-error writer
lock.
---
be/src/runtime/runtime_state.cpp | 87 ++++++++++++++++++++++++++--------------
be/src/runtime/runtime_state.h | 5 ++-
2 files changed, 61 insertions(+), 31 deletions(-)
diff --git a/be/src/runtime/runtime_state.cpp b/be/src/runtime/runtime_state.cpp
index 380bc8f8f72..50740802417 100644
--- a/be/src/runtime/runtime_state.cpp
+++ b/be/src/runtime/runtime_state.cpp
@@ -436,6 +436,21 @@ Status
RuntimeState::append_error_msg_to_file(std::function<std::string()> line,
if (query_type() != TQueryType::LOAD) {
return Status::OK();
}
+
+ const auto error_limit_status = [this]() -> Status {
+ if (_load_zero_tolerance) {
+ return Status::DataQualityError(
+ "Encountered unqualified data, stop processing. Please
check if the source "
+ "data matches the schema, and consider disabling strict
mode or increasing "
+ "max_filter_ratio.");
+ }
+ return Status::OK();
+ };
+ if (_num_print_error_rows.load(std::memory_order_relaxed) > MAX_ERROR_NUM)
{
+ return error_limit_status();
+ }
+
+ std::lock_guard<std::mutex> l(_load_error_log_lock);
// If file haven't been opened, open it here
if (_error_log_file == nullptr) {
Status status = create_error_log_file();
@@ -452,14 +467,7 @@ Status
RuntimeState::append_error_msg_to_file(std::function<std::string()> line,
}
// If num of printed error row exceeds the limit, don't add error messages
to error log file any more
if (_num_print_error_rows.fetch_add(1, std::memory_order_relaxed) >
MAX_ERROR_NUM) {
- // if _load_zero_tolerance, return Error to stop the load process
immediately.
- if (_load_zero_tolerance) {
- return Status::DataQualityError(
- "Encountered unqualified data, stop processing. Please
check if the source "
- "data matches the schema, and consider disabling strict
mode or increasing "
- "max_filter_ratio.");
- }
- return Status::OK();
+ return error_limit_status();
}
fmt::memory_buffer out;
@@ -481,33 +489,52 @@ Status
RuntimeState::append_error_msg_to_file(std::function<std::string()> line,
return Status::OK();
}
+std::string RuntimeState::get_first_error_msg() const {
+ std::lock_guard<std::mutex> l(_load_error_log_lock);
+ return _first_error_msg;
+}
+
std::string RuntimeState::get_error_log_file_path() {
- DBUG_EXECUTE_IF("RuntimeState::get_error_log_file_path.block", {
- if (!_error_log_file_path.empty()) {
- std::this_thread::sleep_for(std::chrono::seconds(1));
+ std::lock_guard<std::mutex> s3_lock(_s3_error_log_file_lock);
+ std::shared_ptr<io::S3FileSystem> s3_error_fs;
+ std::string local_error_log_file_path;
+ std::string remote_error_log_file_path;
+ {
+ std::lock_guard<std::mutex> load_lock(_load_error_log_lock);
+ DBUG_EXECUTE_IF("RuntimeState::get_error_log_file_path.block", {
+ if (!_error_log_file_path.empty()) {
+ std::this_thread::sleep_for(std::chrono::seconds(1));
+ }
+ });
+ if (!_s3_error_fs || !_error_log_file || !_error_log_file->is_open()) {
+ return _error_log_file_path;
}
- });
- std::lock_guard<std::mutex> l(_s3_error_log_file_lock);
- if (_s3_error_fs && _error_log_file && _error_log_file->is_open()) {
+
// close error log file
_error_log_file->close();
- std::string error_log_absolute_path =
-
_exec_env->load_path_mgr()->get_load_error_absolute_path(_error_log_file_path);
- // upload error log file to s3
- Status st = _s3_error_fs->upload(error_log_absolute_path,
_s3_error_log_file_path);
- if (!st.ok()) {
- // upload failed and return local error log file path
- LOG(WARNING) << "Fail to upload error file to s3,
error_log_file_path="
- << _error_log_file_path << ", error=" << st;
- return _error_log_file_path;
- }
- // expiration must be less than a week (in seconds) for presigned url
- static const unsigned EXPIRATION_SECONDS = 7 * 24 * 60 * 60 - 1;
- // Use public or private endpoint based on configuration
- _error_log_file_path =
- _s3_error_fs->generate_presigned_url(_s3_error_log_file_path,
EXPIRATION_SECONDS,
-
config::use_public_endpoint_for_error_log);
+ s3_error_fs = _s3_error_fs;
+ local_error_log_file_path = _error_log_file_path;
+ remote_error_log_file_path = _s3_error_log_file_path;
}
+
+ std::string error_log_absolute_path =
+
_exec_env->load_path_mgr()->get_load_error_absolute_path(local_error_log_file_path);
+ // upload error log file to s3
+ Status st = s3_error_fs->upload(error_log_absolute_path,
remote_error_log_file_path);
+ if (!st.ok()) {
+ // upload failed and return local error log file path
+ LOG(WARNING) << "Fail to upload error file to s3, error_log_file_path="
+ << local_error_log_file_path << ", error=" << st;
+ return local_error_log_file_path;
+ }
+ // expiration must be less than a week (in seconds) for presigned url
+ static const unsigned EXPIRATION_SECONDS = 7 * 24 * 60 * 60 - 1;
+ // Use public or private endpoint based on configuration
+ auto presigned_url =
+ s3_error_fs->generate_presigned_url(remote_error_log_file_path,
EXPIRATION_SECONDS,
+
config::use_public_endpoint_for_error_log);
+ std::lock_guard<std::mutex> load_lock(_load_error_log_lock);
+ _error_log_file_path = std::move(presigned_url);
return _error_log_file_path;
}
diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h
index a3cfc5e4cad..bd9b849ba7a 100644
--- a/be/src/runtime/runtime_state.h
+++ b/be/src/runtime/runtime_state.h
@@ -340,7 +340,7 @@ public:
std::string get_error_log_file_path();
- std::string get_first_error_msg() const { return _first_error_msg; }
+ std::string get_first_error_msg() const;
// append error msg and error line to file when loading data.
// is_summary is true, means we are going to write the summary line
@@ -1029,6 +1029,9 @@ private:
std::shared_ptr<io::S3FileSystem> _s3_error_fs;
// error file path on s3,
${bucket}/${prefix}/error_log/${label}_${fragment_instance_id}
std::string _s3_error_log_file_path;
+ // Protects the load error log stream, paths, and first error message.
+ mutable std::mutex _load_error_log_lock;
+ // Serializes S3 upload and presigned URL publication.
std::mutex _s3_error_log_file_lock;
// used for encoding the global lazy materialize
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
