You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/10/21 10:22:24 UTC
(doris) branch branch-4.2 updated: branch-4.2: [fix](load) Improve cancellation responsiveness and avoid EOS worker starvation (#69274)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 2619cf080b4 branch-4.2: [fix](load) Improve cancellation responsiveness and avoid EOS worker starvation (#69274)
2619cf080b4 is described below
commit 2619cf080b45be254fab7049925c1b76e1c00016
Author: hui lai <la...@selectdb.com>
AuthorDate: Wed Oct 21 18:22:15 2026 +0800
branch-4.2: [fix](load) Improve cancellation responsiveness and avoid EOS worker starvation (#69274)
### What problem does this PR solve?
Backport these three PRs together to `branch-4.2`:
- https://github.com/apache/doris/pull/68628
- https://github.com/apache/doris/pull/68328
- https://github.com/apache/doris/pull/67759
Early EOS requests can occupy all heavy workers while waiting for
remaining senders, and cancellation can queue behind those same workers
or block on channel/writer locks. Use an asynchronous sender-arrival
barrier, isolate cancellation RPCs in a dedicated pool, and publish a
shared cancellation status observed at flush and delete-bitmap work
boundaries. Final-owner cleanup drains running callbacks before
releasing writer resources.
Keep branch-4.2's single-replica writes and row-index writer interfaces.
Retain applicable local/cloud ordinary-writer, EOS, manager,
service-routing, bitmap-token, and memory-reclamation unit coverage.
Upstream group-rowset-only cases and the integration fixture requiring
newer test/storage helpers are not included in this release backport.
### Release note
Improve load cancellation responsiveness and prevent heavy-worker
starvation during incremental-channel EOS close.
### Check List (For Author)
- Test: Compilation and test execution intentionally skipped as
requested. Passed `git diff --check`, changed-line/new-file clang-format
16.0.5 formatting checks, and local include-path verification. This
release branch does not contain `build-support/check-build-hygiene.sh`;
compilation-dependent analysis was not run.
- Behavior changed: Yes; cancellation dispatch is isolated, cancellation
publication avoids writer locks, and waiting EOS RPCs release heavy
workers.
- Does this need documentation: No.
---
be/src/cloud/cloud_delta_writer.cpp | 6 +-
be/src/cloud/cloud_rowset_builder.cpp | 7 +-
be/src/cloud/cloud_rowset_writer.cpp | 3 +-
be/src/cloud/cloud_tablets_channel.cpp | 25 +-
be/src/cloud/cloud_tablets_channel.h | 2 +
be/src/common/config.cpp | 9 +
be/src/common/config.h | 6 +
be/src/common/metrics/doris_metrics.h | 5 +
be/src/load/channel/eos_completion.cpp | 57 +++
be/src/load/channel/eos_completion.h | 53 ++
be/src/load/channel/load_channel.cpp | 59 ++-
be/src/load/channel/load_channel.h | 19 +-
be/src/load/channel/load_channel_mgr.cpp | 112 ++--
be/src/load/channel/load_channel_mgr.h | 13 +-
be/src/load/channel/tablets_channel.cpp | 94 +++-
be/src/load/channel/tablets_channel.h | 42 +-
be/src/load/delta_writer/delta_writer.cpp | 29 +-
be/src/load/delta_writer/delta_writer.h | 5 +-
be/src/load/delta_writer/delta_writer_context.h | 3 +
be/src/load/memtable/memtable_flush_executor.cpp | 15 +-
be/src/load/memtable/memtable_flush_executor.h | 17 +-
be/src/load/memtable/memtable_memory_limiter.cpp | 17 +-
be/src/load/memtable/memtable_writer.cpp | 27 +-
be/src/service/internal_service.cpp | 48 +-
be/src/service/internal_service.h | 2 +
.../storage/delete/calc_delete_bitmap_executor.cpp | 71 ++-
.../storage/delete/calc_delete_bitmap_executor.h | 46 +-
be/src/storage/rowset/beta_rowset_writer.cpp | 27 +-
be/src/storage/rowset/beta_rowset_writer.h | 2 +
be/src/storage/rowset/rowset_writer.h | 3 +
be/src/storage/rowset/rowset_writer_context.h | 1 +
be/src/storage/rowset_builder.cpp | 12 +-
be/src/storage/rowset_builder.h | 2 +-
be/test/load/channel/eos_completion_test.cpp | 151 ++++++
be/test/load/channel/load_channel_eos_test.cpp | 566 +++++++++++++++++++++
be/test/load/channel/load_channel_mgr_test.cpp | 274 ++++++++++
.../load/delta_writer/delta_writer_cancel_test.cpp | 266 ++++++++++
.../load/memtable/memtable_memory_limiter_test.cpp | 91 ++++
.../internal_service_load_work_pool_test.cpp | 175 +++++++
.../delete/calc_delete_bitmap_executor_test.cpp | 152 ++++++
40 files changed, 2337 insertions(+), 177 deletions(-)
diff --git a/be/src/cloud/cloud_delta_writer.cpp b/be/src/cloud/cloud_delta_writer.cpp
index b85e83be20c..8251466a5b5 100644
--- a/be/src/cloud/cloud_delta_writer.cpp
+++ b/be/src/cloud/cloud_delta_writer.cpp
@@ -80,7 +80,7 @@ Status CloudDeltaWriter::write(const Block* block, const DorisVector<uint32_t>&
ExecEnv::GetInstance()->memtable_memory_limiter()->handle_table_memtable_backpressure(
[this]() {
std::lock_guard lock(_mtx);
- return _is_cancelled;
+ return _is_cancelled || !_get_load_cancel_status().ok();
},
table_id());
}
@@ -100,6 +100,7 @@ Status CloudDeltaWriter::write(const Block* block, const DorisVector<uint32_t>&
config::memtable_flush_running_count_limit;
};
while (need_backpressure()) {
+ RETURN_IF_ERROR(_get_load_cancel_status());
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
@@ -143,6 +144,7 @@ const RowsetMetaSharedPtr& CloudDeltaWriter::rowset_meta() {
Status CloudDeltaWriter::commit_rowset() {
g_cloud_commit_rowset_count << 1;
std::lock_guard<bthread::Mutex> lock(_mtx);
+ RETURN_IF_ERROR(_get_load_cancel_status());
// Handle empty rowset (no data written)
if (!_is_init) {
@@ -164,6 +166,7 @@ Status CloudDeltaWriter::_commit_empty_rowset() {
RETURN_IF_ERROR(_rowset_builder->init());
RETURN_IF_ERROR(_rowset_builder->build_rowset());
+ RETURN_IF_ERROR(_get_load_cancel_status());
// If skip writing empty rowset metadata is enabled, we do not commit rowset to meta service.
if (config::skip_writing_empty_rowset_metadata) {
@@ -175,6 +178,7 @@ Status CloudDeltaWriter::_commit_empty_rowset() {
}
Status CloudDeltaWriter::set_txn_related_info() {
+ RETURN_IF_ERROR(_get_load_cancel_status());
return rowset_builder()->set_txn_related_info();
}
diff --git a/be/src/cloud/cloud_rowset_builder.cpp b/be/src/cloud/cloud_rowset_builder.cpp
index a1bbbdca8f3..c2adde120ac 100644
--- a/be/src/cloud/cloud_rowset_builder.cpp
+++ b/be/src/cloud/cloud_rowset_builder.cpp
@@ -23,6 +23,7 @@
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet.h"
#include "cloud/cloud_tablet_mgr.h"
+#include "cpp/sync_point.h"
#include "io/fs/file_system.h"
#include "storage/storage_policy.h"
@@ -37,6 +38,8 @@ CloudRowsetBuilder::CloudRowsetBuilder(CloudStorageEngine& engine, const WriteRe
CloudRowsetBuilder::~CloudRowsetBuilder() {
// Clear file cache immediately when load fails
if (_is_init && _rowset != nullptr && _rowset->rowset_meta()->rowset_state() == PREPARED) {
+ TEST_SYNC_POINT_CALLBACK("CloudRowsetBuilder::~CloudRowsetBuilder:before_clear_cache",
+ this);
_rowset->clear_cache();
}
}
@@ -69,6 +72,7 @@ Status CloudRowsetBuilder::init() {
context.txn_id = _req.txn_id;
context.txn_expiration = _req.txn_expiration;
context.load_id = _req.load_id;
+ context.load_cancel_status = _req.load_cancel_status;
context.rowset_state = PREPARED;
context.segments_overlap = OVERLAPPING;
context.tablet_schema = _tablet_schema;
@@ -89,7 +93,8 @@ Status CloudRowsetBuilder::init() {
_rowset_writer = DORIS_TRY(_tablet->create_rowset_writer(context, false));
- _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor()->create_token();
+ _calc_delete_bitmap_token =
+ _engine.calc_delete_bitmap_executor()->create_token(_req.load_cancel_status);
if (!_skip_writing_rowset_metadata) {
RETURN_IF_ERROR(_engine.meta_mgr().prepare_rowset(*_rowset_writer->rowset_meta(), "",
diff --git a/be/src/cloud/cloud_rowset_writer.cpp b/be/src/cloud/cloud_rowset_writer.cpp
index fdf62a71abc..10540ece259 100644
--- a/be/src/cloud/cloud_rowset_writer.cpp
+++ b/be/src/cloud/cloud_rowset_writer.cpp
@@ -80,7 +80,8 @@ Status CloudRowsetWriter::init(const RowsetWriterContext& rowset_writer_context)
_context.segment_collector = std::make_shared<SegmentCollectorT<BaseBetaRowsetWriter>>(this);
_context.file_writer_creator = std::make_shared<FileWriterCreatorT<BaseBetaRowsetWriter>>(this);
if (_context.mow_context != nullptr) {
- _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor_for_load()->create_token();
+ _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor_for_load()->create_token(
+ _context.load_cancel_status);
}
return Status::OK();
}
diff --git a/be/src/cloud/cloud_tablets_channel.cpp b/be/src/cloud/cloud_tablets_channel.cpp
index 3efedcc6b16..efb66951055 100644
--- a/be/src/cloud/cloud_tablets_channel.cpp
+++ b/be/src/cloud/cloud_tablets_channel.cpp
@@ -23,6 +23,7 @@
#include "cloud/cloud_meta_mgr.h"
#include "cloud/cloud_storage_engine.h"
#include "cloud/config.h"
+#include "cpp/sync_point.h"
#include "load/channel/tablets_channel.h"
#include "load/delta_writer/delta_writer.h"
@@ -50,17 +51,12 @@ Status CloudTabletsChannel::add_batch(const PTabletWriterAddBlockRequest& reques
_add_batch_number_counter->update(1);
}
- auto status = _get_current_seq(cur_seq, request);
- if (UNLIKELY(!status.ok())) {
+ bool should_write = false;
+ auto status = _get_current_seq(cur_seq, request, should_write);
+ if (UNLIKELY(!status.ok()) || !should_write) {
return status;
}
- if (request.packet_seq() < cur_seq) {
- LOG(INFO) << "packet has already recept before, expect_seq=" << cur_seq
- << ", recept_seq=" << request.packet_seq();
- return Status::OK();
- }
-
if (request.is_adaptive_random_bucket()) {
std::unordered_map<int64_t, DorisVector<uint32_t>> partition_to_rowidxs;
RETURN_IF_ERROR(_build_partition_to_rowidxs_for_adaptive_random_bucket(
@@ -133,9 +129,9 @@ Status CloudTabletsChannel::_init_writers_by_partition_ids(
Status CloudTabletsChannel::close(LoadChannel* parent, const PTabletWriterAddBlockRequest& req,
PTabletWriterAddBlockResult* res, bool* finished) {
// FIXME(plat1ko): Too many duplicate code with `TabletsChannel`
- std::lock_guard l(_lock);
+ std::unique_lock l(_lock);
if (_state == kFinished) {
- return _close_status;
+ return _get_close_result(l, req.sender_id(), res);
}
auto sender_id = req.sender_id();
@@ -161,9 +157,15 @@ Status CloudTabletsChannel::close(LoadChannel* parent, const PTabletWriterAddBlo
return Status::OK();
}
+ return _close_and_notify(parent, l, req, res);
+}
+
+Status CloudTabletsChannel::_close_writers(LoadChannel* parent,
+ const PTabletWriterAddBlockRequest& req,
+ PTabletWriterAddBlockResult* res) {
auto* tablet_errors = res->mutable_tablet_errors();
auto* tablet_vec = res->mutable_tablet_vec();
- _state = kFinished;
+ TEST_SYNC_POINT_RETURN_WITH_VALUE("CloudTabletsChannel::close.before_flush", Status::OK(), res);
// All senders are closed
// 1. close all delta writers. under _lock.
@@ -172,6 +174,7 @@ Status CloudTabletsChannel::close(LoadChannel* parent, const PTabletWriterAddBlo
bool success = true;
for (auto&& [tablet_id, base_writer] : _tablet_writers) {
+ RETURN_IF_ERROR(_check_cancelled());
auto* writer = static_cast<CloudDeltaWriter*>(base_writer.get());
// ATTN: the strict mode means strict filtering of column type conversions during import.
// Sometimes all inputs are filtered, but the partition ID is still set, and the writer is
diff --git a/be/src/cloud/cloud_tablets_channel.h b/be/src/cloud/cloud_tablets_channel.h
index 0f9704bb4b4..b2360ace00b 100644
--- a/be/src/cloud/cloud_tablets_channel.h
+++ b/be/src/cloud/cloud_tablets_channel.h
@@ -41,6 +41,8 @@ public:
PTabletWriterAddBlockResult* res, bool* finished) override;
private:
+ Status _close_writers(LoadChannel* parent, const PTabletWriterAddBlockRequest& req,
+ PTabletWriterAddBlockResult* response) override;
Status _prepare_adaptive_random_bucket_writer(BaseDeltaWriter* writer) override;
Status _init_writers_by_partition_ids(const std::unordered_set<int64_t>& partition_ids);
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 295cc4e556a..62476f51bc0 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -642,6 +642,15 @@ DEFINE_mInt64(load_error_log_reserve_hours, "48");
// error log size limit, default 200MB
DEFINE_mInt64(load_error_log_limit_bytes, "209715200");
+// Dedicated load cancellation workers. Requires a restart.
+DEFINE_Int32(brpc_load_light_work_pool_threads, "32");
+DEFINE_Validator(brpc_load_light_work_pool_threads,
+ [](const int config) -> bool { return config > 0; });
+// Queue capacity: -1 selects a CPU-scaled default. Requires a restart.
+DEFINE_Int32(brpc_load_light_work_pool_max_queue_size, "-1");
+DEFINE_Validator(brpc_load_light_work_pool_max_queue_size,
+ [](const int config) -> bool { return config == -1 || config > 0; });
+
DEFINE_Int32(brpc_heavy_work_pool_threads, "-1");
DEFINE_Int32(brpc_peer_fetch_pool_threads, "-1");
DEFINE_Int32(brpc_light_work_pool_threads, "-1");
diff --git a/be/src/common/config.h b/be/src/common/config.h
index f3db5f1a5af..afb0e6c5802 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -702,6 +702,12 @@ DECLARE_mInt64(load_error_log_reserve_hours);
// error log size limit, default 200MB
DECLARE_mInt64(load_error_log_limit_bytes);
+// Dedicated load cancellation workers, default 32. Must be positive; requires a restart.
+DECLARE_Int32(brpc_load_light_work_pool_threads);
+// Queue capacity for the dedicated load cancellation pool.
+// -1 selects max(1024, CPU cores * 32) queued requests. Requires a restart.
+DECLARE_Int32(brpc_load_light_work_pool_max_queue_size);
+
// be brpc interface is classified into two categories: light and heavy
// each category has diffrent thread number
// threads to handle heavy api interface, such as transmit_block etc
diff --git a/be/src/common/metrics/doris_metrics.h b/be/src/common/metrics/doris_metrics.h
index ceb30b850e4..7ba0a1b7e40 100644
--- a/be/src/common/metrics/doris_metrics.h
+++ b/be/src/common/metrics/doris_metrics.h
@@ -229,6 +229,11 @@ public:
IntCounter* upload_rowset_count = nullptr;
IntCounter* upload_fail_count = nullptr;
+ UIntGauge* load_light_work_pool_queue_size = nullptr;
+ UIntGauge* load_light_work_active_threads = nullptr;
+ UIntGauge* load_light_work_pool_max_queue_size = nullptr;
+ UIntGauge* load_light_work_max_threads = nullptr;
+
UIntGauge* light_work_pool_queue_size = nullptr;
UIntGauge* heavy_work_pool_queue_size = nullptr;
UIntGauge* peer_fetch_work_pool_queue_size = nullptr;
diff --git a/be/src/load/channel/eos_completion.cpp b/be/src/load/channel/eos_completion.cpp
new file mode 100644
index 00000000000..c9d1f5b3cd9
--- /dev/null
+++ b/be/src/load/channel/eos_completion.cpp
@@ -0,0 +1,57 @@
+// 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 "load/channel/eos_completion.h"
+
+#include <utility>
+
+namespace doris {
+
+EosCompletion::~EosCompletion() {
+ complete(Status::Cancelled("Load channel destroyed before all EOS senders arrived"));
+}
+
+void EosCompletion::add_waiter(Callback callback) {
+ Status status;
+ {
+ std::lock_guard lock(_mutex);
+ if (!_status.has_value()) {
+ _waiters.push_back(std::move(callback));
+ return;
+ }
+ status = *_status;
+ }
+ callback(status);
+}
+
+void EosCompletion::complete(const Status& status) {
+ std::vector<Callback> waiters;
+ {
+ std::lock_guard lock(_mutex);
+ if (_status.has_value()) {
+ // Sender arrival can race cancellation or load-channel expiration.
+ return;
+ }
+ _status = status;
+ waiters.swap(_waiters);
+ }
+ for (auto& waiter : waiters) {
+ waiter(status);
+ }
+}
+
+} // namespace doris
diff --git a/be/src/load/channel/eos_completion.h b/be/src/load/channel/eos_completion.h
new file mode 100644
index 00000000000..ab1cec1657f
--- /dev/null
+++ b/be/src/load/channel/eos_completion.h
@@ -0,0 +1,53 @@
+// 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.
+
+#pragma once
+
+#include <functional>
+#include <mutex>
+#include <optional>
+#include <vector>
+
+#include "common/status.h"
+
+namespace doris {
+
+// One-shot arrival barrier for all senders of one tablets channel. Success means
+// all senders reached EOS, not that the final flush/commit completed. Sender
+// accounting and final close remain in TabletsChannel/CloudTabletsChannel. This
+// object only owns RPC callbacks, never a channel or request, so pending RPCs
+// cannot keep a cancelled/expired load channel alive.
+class EosCompletion {
+public:
+ using Callback = std::function<void(const Status&)>;
+
+ ~EosCompletion();
+
+ // Register only after all synchronous access to the RPC objects has ended.
+ // May invoke callback inline if all senders arrived or cancellation already won.
+ void add_waiter(Callback callback);
+
+ // The first terminal result wins. Invoke callbacks outside the mutex.
+ void complete(const Status& status);
+
+private:
+ std::mutex _mutex;
+ std::optional<Status> _status;
+ std::vector<Callback> _waiters;
+};
+
+} // namespace doris
diff --git a/be/src/load/channel/load_channel.cpp b/be/src/load/channel/load_channel.cpp
index 1b6af15240c..d98acb81703 100644
--- a/be/src/load/channel/load_channel.cpp
+++ b/be/src/load/channel/load_channel.cpp
@@ -23,6 +23,7 @@
#include "cloud/cloud_tablets_channel.h"
#include "cloud/config.h"
#include "common/logging.h"
+#include "load/channel/eos_completion.h"
#include "load/channel/tablets_channel.h"
#include "runtime/exec_env.h"
#include "runtime/fragment_mgr.h"
@@ -40,6 +41,7 @@ LoadChannel::LoadChannel(const UniqueId& load_id, int64_t timeout_s, bool is_hig
std::string sender_ip, int64_t backend_id, bool enable_profile,
int64_t wg_id)
: _load_id(load_id),
+ _cancel_status(std::make_shared<AtomicStatus>()),
_timeout_s(timeout_s),
_is_high_priority(is_high_priority),
_sender_ip(std::move(sender_ip)),
@@ -78,6 +80,15 @@ LoadChannel::LoadChannel(const UniqueId& load_id, int64_t timeout_s, bool is_hig
}
LoadChannel::~LoadChannel() {
+ // Pending EOS RPCs do not own the channel. Complete them before destroying
+ // writers, which can wait for running flush and delete-bitmap tasks.
+ const auto reason =
+ is_cancelled()
+ ? cancel_status()
+ : Status::Cancelled("Load channel destroyed before all EOS senders arrived");
+ for (const auto& [id, channel] : _tablets_channels) {
+ channel->eos_completion()->complete(reason);
+ }
g_loadchannel_cnt << -1;
std::stringstream rows_str;
for (const auto& entry : _tablets_channels_rows) {
@@ -138,6 +149,7 @@ Status LoadChannel::open(const PTabletWriterOpenRequest& params) {
channel = std::make_shared<TabletsChannel>(engine.to_local(), key, _load_id,
_is_high_priority, _self_profile);
}
+ channel->set_load_cancel_status(_cancel_status);
{
std::lock_guard<std::mutex> l(_tablets_channels_lock);
_tablets_channels.insert({index_id, channel});
@@ -178,7 +190,9 @@ Status LoadChannel::_get_tablets_channel(std::shared_ptr<BaseTabletsChannel>& ch
}
Status LoadChannel::add_batch(const PTabletWriterAddBlockRequest& request,
- PTabletWriterAddBlockResult* response) {
+ PTabletWriterAddBlockResult* response,
+ std::shared_ptr<EosCompletion>* eos_completion) {
+ RETURN_IF_ERROR(cancel_status());
DBUG_EXECUTE_IF("LoadChannel.add_batch.failed",
{ return Status::InternalError("fault injection"); });
SCOPED_TIMER(_add_batch_timer);
@@ -204,8 +218,12 @@ Status LoadChannel::add_batch(const PTabletWriterAddBlockRequest& request,
}
// 3. handle eos
- // if channel is incremental, maybe hang on close until all close request arrived.
+ // Keep the RPC pending without occupying a heavy worker while senders arrive.
if (request.has_eos() && request.eos()) {
+ if (request.hang_wait()) {
+ DCHECK(!channel->is_incremental_channel());
+ *eos_completion = channel->eos_completion();
+ }
st = _handle_eos(channel.get(), request, response);
_report_profile(response);
if (!st.ok()) {
@@ -227,25 +245,11 @@ Status LoadChannel::_handle_eos(BaseTabletsChannel* channel,
bool finished = false;
auto index_id = request.index_id();
+ // close() publishes sender arrival before flushing. Final flush/commit
+ // errors belong to this RPC and cannot revoke earlier EOS responses.
RETURN_IF_ERROR(channel->close(this, request, response, &finished));
- // for init node, we close waiting(hang on) all close request and let them return together.
- if (request.has_hang_wait() && request.hang_wait()) {
- DCHECK(!channel->is_incremental_channel());
- VLOG_DEBUG << fmt::format("txn {}: reciever index {} close waiting by sender {}", _txn_id,
- request.index_id(), request.sender_id());
- int count = 0;
- while (!channel->is_finished()) {
- bthread_usleep(1000);
- count++;
- }
- // now maybe finished or cancelled.
- VLOG_TRACE << "reciever close wait finished!" << request.sender_id();
- if (count >= 1000 * _timeout_s) { // maybe config::streaming_load_rpc_max_alive_time_sec
- return Status::InternalError("Tablets channel didn't wait all close");
- }
- }
-
+ RETURN_IF_ERROR(cancel_status());
if (finished) {
std::lock_guard<std::mutex> l(_lock);
{
@@ -303,13 +307,18 @@ bool LoadChannel::is_finished() {
return _tablets_channels.empty();
}
-Status LoadChannel::cancel() {
- _cancelled.store(true);
- std::lock_guard<std::mutex> l(_lock);
- for (auto& it : _tablets_channels) {
- static_cast<void>(it.second->cancel());
- }
+Status LoadChannel::cancel(const Status& reason) {
+ DCHECK(!reason.ok());
+ _cancel_status->update(reason);
return Status::OK();
}
+Status LoadChannel::cancel_status() const {
+ return _cancel_status->ok() ? Status::OK() : _cancel_status->status();
+}
+
+bool LoadChannel::is_cancelled() const {
+ return !_cancel_status->ok();
+}
+
} // namespace doris
diff --git a/be/src/load/channel/load_channel.h b/be/src/load/channel/load_channel.h
index 2702cf192fe..197615c7d93 100644
--- a/be/src/load/channel/load_channel.h
+++ b/be/src/load/channel/load_channel.h
@@ -34,11 +34,13 @@
namespace doris {
+class EosCompletion;
class PTabletWriterOpenRequest;
class PTabletWriterAddBlockRequest;
class PTabletWriterAddBlockResult;
class OpenPartitionRequest;
class BaseTabletsChannel;
+class AtomicStatus;
// A LoadChannel manages tablets channels for all indexes
// corresponding to a certain load job
@@ -51,14 +53,21 @@ public:
// open a new load channel if not exist
Status open(const PTabletWriterOpenRequest& request);
- // this batch must belong to a index in one transaction
+ // This batch must belong to one index in one transaction. For hang_wait EOS,
+ // return the sender-arrival barrier in eos_completion. Final flush/commit
+ // stays synchronous on the last sender. The caller must register its RPC
+ // completion only after all synchronous request/response access has ended.
Status add_batch(const PTabletWriterAddBlockRequest& request,
- PTabletWriterAddBlockResult* response);
+ PTabletWriterAddBlockResult* response,
+ std::shared_ptr<EosCompletion>* eos_completion);
// return true if this load channel has been opened and all tablets channels are closed then.
bool is_finished();
- Status cancel();
+ // Publish cancellation without taking channel/writer locks. In-flight
+ // requests retain ownership and release writers after their work finishes.
+ Status cancel(const Status& reason = Status::Cancelled("load channel cancelled"));
+ Status cancel_status() const;
time_t last_updated_time() const { return _last_updated_time.load(); }
@@ -68,7 +77,7 @@ public:
bool is_high_priority() const { return _is_high_priority; }
- bool is_cancelled() const { return _cancelled.load(); }
+ bool is_cancelled() const;
WorkloadGroupPtr workload_group() const { return _resource_ctx->workload_group(); }
@@ -111,7 +120,7 @@ private:
std::unordered_set<int64_t> _finished_channel_ids;
// set to true if at least one tablets channel has been opened
bool _opened = false;
- std::atomic<bool> _cancelled {false};
+ const std::shared_ptr<AtomicStatus> _cancel_status;
std::shared_ptr<ResourceContext> _resource_ctx;
diff --git a/be/src/load/channel/load_channel_mgr.cpp b/be/src/load/channel/load_channel_mgr.cpp
index a5f655b5fb6..35ac1754606 100644
--- a/be/src/load/channel/load_channel_mgr.cpp
+++ b/be/src/load/channel/load_channel_mgr.cpp
@@ -88,6 +88,14 @@ Status LoadChannelMgr::open(const PTabletWriterOpenRequest& params) {
if (it != _load_channels.end()) {
channel = it->second;
} else {
+ auto* handle = _load_state_channels->lookup(load_id.to_string());
+ if (handle != nullptr) {
+ auto* value = static_cast<CacheValue*>(_load_state_channels->value(handle));
+ auto status =
+ value != nullptr ? Status::Cancelled(value->_cancel_reason) : Status::OK();
+ _load_state_channels->release(handle);
+ RETURN_IF_ERROR(status);
+ }
// create a new load channel
int64_t timeout_in_req_s =
params.has_load_channel_timeout_s() ? params.load_channel_timeout_s() : -1;
@@ -121,14 +129,11 @@ Status LoadChannelMgr::_get_load_channel(std::shared_ptr<LoadChannel>& channel,
if (handle != nullptr) {
// load is cancelled
if (auto* value = _load_state_channels->value(handle); value != nullptr) {
- const auto& cancel_reason = reinterpret_cast<CacheValue*>(value)->_cancel_reason;
+ const auto cancel_reason = reinterpret_cast<CacheValue*>(value)->_cancel_reason;
_load_state_channels->release(handle);
- if (!cancel_reason.empty()) {
- LOG(INFO) << fmt::format(
- "The channel has been cancelled, load_id = {}, error = {}",
- print_id(load_id), cancel_reason);
- return Status::Cancelled(cancel_reason);
- }
+ LOG(INFO) << fmt::format("The channel has been cancelled, load_id = {}, error = {}",
+ print_id(load_id), cancel_reason);
+ return Status::Cancelled(cancel_reason);
} else {
// load is success, success only when eos be true
_load_state_channels->release(handle);
@@ -149,7 +154,8 @@ Status LoadChannelMgr::_get_load_channel(std::shared_ptr<LoadChannel>& channel,
}
Status LoadChannelMgr::add_batch(const PTabletWriterAddBlockRequest& request,
- PTabletWriterAddBlockResult* response) {
+ PTabletWriterAddBlockResult* response,
+ std::shared_ptr<EosCompletion>* eos_completion) {
UniqueId load_id(request.id());
// 1. get load channel
std::shared_ptr<LoadChannel> channel;
@@ -175,26 +181,47 @@ Status LoadChannelMgr::add_batch(const PTabletWriterAddBlockRequest& request,
// 3. add batch to load channel
// batch may not exist in request(eg: eos request without batch),
// this case will be handled in load channel's add batch method.
- Status st = channel->add_batch(request, response);
+ Status st = channel->add_batch(request, response, eos_completion);
if (UNLIKELY(!st.ok())) {
- RETURN_IF_ERROR(channel->cancel());
+ RETURN_IF_ERROR(_cancel_load_channel(channel, st));
return st;
}
// 4. handle finish
if (channel->is_finished()) {
- _finish_load_channel(load_id);
+ _finish_load_channel(channel);
+ }
+ return Status::OK();
+}
+
+Status LoadChannelMgr::_cancel_load_channel(const std::shared_ptr<LoadChannel>& channel,
+ const Status& reason) {
+ // Publish to the retained instance even if timeout cleanup or EOS removed it.
+ RETURN_IF_ERROR(channel->cancel(reason));
+ std::lock_guard<std::mutex> l(_lock);
+ const auto& load_id = channel->load_id();
+ auto it = _load_channels.find(load_id);
+ if (it == _load_channels.end() || it->second != channel) {
+ // A late RPC must not cancel a replacement opened with the same load ID.
+ return Status::OK();
}
+ _load_channels.erase(it);
+ _record_cancelled_load_channel(load_id, std::string(channel->cancel_status().msg()));
return Status::OK();
}
-void LoadChannelMgr::_finish_load_channel(const UniqueId load_id) {
+void LoadChannelMgr::_finish_load_channel(const std::shared_ptr<LoadChannel>& channel) {
+ const auto& load_id = channel->load_id();
VLOG_NOTICE << "removing load channel " << load_id << " because it's finished";
{
std::lock_guard<std::mutex> l(_lock);
- if (_load_channels.contains(load_id)) {
- _load_channels.erase(load_id);
+ auto it = _load_channels.find(load_id);
+ if (it == _load_channels.end() || it->second != channel || channel->is_cancelled()) {
+ // A concurrent cancel may already have removed the channel and recorded
+ // its failure. Do not replace that tombstone with a successful EOS.
+ return;
}
+ _load_channels.erase(it);
auto* handle = _load_state_channels->insert(load_id.to_string(), nullptr, 1, 1);
_load_state_channels->release(handle);
}
@@ -203,41 +230,53 @@ void LoadChannelMgr::_finish_load_channel(const UniqueId load_id) {
Status LoadChannelMgr::cancel(const PTabletWriterCancelRequest& params) {
UniqueId load_id(params.id());
+ const auto reason = params.has_cancel_reason() && !params.cancel_reason().empty()
+ ? params.cancel_reason()
+ : "load channel cancelled";
std::shared_ptr<LoadChannel> cancelled_channel;
{
std::lock_guard<std::mutex> l(_lock);
if (_load_channels.contains(load_id)) {
cancelled_channel = _load_channels[load_id];
+ // Publish while this exact instance is still mapped, then cache its
+ // effective first failure instead of a later cancel request's reason.
+ // cancel() only updates the shared status; it takes no writer locks.
+ RETURN_IF_ERROR(cancelled_channel->cancel(Status::Cancelled(reason)));
_load_channels.erase(load_id);
- }
- // We just need to record the first cancel msg
- auto* existing_handle = _load_state_channels->lookup(load_id.to_string());
- if (existing_handle == nullptr) {
- if (params.has_cancel_reason() && !params.cancel_reason().empty()) {
- std::unique_ptr<CacheValue> cancel_reason_ptr = std::make_unique<CacheValue>();
- cancel_reason_ptr->_cancel_reason = params.cancel_reason();
- size_t cache_capacity =
- cancel_reason_ptr->_cancel_reason.capacity() + sizeof(CacheValue);
- auto* handle = _load_state_channels->insert(
- load_id.to_string(), cancel_reason_ptr.get(), 1, cache_capacity);
- cancel_reason_ptr.release();
- _load_state_channels->release(handle);
- LOG(INFO) << fmt::format("load_id = {}, record_error reason = {}",
- print_id(load_id), params.cancel_reason());
- }
+ _record_cancelled_load_channel(load_id,
+ std::string(cancelled_channel->cancel_status().msg()));
} else {
- _load_state_channels->release(existing_handle);
+ _record_cancelled_load_channel(load_id, reason);
}
}
if (cancelled_channel != nullptr) {
- RETURN_IF_ERROR(cancelled_channel->cancel());
+ // Keep the final owner's destruction outside the manager lock, since it
+ // may still wait for in-flight work.
LOG(INFO) << "load channel has been cancelled: " << load_id;
}
return Status::OK();
}
+void LoadChannelMgr::_record_cancelled_load_channel(const UniqueId& load_id,
+ const std::string& reason) {
+ // Keep the first terminal state recorded for this load ID.
+ auto* existing_handle = _load_state_channels->lookup(load_id.to_string());
+ if (existing_handle != nullptr) {
+ _load_state_channels->release(existing_handle);
+ return;
+ }
+ auto value = std::make_unique<CacheValue>();
+ value->_cancel_reason = reason;
+ size_t cache_capacity = value->_cancel_reason.capacity() + sizeof(CacheValue);
+ auto* handle =
+ _load_state_channels->insert(load_id.to_string(), value.get(), 1, cache_capacity);
+ value.release();
+ _load_state_channels->release(handle);
+ LOG(INFO) << fmt::format("load_id = {}, record_error reason = {}", print_id(load_id), reason);
+}
+
Status LoadChannelMgr::_start_bg_worker() {
RETURN_IF_ERROR(Thread::create(
"LoadChannelMgr", "cancel_timeout_load_channels",
@@ -275,12 +314,13 @@ Status LoadChannelMgr::_start_load_channels_clean() {
}
}
- // we must cancel these load channels before destroying them.
- // otherwise some object may be invalid before trying to visit it.
- // eg: MemTracker in load channel
+ // Publish cancellation to retained RPCs before releasing the cleanup references.
+ // The last owner performs writer cleanup; memtables and writers retain their
+ // resource contexts until their work finishes. Destruction may still wait here
+ // when cleanup owns the last reference, so keep it outside the manager lock.
for (auto& channel : need_delete_channels) {
RETURN_IF_ERROR(channel->cancel());
- LOG(INFO) << "load channel has been safely deleted: " << channel->load_id()
+ LOG(INFO) << "cancellation published for timed out load channel: " << channel->load_id()
<< ", timeout(s): " << channel->timeout();
}
diff --git a/be/src/load/channel/load_channel_mgr.h b/be/src/load/channel/load_channel_mgr.h
index fdddbee0400..1c06fded153 100644
--- a/be/src/load/channel/load_channel_mgr.h
+++ b/be/src/load/channel/load_channel_mgr.h
@@ -42,6 +42,7 @@
namespace doris {
class PTabletWriterCancelRequest;
+class EosCompletion;
class PTabletWriterOpenRequest;
class Thread;
@@ -56,8 +57,11 @@ public:
// open a new load channel if not exist
Status open(const PTabletWriterOpenRequest& request);
+ // For hang_wait EOS, return a barrier handle to the service, which transfers
+ // the RPC completion after finishing synchronous response writes.
Status add_batch(const PTabletWriterAddBlockRequest& request,
- PTabletWriterAddBlockResult* response);
+ PTabletWriterAddBlockResult* response,
+ std::shared_ptr<EosCompletion>* eos_completion);
// cancel all tablet stream for 'load_id' load
Status cancel(const PTabletWriterCancelRequest& request);
@@ -78,7 +82,10 @@ private:
Status _get_load_channel(std::shared_ptr<LoadChannel>& channel, bool& is_eof,
const UniqueId& load_id, const PTabletWriterAddBlockRequest& request);
- void _finish_load_channel(UniqueId load_id);
+ Status _cancel_load_channel(const std::shared_ptr<LoadChannel>& channel, const Status& reason);
+ void _finish_load_channel(const std::shared_ptr<LoadChannel>& channel);
+ // Caller holds _lock.
+ void _record_cancelled_load_channel(const UniqueId& load_id, const std::string& reason);
Status _start_bg_worker();
@@ -86,6 +93,8 @@ private:
public:
class CacheValue : public LRUCacheValueBase {
public:
+ // Unformatted message, without a Status code prefix or stack trace.
+ // A non-null cache value marks cancellation even when the message is empty.
std::string _cancel_reason;
};
diff --git a/be/src/load/channel/tablets_channel.cpp b/be/src/load/channel/tablets_channel.cpp
index 708ca52bff0..2769cb4bf51 100644
--- a/be/src/load/channel/tablets_channel.cpp
+++ b/be/src/load/channel/tablets_channel.cpp
@@ -41,6 +41,8 @@
#include "common/metrics/doris_metrics.h"
#include "common/metrics/metrics.h"
#include "core/block/block.h"
+#include "cpp/sync_point.h"
+#include "load/channel/eos_completion.h"
#include "load/channel/load_channel.h"
#include "load/delta_writer/delta_writer.h"
#include "storage/storage_engine.h"
@@ -61,6 +63,7 @@ std::atomic<uint64_t> BaseTabletsChannel::_s_tablet_writer_count;
BaseTabletsChannel::BaseTabletsChannel(const TabletsChannelKey& key, const UniqueId& load_id,
bool is_high_priority, RuntimeProfile* profile)
: _key(key),
+ _eos_completion(std::make_shared<EosCompletion>()),
_state(kInitialized),
_load_id(load_id),
_closed_senders(64),
@@ -85,9 +88,19 @@ BaseTabletsChannel::~BaseTabletsChannel() {
TabletsChannel::~TabletsChannel() = default;
+Status BaseTabletsChannel::_check_cancelled() {
+ if (_load_cancel_status && !_load_cancel_status->ok()) {
+ _close_status = _load_cancel_status->status();
+ return _close_status;
+ }
+ return Status::OK();
+}
+
Status BaseTabletsChannel::_get_current_seq(int64_t& cur_seq,
- const PTabletWriterAddBlockRequest& request) {
+ const PTabletWriterAddBlockRequest& request,
+ bool& should_write) {
std::lock_guard<std::mutex> l(_lock);
+ should_write = false;
if (_state != kOpened) {
return _state == kFinished ? _close_status
: Status::InternalError("TabletsChannel {} state: {}",
@@ -100,6 +113,11 @@ Status BaseTabletsChannel::_get_current_seq(int64_t& cur_seq,
<< ", recept_seq=" << request.packet_seq();
return Status::InternalError("lost data packet");
}
+ should_write = request.packet_seq() == cur_seq;
+ if (!should_write) {
+ LOG(INFO) << "packet has already recept before, expect_seq=" << cur_seq
+ << ", recept_seq=" << request.packet_seq();
+ }
return Status::OK();
}
@@ -132,8 +150,9 @@ void TabletsChannel::_init_profile(RuntimeProfile* profile) {
Status BaseTabletsChannel::open(const PTabletWriterOpenRequest& request) {
std::lock_guard<std::mutex> l(_lock);
- // if _state is kOpened, it's a normal case, already open by other sender
- // if _state is kFinished, already cancelled by other sender
+ RETURN_IF_ERROR(_check_cancelled());
+ // Another sender may have already opened or closed this channel.
+ // Shared load cancellation is checked above and does not change _state.
if (_state == kOpened) {
RETURN_IF_ERROR(_init_adaptive_random_bucket_state(request));
return Status::OK();
@@ -192,6 +211,7 @@ Status BaseTabletsChannel::incremental_open(const PTabletWriterOpenRequest& para
}
std::lock_guard<std::mutex> l(_lock);
+ RETURN_IF_ERROR(_check_cancelled());
// one sender may incremental_open many times. but only close one time. so dont count duplicately.
if (_open_by_incremental) {
@@ -231,6 +251,7 @@ Status BaseTabletsChannel::incremental_open(const PTabletWriterOpenRequest& para
incremental_tablet_num++;
WriteRequest wrequest;
+ wrequest.load_cancel_status = _load_cancel_status;
wrequest.index_id = params.index_id();
wrequest.tablet_id = tablet.tablet_id();
wrequest.schema_hash = schema_hash;
@@ -298,15 +319,54 @@ std::unique_ptr<BaseDeltaWriter> TabletsChannel::create_delta_writer(const Write
return std::make_unique<DeltaWriter>(_engine, request, _profile, _load_id);
}
+Status BaseTabletsChannel::_close_and_notify(LoadChannel* parent,
+ std::unique_lock<std::mutex>& lock,
+ const PTabletWriterAddBlockRequest& req,
+ PTabletWriterAddBlockResult* response) {
+ int sender_id = req.sender_id();
+ DCHECK_EQ(_num_remaining_senders, 0);
+ // Stop accepting data before releasing the lock. The arrival barrier allows
+ // earlier senders to close incremental channels while this sender flushes.
+ // A retry of this final sender must instead wait for the final close result.
+ _state = kFinished;
+ _final_sender_id = sender_id;
+ _final_close_in_progress = true;
+ const auto arrival_status = _check_cancelled();
+ lock.unlock();
+ _eos_completion->complete(arrival_status);
+ lock.lock();
+
+ _close_status = _check_cancelled();
+ if (_close_status.ok()) {
+ _close_status = _close_writers(parent, req, response);
+ }
+ _final_close_result = std::make_unique<PTabletWriterAddBlockResult>(*response);
+ _final_close_in_progress = false;
+ _final_close_cv.notify_all();
+ return _close_status;
+}
+
+Status BaseTabletsChannel::_get_close_result(std::unique_lock<std::mutex>& lock, int sender_id,
+ PTabletWriterAddBlockResult* response) {
+ if (sender_id == _final_sender_id) {
+ // The final closer is already running, so waiting here cannot prevent
+ // another sender's EOS from reaching the arrival barrier.
+ TEST_SYNC_POINT("BaseTabletsChannel::close.wait_for_final_result");
+ _final_close_cv.wait(lock, [this] { return !_final_close_in_progress; });
+ DCHECK(_final_close_result != nullptr);
+ response->CopyFrom(*_final_close_result);
+ }
+ return _close_status;
+}
+
Status TabletsChannel::close(LoadChannel* parent, const PTabletWriterAddBlockRequest& req,
PTabletWriterAddBlockResult* res, bool* finished) {
int sender_id = req.sender_id();
int64_t backend_id = req.backend_id();
const auto& partition_ids = req.partition_ids();
- auto* tablet_errors = res->mutable_tablet_errors();
- std::lock_guard<std::mutex> l(_lock);
+ std::unique_lock<std::mutex> l(_lock);
if (_state == kFinished) {
- return _close_status;
+ return _get_close_result(l, sender_id, res);
}
if (_closed_senders.Get(sender_id)) {
// Double close from one sender, just return OK
@@ -330,12 +390,20 @@ Status TabletsChannel::close(LoadChannel* parent, const PTabletWriterAddBlockReq
return Status::OK();
}
- _state = kFinished;
+ return _close_and_notify(parent, l, req, res);
+}
+
+Status TabletsChannel::_close_writers(LoadChannel* parent, const PTabletWriterAddBlockRequest& req,
+ PTabletWriterAddBlockResult* res) {
+ auto* tablet_errors = res->mutable_tablet_errors();
+ TEST_SYNC_POINT_RETURN_WITH_VALUE("TabletsChannel::close.before_flush", Status::OK(), res);
+
// All senders are closed
// 1. close all delta writers
std::set<DeltaWriter*> need_wait_writers;
// under _lock. no need _tablet_writers_lock again.
for (auto&& [tablet_id, writer] : _tablet_writers) {
+ RETURN_IF_ERROR(_check_cancelled());
if (_partition_ids.contains(writer->partition_id())) {
auto st = writer->close();
if (!st.ok()) {
@@ -561,6 +629,7 @@ Status BaseTabletsChannel::_open_all_writers(const PTabletWriterOpenRequest& req
.write_file_cache = request.write_file_cache(),
.storage_vault_id = request.storage_vault_id(),
.enable_table_memtable_backpressure = request.is_adaptive_random_bucket(),
+ .load_cancel_status = _load_cancel_status,
};
auto delta_writer = create_delta_writer(wrequest);
@@ -855,17 +924,12 @@ Status TabletsChannel::add_batch(const PTabletWriterAddBlockRequest& request,
_add_batch_number_counter->update(1);
}
- auto status = _get_current_seq(cur_seq, request);
- if (UNLIKELY(!status.ok())) {
+ bool should_write = false;
+ auto status = _get_current_seq(cur_seq, request, should_write);
+ if (UNLIKELY(!status.ok()) || !should_write) {
return status;
}
- if (request.packet_seq() < cur_seq) {
- LOG(INFO) << "packet has already recept before, expect_seq=" << cur_seq
- << ", recept_seq=" << request.packet_seq();
- return Status::OK();
- }
-
// Adaptive random bucket add-block RPCs carry partition ids. The receiver maps rows to
// the current tablet and advances its selected-tablet state after memtable flush.
if (request.is_adaptive_random_bucket()) {
diff --git a/be/src/load/channel/tablets_channel.h b/be/src/load/channel/tablets_channel.h
index d736cbeb0f9..23b6f554f1c 100644
--- a/be/src/load/channel/tablets_channel.h
+++ b/be/src/load/channel/tablets_channel.h
@@ -20,6 +20,7 @@
#include <glog/logging.h>
#include <atomic>
+#include <condition_variable>
#include <cstdint>
#include <mutex>
#include <ostream>
@@ -81,6 +82,7 @@ class BaseDeltaWriter;
class MemTableWriter;
class OlapTableSchemaParam;
class LoadChannel;
+class EosCompletion;
struct WriteRequest;
// Write channel for a particular (load, index).
@@ -104,6 +106,7 @@ public:
// Mark sender with 'sender_id' as closed.
// If all senders are closed, close this channel, set '*finished' to true, update 'tablet_vec'
// to include all tablets written in this channel.
+ // Release the EOS arrival barrier before final flushing/committing, outside _lock.
// no-op when this channel has been closed or cancelled
virtual Status close(LoadChannel* parent, const PTabletWriterAddBlockRequest& req,
PTabletWriterAddBlockResult* res, bool* finished) = 0;
@@ -111,6 +114,13 @@ public:
// no-op when this channel has been closed or cancelled
virtual Status cancel();
+ // Set once, before publishing the channel or opening any writers.
+ void set_load_cancel_status(std::shared_ptr<AtomicStatus> status) {
+ _load_cancel_status = std::move(status);
+ }
+
+ const std::shared_ptr<EosCompletion>& eos_completion() const { return _eos_completion; }
+
void refresh_profile();
size_t total_received_rows() const { return _total_received_rows; }
@@ -123,6 +133,20 @@ public:
bool is_finished() const { return _state == kFinished; }
protected:
+ // Caller holds _lock. The cancellation publisher never needs this lock.
+ Status _check_cancelled();
+
+ // Called by the final sender under _lock. Publish arrival before flushing,
+ // releasing the channel lock while invoking RPC callbacks.
+ Status _close_and_notify(LoadChannel* parent, std::unique_lock<std::mutex>& lock,
+ const PTabletWriterAddBlockRequest& req,
+ PTabletWriterAddBlockResult* response);
+ Status _get_close_result(std::unique_lock<std::mutex>& lock, int sender_id,
+ PTabletWriterAddBlockResult* response);
+ // Called exactly once by the final sender, with _lock held.
+ virtual Status _close_writers(LoadChannel* parent, const PTabletWriterAddBlockRequest& req,
+ PTabletWriterAddBlockResult* response) = 0;
+
Status _init_adaptive_random_bucket_state(const PTabletWriterOpenRequest& request);
Status _write_block_data(const PTabletWriterAddBlockRequest& request, int64_t cur_seq,
std::unordered_map<int64_t, DorisVector<uint32_t>>& tablet_to_rowidxs,
@@ -137,7 +161,9 @@ protected:
std::unordered_map<int64_t, DorisVector<uint32_t>>* partition_to_rowidxs);
std::shared_ptr<std::mutex> _get_partition_route_lock(int64_t partition_id);
- Status _get_current_seq(int64_t& cur_seq, const PTabletWriterAddBlockRequest& request);
+ // OK does not imply a write: duplicates and closed channels are no-ops.
+ Status _get_current_seq(int64_t& cur_seq, const PTabletWriterAddBlockRequest& request,
+ bool& should_write);
// open all writer
Status _open_all_writers(const PTabletWriterOpenRequest& request);
@@ -155,6 +181,7 @@ protected:
// id of this load channel
TabletsChannelKey _key;
+ std::shared_ptr<EosCompletion> _eos_completion;
// protect _state change. open and close. when add_batch finished, lock to change _next_seqs also
std::mutex _lock;
@@ -166,6 +193,7 @@ protected:
State _state;
UniqueId _load_id;
+ std::shared_ptr<AtomicStatus> _load_cancel_status;
// initialized in open function
int64_t _txn_id = -1;
@@ -179,9 +207,14 @@ protected:
int _num_remaining_senders = 0;
std::vector<int64_t> _next_seqs;
Bitmap _closed_senders;
- // status to return when operate on an already closed/cancelled channel
- // currently it's OK.
+ // Arrival releases earlier senders; only the final sender's retries must
+ // wait for flush/commit and receive its status and tablet results. All these
+ // fields are protected by _lock, including while callbacks run without it.
+ int _final_sender_id = -1;
+ bool _final_close_in_progress = false;
+ std::condition_variable _final_close_cv;
Status _close_status;
+ std::unique_ptr<PTabletWriterAddBlockResult> _final_close_result;
// tablet_id -> TabletChannel. it will only be changed in open() or inc_open()
std::unordered_map<int64_t, std::unique_ptr<BaseDeltaWriter>> _tablet_writers;
@@ -245,6 +278,9 @@ public:
Status cancel() override;
private:
+ Status _close_writers(LoadChannel* parent, const PTabletWriterAddBlockRequest& req,
+ PTabletWriterAddBlockResult* response) override;
+
void _init_profile(RuntimeProfile* profile) override;
// deal with DeltaWriter commit_txn(), add tablet to list for return.
diff --git a/be/src/load/delta_writer/delta_writer.cpp b/be/src/load/delta_writer/delta_writer.cpp
index a0062dd426a..89b174411c1 100644
--- a/be/src/load/delta_writer/delta_writer.cpp
+++ b/be/src/load/delta_writer/delta_writer.cpp
@@ -34,6 +34,7 @@
#include "common/logging.h"
#include "common/status.h"
#include "core/block/block.h"
+#include "cpp/sync_point.h"
#include "io/fs/file_writer.h" // IWYU pragma: keep
#include "load/memtable/memtable_flush_executor.h"
#include "load/memtable/memtable_memory_limiter.h"
@@ -89,13 +90,19 @@ void DeltaWriter::_init_profile(RuntimeProfile* profile) {
}
BaseDeltaWriter::~BaseDeltaWriter() {
+ // Initialization may have created bitmap tokens before memtable init failed.
+ // Stop producers and drain both bitmap phases before any builder cleanup.
+ // A derived constructor can fail before it installs its builder.
+ if (_rowset_builder != nullptr) {
+ auto st = _get_load_cancel_status();
+ WARN_IF_ERROR(BaseDeltaWriter::cancel_with_status(
+ st.ok() ? Status::Cancelled("delta writer destroyed") : st),
+ "failed to cancel work while destroying delta writer");
+ }
if (!_is_init) {
return;
}
- // cancel and wait all memtables in flush queue to be finished
- static_cast<void>(_memtable_writer->cancel());
-
if (_rowset_builder->tablet() != nullptr) {
const FlushStatistic& stat = _memtable_writer->get_flush_token_stats();
_rowset_builder->tablet()->flush_bytes->increment(stat.flush_size_bytes);
@@ -132,15 +139,24 @@ int64_t BaseDeltaWriter::table_id() const {
DeltaWriter::~DeltaWriter() = default;
+Status BaseDeltaWriter::_get_load_cancel_status() const {
+ return _req.load_cancel_status && !_req.load_cancel_status->ok()
+ ? _req.load_cancel_status->status()
+ : Status::OK();
+}
+
Status BaseDeltaWriter::init() {
if (_is_init) {
return Status::OK();
}
+ RETURN_IF_ERROR(_get_load_cancel_status());
std::shared_ptr<WorkloadGroup> wg_sptr = nullptr;
if (doris::thread_context()->is_attach_task()) {
wg_sptr = doris::thread_context()->resource_ctx()->workload_group();
}
RETURN_IF_ERROR(_rowset_builder->init());
+ TEST_SYNC_POINT_RETURN_WITH_VALUE("BaseDeltaWriter::init:before_memtable_init", Status::OK(),
+ this);
RETURN_IF_ERROR(_memtable_writer->init(
_rowset_builder->rowset_writer(), _rowset_builder->tablet_schema(),
_rowset_builder->get_partial_update_info(), wg_sptr,
@@ -162,7 +178,7 @@ Status DeltaWriter::write(const Block* block, const DorisVector<uint32_t>& row_i
ExecEnv::GetInstance()->memtable_memory_limiter()->handle_table_memtable_backpressure(
[this]() {
std::lock_guard<std::mutex> l(_lock);
- return _is_cancelled;
+ return _is_cancelled || !_get_load_cancel_status().ok();
},
table_id());
}
@@ -176,6 +192,7 @@ Status DeltaWriter::write(const Block* block, const DorisVector<uint32_t>& row_i
SCOPED_TIMER(_wait_flush_limit_timer);
while (_memtable_writer->flush_running_count() >=
config::memtable_flush_running_count_limit) {
+ RETURN_IF_ERROR(_get_load_cancel_status());
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
@@ -215,6 +232,7 @@ Status DeltaWriter::close() {
Status BaseDeltaWriter::build_rowset() {
SCOPED_TIMER(_close_wait_timer);
RETURN_IF_ERROR(_memtable_writer->close_wait(_profile));
+ RETURN_IF_ERROR(_get_load_cancel_status());
return _rowset_builder->build_rowset();
}
@@ -226,6 +244,7 @@ Status DeltaWriter::build_rowset() {
}
Status BaseDeltaWriter::submit_calc_delete_bitmap_task() {
+ RETURN_IF_ERROR(_get_load_cancel_status());
return _rowset_builder->submit_calc_delete_bitmap_task();
}
@@ -240,6 +259,7 @@ RowsetBuilder* DeltaWriter::rowset_builder() {
Status DeltaWriter::commit_txn(const PSlaveTabletNodes& slave_tablet_nodes) {
std::lock_guard<std::mutex> l(_lock);
SCOPED_TIMER(_commit_txn_timer);
+ RETURN_IF_ERROR(_get_load_cancel_status());
RETURN_IF_ERROR(rowset_builder()->commit_txn());
for (auto&& node_info : slave_tablet_nodes.slave_nodes()) {
@@ -273,6 +293,7 @@ Status BaseDeltaWriter::cancel_with_status(const Status& st) {
return Status::OK();
}
RETURN_IF_ERROR(_memtable_writer->cancel_with_status(st));
+ RETURN_IF_ERROR(_rowset_builder->cancel(st));
_is_cancelled = true;
return Status::OK();
}
diff --git a/be/src/load/delta_writer/delta_writer.h b/be/src/load/delta_writer/delta_writer.h
index 578692d4d23..5d413580040 100644
--- a/be/src/load/delta_writer/delta_writer.h
+++ b/be/src/load/delta_writer/delta_writer.h
@@ -72,7 +72,8 @@ public:
Status submit_calc_delete_bitmap_task();
Status wait_calc_delete_bitmap();
- // abandon current memtable and wait for all pending-flushing memtables to be destructed.
+ // Abandon the current memtable and cancel queued flush tasks.
+ // Wait for running tasks before releasing their resources.
// mem_consumption() should be 0 after this function returns.
Status cancel();
virtual Status cancel_with_status(const Status& st);
@@ -104,6 +105,8 @@ public:
google::protobuf::RepeatedPtrField<PTabletLoadRowsetInfo>* tablet_info);
protected:
+ Status _get_load_cancel_status() const;
+
virtual void _init_profile(RuntimeProfile* profile);
Status init();
diff --git a/be/src/load/delta_writer/delta_writer_context.h b/be/src/load/delta_writer/delta_writer_context.h
index 834e98bf488..2eb515db41c 100644
--- a/be/src/load/delta_writer/delta_writer_context.h
+++ b/be/src/load/delta_writer/delta_writer_context.h
@@ -28,6 +28,7 @@ namespace doris {
class TupleDescriptor;
class SlotDescriptor;
class OlapTableSchemaParam;
+class AtomicStatus;
struct WriteRequest {
int64_t tablet_id = 0;
@@ -45,6 +46,8 @@ struct WriteRequest {
bool write_file_cache = false;
std::string storage_vault_id;
bool enable_table_memtable_backpressure = false;
+ // Shared by all writers of a load; published before writers can submit work.
+ std::shared_ptr<AtomicStatus> load_cancel_status = nullptr;
};
} // namespace doris
diff --git a/be/src/load/memtable/memtable_flush_executor.cpp b/be/src/load/memtable/memtable_flush_executor.cpp
index ca79b1a8345..c8797a7f2de 100644
--- a/be/src/load/memtable/memtable_flush_executor.cpp
+++ b/be/src/load/memtable/memtable_flush_executor.cpp
@@ -85,6 +85,11 @@ std::ostream& operator<<(std::ostream& os, const FlushStatistic& stat) {
return os;
}
+Status FlushToken::_get_load_cancel_status() const {
+ return _load_cancel_status && !_load_cancel_status->ok() ? _load_cancel_status->status()
+ : Status::OK();
+}
+
Status FlushToken::submit(std::shared_ptr<MemTable> mem_table) {
{
std::shared_lock rdlk(_flush_status_lock);
@@ -144,7 +149,7 @@ Status FlushToken::wait() {
return _flush_status;
}
}
- return Status::OK();
+ return _get_load_cancel_status();
}
Status FlushToken::_try_reserve_memory(const std::shared_ptr<ResourceContext>& resource_context,
@@ -208,6 +213,7 @@ Status FlushToken::_do_flush_memtable(MemTable* memtable, int32_t segment_id, in
// }};
std::unique_ptr<Block> block;
RETURN_IF_ERROR(memtable->to_block(&block));
+ RETURN_IF_ERROR(_get_load_cancel_status());
RETURN_IF_ERROR(_rowset_writer->flush_memtable(block.get(), segment_id, flush_size));
memtable->set_flush_success();
}
@@ -239,7 +245,7 @@ void FlushToken::_flush_memtable(std::shared_ptr<MemTable> memtable_ptr, int32_t
}};
DBUG_EXECUTE_IF("FlushToken.flush_memtable.wait_before_first_shutdown",
{ std::this_thread::sleep_for(std::chrono::milliseconds(10 * 1000)); });
- if (_is_shutdown()) {
+ if (_is_shutdown() || !_get_load_cancel_status().ok()) {
return;
}
DBUG_EXECUTE_IF("FlushToken.flush_memtable.wait_after_first_shutdown",
@@ -349,7 +355,8 @@ void MemTableFlushExecutor::update_memtable_flush_threads() {
Status MemTableFlushExecutor::create_flush_token(std::shared_ptr<FlushToken>& flush_token,
std::shared_ptr<RowsetWriter> rowset_writer,
bool is_high_priority,
- std::shared_ptr<WorkloadGroup> wg_sptr) {
+ std::shared_ptr<WorkloadGroup> wg_sptr,
+ std::shared_ptr<AtomicStatus> load_cancel_status) {
switch (rowset_writer->type()) {
case ALPHA_ROWSET:
// alpha rowset do not support flush in CONCURRENT. and not support alpha rowset now.
@@ -357,7 +364,7 @@ Status MemTableFlushExecutor::create_flush_token(std::shared_ptr<FlushToken>& fl
case BETA_ROWSET: {
// beta rowset can be flush in CONCURRENT, because each memtable using a new segment writer.
ThreadPool* pool = is_high_priority ? _high_prio_flush_pool.get() : _flush_pool.get();
- flush_token = FlushToken::create_shared(pool, wg_sptr);
+ flush_token = FlushToken::create_shared(pool, wg_sptr, std::move(load_cancel_status));
flush_token->set_rowset_writer(rowset_writer);
return Status::OK();
}
diff --git a/be/src/load/memtable/memtable_flush_executor.h b/be/src/load/memtable/memtable_flush_executor.h
index 297500010ac..620b54e3f78 100644
--- a/be/src/load/memtable/memtable_flush_executor.h
+++ b/be/src/load/memtable/memtable_flush_executor.h
@@ -63,13 +63,16 @@ class FlushToken : public std::enable_shared_from_this<FlushToken> {
ENABLE_FACTORY_CREATOR(FlushToken);
public:
- FlushToken(ThreadPool* thread_pool, std::shared_ptr<WorkloadGroup> wg_sptr)
- : _flush_status(Status::OK()), _thread_pool(thread_pool), _wg_wptr(wg_sptr) {}
+ FlushToken(ThreadPool* thread_pool, std::shared_ptr<WorkloadGroup> wg_sptr,
+ std::shared_ptr<AtomicStatus> load_cancel_status = nullptr)
+ : _flush_status(Status::OK()),
+ _load_cancel_status(std::move(load_cancel_status)),
+ _thread_pool(thread_pool),
+ _wg_wptr(wg_sptr) {}
Status submit(std::shared_ptr<MemTable> mem_table);
- // error has happens, so we cancel this token
- // And remove all tasks in the queue.
+ // Stop flush work and wait for running tasks. Queued tasks skip flushing when dispatched.
void cancel();
// wait all tasks in token to be completed.
@@ -85,6 +88,8 @@ public:
const MemTableStat& memtable_stat() { return _memtable_stat; }
private:
+ Status _get_load_cancel_status() const;
+
void _shutdown_flush_token() { _shutdown.store(true); }
bool _is_shutdown() { return _shutdown.load(); }
void _wait_submit_task_finish();
@@ -105,6 +110,7 @@ private:
// Note: Once its value is set to Failed, it cannot return to SUCCESS.
std::shared_mutex _flush_status_lock;
Status _flush_status;
+ const std::shared_ptr<AtomicStatus> _load_cancel_status;
FlushStatistic _stats;
@@ -145,7 +151,8 @@ public:
Status create_flush_token(std::shared_ptr<FlushToken>& flush_token,
std::shared_ptr<RowsetWriter> rowset_writer, bool is_high_priority,
- std::shared_ptr<WorkloadGroup> wg_sptr);
+ std::shared_ptr<WorkloadGroup> wg_sptr,
+ std::shared_ptr<AtomicStatus> load_cancel_status = nullptr);
// return true if it already has any flushing task
bool check_and_inc_has_any_flushing_task() {
diff --git a/be/src/load/memtable/memtable_memory_limiter.cpp b/be/src/load/memtable/memtable_memory_limiter.cpp
index 897e85538b2..1f62c208219 100644
--- a/be/src/load/memtable/memtable_memory_limiter.cpp
+++ b/be/src/load/memtable/memtable_memory_limiter.cpp
@@ -286,9 +286,10 @@ int64_t MemTableMemoryLimiter::_flush_active_memtables(int64_t need_flush) {
}
int64_t mem_flushed = 0;
+ int64_t mem_reclaimed = 0;
int64_t num_flushed = 0;
- while (mem_flushed < need_flush && !heap.empty()) {
+ while (mem_flushed + mem_reclaimed < need_flush && !heap.empty()) {
auto [writer, sort_mem] = heap.top();
heap.pop();
auto w = writer.lock();
@@ -302,6 +303,12 @@ int64_t MemTableMemoryLimiter::_flush_active_memtables(int64_t need_flush) {
continue;
}
Status st = w->flush_async();
+ if (st.is<ErrorCode::CANCELLED>()) {
+ // The active memtable was discarded, not flushed. Running flush
+ // tasks are still owned by the writer and must not be waited here.
+ mem_reclaimed += mem;
+ continue;
+ }
if (!st.ok()) {
auto err_msg = fmt::format(
"tablet writer failed to reduce mem consumption by flushing memtable, "
@@ -316,7 +323,13 @@ int64_t MemTableMemoryLimiter::_flush_active_memtables(int64_t need_flush) {
g_memtable_memory_limit_flush_size_bytes << mem;
}
LOG(INFO) << "flushed " << num_flushed << " out of " << _active_writers.size()
- << " active writers, flushed size: " << PrettyPrinter::print_bytes(mem_flushed);
+ << " active writers, flushed size: " << PrettyPrinter::print_bytes(mem_flushed)
+ << ", reclaimed cancelled memtables: " << PrettyPrinter::print_bytes(mem_reclaimed);
+ if (mem_reclaimed > 0) {
+ // Recheck the hard limit before the caller waits, and wake other waiters
+ // if reclaiming cancelled memtables has relieved memory pressure.
+ _refresh_mem_tracker();
+ }
return mem_flushed;
}
diff --git a/be/src/load/memtable/memtable_writer.cpp b/be/src/load/memtable/memtable_writer.cpp
index a000206990b..7c479f8262c 100644
--- a/be/src/load/memtable/memtable_writer.cpp
+++ b/be/src/load/memtable/memtable_writer.cpp
@@ -80,7 +80,8 @@ Status MemTableWriter::init(std::shared_ptr<RowsetWriter> rowset_writer,
// we can make sure same keys sort in the same order in all replicas.
RETURN_IF_ERROR(
ExecEnv::GetInstance()->storage_engine().memtable_flush_executor()->create_flush_token(
- _flush_token, _rowset_writer, _req.is_high_priority, wg_sptr));
+ _flush_token, _rowset_writer, _req.is_high_priority, wg_sptr,
+ _req.load_cancel_status));
_is_init = true;
return Status::OK();
@@ -97,6 +98,9 @@ Status MemTableWriter::write(const Block* block, const DorisVector<uint32_t>& ro
_lock_watch.start();
std::lock_guard<std::mutex> l(_lock);
_lock_watch.stop();
+ if (_req.load_cancel_status && !_req.load_cancel_status->ok()) {
+ return _req.load_cancel_status->status();
+ }
if (_is_cancelled) {
return _cancel_status;
}
@@ -179,10 +183,20 @@ Status MemTableWriter::_flush_memtable_async() {
Status MemTableWriter::flush_async() {
std::lock_guard<std::mutex> l(_lock);
- // Three calling paths:
- // 1. call by local, from `VTabletWriterV2::_write_memtable`.
- // 2. call by remote, from `LoadChannelMgr::_get_load_channel`.
- // 3. call by daemon thread, from `handle_paused_queries` -> `flush_workload_group_memtables`.
+ if (_req.load_cancel_status && !_req.load_cancel_status->ok()) {
+ // Reclaim the unsubmitted memtable even while an RPC retains this writer.
+ // Do not cancel the flush token here: pressure flushing holds the limiter
+ // lock and must not wait for running flush tasks. Leave _is_cancelled
+ // unset so the final owner's cancel still drains the token.
+ {
+ std::lock_guard<std::mutex> lm(_mem_table_ptr_lock);
+ _mem_table.reset();
+ }
+ return Status::Cancelled("Load has been cancelled: {}",
+ _req.load_cancel_status->status().to_string());
+ }
+ // Memory-pressure flushing can race with writer initialization or close.
+ // DeltaWriter also exposes this operation through flush_memtable_async().
if (!_is_init || _is_closed) {
// This writer is uninitialized or closed before flushing, do nothing.
// We return OK instead of NOT_INITIALIZED or ALREADY_CLOSED.
@@ -237,6 +251,9 @@ Status MemTableWriter::close() {
_lock_watch.start();
std::lock_guard<std::mutex> l(_lock);
_lock_watch.stop();
+ if (_req.load_cancel_status && !_req.load_cancel_status->ok()) {
+ return _req.load_cancel_status->status();
+ }
if (_is_cancelled) {
return _cancel_status;
}
diff --git a/be/src/service/internal_service.cpp b/be/src/service/internal_service.cpp
index fb297867f7f..3f8719acce7 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -83,6 +83,7 @@
#include "io/fs/local_file_system.h"
#include "io/fs/stream_load_pipe.h"
#include "io/io_common.h"
+#include "load/channel/eos_completion.h"
#include "load/channel/load_channel_mgr.h"
#include "load/channel/load_stream_mgr.h"
#include "load/delta_writer/delta_writer.h"
@@ -145,6 +146,11 @@ using namespace ErrorCode;
const uint32_t DOWNLOAD_FILE_MAX_RETRY = 3;
+DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(load_light_work_pool_queue_size, MetricUnit::NOUNIT);
+DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(load_light_work_active_threads, MetricUnit::NOUNIT);
+DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(load_light_work_pool_max_queue_size, MetricUnit::NOUNIT);
+DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(load_light_work_max_threads, MetricUnit::NOUNIT);
+
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(heavy_work_pool_queue_size, MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(peer_fetch_work_pool_queue_size, MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(light_work_pool_queue_size, MetricUnit::NOUNIT);
@@ -166,6 +172,12 @@ DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(arrow_flight_work_max_threads, MetricUnit::NO
static bvar::LatencyRecorder g_process_remote_fetch_rowsets_latency("process_remote_fetch_rowsets");
+static int32_t resolved_brpc_load_light_work_pool_max_queue_size() {
+ return config::brpc_load_light_work_pool_max_queue_size != -1
+ ? config::brpc_load_light_work_pool_max_queue_size
+ : std::max(1024, CpuInfo::num_cores() * 32);
+}
+
static int32_t resolved_brpc_peer_fetch_pool_threads() {
return config::brpc_peer_fetch_pool_threads != -1 ? config::brpc_peer_fetch_pool_threads
: std::max(64, CpuInfo::num_cores() * 2);
@@ -230,6 +242,10 @@ PInternalService::PInternalService(ExecEnv* exec_env)
? config::brpc_heavy_work_pool_max_queue_size
: std::max(10240, CpuInfo::num_cores() * 320),
"brpc_heavy"),
+ // Keep cancellation dispatch independent of potentially blocking opens and writes.
+ _load_light_work_pool(config::brpc_load_light_work_pool_threads,
+ resolved_brpc_load_light_work_pool_max_queue_size(),
+ "brpc_load_light"),
// peer fetch threadpool isolates fetch_peer_data from heavy load traffic to avoid peer reads starving imports.
_peer_fetch_pool(resolved_brpc_peer_fetch_pool_threads(),
resolved_brpc_peer_fetch_pool_max_queue_size(), "brpc_peer_fetch"),
@@ -249,6 +265,15 @@ PInternalService::PInternalService(ExecEnv* exec_env)
? config::brpc_arrow_flight_work_pool_max_queue_size
: std::max(20480, CpuInfo::num_cores() * 640),
"brpc_arrow_flight") {
+ REGISTER_HOOK_METRIC(load_light_work_pool_queue_size,
+ [this]() { return _load_light_work_pool.get_queue_size(); });
+ REGISTER_HOOK_METRIC(load_light_work_active_threads,
+ [this]() { return _load_light_work_pool.get_active_threads(); });
+ REGISTER_HOOK_METRIC(load_light_work_pool_max_queue_size,
+ []() { return resolved_brpc_load_light_work_pool_max_queue_size(); });
+ REGISTER_HOOK_METRIC(load_light_work_max_threads,
+ []() { return config::brpc_load_light_work_pool_threads; });
+
REGISTER_HOOK_METRIC(heavy_work_pool_queue_size,
[this]() { return _heavy_work_pool.get_queue_size(); });
REGISTER_HOOK_METRIC(peer_fetch_work_pool_queue_size,
@@ -295,6 +320,11 @@ PInternalServiceImpl::PInternalServiceImpl(StorageEngine& engine, ExecEnv* exec_
PInternalServiceImpl::~PInternalServiceImpl() = default;
PInternalService::~PInternalService() {
+ DEREGISTER_HOOK_METRIC(load_light_work_pool_queue_size);
+ DEREGISTER_HOOK_METRIC(load_light_work_active_threads);
+ DEREGISTER_HOOK_METRIC(load_light_work_pool_max_queue_size);
+ DEREGISTER_HOOK_METRIC(load_light_work_max_threads);
+
DEREGISTER_HOOK_METRIC(heavy_work_pool_queue_size);
DEREGISTER_HOOK_METRIC(peer_fetch_work_pool_queue_size);
DEREGISTER_HOOK_METRIC(light_work_pool_queue_size);
@@ -519,10 +549,11 @@ void PInternalService::tablet_writer_add_block(google::protobuf::RpcController*
int64_t wait_execution_time_ns = MonotonicNanos() - submit_task_time_ns;
brpc::ClosureGuard closure_guard(done);
int64_t execution_time_ns = 0;
+ std::shared_ptr<EosCompletion> eos_completion;
{
SCOPED_RAW_TIMER(&execution_time_ns);
signal::SignalTaskIdKeeper keeper(request->id());
- auto st = _exec_env->load_channel_mgr()->add_batch(*request, response);
+ auto st = _exec_env->load_channel_mgr()->add_batch(*request, response, &eos_completion);
if (!st.ok()) {
LOG(WARNING) << "tablet writer add block failed, message=" << st
<< ", id=" << request->id() << ", index_id=" << request->index_id()
@@ -533,6 +564,17 @@ void PInternalService::tablet_writer_add_block(google::protobuf::RpcController*
}
response->set_execution_time_us(execution_time_ns / NANOS_PER_MICRO);
response->set_wait_execution_time_us(wait_execution_time_ns / NANOS_PER_MICRO);
+ if (eos_completion && response->status().status_code() == TStatusCode::OK) {
+ // Transfer done before publishing the callback: a completed barrier
+ // can invoke it inline. All request/response access above must end
+ // first, including profiling, timers and SignalTaskIdKeeper.
+ auto* deferred_done = closure_guard.release();
+ eos_completion->add_waiter([response, deferred_done](const Status& status) {
+ status.to_protobuf(response->mutable_status());
+ deferred_done->Run();
+ // The RPC objects (including HTTP's request wrapper) are gone.
+ });
+ }
});
if (!ret) {
offer_failed(response, done, _heavy_work_pool);
@@ -544,7 +586,7 @@ void PInternalService::tablet_writer_cancel(google::protobuf::RpcController* con
const PTabletWriterCancelRequest* request,
PTabletWriterCancelResult* response,
google::protobuf::Closure* done) {
- bool ret = _heavy_work_pool.try_offer([this, request, done]() {
+ bool ret = _load_light_work_pool.try_offer([this, request, done]() {
VLOG_RPC << "tablet writer cancel, id=" << request->id()
<< ", index_id=" << request->index_id() << ", sender_id=" << request->sender_id();
signal::SignalTaskIdKeeper keeper(request->id());
@@ -557,7 +599,7 @@ void PInternalService::tablet_writer_cancel(google::protobuf::RpcController* con
}
});
if (!ret) {
- offer_failed(response, done, _heavy_work_pool);
+ offer_failed(response, done, _load_light_work_pool);
return;
}
}
diff --git a/be/src/service/internal_service.h b/be/src/service/internal_service.h
index 4098c82120a..53b560fdb37 100644
--- a/be/src/service/internal_service.h
+++ b/be/src/service/internal_service.h
@@ -274,6 +274,8 @@ protected:
// define the interface for reading and writing data as heavy interface
// otherwise as light interface
FifoThreadPool _heavy_work_pool;
+ // Dedicated pool for cancellation; open/write/close use the heavy pool.
+ FifoThreadPool _load_light_work_pool;
FifoThreadPool _peer_fetch_pool;
FifoThreadPool _light_work_pool;
FifoThreadPool _arrow_flight_work_pool;
diff --git a/be/src/storage/delete/calc_delete_bitmap_executor.cpp b/be/src/storage/delete/calc_delete_bitmap_executor.cpp
index af1b4bfb23a..d5fb2f87b39 100644
--- a/be/src/storage/delete/calc_delete_bitmap_executor.cpp
+++ b/be/src/storage/delete/calc_delete_bitmap_executor.cpp
@@ -22,6 +22,7 @@
#include <ostream>
#include "common/logging.h"
+#include "cpp/sync_point.h"
#include "load/memtable/memtable.h"
#include "storage/tablet/base_tablet.h"
#include "util/time.h"
@@ -35,16 +36,9 @@ Status CalcDeleteBitmapToken::submit(BaseTabletSPtr tablet, RowsetSharedPtr cur_
int64_t end_version, DeleteBitmapPtr delete_bitmap,
RowsetWriter* rowset_writer,
DeleteBitmapPtr tablet_delete_bitmap) {
- {
- std::shared_lock rlock(_lock);
- RETURN_IF_ERROR(_status);
- _resource_ctx = thread_context()->resource_ctx();
- }
-
const auto submit_time_us = MonotonicMicros();
- return _thread_token->submit_func([=, this]() {
+ return submit_func([=]() {
const auto queue_time_us = MonotonicMicros() - submit_time_us;
- SCOPED_ATTACH_TASK(_resource_ctx);
auto st = tablet->calc_segment_delete_bitmap(cur_rowset, cur_segment, target_rowsets,
delete_bitmap, end_version, rowset_writer,
tablet_delete_bitmap, queue_time_us);
@@ -53,11 +47,8 @@ Status CalcDeleteBitmapToken::submit(BaseTabletSPtr tablet, RowsetSharedPtr cur_
<< tablet->tablet_id() << " rowset: " << cur_rowset->rowset_id()
<< " seg_id: " << cur_segment->id() << " version: " << end_version
<< " error: " << st;
- std::lock_guard wlock(_lock);
- if (_status.ok()) {
- _status = st;
- }
}
+ return st;
});
}
@@ -65,33 +56,57 @@ Status CalcDeleteBitmapToken::submit(BaseTabletSPtr tablet, TabletSchemaSPtr sch
RowsetId rowset_id,
const std::vector<segment_v2::SegmentSharedPtr>& segments,
DeleteBitmapPtr delete_bitmap) {
- {
- std::shared_lock rlock(_lock);
- RETURN_IF_ERROR(_status);
- _resource_ctx = thread_context()->resource_ctx();
- }
const auto submit_time_us = MonotonicMicros();
- return _thread_token->submit_func([=, this]() {
+ return submit_func([=]() {
const auto queue_time_us = MonotonicMicros() - submit_time_us;
- SCOPED_ATTACH_TASK(_resource_ctx);
auto st = tablet->calc_delete_bitmap_between_segments(schema, rowset_id, segments,
delete_bitmap, queue_time_us);
if (!st.ok()) {
LOG(WARNING) << "failed to calc delete bitmap between segments, tablet_id: "
<< tablet->tablet_id() << " rowset: " << rowset_id
<< " segments num: " << segments.size() << " error: " << st;
- std::lock_guard wlock(_lock);
- if (_status.ok()) {
- _status = st;
- }
}
+ return st;
});
}
+Status CalcDeleteBitmapToken::_submit_func(std::function<void()> func) {
+ TEST_SYNC_POINT_CALLBACK("CalcDeleteBitmapToken::submit_func:before_submit", this);
+ auto st = _thread_token->submit_func(std::move(func));
+ if (!st.ok()) {
+ // Teardown can shut down the token after the entry check.
+ RETURN_IF_ERROR(_get_status());
+ }
+ return st;
+}
+
Status CalcDeleteBitmapToken::wait() {
_thread_token->wait();
- // all tasks complete here, don't need lock;
- return _status;
+ return _get_status();
+}
+
+Status CalcDeleteBitmapToken::_get_status() {
+ std::shared_lock rlock(_lock);
+ RETURN_IF_ERROR(_status);
+ return _load_cancel_status && !_load_cancel_status->ok() ? _load_cancel_status->status()
+ : Status::OK();
+}
+
+void CalcDeleteBitmapToken::_set_status(const Status& st) {
+ DCHECK(!st.ok());
+ std::lock_guard wlock(_lock);
+ if (_status.ok()) {
+ // Keep an earlier token failure local. If load cancellation is already
+ // observable, a later callback failure must not replace its reason.
+ _status = _load_cancel_status && !_load_cancel_status->ok() ? _load_cancel_status->status()
+ : st;
+ }
+}
+
+void CalcDeleteBitmapToken::cancel(const Status& st) {
+ _set_status(st);
+ // Running callbacks may still need _lock to publish their result.
+ _thread_token->shutdown();
}
void CalcDeleteBitmapExecutor::init(const std::string& name, int max_threads) {
@@ -101,9 +116,11 @@ void CalcDeleteBitmapExecutor::init(const std::string& name, int max_threads) {
.build(&_thread_pool));
}
-std::unique_ptr<CalcDeleteBitmapToken> CalcDeleteBitmapExecutor::create_token() {
+std::unique_ptr<CalcDeleteBitmapToken> CalcDeleteBitmapExecutor::create_token(
+ std::shared_ptr<AtomicStatus> load_cancel_status) {
return std::make_unique<CalcDeleteBitmapToken>(
- _thread_pool->new_token(ThreadPool::ExecutionMode::CONCURRENT));
+ _thread_pool->new_token(ThreadPool::ExecutionMode::CONCURRENT),
+ std::move(load_cancel_status));
}
} // namespace doris
diff --git a/be/src/storage/delete/calc_delete_bitmap_executor.h b/be/src/storage/delete/calc_delete_bitmap_executor.h
index 4f0f0fd2c7f..22a126a1c26 100644
--- a/be/src/storage/delete/calc_delete_bitmap_executor.h
+++ b/be/src/storage/delete/calc_delete_bitmap_executor.h
@@ -20,6 +20,7 @@
#include <stdint.h>
#include <atomic>
+#include <functional>
#include <iosfwd>
#include <memory>
#include <shared_mutex>
@@ -48,8 +49,14 @@ enum RowsetTypePB : int;
// 4. call `get_delete_bitmap()` to get the result of all tasks
class CalcDeleteBitmapToken {
public:
- explicit CalcDeleteBitmapToken(std::unique_ptr<ThreadPoolToken> thread_token)
- : _thread_token(std::move(thread_token)), _status(Status::OK()) {}
+ explicit CalcDeleteBitmapToken(std::unique_ptr<ThreadPoolToken> thread_token,
+ std::shared_ptr<AtomicStatus> load_cancel_status = nullptr)
+ : _thread_token(std::move(thread_token)),
+ _load_cancel_status(std::move(load_cancel_status)),
+ _status(Status::OK()) {}
+
+ // Drain callbacks before destroying the status and lock they access.
+ ~CalcDeleteBitmapToken() { _thread_token->shutdown(); }
// calculate delete bitmap of `cur_segment` to historical `target_rowsets`
Status submit(BaseTabletSPtr tablet, RowsetSharedPtr cur_rowset,
@@ -66,19 +73,18 @@ public:
// submit a generic function to the thread pool
template <typename Func>
Status submit_func(Func&& func) {
- {
- std::shared_lock rlock(_lock);
- RETURN_IF_ERROR(_status);
- _resource_ctx = thread_context()->resource_ctx();
- }
- return _thread_token->submit_func([this, func = std::forward<Func>(func)]() {
- SCOPED_ATTACH_TASK(_resource_ctx);
- auto st = func();
+ RETURN_IF_ERROR(_get_status());
+ // Each callback owns the context of its submitter, even when submissions overlap.
+ auto resource_ctx = thread_context()->resource_ctx();
+ return _submit_func([this, resource_ctx, func = std::forward<Func>(func)]() {
+ SCOPED_ATTACH_TASK(resource_ctx);
+ // Publication does not remove queued tasks. Skip their calculation when dispatched.
+ auto st = _get_status();
+ if (st.ok()) {
+ st = func();
+ }
if (!st.ok()) {
- std::lock_guard wlock(_lock);
- if (_status.ok()) {
- _status = st;
- }
+ _set_status(st);
}
});
}
@@ -86,16 +92,21 @@ public:
// wait all tasks in token to be completed.
Status wait();
- void cancel() { _thread_token->shutdown(); }
+ // Teardown only: discard pending callbacks and wait for running callbacks.
+ void cancel(const Status& st = Status::Cancelled("delete bitmap calculation cancelled"));
private:
+ Status _submit_func(std::function<void()> func);
+ Status _get_status();
+ void _set_status(const Status& st);
+
std::unique_ptr<ThreadPoolToken> _thread_token;
+ const std::shared_ptr<AtomicStatus> _load_cancel_status;
std::shared_mutex _lock;
// Records the current status of the calc delete bitmap job.
// Note: Once its value is set to Failed, it cannot return to SUCCESS.
Status _status;
- std::shared_ptr<ResourceContext> _resource_ctx;
};
// CalcDeleteBitmapExecutor is responsible for calc delete bitmap concurrently.
@@ -108,7 +119,8 @@ public:
// init should be called after storage engine is opened,
void init(const std::string& name, int max_threads);
- std::unique_ptr<CalcDeleteBitmapToken> create_token();
+ std::unique_ptr<CalcDeleteBitmapToken> create_token(
+ std::shared_ptr<AtomicStatus> load_cancel_status = nullptr);
private:
std::unique_ptr<ThreadPool> _thread_pool;
diff --git a/be/src/storage/rowset/beta_rowset_writer.cpp b/be/src/storage/rowset/beta_rowset_writer.cpp
index 1fdb50f08fe..676b258bcaa 100644
--- a/be/src/storage/rowset/beta_rowset_writer.cpp
+++ b/be/src/storage/rowset/beta_rowset_writer.cpp
@@ -326,7 +326,13 @@ BetaRowsetWriter::BetaRowsetWriter(StorageEngine& engine)
: _engine(engine), _segcompaction_worker(std::make_shared<SegcompactionWorker>(this)) {}
BaseBetaRowsetWriter::~BaseBetaRowsetWriter() {
- if (!_already_built && _rowset_meta->is_local()) {
+ // Finish callbacks before removing files they can still read on cancellation.
+ if (_calc_delete_bitmap_token) {
+ _calc_delete_bitmap_token->cancel();
+ }
+ if (!_already_built && _rowset_meta && _rowset_meta->is_local()) {
+ TEST_SYNC_POINT_CALLBACK("BaseBetaRowsetWriter::~BaseBetaRowsetWriter:before_file_cleanup",
+ this);
// abnormal exit, remove all files generated
auto& fs = io::global_local_filesystem();
for (int i = _segment_start_id; i < _segment_creator.next_segment_id(); ++i) {
@@ -339,12 +345,13 @@ BaseBetaRowsetWriter::~BaseBetaRowsetWriter() {
fmt::format("Failed to delete file={}", seg_path));
}
}
- if (_calc_delete_bitmap_token) {
- _calc_delete_bitmap_token->cancel();
- }
}
BetaRowsetWriter::~BetaRowsetWriter() {
+ // Bitmap callbacks capture this and can call virtual methods using derived state.
+ if (_calc_delete_bitmap_token != nullptr) {
+ _calc_delete_bitmap_token->cancel();
+ }
/* Note that segcompaction is async and in parallel with load job. So we should handle carefully
* when the job is cancelled. Although it is meaningless to continue segcompaction when the job
* is cancelled, the objects involved in the job should be preserved during segcompaction to
@@ -352,6 +359,12 @@ BetaRowsetWriter::~BetaRowsetWriter() {
WARN_IF_ERROR(_wait_flying_segcompaction(), "segment compaction failed");
}
+void BaseBetaRowsetWriter::cancel_calc_delete_bitmap(const Status& st) {
+ if (_calc_delete_bitmap_token != nullptr) {
+ _calc_delete_bitmap_token->cancel(st);
+ }
+}
+
Status BaseBetaRowsetWriter::init(const RowsetWriterContext& rowset_writer_context) {
_context = rowset_writer_context;
_rowset_meta.reset(new RowsetMeta);
@@ -432,6 +445,8 @@ Status BaseBetaRowsetWriter::_generate_delete_bitmap(int32_t segment_id) {
OlapStopWatch watch;
// Step 2: Build tmp rowset (needs file_writer to be closed)
+ TEST_SYNC_POINT_CALLBACK("BaseBetaRowsetWriter::_generate_delete_bitmap:before_build_tmp",
+ this);
RowsetSharedPtr rowset_ptr;
st = _build_tmp(rowset_ptr);
if (!st.ok()) {
@@ -473,6 +488,7 @@ Status BaseBetaRowsetWriter::_generate_delete_bitmap(int32_t segment_id) {
<< _context.mow_context->delete_bitmap->cardinality()
<< ", queue_time_us: " << queue_time_us
<< ", cost: " << watch.get_elapse_time_us() << "(us), total rows: " << total_rows;
+ TEST_SYNC_POINT_CALLBACK("BaseBetaRowsetWriter::_generate_delete_bitmap:finished", this);
return Status::OK();
});
}
@@ -483,7 +499,8 @@ Status BetaRowsetWriter::init(const RowsetWriterContext& rowset_writer_context)
_segcompaction_worker->init_mem_tracker(rowset_writer_context);
}
if (_context.mow_context != nullptr) {
- _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor_for_load()->create_token();
+ _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor_for_load()->create_token(
+ _context.load_cancel_status);
}
return Status::OK();
}
diff --git a/be/src/storage/rowset/beta_rowset_writer.h b/be/src/storage/rowset/beta_rowset_writer.h
index 8a8642c2e5b..e937d7a9cbb 100644
--- a/be/src/storage/rowset/beta_rowset_writer.h
+++ b/be/src/storage/rowset/beta_rowset_writer.h
@@ -118,6 +118,8 @@ public:
~BaseBetaRowsetWriter() override;
+ void cancel_calc_delete_bitmap(const Status& st) override;
+
Status init(const RowsetWriterContext& rowset_writer_context) override;
Status add_block(const Block* block) override;
diff --git a/be/src/storage/rowset/rowset_writer.h b/be/src/storage/rowset/rowset_writer.h
index 75c0bff084b..1150b79372f 100644
--- a/be/src/storage/rowset/rowset_writer.h
+++ b/be/src/storage/rowset/rowset_writer.h
@@ -72,6 +72,9 @@ public:
RowsetWriter() = default;
virtual ~RowsetWriter() = default;
+ // Called after flush producers stop and before owner cleanup.
+ virtual void cancel_calc_delete_bitmap(const Status& st) {}
+
virtual Status init(const RowsetWriterContext& rowset_writer_context) = 0;
virtual Status add_block(const Block* block) {
diff --git a/be/src/storage/rowset/rowset_writer_context.h b/be/src/storage/rowset/rowset_writer_context.h
index 52b0acd1b20..5e0eba14474 100644
--- a/be/src/storage/rowset/rowset_writer_context.h
+++ b/be/src/storage/rowset/rowset_writer_context.h
@@ -76,6 +76,7 @@ struct RowsetWriterContext {
int64_t txn_id {0};
int64_t txn_expiration {0}; // For cloud mode
PUniqueId load_id;
+ std::shared_ptr<AtomicStatus> load_cancel_status;
TabletUid tablet_uid {0, 0};
// indicate whether the data among segments is overlapping.
// default is OVERLAP_UNKNOWN.
diff --git a/be/src/storage/rowset_builder.cpp b/be/src/storage/rowset_builder.cpp
index 4ed9709aea5..d516ef62eb9 100644
--- a/be/src/storage/rowset_builder.cpp
+++ b/be/src/storage/rowset_builder.cpp
@@ -220,6 +220,7 @@ Status RowsetBuilder::init() {
RowsetWriterContext context;
context.txn_id = _req.txn_id;
context.load_id = _req.load_id;
+ context.load_cancel_status = _req.load_cancel_status;
context.rowset_state = PREPARED;
context.segments_overlap = OVERLAPPING;
context.tablet_schema = _tablet_schema;
@@ -235,7 +236,8 @@ Status RowsetBuilder::init() {
_rowset_writer = DORIS_TRY(_tablet->create_rowset_writer(context, false));
_pending_rs_guard = _engine.pending_local_rowsets().add(context.rowset_id);
- _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor()->create_token();
+ _calc_delete_bitmap_token =
+ _engine.calc_delete_bitmap_executor()->create_token(_req.load_cancel_status);
_is_init = true;
return Status::OK();
@@ -356,13 +358,17 @@ Status RowsetBuilder::commit_txn() {
return Status::OK();
}
-Status BaseRowsetBuilder::cancel() {
+Status BaseRowsetBuilder::cancel(const Status& st) {
std::lock_guard<std::mutex> l(_lock);
if (_is_cancelled) {
return Status::OK();
}
if (_calc_delete_bitmap_token != nullptr) {
- _calc_delete_bitmap_token->cancel();
+ _calc_delete_bitmap_token->cancel(st);
+ }
+ // Flush-time bitmap work belongs to the rowset writer, not the builder token.
+ if (_rowset_writer != nullptr) {
+ _rowset_writer->cancel_calc_delete_bitmap(st);
}
_is_cancelled = true;
return Status::OK();
diff --git a/be/src/storage/rowset_builder.h b/be/src/storage/rowset_builder.h
index 205447d8faa..d34401ef2fc 100644
--- a/be/src/storage/rowset_builder.h
+++ b/be/src/storage/rowset_builder.h
@@ -62,7 +62,7 @@ public:
Status wait_calc_delete_bitmap();
- Status cancel();
+ virtual Status cancel(const Status& st = Status::Cancelled("rowset builder cancelled"));
const std::shared_ptr<RowsetWriter>& rowset_writer() const { return _rowset_writer; }
diff --git a/be/test/load/channel/eos_completion_test.cpp b/be/test/load/channel/eos_completion_test.cpp
new file mode 100644
index 00000000000..80c3df32215
--- /dev/null
+++ b/be/test/load/channel/eos_completion_test.cpp
@@ -0,0 +1,151 @@
+// 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 "load/channel/eos_completion.h"
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "util/countdown_latch.h"
+
+namespace doris {
+
+TEST(EosCompletionTest, MoreSendersThanWorkers) {
+ EosCompletion completion;
+ int replies = 0;
+ // A single worker can process all 100 EOS tasks: registering a waiter must
+ // return without waiting for any other sender or sending a successful RPC.
+ for (int sender = 0; sender < 100; ++sender) {
+ completion.add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ });
+ EXPECT_EQ(replies, 0);
+ }
+ completion.complete(Status::OK());
+ EXPECT_EQ(replies, 100);
+}
+
+TEST(EosCompletionTest, FailureAndLateRegistration) {
+ EosCompletion completion;
+ int replies = 0;
+ auto callback = [&](const Status& status) {
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("flush failed"), std::string::npos);
+ ++replies;
+ };
+ completion.add_waiter(callback);
+ completion.complete(Status::InternalError("flush failed"));
+ // Models an EOS task finishing its response/profile writes after failure.
+ completion.add_waiter(callback);
+ completion.complete(Status::OK());
+ EXPECT_EQ(replies, 2);
+}
+
+TEST(EosCompletionTest, ArrivalSuccessSurvivesLaterCloseFailure) {
+ EosCompletion completion;
+ int replies = 0;
+ auto reply = [&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ };
+ completion.add_waiter(reply);
+ completion.complete(Status::OK());
+ completion.complete(Status::InternalError("final flush failed"));
+ completion.complete(Status::Cancelled("late cancellation"));
+ completion.add_waiter(reply);
+ EXPECT_EQ(replies, 2);
+}
+
+TEST(EosCompletionTest, CancellationWinsBeforeRegistration) {
+ EosCompletion completion;
+ completion.complete(Status::Cancelled("load expired"));
+ completion.complete(Status::OK());
+ int replies = 0;
+ completion.add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.is<ErrorCode::CANCELLED>());
+ ++replies;
+ });
+ EXPECT_EQ(replies, 1);
+}
+
+TEST(EosCompletionTest, DestroyingUnfinishedChannelReleasesWaiters) {
+ int replies = 0;
+ {
+ EosCompletion completion;
+ completion.add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.is<ErrorCode::CANCELLED>());
+ ++replies;
+ });
+ }
+ EXPECT_EQ(replies, 1);
+}
+
+TEST(EosCompletionTest, CallbacksRunOutsideLock) {
+ EosCompletion completion;
+ int replies = 0;
+ completion.add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ completion.add_waiter([&](const Status& result) {
+ EXPECT_TRUE(result.ok());
+ ++replies;
+ });
+ completion.complete(Status::Cancelled("late cancellation"));
+ });
+ completion.complete(Status::OK());
+ EXPECT_EQ(replies, 2);
+}
+
+TEST(EosCompletionTest, ConcurrentRegistrationArrivalAndCancellationCompleteOnce) {
+ EosCompletion completion;
+ CountDownLatch start(1);
+ std::atomic<int> replies {0};
+ std::atomic<int> successes {0};
+ std::vector<std::thread> tasks;
+ for (int i = 0; i < 100; ++i) {
+ tasks.emplace_back([&] {
+ start.wait();
+ completion.add_waiter([&](const Status& status) {
+ if (status.ok()) {
+ ++successes;
+ }
+ ++replies;
+ });
+ });
+ }
+ tasks.emplace_back([&] {
+ start.wait();
+ completion.complete(Status::OK());
+ });
+ tasks.emplace_back([&] {
+ start.wait();
+ completion.complete(Status::Cancelled("load expired"));
+ });
+ start.count_down();
+ for (auto& task : tasks) {
+ task.join();
+ }
+ EXPECT_EQ(replies.load(), 100);
+ EXPECT_TRUE(successes.load() == 0 || successes.load() == 100);
+}
+
+} // namespace doris
diff --git a/be/test/load/channel/load_channel_eos_test.cpp b/be/test/load/channel/load_channel_eos_test.cpp
new file mode 100644
index 00000000000..624c97ab4b3
--- /dev/null
+++ b/be/test/load/channel/load_channel_eos_test.cpp
@@ -0,0 +1,566 @@
+// 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 <gen_cpp/internal_service.pb.h>
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <chrono>
+#include <ctime>
+#include <memory>
+#include <string>
+#include <thread>
+#include <utility>
+
+#include "agent/be_exec_version_manager.h"
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablets_channel.h"
+#include "core/block/block.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "cpp/sync_point.h"
+#include "load/channel/eos_completion.h"
+#include "load/channel/load_channel.h"
+#include "load/channel/load_channel_mgr.h"
+#include "load/channel/tablets_channel.h"
+#include "runtime/exec_env.h"
+#include "runtime/fragment_mgr.h"
+#include "storage/storage_engine.h"
+#include "util/countdown_latch.h"
+#include "util/defer_op.h"
+
+namespace doris {
+
+class LoadChannelEosTest : public testing::TestWithParam<bool> {
+protected:
+ void SetUp() override {
+ auto* env = ExecEnv::GetInstance();
+ _previous_fragment_mgr = env->_fragment_mgr;
+ _fragment_mgr = std::make_unique<FragmentMgr>(env);
+ env->_fragment_mgr = _fragment_mgr.get();
+ _load = std::make_shared<LoadChannel>(UniqueId(1, 2), 60, true, "", 0, false, 0);
+ }
+
+ void TearDown() override {
+ _load.reset();
+ _fragment_mgr->stop();
+ ExecEnv::GetInstance()->_fragment_mgr = _previous_fragment_mgr;
+ }
+
+ std::shared_ptr<BaseTabletsChannel> make_channel(int senders) {
+ PUniqueId id;
+ id.set_hi(1);
+ id.set_lo(2);
+ std::shared_ptr<BaseTabletsChannel> channel;
+ if (GetParam()) {
+ _cloud_engine = std::make_unique<CloudStorageEngine>(EngineOptions {});
+ channel = std::make_shared<CloudTabletsChannel>(
+ *_cloud_engine, TabletsChannelKey(id, 10), UniqueId(1, 2), true, nullptr);
+ } else {
+ _engine = std::make_unique<StorageEngine>(EngineOptions {});
+ channel = std::make_shared<TabletsChannel>(*_engine, TabletsChannelKey(id, 10),
+ UniqueId(1, 2), true, nullptr);
+ }
+ channel->set_load_cancel_status(_load->_cancel_status);
+ // No writers are needed to exercise real sender accounting and close.
+ channel->_state = BaseTabletsChannel::kOpened;
+ channel->_num_remaining_senders = senders;
+ channel->_closed_senders.Reset(senders);
+ channel->_next_seqs.resize(senders);
+ _load->_tablets_channels.emplace(10, channel);
+ _load->_opened = true;
+ return channel;
+ }
+
+ static PTabletWriterAddBlockRequest eos(int sender, bool hang_wait = true) {
+ PTabletWriterAddBlockRequest request;
+ request.mutable_id()->set_hi(1);
+ request.mutable_id()->set_lo(2);
+ request.set_index_id(10);
+ request.set_sender_id(sender);
+ request.set_backend_id(1);
+ request.set_eos(true);
+ request.set_hang_wait(hang_wait);
+ return request;
+ }
+
+ static PTabletWriterAddBlockRequest eos_with_block(int sender) {
+ auto request = eos(sender);
+ request.set_packet_seq(7);
+ request.add_tablet_ids(123);
+ auto column = ColumnInt32::create();
+ column->insert_value(42);
+ Block block {{std::move(column), std::make_shared<DataTypeInt32>(), "value"}};
+ size_t uncompressed_bytes = 0;
+ size_t compressed_bytes = 0;
+ int64_t compress_time = 0;
+ EXPECT_TRUE(block.serialize(BeExecVersionManager::get_newest_version(),
+ request.mutable_block(), &uncompressed_bytes, &compressed_bytes,
+ &compress_time, segment_v2::CompressionTypePB::SNAPPY)
+ .ok());
+ return request;
+ }
+
+ Status retry_on_rpc_thread(const PTabletWriterAddBlockRequest& request,
+ PTabletWriterAddBlockResult* response,
+ std::shared_ptr<EosCompletion>* completion) {
+ // Arrival callbacks run inside the final closer's attached load task.
+ // A real retry runs on another RPC worker: calling add_batch inline
+ // would illegally nest SCOPED_ATTACH_TASK on the callback thread.
+ Status status;
+ std::thread rpc([&] { status = _load->add_batch(request, response, completion); });
+ rpc.join();
+ return status;
+ }
+
+ void check_final_sender_retry(bool with_block, bool close_fails) {
+ auto channel = make_channel(2);
+ auto request = with_block ? eos_with_block(1) : eos(1);
+ // Model the packet already accepted by add_batch before entering EOS
+ // accounting. No writer is installed: reaching the writer lookup on a
+ // retry would fail with "unknown tablet" instead of succeeding silently.
+ channel->_next_seqs[1] = 8;
+ PTabletWriterAddBlockResult first_response;
+ std::shared_ptr<EosCompletion> first;
+ ASSERT_TRUE(_load->add_batch(eos(0), &first_response, &first).ok());
+
+ CountDownLatch arrival_callback(1);
+ CountDownLatch finish_callback(1);
+ CountDownLatch retry_waiting(1);
+ first->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ arrival_callback.count_down();
+ finish_callback.wait();
+ });
+ auto* sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard wait_guard;
+ SyncPoint::CallbackGuard close_guard;
+ sp->set_call_back(
+ "BaseTabletsChannel::close.wait_for_final_result",
+ [&](auto&&) { retry_waiting.count_down(); }, &wait_guard);
+ int final_closes = 0;
+ sp->set_call_back(
+ before_flush_sync_point(),
+ [&](auto&& args) {
+ ++final_closes;
+ auto* response = try_any_cast<PTabletWriterAddBlockResult*>(args[0]);
+ auto* tablet = response->add_tablet_vec();
+ tablet->set_tablet_id(123);
+ tablet->set_schema_hash(0);
+ tablet->set_received_rows(1);
+ if (close_fails) {
+ auto* ret = try_any_cast_ret<Status>(args);
+ ret->first = Status::InternalError("final commit failed");
+ ret->second = true;
+ }
+ },
+ &close_guard);
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+
+ Status final_status;
+ PTabletWriterAddBlockResult final_response;
+ std::thread finalizer([&] {
+ std::shared_ptr<EosCompletion> completion;
+ final_status = _load->add_batch(request, &final_response, &completion);
+ });
+ bool arrived = arrival_callback.wait_for(std::chrono::seconds(10));
+ EXPECT_TRUE(arrived);
+ Status retry_status;
+ PTabletWriterAddBlockResult retry_response;
+ std::atomic<bool> retry_returned {false};
+ std::thread retry([&] {
+ std::shared_ptr<EosCompletion> completion;
+ retry_status = _load->add_batch(request, &retry_response, &completion);
+ retry_returned = true;
+ });
+ // The finalizer is still inside an arrival callback, with _lock released.
+ // Require the retry to actually enter this window before allowing close.
+ EXPECT_TRUE(retry_waiting.wait_for(std::chrono::seconds(10)));
+ EXPECT_FALSE(retry_returned.load());
+ finish_callback.count_down();
+ finalizer.join();
+ retry.join();
+ EXPECT_TRUE(retry_returned.load());
+ EXPECT_EQ(final_closes, 1);
+ EXPECT_EQ(channel->_num_remaining_senders, 0);
+ EXPECT_EQ(channel->_next_seqs[1], 8);
+ EXPECT_EQ(final_status.ok(), !close_fails);
+ EXPECT_EQ(retry_status.to_string(), final_status.to_string());
+ EXPECT_EQ(retry_response.tablet_errors_size(), 0);
+ ASSERT_EQ(retry_response.tablet_vec_size(), 1);
+ EXPECT_EQ(retry_response.tablet_vec(0).tablet_id(), 123);
+ EXPECT_EQ(retry_response.tablet_vec(0).received_rows(), 1);
+ if (close_fails) {
+ EXPECT_NE(retry_status.to_string().find("final commit failed"), std::string::npos);
+ }
+ }
+
+ const char* before_flush_sync_point() const {
+ return GetParam() ? "CloudTabletsChannel::close.before_flush"
+ : "TabletsChannel::close.before_flush";
+ }
+
+ FragmentMgr* _previous_fragment_mgr = nullptr;
+ std::unique_ptr<FragmentMgr> _fragment_mgr;
+ std::unique_ptr<StorageEngine> _engine;
+ std::unique_ptr<CloudStorageEngine> _cloud_engine;
+ std::shared_ptr<LoadChannel> _load;
+};
+
+TEST_P(LoadChannelEosTest, SingleWorkerProcessesOneHundredSendersAndDuplicateEos) {
+ auto channel = make_channel(100);
+ std::atomic<int> replies {0};
+ CountDownLatch processed(1);
+ std::thread worker([&] {
+ // Retry sender zero before any other sender. It must not decrement the
+ // remaining count twice, and both RPCs must wait for all senders to arrive.
+ for (int i = 0; i <= 100; ++i) {
+ int sender = i == 0 ? 0 : i - 1;
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> completion;
+ EXPECT_TRUE(_load->add_batch(eos(sender), &response, &completion).ok());
+ EXPECT_NE(completion, nullptr);
+ completion->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ EXPECT_EQ(channel->_num_remaining_senders, 0);
+ ++replies;
+ });
+ if (sender < 99) {
+ EXPECT_EQ(replies.load(), 0);
+ }
+ }
+ processed.count_down();
+ });
+ bool finished = processed.wait_for(std::chrono::seconds(10));
+ if (!finished) {
+ // Also makes the old blocking implementation fail instead of hanging UT.
+ EXPECT_TRUE(_load->cancel().ok());
+ }
+ worker.join();
+ EXPECT_TRUE(finished);
+ EXPECT_EQ(replies.load(), 101);
+ EXPECT_TRUE(_load->is_finished());
+ EXPECT_EQ(channel->_num_remaining_senders, 0);
+}
+
+TEST_P(LoadChannelEosTest, NonWaitingLastSenderReleasesEarlierWaiter) {
+ auto channel = make_channel(2);
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> first;
+ ASSERT_TRUE(_load->add_batch(eos(0), &response, &first).ok());
+ int replies = 0;
+ first->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ });
+ EXPECT_EQ(replies, 0);
+ bool reached_close = false;
+ auto* sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ sp->set_call_back(
+ before_flush_sync_point(),
+ [&](auto&&) {
+ reached_close = true;
+ EXPECT_EQ(replies, 1);
+ },
+ &guard);
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+ std::shared_ptr<EosCompletion> last;
+ EXPECT_TRUE(_load->add_batch(eos(1, false), &response, &last).ok());
+ EXPECT_TRUE(reached_close);
+ EXPECT_EQ(last, nullptr);
+ EXPECT_EQ(replies, 1);
+}
+
+TEST_P(LoadChannelEosTest, FinalOwnerReleasesRegisteredAndNotYetRegisteredRpcsAfterCancel) {
+ auto channel = make_channel(3);
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> first;
+ std::shared_ptr<EosCompletion> second;
+ ASSERT_TRUE(_load->add_batch(eos(0), &response, &first).ok());
+ ASSERT_TRUE(_load->add_batch(eos(1), &response, &second).ok());
+ int replies = 0;
+ auto callback = [&](const Status& status) {
+ EXPECT_TRUE(status.is<ErrorCode::CANCELLED>());
+ // Final-owner cleanup invokes RPC callbacks outside the tablets-channel lock.
+ std::lock_guard lock(channel->_lock);
+ ++replies;
+ };
+ first->add_waiter(callback);
+ EXPECT_TRUE(_load->cancel().ok());
+ EXPECT_TRUE(_load->cancel().ok());
+ EXPECT_EQ(replies, 0);
+ _load.reset();
+ EXPECT_EQ(replies, 1);
+ second->add_waiter(callback);
+ EXPECT_EQ(replies, 2);
+}
+
+TEST_P(LoadChannelEosTest, TimeoutCleanerReleasesPendingRpc) {
+ auto channel = make_channel(2);
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> completion;
+ ASSERT_TRUE(_load->add_batch(eos(0), &response, &completion).ok());
+ LoadChannelMgr manager;
+ manager._load_channels.emplace(_load->load_id(), _load);
+ _load->_last_updated_time.store(time(nullptr) - 61);
+ int replies = 0;
+ completion->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.is<ErrorCode::CANCELLED>());
+ EXPECT_TRUE(manager.get_all_load_channel_ids().empty());
+ ++replies;
+ });
+ _load.reset(); // The manager is the final owner, as with pending asynchronous RPCs.
+ EXPECT_TRUE(manager._start_load_channels_clean().ok());
+ EXPECT_EQ(replies, 1);
+ manager.stop();
+}
+
+// Exercise the real local/cloud close paths. Hold the final closer immediately
+// before writer flushing, after sender accounting and barrier publication.
+TEST_P(LoadChannelEosTest, ArrivalReleasesWaitersBeforeFinalCloseFailure) {
+ auto channel = make_channel(3);
+ CountDownLatch closing(1);
+ CountDownLatch finish_close(1);
+ auto* sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ sp->set_call_back(
+ before_flush_sync_point(),
+ [&](auto&& args) {
+ closing.count_down();
+ finish_close.wait();
+ auto* ret = try_any_cast_ret<Status>(args);
+ ret->first = Status::InternalError("final flush failed");
+ ret->second = true;
+ },
+ &guard);
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+
+ PTabletWriterAddBlockResult first_response;
+ PTabletWriterAddBlockResult second_response;
+ std::shared_ptr<EosCompletion> first;
+ std::shared_ptr<EosCompletion> second;
+ ASSERT_TRUE(_load->add_batch(eos(0), &first_response, &first).ok());
+ ASSERT_TRUE(_load->add_batch(eos(1), &second_response, &second).ok());
+ std::atomic<int> replies {0};
+ auto reply = [&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ };
+ first->add_waiter(reply);
+ std::atomic<bool> final_rpc_returned {false};
+ std::thread finalizer([&] {
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> last;
+ auto status = _load->add_batch(eos(2), &response, &last);
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("final flush failed"), std::string::npos);
+ // The service returns this error directly instead of registering the
+ // final RPC on the already successful arrival barrier.
+ final_rpc_returned = true;
+ });
+ bool reached_close = closing.wait_for(std::chrono::seconds(10));
+ EXPECT_TRUE(reached_close);
+ if (reached_close) {
+ EXPECT_TRUE(channel->is_finished());
+ EXPECT_FALSE(_load->is_finished());
+ EXPECT_FALSE(final_rpc_returned.load());
+ EXPECT_EQ(replies.load(), 1);
+ // A sender may finish its response writes after the last EOS arrived.
+ second->add_waiter(reply);
+ EXPECT_EQ(replies.load(), 2);
+ }
+ finish_close.count_down();
+ finalizer.join();
+ EXPECT_TRUE(final_rpc_returned.load());
+ EXPECT_TRUE(_load->cancel().ok());
+ // Neither the later close failure nor cancellation can revoke arrival.
+ first->add_waiter(reply);
+ EXPECT_EQ(replies.load(), 3);
+}
+
+TEST_P(LoadChannelEosTest, FinalRpcKeepsItsResponseUntilCloseReturns) {
+ auto channel = make_channel(2);
+ auto first_response = std::make_unique<PTabletWriterAddBlockResult>();
+ std::shared_ptr<EosCompletion> first;
+ ASSERT_TRUE(_load->add_batch(eos(0), first_response.get(), &first).ok());
+ int replies = 0;
+ first->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ first_response.reset();
+ ++replies;
+ });
+
+ bool reached_close = false;
+ auto* sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ sp->set_call_back(
+ before_flush_sync_point(),
+ [&](auto&& args) {
+ reached_close = true;
+ EXPECT_EQ(replies, 1);
+ EXPECT_EQ(first_response, nullptr);
+ EXPECT_FALSE(_load->is_finished());
+ // Model tablet results written by the final closer after all
+ // earlier RPCs have already been released.
+ auto* response = try_any_cast<PTabletWriterAddBlockResult*>(args[0]);
+ auto* tablet = response->add_tablet_vec();
+ tablet->set_tablet_id(123);
+ tablet->set_schema_hash(0);
+ tablet->set_received_rows(456);
+ tablet->set_num_rows_filtered(7);
+ },
+ &guard);
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+
+ auto response = std::make_unique<PTabletWriterAddBlockResult>();
+ std::shared_ptr<EosCompletion> last;
+ ASSERT_TRUE(_load->add_batch(eos(1), response.get(), &last).ok());
+ EXPECT_TRUE(reached_close);
+ EXPECT_TRUE(_load->is_finished());
+ // Same ownership boundary as the service: finish synchronous writes before
+ // registering the final callback, which can destroy the response inline.
+ response->set_execution_time_us(123);
+ last->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ EXPECT_EQ(response->execution_time_us(), 123);
+ ASSERT_EQ(response->tablet_vec_size(), 1);
+ EXPECT_EQ(response->tablet_vec(0).tablet_id(), 123);
+ EXPECT_EQ(response->tablet_vec(0).received_rows(), 456);
+ EXPECT_EQ(response->tablet_vec(0).num_rows_filtered(), 7);
+ response.reset();
+ ++replies;
+ });
+ EXPECT_EQ(response, nullptr);
+ EXPECT_EQ(replies, 2);
+}
+
+TEST_P(LoadChannelEosTest, ArrivalCallbacksAllowConcurrentRetryAndCancel) {
+ auto channel = make_channel(2);
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> first;
+ ASSERT_TRUE(_load->add_batch(eos(0), &response, &first).ok());
+ int replies = 0;
+ first->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ EXPECT_FALSE(_load->is_finished());
+ ++replies;
+ // The independent RPC worker must acquire the load/channel locks and
+ // finish while this callback is running. Joining it also checks that
+ // the final closer invokes callbacks without holding either lock.
+ PTabletWriterAddBlockResult duplicate_response;
+ std::shared_ptr<EosCompletion> duplicate;
+ EXPECT_TRUE(retry_on_rpc_thread(eos(0), &duplicate_response, &duplicate).ok());
+ ASSERT_NE(duplicate, nullptr);
+ EXPECT_EQ(channel->_num_remaining_senders, 0);
+ duplicate->add_waiter([&](const Status& duplicate_status) {
+ EXPECT_TRUE(duplicate_status.ok());
+ ++replies;
+ });
+ EXPECT_TRUE(_load->cancel().ok());
+ });
+ int final_closes = 0;
+ auto* sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ sp->set_call_back(
+ before_flush_sync_point(), [&](auto&&) { ++final_closes; }, &guard);
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+
+ std::shared_ptr<EosCompletion> last;
+ EXPECT_TRUE(_load->add_batch(eos(1), &response, &last).is<ErrorCode::CANCELLED>());
+ last->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ });
+ EXPECT_EQ(replies, 3);
+ EXPECT_EQ(final_closes, 0);
+ EXPECT_TRUE(_load->is_cancelled());
+ EXPECT_FALSE(_load->is_finished());
+}
+
+TEST_P(LoadChannelEosTest, FinalSenderRetryWaitsForCommitFailure) {
+ check_final_sender_retry(false, true);
+}
+
+TEST_P(LoadChannelEosTest, FinalSenderRetryReceivesTabletResults) {
+ check_final_sender_retry(false, false);
+}
+
+TEST_P(LoadChannelEosTest, FinalSenderBlockRetryWaitsForCommitFailure) {
+ check_final_sender_retry(true, true);
+}
+
+TEST_P(LoadChannelEosTest, FinalSenderBlockRetryReceivesTabletResults) {
+ check_final_sender_retry(true, false);
+}
+
+TEST_P(LoadChannelEosTest, EarlierSenderBlockRetryDoesNotWriteDuringArrivalCallback) {
+ auto channel = make_channel(2);
+ channel->_next_seqs[0] = 8;
+ auto request = eos_with_block(0);
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> first;
+ ASSERT_TRUE(_load->add_batch(request, &response, &first).ok());
+ int replies = 0;
+ first->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ PTabletWriterAddBlockResult retry_response;
+ std::shared_ptr<EosCompletion> retry;
+ auto retry_status = retry_on_rpc_thread(request, &retry_response, &retry);
+ EXPECT_TRUE(retry_status.ok()) << retry_status;
+ EXPECT_EQ(retry_response.tablet_errors_size(), 0);
+ EXPECT_EQ(channel->_next_seqs[0], 8);
+ ASSERT_NE(retry, nullptr);
+ retry->add_waiter([&](const Status& st) {
+ EXPECT_TRUE(st.ok());
+ ++replies;
+ });
+ });
+ std::shared_ptr<EosCompletion> last;
+ EXPECT_TRUE(_load->add_batch(eos(1), &response, &last).ok());
+ EXPECT_EQ(replies, 1);
+}
+
+TEST_P(LoadChannelEosTest, PacketSequenceAdmission) {
+ auto channel = make_channel(1);
+ channel->_next_seqs[0] = 8;
+ auto request = eos_with_block(0);
+ int64_t current_seq = 0;
+ bool should_write = true;
+ ASSERT_TRUE(channel->_get_current_seq(current_seq, request, should_write).ok());
+ EXPECT_FALSE(should_write);
+ EXPECT_EQ(current_seq, 8);
+ request.set_packet_seq(8);
+ ASSERT_TRUE(channel->_get_current_seq(current_seq, request, should_write).ok());
+ EXPECT_TRUE(should_write);
+ request.set_packet_seq(9);
+ EXPECT_FALSE(channel->_get_current_seq(current_seq, request, should_write).ok());
+ EXPECT_FALSE(should_write);
+ EXPECT_TRUE(channel->cancel().ok());
+ request.set_packet_seq(8);
+ ASSERT_TRUE(channel->_get_current_seq(current_seq, request, should_write).ok());
+ EXPECT_FALSE(should_write);
+}
+
+INSTANTIATE_TEST_SUITE_P(LocalAndCloud, LoadChannelEosTest, testing::Bool());
+
+} // namespace doris
diff --git a/be/test/load/channel/load_channel_mgr_test.cpp b/be/test/load/channel/load_channel_mgr_test.cpp
new file mode 100644
index 00000000000..007592d9641
--- /dev/null
+++ b/be/test/load/channel/load_channel_mgr_test.cpp
@@ -0,0 +1,274 @@
+// 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 "load/channel/load_channel_mgr.h"
+
+#include <gtest/gtest.h>
+
+#include <thread>
+
+#include "runtime/exec_env.h"
+#include "runtime/fragment_mgr.h"
+#include "util/countdown_latch.h"
+
+namespace doris {
+
+class LoadChannelMgrTest : public testing::Test {
+protected:
+ void SetUp() override {
+ auto* exec_env = ExecEnv::GetInstance();
+ if (exec_env->fragment_mgr() == nullptr) {
+ _fragment_mgr = std::make_unique<FragmentMgr>(exec_env);
+ exec_env->_fragment_mgr = _fragment_mgr.get();
+ }
+ _mgr = std::make_unique<LoadChannelMgr>();
+ // Drive timeout cleanup explicitly so the interleavings are deterministic.
+ _mgr->_load_state_channels = std::make_unique<LoadChannelMgr::LoadStateChannelCache>(1024);
+ }
+
+ void TearDown() override {
+ _mgr->stop();
+ _mgr.reset();
+ if (_fragment_mgr != nullptr) {
+ _fragment_mgr->stop();
+ ExecEnv::GetInstance()->_fragment_mgr = nullptr;
+ _fragment_mgr.reset();
+ }
+ }
+
+ std::shared_ptr<LoadChannel> create_channel() {
+ auto channel = std::make_shared<LoadChannel>(_load_id, 1, true, "127.0.0.1", 0, false, -1);
+ std::lock_guard<std::mutex> lock(_mgr->_lock);
+ _mgr->_load_channels[_load_id] = channel;
+ return channel;
+ }
+
+ void expect_no_cached_state() {
+ auto* handle = _mgr->_load_state_channels->lookup(_load_id.to_string());
+ EXPECT_EQ(handle, nullptr);
+ if (handle != nullptr) {
+ _mgr->_load_state_channels->release(handle);
+ }
+ }
+
+ void expect_cached_failure(const Status& reason) {
+ auto* handle = _mgr->_load_state_channels->lookup(_load_id.to_string());
+ ASSERT_NE(handle, nullptr);
+ auto* value =
+ static_cast<LoadChannelMgr::CacheValue*>(_mgr->_load_state_channels->value(handle));
+ EXPECT_NE(value, nullptr);
+ if (value != nullptr) {
+ EXPECT_EQ(value->_cancel_reason, reason.msg());
+ }
+ _mgr->_load_state_channels->release(handle);
+ }
+
+ UniqueId _load_id {1, 2};
+ std::unique_ptr<FragmentMgr> _fragment_mgr;
+ std::unique_ptr<LoadChannelMgr> _mgr;
+};
+
+TEST_F(LoadChannelMgrTest, FailedBatchCancelsRetainedChannelAndCachesReason) {
+ auto channel = create_channel();
+ PTabletWriterAddBlockRequest request;
+ *request.mutable_id() = _load_id.to_proto();
+ request.set_index_id(100);
+ PTabletWriterAddBlockResult response;
+ // No index was opened, so this exercises the actual add_batch failure path.
+ auto st = _mgr->add_batch(request, &response, nullptr);
+ ASSERT_FALSE(st.ok());
+ EXPECT_EQ(channel->cancel_status().to_string(), st.to_string());
+ EXPECT_TRUE(_mgr->_load_channels.empty());
+
+ request.set_eos(true);
+ auto late_status = _mgr->add_batch(request, &response, nullptr);
+ EXPECT_TRUE(late_status.is<ErrorCode::CANCELLED>()) << late_status;
+ EXPECT_EQ(late_status.msg(), st.msg());
+
+ PTabletWriterOpenRequest open_request;
+ *open_request.mutable_id() = _load_id.to_proto();
+ EXPECT_TRUE(_mgr->open(open_request).is<ErrorCode::CANCELLED>());
+}
+
+TEST_F(LoadChannelMgrTest, PublicCancelCachesAlreadyPublishedFailure) {
+ auto channel = create_channel();
+ const auto first_failure = Status::InternalError("first writer failure");
+ CountDownLatch published(1);
+ CountDownLatch resume_failure(1);
+ Status publish_status;
+ Status failure_status;
+ std::thread failed_request([&] {
+ SCOPED_INIT_THREAD_CONTEXT();
+ // Stage the publication separately to model _cancel_load_channel paused
+ // before taking the manager lock. Its repeated publication is idempotent.
+ publish_status = channel->cancel(first_failure);
+ published.count_down();
+ resume_failure.wait();
+ failure_status = _mgr->_cancel_load_channel(channel, first_failure);
+ });
+ published.wait();
+
+ PTabletWriterCancelRequest cancel_request;
+ *cancel_request.mutable_id() = _load_id.to_proto();
+ cancel_request.set_cancel_reason("later upstream cancellation");
+ const auto cancel_status = _mgr->cancel(cancel_request);
+ resume_failure.count_down();
+ failed_request.join();
+
+ ASSERT_TRUE(publish_status.ok()) << publish_status;
+ ASSERT_TRUE(cancel_status.ok()) << cancel_status;
+ ASSERT_TRUE(failure_status.ok()) << failure_status;
+ EXPECT_EQ(channel->cancel_status().to_string(), first_failure.to_string());
+ EXPECT_TRUE(_mgr->_load_channels.empty());
+
+ expect_cached_failure(first_failure);
+
+ PTabletWriterOpenRequest open_request;
+ *open_request.mutable_id() = _load_id.to_proto();
+ const auto open_status = _mgr->open(open_request);
+ EXPECT_TRUE(open_status.is<ErrorCode::CANCELLED>()) << open_status;
+ EXPECT_EQ(open_status.msg(), first_failure.msg());
+
+ PTabletWriterAddBlockRequest add_request;
+ *add_request.mutable_id() = _load_id.to_proto();
+ add_request.set_eos(true);
+ PTabletWriterAddBlockResult response;
+ const auto add_status = _mgr->add_batch(add_request, &response, nullptr);
+ EXPECT_TRUE(add_status.is<ErrorCode::CANCELLED>()) << add_status;
+ EXPECT_EQ(add_status.msg(), first_failure.msg());
+}
+
+TEST_F(LoadChannelMgrTest, PublicCancelKeepsTheSameMessageWithOrWithoutChannel) {
+ PTabletWriterCancelRequest request;
+ *request.mutable_id() = _load_id.to_proto();
+ request.set_cancel_reason("upstream load failed");
+ const auto reason = Status::Cancelled(request.cancel_reason());
+
+ // The no-channel path also covers cancellation arriving before open.
+ for (bool has_channel : {false, true}) {
+ if (has_channel) {
+ create_channel();
+ }
+ ASSERT_TRUE(_mgr->cancel(request).ok());
+ expect_cached_failure(reason);
+
+ PTabletWriterOpenRequest open_request;
+ *open_request.mutable_id() = _load_id.to_proto();
+ auto open_status = _mgr->open(open_request);
+ EXPECT_TRUE(open_status.is<ErrorCode::CANCELLED>());
+ EXPECT_EQ(open_status.msg(), reason.msg());
+
+ PTabletWriterAddBlockRequest add_request;
+ *add_request.mutable_id() = _load_id.to_proto();
+ add_request.set_eos(true);
+ PTabletWriterAddBlockResult response;
+ auto add_status = _mgr->add_batch(add_request, &response, nullptr);
+ EXPECT_TRUE(add_status.is<ErrorCode::CANCELLED>());
+ EXPECT_EQ(add_status.msg(), reason.msg());
+
+ PTabletWriterCancelRequest later_request = request;
+ later_request.set_cancel_reason("later cancellation");
+ ASSERT_TRUE(_mgr->cancel(later_request).ok());
+ expect_cached_failure(reason);
+ _mgr->_load_state_channels->erase(_load_id.to_string());
+ }
+}
+
+TEST_F(LoadChannelMgrTest, EmptyFailureMessageStillRejectsLateRequests) {
+ auto channel = create_channel();
+ const auto reason = Status::InternalError("");
+ ASSERT_TRUE(_mgr->_cancel_load_channel(channel, reason).ok());
+ expect_cached_failure(reason);
+
+ PTabletWriterOpenRequest open_request;
+ *open_request.mutable_id() = _load_id.to_proto();
+ auto open_status = _mgr->open(open_request);
+ EXPECT_TRUE(open_status.is<ErrorCode::CANCELLED>());
+ EXPECT_TRUE(open_status.msg().empty());
+
+ PTabletWriterAddBlockRequest add_request;
+ *add_request.mutable_id() = _load_id.to_proto();
+ add_request.set_eos(true);
+ PTabletWriterAddBlockResult response;
+ auto add_status = _mgr->add_batch(add_request, &response, nullptr);
+ EXPECT_TRUE(add_status.is<ErrorCode::CANCELLED>());
+ EXPECT_TRUE(add_status.msg().empty());
+}
+
+TEST_F(LoadChannelMgrTest, LateFailureDoesNotCancelReplacementAfterTimeout) {
+ auto original = create_channel();
+ CountDownLatch captured(1);
+ CountDownLatch resume_failure(1);
+ Status failure_status;
+ std::thread failed_request([&, channel = original] {
+ SCOPED_INIT_THREAD_CONTEXT();
+ captured.count_down();
+ resume_failure.wait();
+ failure_status = _mgr->_cancel_load_channel(channel, Status::InternalError("old request"));
+ });
+ captured.wait();
+ original->_last_updated_time.store(0);
+ EXPECT_TRUE(_mgr->_start_load_channels_clean().ok());
+ auto replacement = create_channel();
+ resume_failure.count_down();
+ failed_request.join();
+
+ EXPECT_TRUE(failure_status.ok()) << failure_status;
+ EXPECT_TRUE(original->is_cancelled());
+ EXPECT_FALSE(replacement->is_cancelled());
+ EXPECT_EQ(_mgr->_load_channels.at(_load_id), replacement);
+ expect_no_cached_state();
+}
+
+TEST_F(LoadChannelMgrTest, LateFinishDoesNotRemoveReplacement) {
+ auto original = create_channel();
+ auto replacement = create_channel();
+ _mgr->_finish_load_channel(original);
+ EXPECT_EQ(_mgr->_load_channels.at(_load_id), replacement);
+ EXPECT_FALSE(replacement->is_cancelled());
+ expect_no_cached_state();
+}
+
+TEST_F(LoadChannelMgrTest, FinishDoesNotOverwritePublishedFailure) {
+ auto channel = create_channel();
+ auto reason = Status::InternalError("first failure");
+ ASSERT_TRUE(channel->cancel(reason).ok());
+ // EOS arrives between cancellation publication and removal from the manager.
+ _mgr->_finish_load_channel(channel);
+ EXPECT_EQ(_mgr->_load_channels.at(_load_id), channel);
+ expect_no_cached_state();
+ ASSERT_TRUE(_mgr->_cancel_load_channel(channel, Status::InternalError("later failure")).ok());
+ _mgr->_finish_load_channel(channel);
+
+ expect_cached_failure(reason);
+}
+
+TEST_F(LoadChannelMgrTest, FailureStillCancelsOriginalAfterFinish) {
+ auto channel = create_channel();
+ _mgr->_finish_load_channel(channel);
+ ASSERT_TRUE(_mgr->_load_channels.empty());
+ auto reason = Status::InternalError("late failure");
+ ASSERT_TRUE(_mgr->_cancel_load_channel(channel, reason).ok());
+ EXPECT_EQ(channel->cancel_status().to_string(), reason.to_string());
+ // A previously completed manager transition keeps its terminal state.
+ auto* handle = _mgr->_load_state_channels->lookup(_load_id.to_string());
+ ASSERT_NE(handle, nullptr);
+ EXPECT_EQ(_mgr->_load_state_channels->value(handle), nullptr);
+ _mgr->_load_state_channels->release(handle);
+}
+
+} // namespace doris
diff --git a/be/test/load/delta_writer/delta_writer_cancel_test.cpp b/be/test/load/delta_writer/delta_writer_cancel_test.cpp
new file mode 100644
index 00000000000..6c58a9ac962
--- /dev/null
+++ b/be/test/load/delta_writer/delta_writer_cancel_test.cpp
@@ -0,0 +1,266 @@
+// 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 <gtest/gtest.h>
+
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+#include "cloud/cloud_delta_writer.h"
+#include "cloud/cloud_rowset_builder.h"
+#include "cloud/cloud_rowset_writer.h"
+#include "cloud/cloud_storage_engine.h"
+#include "load/channel/load_channel.h"
+#include "load/channel/tablets_channel.h"
+#include "load/delta_writer/delta_writer.h"
+#include "runtime/exec_env.h"
+#include "runtime/fragment_mgr.h"
+#include "runtime/memory/mem_tracker_limiter.h"
+#include "runtime/thread_context.h"
+#include "storage/delete/calc_delete_bitmap_executor.h"
+#include "storage/options.h"
+#include "storage/rowset/beta_rowset_writer.h"
+#include "storage/rowset/rowset_writer_context.h"
+#include "storage/rowset_builder.h"
+#include "storage/storage_engine.h"
+#include "util/countdown_latch.h"
+#include "util/defer_op.h"
+
+namespace doris {
+
+// Exercise local/cloud writer cancellation without tablet I/O.
+class DeltaWriterCancelTest : public testing::TestWithParam<int> {
+protected:
+ void SetUp() override {
+ auto tracker = MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+ "DeltaWriterCancelTest");
+ _attach_task = std::make_unique<AttachTask>(tracker);
+ ASSERT_TRUE(ThreadPoolBuilder("DeltaWriterCancelTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&_pool)
+ .ok());
+ _previous_fragment_mgr = ExecEnv::GetInstance()->_fragment_mgr;
+ _fragment_mgr = std::make_unique<FragmentMgr>(ExecEnv::GetInstance());
+ ExecEnv::GetInstance()->_fragment_mgr = _fragment_mgr.get();
+ _load_channel = std::make_shared<LoadChannel>(UniqueId {}, 60, false, "", 0, false, -1);
+ WriteRequest data_req;
+ data_req.load_cancel_status = _load_channel->_cancel_status;
+ if (is_cloud()) {
+ _cloud_engine = std::make_unique<CloudStorageEngine>(EngineOptions {});
+ _writer = std::make_unique<CloudDeltaWriter>(*_cloud_engine, data_req, nullptr,
+ UniqueId {});
+ } else {
+ _local_engine = std::make_unique<StorageEngine>(EngineOptions {});
+ _writer = std::make_unique<DeltaWriter>(*_local_engine, data_req, nullptr, UniqueId {});
+ }
+ _builders = {_writer->_rowset_builder.get()};
+ }
+
+ void TearDown() override {
+ // Release the unrelated task before destroying writers, even after a failed assertion.
+ _release_worker.count_down();
+ if (_pool) {
+ _pool->wait();
+ }
+ _writer.reset();
+ _tokens.clear();
+ _pool.reset();
+ _cloud_engine.reset();
+ _local_engine.reset();
+ _load_channel.reset();
+ if (_fragment_mgr) {
+ _fragment_mgr->stop();
+ ExecEnv::GetInstance()->_fragment_mgr = _previous_fragment_mgr;
+ _fragment_mgr.reset();
+ }
+ _attach_task.reset();
+ }
+
+ bool is_cloud() const { return GetParam() & 1; }
+
+ void install_tokens() {
+ for (auto* builder : _builders) {
+ std::shared_ptr<BaseBetaRowsetWriter> rowset_writer;
+ if (is_cloud()) {
+ rowset_writer = std::make_shared<CloudRowsetWriter>(*_cloud_engine);
+ rowset_writer->_rowset_meta = std::make_shared<RowsetMeta>();
+ } else {
+ rowset_writer = std::make_shared<BetaRowsetWriter>(*_local_engine);
+ // Local writer destruction requires the segcompaction tracker that init()
+ // creates, even when these cancellation tests never create segment files.
+ RowsetWriterContext context;
+ context.tablet_schema = std::make_shared<TabletSchema>();
+ ASSERT_TRUE(rowset_writer->init(context).ok());
+ }
+ // Set up only the state needed by cancellation and destruction. No files are created.
+ rowset_writer->_calc_delete_bitmap_token = make_token();
+ builder->_calc_delete_bitmap_token = make_token();
+ builder->_rowset_writer = rowset_writer;
+ _tokens.push_back(rowset_writer->_calc_delete_bitmap_token.get());
+ _tokens.push_back(builder->_calc_delete_bitmap_token.get());
+ }
+ }
+
+ std::unique_ptr<CalcDeleteBitmapToken> make_token() {
+ return std::make_unique<CalcDeleteBitmapToken>(
+ _pool->new_token(ThreadPool::ExecutionMode::CONCURRENT),
+ _load_channel->_cancel_status);
+ }
+
+ FragmentMgr* _previous_fragment_mgr = nullptr;
+ std::unique_ptr<FragmentMgr> _fragment_mgr;
+ std::shared_ptr<LoadChannel> _load_channel;
+ std::unique_ptr<AttachTask> _attach_task;
+ std::unique_ptr<StorageEngine> _local_engine;
+ std::unique_ptr<CloudStorageEngine> _cloud_engine;
+ std::unique_ptr<ThreadPool> _pool;
+ std::unique_ptr<BaseDeltaWriter> _writer;
+ std::vector<BaseRowsetBuilder*> _builders;
+ std::vector<CalcDeleteBitmapToken*> _tokens;
+ CountDownLatch _worker_started {1};
+ CountDownLatch _release_worker {1};
+ std::atomic<int> _executed {0};
+};
+
+TEST_P(DeltaWriterCancelTest, PublicationSkipsQueuedWorkWithoutDraining) {
+ install_tokens();
+ ASSERT_TRUE(_pool->submit_func([this] {
+ _worker_started.count_down();
+ _release_worker.wait();
+ }).ok());
+ ASSERT_TRUE(_worker_started.wait_for(std::chrono::seconds(10)));
+ for (auto* token : _tokens) {
+ ASSERT_TRUE(token->submit_func([this] {
+ ++_executed;
+ return Status::OK();
+ }).ok());
+ }
+ const auto queued = _pool->get_queue_size();
+ const auto reason = Status::Cancelled("sender cancelled the load");
+ // Holding owner locks must not prevent publishing cancellation.
+ {
+ std::lock_guard channel_lock(_load_channel->_lock);
+ std::lock_guard builder_lock(_builders.front()->_lock);
+ ASSERT_TRUE(_load_channel->cancel(reason).ok());
+ }
+ EXPECT_EQ(_pool->get_queue_size(), queued);
+ for (auto* token : _tokens) {
+ EXPECT_EQ(token->submit_func([] { return Status::OK(); }), reason);
+ }
+ _release_worker.count_down();
+ for (auto* token : _tokens) {
+ EXPECT_EQ(token->wait(), reason);
+ }
+ EXPECT_EQ(_executed.load(), 0);
+}
+
+TEST_P(DeltaWriterCancelTest, FinalOwnerDiscardsBothQueuesAfterPartialInit) {
+ install_tokens();
+ ASSERT_FALSE(_writer->_is_init);
+ ASSERT_TRUE(_pool->submit_func([this] {
+ _worker_started.count_down();
+ _release_worker.wait();
+ }).ok());
+ ASSERT_TRUE(_worker_started.wait_for(std::chrono::seconds(10)));
+ for (auto* token : _tokens) {
+ ASSERT_TRUE(token->submit_func([this] {
+ ++_executed;
+ return Status::OK();
+ }).ok());
+ }
+ ASSERT_TRUE(_load_channel->cancel().ok());
+ // No bitmap callback is running. Final-owner cleanup must discard both phases
+ // without waiting for the unrelated task to release the sole pool worker.
+ _writer.reset();
+ _tokens.clear();
+ EXPECT_EQ(_pool->get_queue_size(), 0);
+ EXPECT_EQ(_executed.load(), 0);
+}
+
+TEST_P(DeltaWriterCancelTest, FinalOwnerWaitsForRunningCallback) {
+ install_tokens();
+ CountDownLatch destroying(1);
+ std::future<void> destructor;
+ Defer cleanup {[&] {
+ _release_worker.count_down();
+ if (destructor.valid()) {
+ destructor.get();
+ }
+ }};
+ ASSERT_TRUE(_tokens.front()
+ ->submit_func([this] {
+ _worker_started.count_down();
+ _release_worker.wait();
+ ++_executed;
+ return Status::InternalError("late bitmap failure");
+ })
+ .ok());
+ ASSERT_TRUE(_worker_started.wait_for(std::chrono::seconds(10)));
+ ASSERT_TRUE(_load_channel->cancel().ok());
+ destructor = std::async(std::launch::async, [&, resource_ctx = _load_channel->_resource_ctx,
+ writer = std::move(_writer)]() mutable {
+ SCOPED_ATTACH_TASK(resource_ctx);
+ destroying.count_down();
+ writer.reset();
+ });
+ ASSERT_TRUE(destroying.wait_for(std::chrono::seconds(10)));
+ EXPECT_EQ(destructor.wait_for(std::chrono::milliseconds(50)), std::future_status::timeout);
+ _release_worker.count_down();
+ destructor.get();
+ _tokens.clear();
+ EXPECT_EQ(_executed.load(), 1);
+}
+
+TEST_P(DeltaWriterCancelTest, KeepsEarlierTokenFailureLocal) {
+ install_tokens();
+ const auto failure = Status::InternalError("bitmap failure");
+ ASSERT_TRUE(_tokens.front()->submit_func([&] { return failure; }).ok());
+ ASSERT_EQ(_tokens.front()->wait(), failure);
+ EXPECT_TRUE(_load_channel->cancel_status().ok());
+ const auto reason = Status::Cancelled("load cancelled");
+ ASSERT_TRUE(_load_channel->cancel(reason).ok());
+ EXPECT_EQ(_tokens.front()->wait(), failure);
+ for (size_t i = 1; i < _tokens.size(); ++i) {
+ EXPECT_EQ(_tokens[i]->wait(), reason);
+ }
+}
+
+TEST_P(DeltaWriterCancelTest, CancellationWinsOverLaterCallbackFailure) {
+ install_tokens();
+ ASSERT_TRUE(_tokens.front()
+ ->submit_func([this] {
+ _worker_started.count_down();
+ _release_worker.wait();
+ return Status::InternalError("late bitmap failure");
+ })
+ .ok());
+ ASSERT_TRUE(_worker_started.wait_for(std::chrono::seconds(10)));
+ const auto reason = Status::Cancelled("load cancelled first");
+ ASSERT_TRUE(_load_channel->cancel(reason).ok());
+ _release_worker.count_down();
+ EXPECT_EQ(_tokens.front()->wait(), reason);
+}
+
+INSTANTIATE_TEST_SUITE_P(LocalCloud, DeltaWriterCancelTest, testing::Range(0, 2));
+
+} // namespace doris
diff --git a/be/test/load/memtable/memtable_memory_limiter_test.cpp b/be/test/load/memtable/memtable_memory_limiter_test.cpp
index 551010e0709..ccfd16e2734 100644
--- a/be/test/load/memtable/memtable_memory_limiter_test.cpp
+++ b/be/test/load/memtable/memtable_memory_limiter_test.cpp
@@ -20,6 +20,7 @@
#include "gtest/gtest_pred_impl.h"
#include "io/fs/local_file_system.h"
#include "load/delta_writer/delta_writer.h"
+#include "load/memtable/memtable_flush_executor.h"
#include "runtime/descriptor_helper.h"
#include "runtime/descriptors.h"
#include "storage/storage_engine.h"
@@ -182,4 +183,94 @@ TEST_F(MemTableMemoryLimiterTest, handle_memtable_flush_test) {
res = _engine_ref->tablet_manager()->drop_tablet(request.tablet_id, request.replica_id, false);
EXPECT_EQ(Status::OK(), res);
}
+
+class CancelledMemTableMemoryLimiterTest : public MemTableMemoryLimiterTest,
+ public testing::WithParamInterface<bool> {};
+
+TEST_P(CancelledMemTableMemoryLimiterTest, ReclaimActiveMemoryWhileWriterIsRetained) {
+ RuntimeProfile profile("CancelledMemTableWriter");
+ TCreateTabletReq tablet_request;
+ create_tablet_request(10001, 270068373, &tablet_request);
+ ASSERT_TRUE(_engine_ref->create_tablet(tablet_request, &profile).ok());
+
+ TDescriptorTable tdesc_tbl = create_descriptor_tablet();
+ ObjectPool obj_pool;
+ DescriptorTbl* desc_tbl = nullptr;
+ ASSERT_TRUE(DescriptorTbl::create(&obj_pool, tdesc_tbl, &desc_tbl).ok());
+ auto* tuple_desc = desc_tbl->get_tuple_descriptor(0);
+ auto cancel_status = std::make_shared<AtomicStatus>();
+ WriteRequest write_req;
+ write_req.tablet_id = tablet_request.tablet_id;
+ write_req.schema_hash = tablet_request.tablet_schema.schema_hash;
+ write_req.txn_id = 20003;
+ write_req.partition_id = tablet_request.partition_id;
+ write_req.tuple_desc = tuple_desc;
+ write_req.slots = &tuple_desc->slots();
+ write_req.table_schema_param = std::make_shared<OlapTableSchemaParam>();
+ write_req.load_cancel_status = cancel_status;
+ auto delta_writer =
+ std::make_unique<DeltaWriter>(*_engine_ref, write_req, &profile, TUniqueId {});
+ Block block;
+ for (const auto* slot : tuple_desc->slots()) {
+ block.insert(ColumnWithTypeAndName(slot->get_empty_mutable_column(), slot->type(),
+ slot->col_name()));
+ }
+ auto columns = std::move(block).mutate_columns();
+ for (auto& column : columns) {
+ column->insert_default();
+ }
+ block.set_columns(std::move(columns));
+ ASSERT_TRUE(delta_writer->write(&block, {0}).ok());
+ auto writer = delta_writer->_memtable_writer;
+ std::weak_ptr<MemTable> memtable = writer->_mem_table;
+ auto token = writer->_flush_token;
+ auto active_memory = writer->active_memtable_mem_consumption();
+ ASSERT_GT(active_memory, 0);
+ auto segment_num = writer->_segment_num;
+
+ // add_batch failures can publish a non-CANCELLED status to the shared load.
+ const auto failure = Status::InternalError("load failed");
+ ASSERT_TRUE(cancel_status->update(failure));
+ EXPECT_FALSE(writer->_is_cancelled);
+
+ auto* limiter = ExecEnv::GetInstance()->memtable_memory_limiter();
+ ASSERT_TRUE(limiter->init(100).ok());
+ limiter->refresh_mem_tracker();
+ ASSERT_GT(limiter->mem_usage(), limiter->_load_hard_mem_limit);
+ if (GetParam()) {
+ // An unrelated load must leave the hard-limit wait after reclaiming the
+ // cancelled load's memory, without waiting for its retained writer.
+ // Bound retries so a regression fails instead of hanging the test.
+ int cancel_checks = 0;
+ limiter->handle_memtable_flush([&]() { return ++cancel_checks > 1; });
+ EXPECT_EQ(cancel_checks, 1);
+ } else {
+ // Discarded memory must not be reported as flushed memory.
+ EXPECT_EQ(limiter->_flush_active_memtables(active_memory), 0);
+ }
+ EXPECT_TRUE(memtable.expired());
+ EXPECT_EQ(writer->_mem_table, nullptr);
+ EXPECT_FALSE(writer->_is_cancelled);
+ EXPECT_FALSE(token->_is_shutdown());
+ EXPECT_EQ(writer->_segment_num, segment_num);
+ EXPECT_TRUE(writer->_freezed_mem_tables.empty());
+ EXPECT_EQ(writer->active_memtable_mem_consumption(), 0);
+ EXPECT_EQ(limiter->mem_usage(), 0);
+ EXPECT_TRUE(limiter->_active_writers.empty());
+
+ // An RPC may have passed DeltaWriter's cancellation check before reclamation.
+ // The memtable-level checks must reject it before dereferencing _mem_table.
+ EXPECT_EQ(writer->write(&block, {0}, nullptr), failure);
+ EXPECT_EQ(writer->close(), failure);
+ EXPECT_TRUE(writer->flush_async().is<ErrorCode::CANCELLED>());
+ EXPECT_EQ(writer->_segment_num, segment_num);
+ EXPECT_EQ(writer->_mem_table, nullptr);
+
+ // Reclaiming active memory must not suppress the final owner's token cleanup.
+ delta_writer.reset();
+ EXPECT_TRUE(writer->_is_cancelled);
+ EXPECT_TRUE(token->_is_shutdown());
+}
+
+INSTANTIATE_TEST_SUITE_P(PressureFlush, CancelledMemTableMemoryLimiterTest, testing::Bool());
} // namespace doris
diff --git a/be/test/service/internal_service_load_work_pool_test.cpp b/be/test/service/internal_service_load_work_pool_test.cpp
new file mode 100644
index 00000000000..429cfccbd57
--- /dev/null
+++ b/be/test/service/internal_service_load_work_pool_test.cpp
@@ -0,0 +1,175 @@
+// 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 <gtest/gtest.h>
+
+#include <atomic>
+#include <future>
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "common/config.h"
+#include "load/channel/load_stream_mgr.h"
+#include "runtime/exec_env.h"
+#include "service/internal_service.h"
+
+namespace doris {
+namespace {
+
+// Hold every worker so routing and queue rejection can be checked without running
+// storage handlers. Discard queued RPCs before releasing the worker at teardown.
+class PausedLoadRpcPool {
+public:
+ explicit PausedLoadRpcPool(FifoThreadPool& pool) : _pool(pool) {
+ auto resume = _resume.get_future().share();
+ for (size_t i = 0; i < _pool._threads.size(); ++i) {
+ auto started = std::make_shared<std::promise<void>>();
+ auto ready = started->get_future();
+ CHECK(_pool.try_offer([started, resume]() {
+ started->set_value();
+ resume.wait();
+ }));
+ ready.wait();
+ }
+ }
+
+ ~PausedLoadRpcPool() {
+ _pool.shutdown();
+ _resume.set_value();
+ _pool.join();
+ }
+
+private:
+ FifoThreadPool& _pool;
+ std::promise<void> _resume;
+};
+
+class LoadRpcCountingClosure : public google::protobuf::Closure {
+public:
+ void Run() override { ++calls; }
+ std::atomic<int> calls {0};
+};
+
+} // namespace
+
+class InternalServiceLoadWorkPoolTest : public testing::TestWithParam<int> {
+protected:
+ void SetUp() override {
+ // Keep pools and queues small and restore configuration after each test.
+ for (auto* setting :
+ {&config::brpc_heavy_work_pool_threads, &config::brpc_heavy_work_pool_max_queue_size,
+ &config::brpc_light_work_pool_threads, &config::brpc_light_work_pool_max_queue_size,
+ &config::brpc_peer_fetch_pool_threads, &config::brpc_peer_fetch_pool_max_queue_size,
+ &config::brpc_arrow_flight_work_pool_threads,
+ &config::brpc_arrow_flight_work_pool_max_queue_size,
+ &config::brpc_load_light_work_pool_threads,
+ &config::brpc_load_light_work_pool_max_queue_size}) {
+ _saved_config.emplace_back(setting, *setting);
+ *setting = 1;
+ }
+ // Use a non-default value to verify that the cancellation pool honors configuration.
+ config::brpc_load_light_work_pool_threads = 3;
+ _exec_env._load_stream_mgr = std::make_unique<LoadStreamMgr>(1);
+ _service = std::make_unique<PInternalService>(&_exec_env);
+ for (auto* pool : {&_service->_heavy_work_pool, &_service->_light_work_pool,
+ &_service->_load_light_work_pool}) {
+ _paused_pools.push_back(std::make_unique<PausedLoadRpcPool>(*pool));
+ }
+ }
+
+ void TearDown() override {
+ _paused_pools.clear();
+ _exec_env.load_stream_mgr()->set_heavy_work_pool(nullptr);
+ _service.reset();
+ _exec_env._load_stream_mgr.reset();
+ for (const auto& [setting, value] : _saved_config) {
+ *setting = value;
+ }
+ }
+
+ ExecEnv _exec_env;
+ std::unique_ptr<PInternalService> _service;
+ std::vector<std::unique_ptr<PausedLoadRpcPool>> _paused_pools;
+ std::vector<std::pair<int32_t*, int32_t>> _saved_config;
+};
+
+TEST_F(InternalServiceLoadWorkPoolTest, CancelBypassesFullHeavyPool) {
+ EXPECT_EQ(_service->_load_light_work_pool.get_active_threads(), 3);
+ ASSERT_TRUE(_service->_heavy_work_pool.try_offer([] {}));
+
+ PTabletWriterCancelRequest request;
+ PTabletWriterCancelResult response;
+ LoadRpcCountingClosure done;
+ _service->tablet_writer_cancel(nullptr, &request, &response, &done);
+ EXPECT_EQ(done.calls.load(), 0);
+ EXPECT_EQ(_service->_load_light_work_pool.get_queue_size(), 1);
+ EXPECT_EQ(_service->_light_work_pool.get_queue_size(), 0);
+
+ // Cancel's protobuf response is empty; queue rejection must still run the closure once.
+ _service->tablet_writer_cancel(nullptr, &request, &response, &done);
+ EXPECT_EQ(done.calls.load(), 1);
+}
+
+TEST_P(InternalServiceLoadWorkPoolTest, OpenAndAddBlockKeepUsingHeavyPool) {
+ ASSERT_TRUE(_service->_load_light_work_pool.try_offer([] {}));
+
+ PTabletWriterOpenRequest open_request;
+ PTabletWriterOpenResult open_response;
+ POpenLoadStreamRequest stream_request;
+ POpenLoadStreamResponse stream_response;
+ PTabletWriterAddBlockRequest block_request;
+ PTabletWriterAddBlockResult block_response;
+ LoadRpcCountingClosure done;
+ auto submit = [&]() {
+ switch (GetParam()) {
+ case 0:
+ _service->tablet_writer_open(nullptr, &open_request, &open_response, &done);
+ break;
+ case 1:
+ _service->open_load_stream(nullptr, &stream_request, &stream_response, &done);
+ break;
+ case 2:
+ _service->tablet_writer_add_block(nullptr, &block_request, &block_response, &done);
+ break;
+ }
+ };
+
+ submit();
+ EXPECT_EQ(done.calls.load(), 0);
+ EXPECT_EQ(_service->_heavy_work_pool.get_queue_size(), 1);
+ EXPECT_EQ(_service->_light_work_pool.get_queue_size(), 0);
+
+ submit();
+ EXPECT_EQ(done.calls.load(), 1);
+ const auto& status = GetParam() == 0 ? open_response.status()
+ : GetParam() == 1 ? stream_response.status()
+ : block_response.status();
+ EXPECT_EQ(status.status_code(), TStatusCode::CANCELLED);
+ ASSERT_EQ(status.error_msgs_size(), 1);
+ EXPECT_NE(status.error_msgs(0).find("brpc_heavy"), std::string::npos);
+}
+
+INSTANTIATE_TEST_SUITE_P(HeavyLoadRequests, InternalServiceLoadWorkPoolTest,
+ testing::Values(0, 1, 2));
+
+TEST_F(InternalServiceLoadWorkPoolTest, StreamingCloseKeepsUsingHeavyPool) {
+ EXPECT_EQ(_exec_env.load_stream_mgr()->heavy_work_pool(), &_service->_heavy_work_pool);
+ EXPECT_NE(_exec_env.load_stream_mgr()->heavy_work_pool(), &_service->_load_light_work_pool);
+}
+
+} // namespace doris
diff --git a/be/test/storage/delete/calc_delete_bitmap_executor_test.cpp b/be/test/storage/delete/calc_delete_bitmap_executor_test.cpp
new file mode 100644
index 00000000000..988dbb9cc0f
--- /dev/null
+++ b/be/test/storage/delete/calc_delete_bitmap_executor_test.cpp
@@ -0,0 +1,152 @@
+// 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 "storage/delete/calc_delete_bitmap_executor.h"
+
+#include <gtest/gtest.h>
+
+#include <chrono>
+#include <future>
+
+#include "cpp/sync_point.h"
+#include "runtime/memory/mem_tracker_limiter.h"
+#include "runtime/thread_context.h"
+#include "runtime/workload_management/resource_context.h"
+#include "util/countdown_latch.h"
+#include "util/defer_op.h"
+
+namespace doris {
+
+TEST(CalcDeleteBitmapTokenTest, QueuedCallbacksKeepTheirSubmitterResourceContext) {
+ SCOPED_INIT_THREAD_CONTEXT();
+ std::unique_ptr<ThreadPool> pool;
+ ASSERT_TRUE(ThreadPoolBuilder("BitmapResourceContextTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&pool)
+ .ok());
+ CalcDeleteBitmapToken token(pool->new_token(ThreadPool::ExecutionMode::CONCURRENT));
+ CountDownLatch entered(1);
+ CountDownLatch release(1);
+ Defer cleanup {[&] {
+ release.count_down();
+ pool->wait();
+ }};
+ ASSERT_TRUE(pool->submit_func([&] {
+ entered.count_down();
+ release.wait();
+ }).ok());
+ ASSERT_TRUE(entered.wait_for(std::chrono::seconds(10)));
+
+ // Queue callbacks from different submitters before allowing either to run.
+ // A token-wide context would incorrectly attach the last submitter to both.
+ for (const auto* label : {"BitmapSubmitter1", "BitmapSubmitter2"}) {
+ auto resource_ctx = ResourceContext::create_shared();
+ resource_ctx->memory_context()->set_mem_tracker(
+ MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER, label));
+ SCOPED_ATTACH_TASK(resource_ctx);
+ ASSERT_TRUE(token.submit_func([resource_ctx] {
+ EXPECT_EQ(thread_context()->resource_ctx(), resource_ctx);
+ return Status::OK();
+ }).ok());
+ }
+ release.count_down();
+ EXPECT_TRUE(token.wait().ok());
+}
+
+TEST(CalcDeleteBitmapTokenTest, PreservesTaskFailure) {
+ SCOPED_INIT_THREAD_CONTEXT();
+ CalcDeleteBitmapExecutor executor;
+ executor.init("BitmapFailureTest", 1);
+ auto token = executor.create_token();
+ const auto failure = Status::InternalError("bitmap calculation failed");
+ ASSERT_TRUE(token->submit_func([&] { return failure; }).ok());
+ EXPECT_EQ(token->wait().to_string(), failure.to_string());
+ EXPECT_EQ(token->submit_func([] { return Status::OK(); }).to_string(), failure.to_string());
+}
+
+TEST(CalcDeleteBitmapTokenTest, CreatedAfterCancellationRejectsEverySubmissionEntry) {
+ SCOPED_INIT_THREAD_CONTEXT();
+ CalcDeleteBitmapExecutor executor;
+ executor.init("BitmapLateTokenTest", 1);
+ auto status = std::make_shared<AtomicStatus>();
+ const auto reason = Status::Cancelled("load already cancelled");
+ status->update(reason);
+ auto token = executor.create_token(status);
+ EXPECT_EQ(token->submit_func([] { return Status::OK(); }), reason);
+ // Neither typed callback may dereference its inputs after cancellation.
+ EXPECT_EQ(token->submit(nullptr, nullptr, nullptr, {}, 0, nullptr, nullptr, nullptr), reason);
+ EXPECT_EQ(token->submit(nullptr, nullptr, RowsetId {}, {}, nullptr), reason);
+ EXPECT_EQ(token->wait(), reason);
+}
+
+TEST(CalcDeleteBitmapTokenTest, FailedSubmitKeepsCancellationReason) {
+ SCOPED_INIT_THREAD_CONTEXT();
+ CalcDeleteBitmapExecutor executor;
+ executor.init("BitmapSubmitCancelRaceTest", 1);
+ auto status = std::make_shared<AtomicStatus>();
+ auto token = executor.create_token(status);
+ const auto reason = Status::Cancelled("cancelled between check and submit");
+ auto* sync_point = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ Defer cleanup {[&] { sync_point->disable_processing(); }};
+ sync_point->set_call_back(
+ "CalcDeleteBitmapToken::submit_func:before_submit",
+ [&](auto&& args) {
+ if (try_any_cast<CalcDeleteBitmapToken*>(args[0]) == token.get()) {
+ status->update(reason);
+ token->cancel();
+ }
+ },
+ &guard);
+ sync_point->enable_processing();
+ EXPECT_EQ(token->submit_func([] { return Status::OK(); }), reason);
+ EXPECT_EQ(token->wait(), reason);
+}
+
+TEST(CalcDeleteBitmapTokenTest, DestructionWaitsForRunningCallback) {
+ SCOPED_INIT_THREAD_CONTEXT();
+ CalcDeleteBitmapExecutor executor;
+ executor.init("BitmapDestructionTest", 1);
+ auto token = executor.create_token();
+ CountDownLatch entered(1);
+ CountDownLatch release(1);
+ CountDownLatch destroying(1);
+ Defer cleanup {[&] {
+ release.count_down();
+ if (token) {
+ token->cancel();
+ }
+ }};
+ ASSERT_TRUE(token->submit_func([&] {
+ entered.count_down();
+ release.wait();
+ // The wrapper still accesses the token's status after this callback.
+ return Status::InternalError("callback failed during destruction");
+ }).ok());
+ ASSERT_TRUE(entered.wait_for(std::chrono::seconds(10)));
+ auto destructor = std::async(std::launch::async, [&, owned = std::move(token)]() mutable {
+ destroying.count_down();
+ owned.reset();
+ });
+ EXPECT_TRUE(destroying.wait_for(std::chrono::seconds(10)));
+ EXPECT_EQ(destructor.wait_for(std::chrono::milliseconds(50)), std::future_status::timeout);
+ release.count_down();
+ destructor.get();
+}
+
+} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org