This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 6f3af1a9d56 [enhance](load) Support lock-free load cancellation
(#67759)
6f3af1a9d56 is described below
commit 6f3af1a9d56a6ff2c70f4dabfd703f5e71f56451
Author: hui lai <[email protected]>
AuthorDate: Wed Oct 14 17:27:40 2026 +0800
[enhance](load) Support lock-free load cancellation (#67759)
### What problem does this PR solve?
Support lock-free load cancellation by publishing the load cancellation
status without acquiring channel or writer locks or synchronously
traversing writers. Previously, close could hold those locks while
waiting for flush or delete bitmap work, delaying cancellation.
A shared per-load `AtomicStatus` separates cancellation publication from
resource cleanup. Local/cloud write and close paths observe the status
at phase boundaries, and pending asynchronous EOS RPCs are completed
during final-owner cleanup. Flush and delete bitmap tasks cooperate with
that status; final owners drain running work before reclaiming
resources.
The same status is propagated through ordinary and incremental opens,
local/cloud writers, group children, rowset-writer contexts, and both
delete bitmap executors. Queued memtable flush tasks check it before
flushing. Bitmap tokens reject submissions when cancellation is
observable and check again when a queued callback is dispatched,
skipping the calculation after cancellation. Each callback owns its
submitter's resource context.
Final-owner cleanup first completes pending EOS RPC callbacks before
destroying writers. The final sender checks the shared cancellation
status before publishing EOS arrival and again before flushing;
successful arrival is not revoked by later cancellation. Cleanup then
stops and drains memtable flush producers, then cancels builder and
rowset-writer bitmap tokens. Token cancellation discards remaining
queued callbacks and waits for running callbacks before builder
rollback, cloud cache cleanup, or local file cleanup. This also covers
group children and initialization that created bitmap work before
memtable initialization failed. The local rowset writer drains bitmap
callbacks before its derived state is destroyed.
Earlier token calculation failures remain local to that token. A
callback failure observed after load cancellation preserves the
cancellation reason. Failed submissions recheck the effective status
instead of replacing cancellation with a generic thread-pool shutdown
error.
The existing manager changes retain cancellation reasons for failed
requests and explicit cancellation, protect cached failures from stale
successful EOS transitions, and compare channel instances before
removing mappings. Cached reasons preserve the original message,
including empty failure messages.
### Cancellation contract and limits
- **Cancel:** publish the shared status; no bitmap registry, token
traversal, or shutdown is added to this publication path.
- **Submit bitmap:** reject work when cancellation is observed.
- **Dispatch queued bitmap:** check the shared status and skip the
calculation.
- **Final-owner cleanup:** cancel tokens, discard pending callbacks, and
wait for running callbacks before reclaiming resources.
This does not proactively remove queued tasks when cancellation is
published, wake bitmap waiters immediately, or interrupt
calculations/I/O that have already started. A retained close can still
wait for queued callbacks to be dispatched. Final-owner destruction can
still extend cancel RPC completion. `tablet_writer_cancel` uses master's
dedicated load-light work pool. Batched queue removal is deferred.
Related: #67778. This PR integrates bitmap status checks and cleanup
with shared-status publication; it does not introduce the load-wide
synchronous token-shutdown coordinator.
### Release note
Load cancellation publishes a shared status without waiting for channel
or writer locks held by close. Queued flush and delete bitmap work
observes cancellation, new bitmap submissions are rejected, and final
writer cleanup discards pending bitmap callbacks while safely draining
running work.
---
be/src/cloud/cloud_delta_writer.cpp | 6 +-
be/src/cloud/cloud_rowset_builder.cpp | 13 +-
be/src/cloud/cloud_rowset_builder.h | 2 +
be/src/cloud/cloud_rowset_writer.cpp | 3 +-
be/src/cloud/cloud_tablets_channel.cpp | 1 +
be/src/load/channel/load_channel.cpp | 45 +-
be/src/load/channel/load_channel.h | 10 +-
be/src/load/channel/load_channel_mgr.cpp | 107 ++--
be/src/load/channel/load_channel_mgr.h | 7 +-
be/src/load/channel/tablets_channel.cpp | 25 +-
be/src/load/channel/tablets_channel.h | 9 +
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 | 26 +-
.../storage/delete/calc_delete_bitmap_executor.cpp | 71 ++-
.../storage/delete/calc_delete_bitmap_executor.h | 46 +-
be/src/storage/rowset/beta_rowset_writer.cpp | 25 +-
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 | 18 +-
be/src/storage/rowset_builder.h | 4 +-
be/test/load/channel/load_channel_eos_test.cpp | 19 +-
be/test/load/channel/load_channel_mgr_test.cpp | 274 ++++++++++
.../delta_writer_cancel_integration_test.cpp | 591 +++++++++++++++++++++
.../load/delta_writer/delta_writer_cancel_test.cpp | 295 ++++++++++
.../load/memtable/memtable_flush_executor_test.cpp | 119 +++++
.../load/memtable/memtable_memory_limiter_test.cpp | 91 ++++
.../delete/calc_delete_bitmap_executor_test.cpp | 152 ++++++
33 files changed, 1905 insertions(+), 146 deletions(-)
diff --git a/be/src/cloud/cloud_delta_writer.cpp
b/be/src/cloud/cloud_delta_writer.cpp
index 19eaef0f00a..83553e176cd 100644
--- a/be/src/cloud/cloud_delta_writer.cpp
+++ b/be/src/cloud/cloud_delta_writer.cpp
@@ -93,7 +93,7 @@ Status CloudDeltaWriter::write(const Block* block, const
TabletAddRowsPayload& r
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());
}
@@ -115,6 +115,7 @@ Status CloudDeltaWriter::write(const Block* block, const
TabletAddRowsPayload& r
return _memtable_writer->flush_running_count() >=
effective_flush_running_count_limit;
};
while (need_backpressure()) {
+ RETURN_IF_ERROR(_get_load_cancel_status());
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
@@ -154,6 +155,7 @@ void CloudDeltaWriter::update_tablet_stats() {
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) {
@@ -173,6 +175,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) {
@@ -183,6 +186,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 d9065c340ec..5660fb6bbb3 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/rowset/group_rowset_writer.h"
#include "storage/rowset/rowset_factory.h"
@@ -53,6 +54,8 @@
CloudGroupRowsetBuilder::CloudGroupRowsetBuilder(CloudStorageEngine& engine,
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();
}
}
@@ -89,6 +92,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.db_id = _req.table_schema_param->db_id();
context.table_id = _req.table_schema_param->table_id();
context.rowset_state = PREPARED;
@@ -117,7 +121,8 @@ Status CloudRowsetBuilder::init() {
_rowset_writer = DORIS_TRY(_tablet->create_rowset_writer(context, false));
_rowset_id = 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);
if (!_skip_writing_rowset_metadata) {
RETURN_IF_ERROR(_engine.meta_mgr().prepare_rowset(*_rowset_writer->rowset_meta(),
"",
@@ -172,6 +177,12 @@ Status CloudGroupRowsetBuilder::wait_calc_delete_bitmap() {
return _data_builder->wait_calc_delete_bitmap();
}
+Status CloudGroupRowsetBuilder::cancel(const Status& st) {
+ RETURN_IF_ERROR(_data_builder->cancel(st));
+ RETURN_IF_ERROR(_row_binlog_builder->cancel(st));
+ return BaseRowsetBuilder::cancel(st);
+}
+
void CloudGroupRowsetBuilder::update_tablet_stats() {
_data_builder->update_tablet_stats();
_row_binlog_builder->update_tablet_stats();
diff --git a/be/src/cloud/cloud_rowset_builder.h
b/be/src/cloud/cloud_rowset_builder.h
index c65e1924448..8d70a36fd64 100644
--- a/be/src/cloud/cloud_rowset_builder.h
+++ b/be/src/cloud/cloud_rowset_builder.h
@@ -74,6 +74,8 @@ public:
Status wait_calc_delete_bitmap() override;
+ Status cancel(const Status& st) override;
+
void update_tablet_stats() override;
Status commit_rowset(const std::string& job_id, int64_t table_id) override;
diff --git a/be/src/cloud/cloud_rowset_writer.cpp
b/be/src/cloud/cloud_rowset_writer.cpp
index 9753ca3dd99..20aa151a341 100644
--- a/be/src/cloud/cloud_rowset_writer.cpp
+++ b/be/src/cloud/cloud_rowset_writer.cpp
@@ -89,7 +89,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 28caf74f79d..4d9f76a145d 100644
--- a/be/src/cloud/cloud_tablets_channel.cpp
+++ b/be/src/cloud/cloud_tablets_channel.cpp
@@ -205,6 +205,7 @@ Status
CloudTabletsChannel::_close_writers(PTabletWriterAddBlockResult* res) {
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/load/channel/load_channel.cpp
b/be/src/load/channel/load_channel.cpp
index e1066347c87..7cf31092087 100644
--- a/be/src/load/channel/load_channel.cpp
+++ b/be/src/load/channel/load_channel.cpp
@@ -20,8 +20,6 @@
#include <gen_cpp/internal_service.pb.h>
#include <glog/logging.h>
-#include <vector>
-
#include "cloud/cloud_tablets_channel.h"
#include "cloud/config.h"
#include "common/logging.h"
@@ -44,6 +42,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)),
@@ -82,6 +81,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) {
@@ -142,6 +150,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> lt(_tablets_channels_lock);
_tablets_channels.insert({index_id, channel});
@@ -184,6 +193,7 @@ Status
LoadChannel::_get_tablets_channel(std::shared_ptr<BaseTabletsChannel>& ch
Status LoadChannel::add_batch(const PTabletWriterAddBlockRequest& request,
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);
@@ -240,6 +250,7 @@ Status LoadChannel::_handle_eos(BaseTabletsChannel* channel,
// errors belong to this RPC and cannot revoke earlier EOS responses.
RETURN_IF_ERROR(channel->close(this, request, response, &finished));
+ RETURN_IF_ERROR(cancel_status());
if (finished) {
std::lock_guard<std::mutex> l(_lock);
{
@@ -297,26 +308,18 @@ bool LoadChannel::is_finished() {
return _tablets_channels.empty();
}
-Status LoadChannel::cancel() {
- _cancelled.store(true);
- std::vector<std::shared_ptr<BaseTabletsChannel>> channels;
- {
- std::lock_guard<std::mutex> l(_lock);
- channels.reserve(_tablets_channels.size());
- for (auto& [id, channel] : _tablets_channels) {
- channels.push_back(channel);
- }
- }
- // Complete pending RPCs before waiting for writer cancellation. No
callback
- // may run under _lock, including when the timeout cleaner calls cancel().
- for (auto& channel : channels) {
- channel->eos_completion()->complete(
- Status::Cancelled("Load channel {} cancelled",
_load_id.to_string()));
- }
- for (auto& channel : channels) {
- static_cast<void>(channel->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 108abeedac4..be3ece6d570 100644
--- a/be/src/load/channel/load_channel.h
+++ b/be/src/load/channel/load_channel.h
@@ -41,6 +41,7 @@ class PTabletWriterAddBlockRequest;
class PTabletWriterAddBlockResult;
class OpenPartitionRequest;
class BaseTabletsChannel;
+class AtomicStatus;
// A LoadChannel manages tablets channels for all indexes
// corresponding to a certain load job
@@ -64,7 +65,10 @@ public:
// 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(); }
@@ -74,7 +78,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(); }
@@ -117,7 +121,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 5682edeb99e..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);
@@ -178,24 +183,45 @@ Status LoadChannelMgr::add_batch(const
PTabletWriterAddBlockRequest& request,
// this case will be handled in load channel's add batch method.
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);
}
@@ -204,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",
@@ -276,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 255f8b1c223..1c06fded153 100644
--- a/be/src/load/channel/load_channel_mgr.h
+++ b/be/src/load/channel/load_channel_mgr.h
@@ -82,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();
@@ -90,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 ab540459727..01e19042245 100644
--- a/be/src/load/channel/tablets_channel.cpp
+++ b/be/src/load/channel/tablets_channel.cpp
@@ -90,6 +90,14 @@ 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,
bool& should_write) {
@@ -138,8 +146,9 @@ void BaseTabletsChannel::_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();
@@ -198,6 +207,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) {
@@ -238,6 +248,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;
@@ -357,11 +368,15 @@ Status
BaseTabletsChannel::_close_and_notify(std::unique_lock<std::mutex>& lock,
_state = kFinished;
_final_sender_id = sender_id;
_final_close_in_progress = true;
+ const auto arrival_status = _check_cancelled();
lock.unlock();
- _eos_completion->complete(Status::OK());
+ _eos_completion->complete(arrival_status);
lock.lock();
- _close_status = _close_writers(response);
+ _close_status = _check_cancelled();
+ if (_close_status.ok()) {
+ _close_status = _close_writers(response);
+ }
_final_close_result =
std::make_unique<PTabletWriterAddBlockResult>(*response);
_final_close_in_progress = false;
_final_close_cv.notify_all();
@@ -424,6 +439,7 @@ Status
TabletsChannel::_close_writers(PTabletWriterAddBlockResult* res) {
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()) {
@@ -620,6 +636,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,
};
if (tablet.has_binlog_tablet_id()) {
wrequest.binlog_tablet_id = tablet.binlog_tablet_id();
diff --git a/be/src/load/channel/tablets_channel.h
b/be/src/load/channel/tablets_channel.h
index 51a1de5c7db..0bb734cf3fd 100644
--- a/be/src/load/channel/tablets_channel.h
+++ b/be/src/load/channel/tablets_channel.h
@@ -113,6 +113,11 @@ 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();
@@ -127,6 +132,9 @@ 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(std::unique_lock<std::mutex>& lock, int sender_id,
@@ -181,6 +189,7 @@ protected:
State _state;
UniqueId _load_id;
+ std::shared_ptr<AtomicStatus> _load_cancel_status;
// initialized in open function
int64_t _txn_id = -1;
diff --git a/be/src/load/delta_writer/delta_writer.cpp
b/be/src/load/delta_writer/delta_writer.cpp
index abcda75afa4..84504805e1e 100644
--- a/be/src/load/delta_writer/delta_writer.cpp
+++ b/be/src/load/delta_writer/delta_writer.cpp
@@ -31,6 +31,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"
@@ -93,13 +94,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);
@@ -136,15 +143,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,
@@ -166,7 +182,7 @@ Status DeltaWriter::write(const Block* block, const
TabletAddRowsPayload& rows,
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());
}
@@ -182,6 +198,7 @@ Status DeltaWriter::write(const Block* block, const
TabletAddRowsPayload& rows,
config::memtable_flush_running_count_limit *
(_req.write_req_type == WriteRequestType::GROUP ? 2 : 1);
while (_memtable_writer->flush_running_count() >=
effective_flush_running_count_limit) {
+ RETURN_IF_ERROR(_get_load_cancel_status());
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
@@ -221,6 +238,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();
}
@@ -232,6 +250,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();
}
@@ -242,6 +261,7 @@ Status BaseDeltaWriter::wait_calc_delete_bitmap() {
Status DeltaWriter::commit_txn() {
std::lock_guard<std::mutex> l(_lock);
SCOPED_TIMER(_commit_txn_timer);
+ RETURN_IF_ERROR(_get_load_cancel_status());
return _rowset_builder->commit_txn();
}
@@ -254,6 +274,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 4bed76d80cd..a405349161a 100644
--- a/be/src/load/delta_writer/delta_writer.h
+++ b/be/src/load/delta_writer/delta_writer.h
@@ -70,7 +70,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 06b4659c79b..88de1a171bf 100644
--- a/be/src/load/delta_writer/delta_writer_context.h
+++ b/be/src/load/delta_writer/delta_writer_context.h
@@ -30,6 +30,7 @@ namespace doris {
class TupleDescriptor;
class SlotDescriptor;
class OlapTableSchemaParam;
+class AtomicStatus;
enum class WriteRequestType {
DATA = 0, // data write
@@ -56,6 +57,8 @@ struct WriteRequest {
WriteRequestType write_req_type = WriteRequestType::DATA;
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;
};
struct TabletAddRowsPayload {
diff --git a/be/src/load/memtable/memtable_flush_executor.cpp
b/be/src/load/memtable/memtable_flush_executor.cpp
index 2ad61661d4f..6345e5121e1 100644
--- a/be/src/load/memtable/memtable_flush_executor.cpp
+++ b/be/src/load/memtable/memtable_flush_executor.cpp
@@ -161,6 +161,11 @@ Status FlushToken::_submit_sub_tasks(ThreadPool* pool,
return Status::OK();
}
+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);
@@ -259,7 +264,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,
@@ -356,7 +361,7 @@ void FlushToken::_flush_memtable_impl(RowsetWriter*
flush_writer, MemTable* memt
}};
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",
@@ -406,6 +411,7 @@ void FlushToken::_flush_memtable_impl(RowsetWriter*
flush_writer, MemTable* memt
// }};
std::shared_ptr<Block> flush_block;
RETURN_IF_ERROR(_memtable2block(memtable, shared_memtable,
flush_block));
+ RETURN_IF_ERROR(_get_load_cancel_status());
RETURN_IF_ERROR(
flush_writer->flush_memtable(flush_block.get(),
segment_id, &flush_size));
memtable->set_flush_success();
@@ -518,7 +524,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<OlapTableSchemaParam> table_schema_param) {
+ std::shared_ptr<OlapTableSchemaParam> table_schema_param,
+ 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.
@@ -526,7 +533,7 @@ Status MemTableFlushExecutor::create_flush_token(
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);
flush_token->set_table_schema_param(std::move(table_schema_param));
return Status::OK();
diff --git a/be/src/load/memtable/memtable_flush_executor.h
b/be/src/load/memtable/memtable_flush_executor.h
index c3053d483e5..3c707439b00 100644
--- a/be/src/load/memtable/memtable_flush_executor.h
+++ b/be/src/load/memtable/memtable_flush_executor.h
@@ -99,13 +99,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.
@@ -125,6 +128,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();
@@ -155,6 +160,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;
@@ -198,7 +204,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<OlapTableSchemaParam>
table_schema_param = nullptr);
+ std::shared_ptr<OlapTableSchemaParam>
table_schema_param = nullptr,
+ 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 f98aea05f71..d32ed37d18d 100644
--- a/be/src/load/memtable/memtable_memory_limiter.cpp
+++ b/be/src/load/memtable/memtable_memory_limiter.cpp
@@ -282,9 +282,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();
@@ -298,6 +299,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, "
@@ -312,7 +319,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 09ab384b847..f05e5a5cde4 100644
--- a/be/src/load/memtable/memtable_writer.cpp
+++ b/be/src/load/memtable/memtable_writer.cpp
@@ -114,7 +114,7 @@ Status MemTableWriter::init(std::shared_ptr<RowsetWriter>
rowset_writer,
RETURN_IF_ERROR(
ExecEnv::GetInstance()->storage_engine().memtable_flush_executor()->create_flush_token(
_flush_token, _rowset_writer, _req.is_high_priority,
wg_sptr,
- _req.table_schema_param));
+ _req.table_schema_param, _req.load_cancel_status));
_is_init = true;
return Status::OK();
@@ -132,6 +132,9 @@ Status MemTableWriter::write(const Block* block, const
TabletAddRowsPayload& row
_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;
}
@@ -228,10 +231,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.
@@ -287,6 +300,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/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 265448a8314..70c855901ad 100644
--- a/be/src/storage/rowset/beta_rowset_writer.cpp
+++ b/be/src/storage/rowset/beta_rowset_writer.cpp
@@ -339,7 +339,13 @@ BetaRowsetWriter::BetaRowsetWriter(StorageEngine& engine)
RowBinlogRowsetWriter::RowBinlogRowsetWriter(StorageEngine& engine) :
BetaRowsetWriter(engine) {}
BaseBetaRowsetWriter::~BaseBetaRowsetWriter() {
+ // 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->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) {
@@ -352,12 +358,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
@@ -365,6 +372,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;
DCHECK(_context.tablet_schema != nullptr);
@@ -457,6 +470,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()) {
@@ -497,6 +512,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();
});
}
@@ -507,7 +523,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 49811ac7714..3887919f0da 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 8bb6cb057a1..a1f38ee4768 100644
--- a/be/src/storage/rowset/rowset_writer.h
+++ b/be/src/storage/rowset/rowset_writer.h
@@ -75,6 +75,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 7c9d4add5ce..3a29a8a4b66 100644
--- a/be/src/storage/rowset/rowset_writer_context.h
+++ b/be/src/storage/rowset/rowset_writer_context.h
@@ -91,6 +91,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 2620eedf321..c4b7530ce97 100644
--- a/be/src/storage/rowset_builder.cpp
+++ b/be/src/storage/rowset_builder.cpp
@@ -256,7 +256,8 @@ Status RowsetBuilder::init() {
tmp_pending_rowset_ids.begin() + 1);
_pending_rs_guard =
_engine.pending_local_rowsets().add(tmp_pending_rowset_ids);
- _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();
@@ -267,6 +268,7 @@ Status
BaseRowsetBuilder::_init_context_common_fields(RowsetWriterContext& conte
context.txn_id = _req.txn_id;
context.load_id = _req.load_id;
+ context.load_cancel_status = _req.load_cancel_status;
context.db_id = _req.table_schema_param->db_id();
context.table_id = _req.table_schema_param->table_id();
context.rowset_state = PREPARED;
@@ -413,13 +415,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();
@@ -598,6 +604,12 @@ Status GroupRowsetBuilder::wait_calc_delete_bitmap() {
return _txn_rs_builder->wait_calc_delete_bitmap();
}
+Status GroupRowsetBuilder::cancel(const Status& st) {
+ RETURN_IF_ERROR(_txn_rs_builder->cancel(st));
+ RETURN_IF_ERROR(_row_binlog_rowset_builder->cancel(st));
+ return BaseRowsetBuilder::cancel(st);
+}
+
Status GroupRowsetBuilder::commit_txn() {
// Attach binlog rowset (and its independent binlog tablet) to txn rowset,
so that
// commit/rollback and clean-up are all handled by txn rowset builder.
diff --git a/be/src/storage/rowset_builder.h b/be/src/storage/rowset_builder.h
index 0ff74f5121d..a0eb9f40d54 100644
--- a/be/src/storage/rowset_builder.h
+++ b/be/src/storage/rowset_builder.h
@@ -67,7 +67,7 @@ public:
return Status::NotSupported("BaseRowsetBuilder::commit_txn not
implemented");
}
- Status cancel();
+ virtual Status cancel(const Status& st = Status::Cancelled("rowset builder
cancelled"));
const std::shared_ptr<RowsetWriter>& rowset_writer() const { return
_rowset_writer; }
@@ -206,6 +206,8 @@ public:
Status wait_calc_delete_bitmap() override;
+ Status cancel(const Status& st) override;
+
Status commit_txn() override;
const BaseTabletSPtr& tablet_sptr() const override { return
_txn_rs_builder->tablet_sptr(); }
diff --git a/be/test/load/channel/load_channel_eos_test.cpp
b/be/test/load/channel/load_channel_eos_test.cpp
index ca76dac5c83..624c97ab4b3 100644
--- a/be/test/load/channel/load_channel_eos_test.cpp
+++ b/be/test/load/channel/load_channel_eos_test.cpp
@@ -75,6 +75,7 @@ protected:
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;
@@ -287,7 +288,7 @@ TEST_P(LoadChannelEosTest,
NonWaitingLastSenderReleasesEarlierWaiter) {
EXPECT_EQ(replies, 1);
}
-TEST_P(LoadChannelEosTest, CancelReleasesRegisteredAndNotYetRegisteredRpcs) {
+TEST_P(LoadChannelEosTest,
FinalOwnerReleasesRegisteredAndNotYetRegisteredRpcsAfterCancel) {
auto channel = make_channel(3);
PTabletWriterAddBlockResult response;
std::shared_ptr<EosCompletion> first;
@@ -297,14 +298,17 @@ TEST_P(LoadChannelEosTest,
CancelReleasesRegisteredAndNotYetRegisteredRpcs) {
int replies = 0;
auto callback = [&](const Status& status) {
EXPECT_TRUE(status.is<ErrorCode::CANCELLED>());
- // Re-enter the load lock; cancellation must invoke callbacks outside
it.
- EXPECT_FALSE(_load->is_finished());
+ // 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());
- second->add_waiter(callback);
EXPECT_TRUE(_load->cancel().ok());
+ EXPECT_EQ(replies, 0);
+ _load.reset();
+ EXPECT_EQ(replies, 1);
+ second->add_waiter(callback);
EXPECT_EQ(replies, 2);
}
@@ -322,6 +326,7 @@ TEST_P(LoadChannelEosTest,
TimeoutCleanerReleasesPendingRpc) {
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();
@@ -481,15 +486,15 @@ TEST_P(LoadChannelEosTest,
ArrivalCallbacksAllowConcurrentRetryAndCancel) {
Defer disable_sync_points {[&] { sp->disable_processing(); }};
std::shared_ptr<EosCompletion> last;
- EXPECT_TRUE(_load->add_batch(eos(1), &response, &last).ok());
+ 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, 1);
+ EXPECT_EQ(final_closes, 0);
EXPECT_TRUE(_load->is_cancelled());
- EXPECT_TRUE(_load->is_finished());
+ EXPECT_FALSE(_load->is_finished());
}
TEST_P(LoadChannelEosTest, FinalSenderRetryWaitsForCommitFailure) {
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_integration_test.cpp
b/be/test/load/delta_writer/delta_writer_cancel_integration_test.cpp
new file mode 100644
index 00000000000..56c12708956
--- /dev/null
+++ b/be/test/load/delta_writer/delta_writer_cancel_integration_test.cpp
@@ -0,0 +1,591 @@
+// 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/AgentService_types.h>
+#include <gen_cpp/Descriptors_types.h>
+#include <gen_cpp/Types_types.h>
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <chrono>
+#include <filesystem>
+#include <future>
+#include <memory>
+#include <string>
+#include <thread>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "cloud/cloud_delta_writer.h"
+#include "cloud/cloud_rowset_builder.h"
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablet.h"
+#include "cloud/cloud_tablet_mgr.h"
+#include "cloud/config.h"
+#include "common/config.h"
+#include "core/block/block.h"
+#include "core/field.h"
+#include "cpp/sync_point.h"
+#include "io/fs/local_file_system.h"
+#include "io/fs/remote_file_system.h"
+#include "load/channel/eos_completion.h"
+#include "load/channel/load_channel.h"
+#include "load/channel/tablets_channel.h"
+#include "load/delta_writer/delta_writer.h"
+#include "load/memtable/memtable_flush_executor.h"
+#include "load/memtable/memtable_memory_limiter.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/rowset/beta_rowset_writer.h"
+#include "storage/rowset/rowset_factory.h"
+#include "storage/rowset_builder.h"
+#include "storage/storage_engine.h"
+#include "storage/storage_policy.h"
+#include "storage/tablet/tablet.h"
+#include "storage/tablet/tablet_manager.h"
+#include "storage/tablet/tablet_meta.h"
+#include "storage/tablet_info.h"
+#include "testutil/creators.h"
+#include "util/countdown_latch.h"
+
+namespace doris {
+namespace {
+
+constexpr int64_t kStorageResourceId = 6775971001;
+
+// Exercise cloud writer I/O against real temporary files without an
object-store service.
+class LocalRemoteFileSystem final : public io::RemoteFileSystem {
+public:
+ explicit LocalRemoteFileSystem(std::string root_path)
+ : RemoteFileSystem(std::move(root_path),
std::to_string(kStorageResourceId),
+ io::FileSystemType::BROKER) {}
+
+private:
+ Status create_file_impl(const io::Path& file, io::FileWriterPtr* writer,
+ const io::FileWriterOptions* opts) override {
+ return io::global_local_filesystem()->create_file(file, writer, opts);
+ }
+
+ Status create_directory_impl(const io::Path& dir, bool failed_if_exists)
override {
+ return io::global_local_filesystem()->create_directory(dir,
failed_if_exists);
+ }
+
+ Status delete_file_impl(const io::Path& file) override {
+ return io::global_local_filesystem()->delete_file(file);
+ }
+
+ Status batch_delete_impl(const std::vector<io::Path>& files) override {
+ return io::global_local_filesystem()->batch_delete(files);
+ }
+
+ Status delete_directory_impl(const io::Path& dir) override {
+ return io::global_local_filesystem()->delete_directory(dir);
+ }
+
+ Status exists_impl(const io::Path& path, bool* res) const override {
+ return io::global_local_filesystem()->exists(path, res);
+ }
+
+ Status file_size_impl(const io::Path& file, int64_t* file_size) const
override {
+ return io::global_local_filesystem()->file_size(file, file_size);
+ }
+
+ Status list_impl(const io::Path& dir, bool only_file,
std::vector<io::FileInfo>* files,
+ bool* exists) override {
+ return io::global_local_filesystem()->list(dir, only_file, files,
exists);
+ }
+
+ Status rename_impl(const io::Path& orig_name, const io::Path& new_name)
override {
+ return io::global_local_filesystem()->rename(orig_name, new_name);
+ }
+
+ Status upload_impl(const io::Path& local_file, const io::Path&
remote_file) override {
+ return io::global_local_filesystem()->link_file(local_file,
remote_file);
+ }
+
+ Status batch_upload_impl(const std::vector<io::Path>& local_files,
+ const std::vector<io::Path>& remote_files)
override {
+ DCHECK_EQ(local_files.size(), remote_files.size());
+ for (size_t i = 0; i < local_files.size(); ++i) {
+ RETURN_IF_ERROR(upload_impl(local_files[i], remote_files[i]));
+ }
+ return Status::OK();
+ }
+
+ Status download_impl(const io::Path& remote_file, const io::Path&
local_file) override {
+ return io::global_local_filesystem()->link_file(remote_file,
local_file);
+ }
+
+ Status open_file_internal(const io::Path& file, io::FileReaderSPtr* reader,
+ const io::FileReaderOptions& opts) override {
+ return io::global_local_filesystem()->open_file(file, reader, &opts);
+ }
+};
+
+} // namespace
+
+class DeltaWriterCancelIntegrationTest : public testing::TestWithParam<bool> {
+protected:
+ static constexpr int64_t kTabletId = 71001;
+ static constexpr int64_t kIndexId = 72001;
+
+ void SetUp() override {
+ auto* env = ExecEnv::GetInstance();
+ _saved_engine = std::move(env->_storage_engine);
+ _saved_limiter = std::move(env->_memtable_memory_limiter);
+ _saved_fragment_mgr = env->_fragment_mgr;
+ _saved_cloud_id = config::cloud_unique_id;
+ _saved_deploy_mode = config::deploy_mode;
+ _saved_storage_root = config::storage_root_path;
+ _saved_sync_rowsets =
config::cloud_mow_sync_rowsets_when_load_txn_begin;
+ _saved_file_cache = config::enable_file_cache;
+ _saved_between_segments =
config::enable_calc_delete_bitmap_between_segments_concurrently;
+ config::cloud_unique_id = GetParam() ? "bitmap-cancel-ut" : "";
+ config::deploy_mode = GetParam() ? "cloud" : "local";
+ config::cloud_mow_sync_rowsets_when_load_txn_begin = false;
+ config::enable_file_cache = false;
+ config::enable_calc_delete_bitmap_between_segments_concurrently = true;
+ _root = (std::filesystem::current_path() / "ut_dir" /
+ (GetParam() ? "bitmap_cancel_cloud" : "bitmap_cancel_local"))
+ .string();
+ config::storage_root_path = _root;
+
ASSERT_TRUE(io::global_local_filesystem()->delete_directory(_root).ok());
+
ASSERT_TRUE(io::global_local_filesystem()->create_directory(_root).ok());
+ _attach =
std::make_unique<AttachTask>(MemTrackerLimiter::create_shared(
+ MemTrackerLimiter::Type::OTHER,
"DeltaWriterCancelIntegrationTest"));
+
+ _fragment_mgr = std::make_unique<FragmentMgr>(env);
+ env->_fragment_mgr = _fragment_mgr.get();
+ env->set_memtable_memory_limiter(new MemTableMemoryLimiter());
+ if (GetParam()) {
+ auto engine = std::make_unique<CloudStorageEngine>(EngineOptions
{});
+ engine->init_calc_delete_bitmap_executor_for_UT();
+ engine->_memtable_flush_executor =
std::make_unique<MemTableFlushExecutor>();
+ engine->_memtable_flush_executor->init(1);
+ auto fs = std::make_shared<LocalRemoteFileSystem>(_root);
+ engine->set_latest_fs(fs);
+ // Temporary rowsets copy the resource ID through protobuf
metadata and
+ // resolve it from the registry when bitmap callbacks reopen their
segments.
+ put_storage_resource(kStorageResourceId, StorageResource {fs}, 0);
+ env->set_storage_engine(std::move(engine));
+ } else {
+ EngineOptions options;
+ options.store_paths.emplace_back(_root, -1);
+ auto engine = std::make_unique<StorageEngine>(options);
+ ASSERT_TRUE(engine->open().ok());
+ env->set_storage_engine(std::move(engine));
+ }
+ _engine = &env->storage_engine();
+ _load_channel = std::make_shared<LoadChannel>(UniqueId {71, 1}, 60,
false, "", 0, true, -1);
+ create_tablet(kTabletId);
+ }
+
+ void TearDown() override {
+ // Unblock callbacks on assertion failure, but do not pre-drain the
executor:
+ // writer destruction itself must enforce the lifetime guarantee.
+ _release_bitmap.count_down();
+ if (_cleanup.valid()) {
+ _cleanup.get();
+ }
+ _load_channel.reset();
+ _tablet.reset();
+ auto* sp = SyncPoint::get_instance();
+ sp->disable_processing();
+ _guards.clear();
+ sp->clear_trace();
+ auto* env = ExecEnv::GetInstance();
+ env->set_memtable_memory_limiter(nullptr);
+ env->set_storage_engine(nullptr);
+ // All cloud writers and callbacks must finish before their resource
is removed.
+ if (GetParam()) {
+ delete_storage_resource(kStorageResourceId);
+ }
+ env->_storage_engine = std::move(_saved_engine);
+ env->_memtable_memory_limiter = std::move(_saved_limiter);
+ if (_fragment_mgr) {
+ _fragment_mgr->stop();
+ env->_fragment_mgr = _saved_fragment_mgr;
+ _fragment_mgr.reset();
+ }
+ _attach.reset();
+
EXPECT_TRUE(io::global_local_filesystem()->delete_directory(_root).ok());
+ config::cloud_unique_id = _saved_cloud_id;
+ config::deploy_mode = _saved_deploy_mode;
+ config::storage_root_path = _saved_storage_root;
+ config::cloud_mow_sync_rowsets_when_load_txn_begin =
_saved_sync_rowsets;
+ config::enable_file_cache = _saved_file_cache;
+ config::enable_calc_delete_bitmap_between_segments_concurrently =
_saved_between_segments;
+ }
+
+ void create_tablet(int64_t id) {
+ _request = testutil::create_tablet_request(
+ id, 73001, 74001, 1, TKeysType::UNIQUE_KEYS,
+ {{"k1", TPrimitiveType::INT, true}, {"v1",
TPrimitiveType::INT, false}});
+ _request.__set_enable_unique_key_merge_on_write(true);
+ if (GetParam()) {
+ std::unordered_map<uint32_t, uint32_t> ids {{0, 0}, {1, 1}};
+ auto meta = TabletMeta::create(_request, TabletUid::gen_uid(), 0,
2, ids);
+ auto tablet = std::make_shared<CloudTablet>(_engine->to_cloud(),
meta);
+ auto rs_meta = std::make_shared<RowsetMeta>();
+ rs_meta->set_rowset_type(BETA_ROWSET);
+ rs_meta->set_rowset_state(VISIBLE);
+ rs_meta->set_tablet_id(id);
+
rs_meta->set_tablet_schema_hash(_request.tablet_schema.schema_hash);
+ rs_meta->set_tablet_schema(tablet->tablet_schema());
+ rs_meta->set_version(Version(0, 1));
+ rs_meta->set_segments_overlap(OVERLAP_UNKNOWN);
+ RowsetSharedPtr rowset;
+ ASSERT_TRUE(RowsetFactory::create_rowset(tablet->tablet_schema(),
"", rs_meta, &rowset)
+ .ok());
+ {
+ std::unique_lock meta_lock(tablet->get_header_lock());
+ tablet->add_rowsets({rowset}, false, meta_lock);
+ tablet->set_cumulative_layer_point(2);
+ }
+ _engine->to_cloud().tablet_mgr().put_tablet_for_UT(tablet);
+ ASSERT_TRUE(io::global_local_filesystem()
+ ->create_directory(fmt::format("{}/data/{}",
_root, id))
+ .ok());
+ _tablet = tablet;
+ } else {
+ ASSERT_TRUE(
+ _engine->to_local().create_tablet(_request,
_load_channel->_self_profile).ok());
+ _tablet = _engine->to_local().tablet_manager()->get_tablet(id);
+ }
+ ASSERT_NE(_tablet, nullptr);
+ }
+
+ PTabletWriterOpenRequest open_request(int64_t id, bool incremental =
false) {
+ auto desc = testutil::create_descriptor_table(
+ {{TYPE_INT, "k1", false}, {TYPE_INT, "v1", false}});
+ auto schema = testutil::create_table_schema_param(
+ desc, kIndexId, _request.tablet_schema.schema_hash,
_request.tablet_schema.columns);
+ PTabletWriterOpenRequest request;
+ request.mutable_id()->set_hi(71);
+ request.mutable_id()->set_lo(1);
+ request.set_index_id(kIndexId);
+ request.set_txn_id(75001);
+ request.set_txn_expiration(4102444800);
+ request.set_num_senders(1);
+ request.set_sender_id(0);
+ request.set_is_incremental(incremental);
+ request.set_is_high_priority(_load_channel->is_high_priority());
+ request.set_enable_profile(true);
+ schema->to_protobuf(request.mutable_schema());
+ auto* tablet = request.add_tablets();
+ tablet->set_tablet_id(id);
+ tablet->set_partition_id(_request.partition_id);
+ return request;
+ }
+
+ Status open_writer(int64_t id, bool incremental = false) {
+ auto request = open_request(id, incremental);
+ Status status;
+ // open() attaches the load task itself. Model its RPC worker
separately
+ // from the fixture's attached task to avoid nesting AttachTask.
+ std::thread rpc([&] { status = _load_channel->open(request); });
+ rpc.join();
+ RETURN_IF_ERROR(status);
+ _writer =
_load_channel->_tablets_channels.at(kIndexId)->_tablet_writers.at(id).get();
+ _builder = _writer->_rowset_builder.get();
+ if (GetParam()) {
+ // Only metadata RPCs are skipped; builders and both executors
initialize normally.
+
static_cast<CloudRowsetBuilder*>(_builder)->set_skip_writing_rowset_metadata(true);
+ }
+ return Status::OK();
+ }
+
+ Block block() {
+ auto result = _tablet->tablet_schema()->create_storage_block();
+ {
+ auto guard = result.mutate_columns_scoped();
+ auto& columns = guard.mutable_columns();
+
columns[0]->insert(Field::create_field<PrimitiveType::TYPE_INT>(1));
+
columns[1]->insert(Field::create_field<PrimitiveType::TYPE_INT>(10));
+ }
+ return result;
+ }
+
+ void check_wiring() {
+ _rowset_writer =
static_cast<BaseBetaRowsetWriter*>(_builder->rowset_writer().get());
+ const auto& cancellation = _load_channel->_cancel_status;
+ ASSERT_EQ(_writer->_req.load_cancel_status, cancellation);
+ ASSERT_EQ(_builder->_req.load_cancel_status, cancellation);
+ ASSERT_EQ(_rowset_writer->context().load_cancel_status, cancellation);
+ ASSERT_NE(_builder->_calc_delete_bitmap_token, nullptr);
+ ASSERT_NE(_rowset_writer->_calc_delete_bitmap_token, nullptr);
+ EXPECT_EQ(_builder->_calc_delete_bitmap_token->_load_cancel_status,
cancellation);
+
EXPECT_EQ(_rowset_writer->_calc_delete_bitmap_token->_load_cancel_status,
cancellation);
+ EXPECT_EQ(_builder->_calc_delete_bitmap_token->_thread_token->_pool,
+ _engine->calc_delete_bitmap_executor()->_thread_pool.get());
+
EXPECT_EQ(_rowset_writer->_calc_delete_bitmap_token->_thread_token->_pool,
+
_engine->calc_delete_bitmap_executor_for_load()->_thread_pool.get());
+ EXPECT_NE(_engine->calc_delete_bitmap_executor()->_thread_pool.get(),
+
_engine->calc_delete_bitmap_executor_for_load()->_thread_pool.get());
+ }
+
+ template <typename Callback>
+ void on(const std::string& point, Callback&& callback) {
+ _guards.emplace_back();
+ SyncPoint::get_instance()->set_call_back(point,
std::forward<Callback>(callback),
+ &_guards.back());
+ SyncPoint::get_instance()->enable_processing();
+ }
+
+ void block_bitmap_and_observe_cleanup() {
+ _segment_path = _rowset_writer->context().segment_path(0);
+ if (GetParam()) {
+ _segment_path = (_engine->to_cloud().latest_fs()->root_path() /
_segment_path).string();
+ }
+ on("BaseBetaRowsetWriter::_generate_delete_bitmap:before_build_tmp",
[&](auto&& args) {
+ if (try_any_cast<BaseBetaRowsetWriter*>(args[0]) ==
_rowset_writer) {
+ _bitmap_started.count_down();
+ _release_bitmap.wait();
+ bool exists = false;
+
EXPECT_TRUE(io::global_local_filesystem()->exists(_segment_path, &exists).ok());
+ EXPECT_TRUE(exists);
+ EXPECT_FALSE(_cleanup_started.load());
+ }
+ });
+ on("BaseBetaRowsetWriter::_generate_delete_bitmap:finished",
[&](auto&& args) {
+ if (try_any_cast<BaseBetaRowsetWriter*>(args[0]) ==
_rowset_writer) {
+ _bitmap_finished = true;
+ }
+ });
+ on("BaseBetaRowsetWriter::~BaseBetaRowsetWriter:before_file_cleanup",
[&](auto&& args) {
+ if (try_any_cast<BaseBetaRowsetWriter*>(args[0]) ==
_rowset_writer) {
+ EXPECT_TRUE(_bitmap_finished.load());
+ _cleanup_started = true;
+ }
+ });
+ on("CloudRowsetBuilder::~CloudRowsetBuilder:before_clear_cache",
[&](auto&& args) {
+ if (try_any_cast<CloudRowsetBuilder*>(args[0]) == _builder) {
+ EXPECT_TRUE(_bitmap_finished.load());
+ _cleanup_started = true;
+ }
+ });
+ }
+
+ void start_cleanup() {
+ auto resource_ctx = _load_channel->_resource_ctx;
+ auto& owner =
_load_channel->_tablets_channels.at(kIndexId)->_tablet_writers.at(kTabletId);
+ _cleanup =
+ std::async(std::launch::async, [resource_ctx, writer =
std::move(owner)]() mutable {
+ SCOPED_ATTACH_TASK(resource_ctx);
+ writer.reset();
+ });
+ EXPECT_EQ(_cleanup.wait_for(std::chrono::milliseconds(100)),
std::future_status::timeout);
+ EXPECT_FALSE(_cleanup_started.load());
+ EXPECT_FALSE(_bitmap_finished.load());
+ _release_bitmap.count_down();
+ _cleanup.get();
+ EXPECT_TRUE(_bitmap_finished.load());
+ }
+
+ Status flush_two_memtables() {
+ auto data = block();
+ for (int i = 0; i < 2; ++i) {
+ RETURN_IF_ERROR(_writer->write(&data, TabletAddRowsPayload
{.row_idxs = {0}}));
+ RETURN_IF_ERROR(_writer->flush_memtable_async());
+ RETURN_IF_ERROR(_writer->wait_flush());
+ }
+ return _rowset_writer->_calc_delete_bitmap_token->wait();
+ }
+
+ void exercise_both_phases() {
+ ASSERT_TRUE(_writer->init().ok());
+ check_wiring();
+ auto st = flush_two_memtables();
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_TRUE(_writer->close().ok());
+ ASSERT_TRUE(_writer->build_rowset().ok());
+ std::atomic<int> builder_submissions {0};
+ on("CalcDeleteBitmapToken::submit_func:before_submit", [&](auto&&
args) {
+ if (try_any_cast<CalcDeleteBitmapToken*>(args[0]) ==
+ _builder->_calc_delete_bitmap_token.get()) {
+ ++builder_submissions;
+ }
+ });
+ ASSERT_TRUE(_writer->submit_calc_delete_bitmap_task().ok());
+ ASSERT_TRUE(_writer->wait_calc_delete_bitmap().ok());
+ EXPECT_GT(builder_submissions.load(), 0);
+ SyncPoint::get_instance()->clear_call_back(
+ "CalcDeleteBitmapToken::submit_func:before_submit");
+ }
+
+ Status prepare_partial_init_callback() {
+ check_wiring();
+ auto data = block();
+ if (GetParam()) {
+ // Keep a real temporary PREPARED rowset in the builder to
exercise its
+ // cache-cleanup branch. build() would instead mark the rowset
COMMITTED.
+ RETURN_IF_ERROR(_rowset_writer->flush_single_block(&data));
+ RETURN_IF_ERROR(_rowset_writer->_calc_delete_bitmap_token->wait());
+ RETURN_IF_ERROR(_rowset_writer->_build_tmp(_builder->_rowset));
+ EXPECT_EQ(_builder->rowset()->rowset_meta()->rowset_state(),
PREPARED);
+ block_bitmap_and_observe_cleanup();
+ // Resubmit the real callback against the already-created segment.
+ return _rowset_writer->_generate_delete_bitmap(0);
+ }
+ block_bitmap_and_observe_cleanup();
+ return _rowset_writer->flush_single_block(&data);
+ }
+
+ std::string _root;
+ std::string _saved_cloud_id;
+ std::string _saved_deploy_mode;
+ std::string _saved_storage_root;
+ bool _saved_sync_rowsets = false;
+ bool _saved_file_cache = false;
+ bool _saved_between_segments = false;
+ std::unique_ptr<BaseStorageEngine> _saved_engine;
+ std::unique_ptr<MemTableMemoryLimiter> _saved_limiter;
+ FragmentMgr* _saved_fragment_mgr = nullptr;
+ std::unique_ptr<FragmentMgr> _fragment_mgr;
+ std::unique_ptr<AttachTask> _attach;
+ BaseStorageEngine* _engine = nullptr;
+ BaseTabletSPtr _tablet;
+ TCreateTabletReq _request;
+ std::shared_ptr<LoadChannel> _load_channel;
+ BaseDeltaWriter* _writer = nullptr;
+ BaseRowsetBuilder* _builder = nullptr;
+ BaseBetaRowsetWriter* _rowset_writer = nullptr;
+ std::string _segment_path;
+ std::vector<SyncPoint::CallbackGuard> _guards;
+ CountDownLatch _bitmap_started {1};
+ CountDownLatch _release_bitmap {1};
+ std::atomic<bool> _bitmap_finished {false};
+ std::atomic<bool> _cleanup_started {false};
+ std::future<void> _cleanup;
+};
+
+TEST_P(DeltaWriterCancelIntegrationTest,
OrdinaryAndIncrementalOpenWireBothPhases) {
+ for (int i = 0; i < 2; ++i) {
+ if (i != 0) {
+ create_tablet(kTabletId + i);
+ }
+ ASSERT_TRUE(open_writer(kTabletId + i, i != 0).ok());
+ ASSERT_NO_FATAL_FAILURE(exercise_both_phases());
+ }
+ ASSERT_TRUE(_load_channel->cancel().ok());
+ for (const auto& [_, writer] :
_load_channel->_tablets_channels.at(kIndexId)->_tablet_writers) {
+ auto* builder = writer->_rowset_builder.get();
+ auto* rowset_writer =
static_cast<BaseBetaRowsetWriter*>(builder->rowset_writer().get());
+
EXPECT_TRUE(builder->_calc_delete_bitmap_token->wait().is<ErrorCode::CANCELLED>());
+
EXPECT_TRUE(rowset_writer->_calc_delete_bitmap_token->wait().is<ErrorCode::CANCELLED>());
+ }
+}
+
+TEST_P(DeltaWriterCancelIntegrationTest,
PublicationReturnsBeforeFinalOwnerDrainsRealCallback) {
+ ASSERT_TRUE(open_writer(kTabletId).ok());
+ ASSERT_TRUE(_writer->init().ok());
+ check_wiring();
+ auto data = block();
+ block_bitmap_and_observe_cleanup();
+ ASSERT_TRUE(_rowset_writer->flush_single_block(&data).ok());
+ ASSERT_TRUE(_bitmap_started.wait_for(std::chrono::seconds(10)));
+ ASSERT_TRUE(_load_channel->cancel().ok());
+ EXPECT_FALSE(_bitmap_finished.load());
+ EXPECT_FALSE(_cleanup_started.load());
+ start_cleanup();
+ _load_channel.reset();
+ if (!GetParam()) {
+ EXPECT_TRUE(_cleanup_started.load());
+ bool exists = true;
+ ASSERT_TRUE(io::global_local_filesystem()->exists(_segment_path,
&exists).ok());
+ EXPECT_FALSE(exists);
+ }
+}
+
+TEST_P(DeltaWriterCancelIntegrationTest,
PartialInitDestructionDrainsBeforeFileOrCacheCleanup) {
+ ASSERT_TRUE(open_writer(kTabletId).ok());
+ const auto failure = Status::InternalError("memtable initialization
failed");
+ on("BaseDeltaWriter::init:before_memtable_init", [&](auto&& args) {
+ if (try_any_cast<BaseDeltaWriter*>(args[0]) != _writer) {
+ return;
+ }
+ auto st = prepare_partial_init_callback();
+ EXPECT_TRUE(st.ok()) << st;
+ auto* ret = try_any_cast_ret<Status>(args);
+ ret->first = st.ok() ? failure : st;
+ ret->second = true;
+ });
+ EXPECT_EQ(_writer->init(), failure);
+ ASSERT_FALSE(_writer->_is_init);
+ ASSERT_TRUE(_builder->_is_init);
+ ASSERT_TRUE(_bitmap_started.wait_for(std::chrono::seconds(10)));
+ start_cleanup();
+ EXPECT_TRUE(_cleanup_started.load());
+ if (!GetParam()) {
+ bool exists = true;
+ ASSERT_TRUE(io::global_local_filesystem()->exists(_segment_path,
&exists).ok());
+ EXPECT_FALSE(exists);
+ }
+}
+
+TEST_P(DeltaWriterCancelIntegrationTest,
FinalOwnerCompletesPendingEosBeforeDestroyingWriters) {
+ ASSERT_TRUE(open_writer(kTabletId).ok());
+ auto* tablets = _load_channel->_tablets_channels.at(kIndexId).get();
+ auto completion = tablets->eos_completion();
+ const auto reason = Status::InternalError("cancel before all EOS senders
arrive");
+ int replies = 0;
+ completion->add_waiter([&](const Status& status) {
+ EXPECT_EQ(status, reason);
+ EXPECT_EQ(tablets->_tablet_writers.at(kTabletId).get(), _writer);
+ ++replies;
+ });
+ {
+ // Publishing cancellation must not take either channel lock or run
RPC callbacks.
+ std::lock_guard load_lock(_load_channel->_lock);
+ std::lock_guard tablets_lock(tablets->_lock);
+ ASSERT_TRUE(_load_channel->cancel(reason).ok());
+ EXPECT_EQ(replies, 0);
+ }
+ _load_channel.reset();
+ EXPECT_EQ(replies, 1);
+ // The service may register after add_batch released the last channel
owner.
+ completion->add_waiter([&](const Status& status) {
+ EXPECT_EQ(status, reason);
+ ++replies;
+ });
+ EXPECT_EQ(replies, 2);
+}
+
+TEST_P(DeltaWriterCancelIntegrationTest,
PublishedCancellationWinsBeforeEosArrival) {
+ ASSERT_TRUE(open_writer(kTabletId).ok());
+ auto* tablets = _load_channel->_tablets_channels.at(kIndexId).get();
+ const auto reason = Status::Cancelled("cancel before final EOS sender");
+ ASSERT_TRUE(_load_channel->cancel(reason).ok());
+ int replies = 0;
+ tablets->eos_completion()->add_waiter([&](const Status& status) {
+ EXPECT_EQ(status, reason);
+ ++replies;
+ });
+ PTabletWriterAddBlockResult response;
+ std::unique_lock lock(tablets->_lock);
+ tablets->_num_remaining_senders = 0;
+ EXPECT_EQ(tablets->_close_and_notify(lock, 0, &response), reason);
+ EXPECT_EQ(replies, 1);
+}
+
+INSTANTIATE_TEST_SUITE_P(LocalCloud, DeltaWriterCancelIntegrationTest,
testing::Bool());
+
+} // 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..a1170c48c3f
--- /dev/null
+++ b/be/test/load/delta_writer/delta_writer_cancel_test.cpp
@@ -0,0 +1,295 @@
+// 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/group_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 both local/cloud writers, with and without row binlog, 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;
+ WriteRequest group_req = data_req;
+ group_req.write_req_type = WriteRequestType::GROUP;
+ WriteRequest binlog_req = data_req;
+ binlog_req.write_req_type = WriteRequestType::ROW_BINLOG;
+ if (is_cloud()) {
+ _cloud_engine = std::make_unique<CloudStorageEngine>(EngineOptions
{});
+ if (is_group()) {
+ _writer = std::make_unique<CloudDeltaWriter>(*_cloud_engine,
group_req, data_req,
+ binlog_req,
nullptr, UniqueId {});
+ auto* group =
static_cast<CloudGroupRowsetBuilder*>(_writer->_rowset_builder.get());
+ _builders = {group->data_builder(),
group->row_binlog_builder()};
+ } else {
+ _writer = std::make_unique<CloudDeltaWriter>(*_cloud_engine,
data_req, nullptr,
+ UniqueId {});
+ }
+ } else {
+ _local_engine = std::make_unique<StorageEngine>(EngineOptions {});
+ if (is_group()) {
+ _writer = std::make_unique<DeltaWriter>(*_local_engine,
group_req, data_req,
+ binlog_req, nullptr,
UniqueId {});
+ auto* group =
static_cast<GroupRowsetBuilder*>(_writer->_rowset_builder.get());
+ _builders = {group->txn_rowset_builder(),
group->row_binlog_builder()};
+ } else {
+ _writer = std::make_unique<DeltaWriter>(*_local_engine,
data_req, nullptr,
+ UniqueId {});
+ }
+ }
+ if (!is_group()) {
+ _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; }
+ bool is_group() const { return GetParam() & 2; }
+
+ 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());
+ }
+ if (is_group()) {
+ auto group_writer = std::make_shared<GroupRowsetWriter>();
+ group_writer->set_data_writer(_builders[0]->rowset_writer());
+ group_writer->set_row_binlog_writer(_builders[1]->rowset_writer());
+ _writer->_rowset_builder->_rowset_writer = std::move(group_writer);
+ }
+ }
+
+ 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(LocalCloudOrdinaryGroup, DeltaWriterCancelTest,
testing::Range(0, 4));
+
+} // namespace doris
diff --git a/be/test/load/memtable/memtable_flush_executor_test.cpp
b/be/test/load/memtable/memtable_flush_executor_test.cpp
index ad6e27dd0c5..8053cfcf9aa 100644
--- a/be/test/load/memtable/memtable_flush_executor_test.cpp
+++ b/be/test/load/memtable/memtable_flush_executor_test.cpp
@@ -26,6 +26,8 @@
#include <atomic>
#include <chrono>
+#include <functional>
+#include <future>
#include <string>
#include <thread>
@@ -48,6 +50,8 @@
#include "storage/tablet/tablet_meta_manager.h"
#include "storage/utils.h"
#include "testutil/creators.h"
+#include "util/countdown_latch.h"
+#include "util/defer_op.h"
namespace doris {
@@ -82,6 +86,9 @@ public:
Status flush_memtable(Block* block, int32_t segment_id, int64_t*
flush_size) override {
EXPECT_GT(block->rows(), 0);
+ if (_flush_hook) {
+ _flush_hook();
+ }
if (_flush_enter_cnt != nullptr) {
++(*_flush_enter_cnt);
}
@@ -126,11 +133,14 @@ public:
std::shared_ptr<PartialUpdateInfo> get_partial_update_info() override {
return nullptr; }
bool is_partial_update() override { return false; }
+ void set_flush_hook(std::function<void()> hook) { _flush_hook =
std::move(hook); }
+
int32_t last_segment_id() const { return _last_segment_id; }
ConstAllocatedLsnVectorSharedPtr last_seg_lsn() const { return
_last_seg_lsn; }
private:
+ std::function<void()> _flush_hook;
std::atomic<int>* _flush_cnt;
bool _fail_on_flush;
std::string _flush_error_msg;
@@ -723,4 +733,113 @@ TEST_F(MemTableFlushExecutorGroupFlushTest,
TestGroupFlushTokenCancelledCleanup)
drop_tablet(ctx.request);
}
+class MemTableFlushCancellationTest : public
MemTableFlushExecutorGroupFlushTest,
+ public testing::WithParamInterface<bool>
{
+protected:
+ static void expect_skipped_flush(const FlushStatistic& stats) {
+ EXPECT_EQ(stats.flush_submit_count.load(), 0);
+ EXPECT_EQ(stats.flush_running_count.load(), 0);
+ EXPECT_EQ(stats.flush_finish_count.load(), 0);
+ }
+};
+
+TEST_P(MemTableFlushCancellationTest, SharedCancellationSkipsQueuedFlushes) {
+ SCOPED_INIT_THREAD_CONTEXT();
+ const bool group_flush = GetParam();
+ GroupFlushTestContext ctx;
+ prepare_group_flush_test_context(10007 + group_flush, 270068379 +
group_flush, {7000, 7001},
+ &ctx);
+ std::atomic<int> data_flush_cnt = 0;
+ std::atomic<int> binlog_flush_cnt = 0;
+ auto data_writer = std::make_shared<MockRowsetWriter>(&data_flush_cnt);
+ auto binlog_writer = std::make_shared<MockRowsetWriter>(&binlog_flush_cnt);
+ std::shared_ptr<GroupRowsetWriter> group_writer;
+ ASSERT_TRUE(create_group_rowset_writer(ctx, 7, data_writer, binlog_writer,
&group_writer).ok());
+
+ std::unique_ptr<ThreadPool> pool;
+ ASSERT_TRUE(ThreadPoolBuilder("CancelledFlushTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&pool)
+ .ok());
+ auto status = std::make_shared<AtomicStatus>();
+ auto token = FlushToken::create_shared(pool.get(), nullptr, status);
+ token->set_rowset_writer(group_flush ?
std::static_pointer_cast<RowsetWriter>(group_writer)
+ : data_writer);
+ token->set_table_schema_param(create_group_flush_table_schema_param(ctx));
+ 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)));
+ ASSERT_TRUE(token->submit(ctx.memtable).ok());
+ EXPECT_EQ(token->get_stats().flush_submit_count.load(), group_flush ? 2 :
1);
+ status->update(Status::Cancelled("cancel queued flush"));
+ // Both pre-cancel and post-cancel submissions must skip flushing when
dispatched.
+ ASSERT_TRUE(token->submit(ctx.memtable).ok());
+ release.count_down();
+ EXPECT_TRUE(token->wait().is<ErrorCode::CANCELLED>());
+ pool->wait();
+ EXPECT_EQ(data_flush_cnt.load(), 0);
+ EXPECT_EQ(binlog_flush_cnt.load(), 0);
+ expect_skipped_flush(token->get_stats());
+
EXPECT_FALSE(group_writer->context().allocated_lsn_map->contains_segment(0));
+ drop_tablet(ctx.request);
+}
+
+INSTANTIATE_TEST_SUITE_P(LoadCancellation, MemTableFlushCancellationTest,
testing::Bool());
+
+TEST_F(MemTableFlushExecutorGroupFlushTest,
SharedCancellationDuringGroupFlush) {
+ SCOPED_INIT_THREAD_CONTEXT();
+ GroupFlushTestContext ctx;
+ prepare_group_flush_test_context(10009, 270068381, {9000, 9001}, &ctx);
+ std::atomic<int> data_flush_cnt = 0;
+ std::atomic<int> binlog_flush_cnt = 0;
+ auto data_writer = std::make_shared<MockRowsetWriter>(&data_flush_cnt);
+ auto binlog_writer = std::make_shared<MockRowsetWriter>(&binlog_flush_cnt);
+ CountDownLatch entered(1);
+ CountDownLatch release(1);
+ data_writer->set_flush_hook([&] {
+ entered.count_down();
+ release.wait();
+ });
+ std::shared_ptr<GroupRowsetWriter> group_writer;
+ ASSERT_TRUE(create_group_rowset_writer(ctx, 9, data_writer, binlog_writer,
&group_writer).ok());
+ std::unique_ptr<ThreadPool> pool;
+ ASSERT_TRUE(ThreadPoolBuilder("RunningFlushCancellationTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&pool)
+ .ok());
+ auto status = std::make_shared<AtomicStatus>();
+ auto token = FlushToken::create_shared(pool.get(), nullptr, status);
+ token->set_rowset_writer(group_writer);
+ token->set_table_schema_param(create_group_flush_table_schema_param(ctx));
+ Defer cleanup {[&] {
+ release.count_down();
+ pool->wait();
+ }};
+ ASSERT_TRUE(token->submit(ctx.memtable).ok());
+ ASSERT_TRUE(entered.wait_for(std::chrono::seconds(10)));
+ status->update(Status::Cancelled("cancel during data flush"));
+ auto waiter = std::async(std::launch::async, [&] { return token->wait();
});
+ EXPECT_EQ(waiter.wait_for(std::chrono::milliseconds(50)),
std::future_status::timeout);
+ release.count_down();
+ EXPECT_TRUE(waiter.get().is<ErrorCode::CANCELLED>());
+ pool->wait();
+ EXPECT_EQ(data_flush_cnt.load(), 1);
+ EXPECT_EQ(binlog_flush_cnt.load(), 0);
+ EXPECT_EQ(token->get_stats().flush_submit_count.load(), 0);
+ EXPECT_EQ(token->get_stats().flush_running_count.load(), 0);
+ EXPECT_FALSE(group_writer->context().allocated_lsn_map->contains_segment(
+ data_writer->last_segment_id()));
+ drop_tablet(ctx.request);
+}
+
} // 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 141f2dd4041..457855d041c 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, TabletAddRowsPayload {.row_idxs =
{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, TabletAddRowsPayload {.row_idxs = {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/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: [email protected]
For additional commands, e-mail: [email protected]
