You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by da...@apache.org on 2026/02/08 00:52:16 UTC
(doris) 01/01: [feature](read-uncommitted) Add READ UNCOMMITTED isolation level for unpublished rowsets (Phase 1)
This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch feature/read-uncommitted-phase1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 473aa6a8a69906fa31d673f0640a09a1cdeb1b16
Author: Yongqiang YANG <da...@gmail.com>
AuthorDate: Sat Feb 7 16:51:57 2026 -0800
[feature](read-uncommitted) Add READ UNCOMMITTED isolation level for unpublished rowsets (Phase 1)
Enable queries to read uncommitted but committed-to-storage rowsets via
SET read_uncommitted = true. Supports DUP_KEYS and UNIQUE_KEYS tables
in both local and cloud modes. Includes async cross-uncommitted dedup
for MoW tables with compaction and schema change awareness.
---
be/src/cloud/cloud_rowset_builder.cpp | 17 ++
be/src/cloud/cloud_storage_engine.cpp | 5 +
be/src/cloud/cloud_storage_engine.h | 5 +
be/src/cloud/cloud_txn_delete_bitmap_cache.cpp | 10 +
be/src/olap/base_tablet.cpp | 9 +
be/src/olap/compaction.cpp | 8 +
be/src/olap/rowset_builder.cpp | 16 +
be/src/olap/storage_engine.cpp | 5 +
be/src/olap/storage_engine.h | 5 +
be/src/olap/txn_manager.cpp | 13 +
be/src/olap/uncommitted_rowset_registry.cpp | 333 +++++++++++++++++++++
be/src/olap/uncommitted_rowset_registry.h | 121 ++++++++
be/src/pipeline/exec/olap_scan_operator.cpp | 65 ++++
be/src/runtime/runtime_state.h | 4 +
.../java/org/apache/doris/qe/SessionVariable.java | 17 +-
gensrc/thrift/PaloInternalService.thrift | 2 +
16 files changed, 633 insertions(+), 2 deletions(-)
diff --git a/be/src/cloud/cloud_rowset_builder.cpp b/be/src/cloud/cloud_rowset_builder.cpp
index 8ef15424a3d..b9c528af88e 100644
--- a/be/src/cloud/cloud_rowset_builder.cpp
+++ b/be/src/cloud/cloud_rowset_builder.cpp
@@ -19,6 +19,7 @@
#include "cloud/cloud_meta_mgr.h"
#include "cloud/cloud_storage_engine.h"
+#include "olap/uncommitted_rowset_registry.h"
#include "cloud/cloud_tablet.h"
#include "cloud/cloud_tablet_mgr.h"
#include "olap/storage_policy.h"
@@ -157,6 +158,22 @@ Status CloudRowsetBuilder::set_txn_related_delete_bitmap() {
_req.txn_id, _tablet->tablet_id(), _delete_bitmap, *_rowset_ids, _rowset,
_req.txn_expiration, _partial_update_info);
}
+
+ // Register uncommitted rowset for READ UNCOMMITTED visibility
+ if (auto* registry = _engine.uncommitted_rowset_registry()) {
+ auto entry = std::make_shared<UncommittedRowsetEntry>();
+ entry->rowset = _rowset;
+ entry->transaction_id = _req.txn_id;
+ entry->partition_id = _req.partition_id;
+ entry->tablet_id = _tablet->tablet_id();
+ entry->unique_key_merge_on_write = _tablet->enable_unique_key_merge_on_write();
+ entry->creation_time = _rowset->creation_time();
+ if (entry->unique_key_merge_on_write && _delete_bitmap) {
+ entry->committed_delete_bitmap = std::make_shared<DeleteBitmap>(*_delete_bitmap);
+ }
+ registry->register_rowset(std::move(entry));
+ }
+
return Status::OK();
}
#include "common/compile_check_end.h"
diff --git a/be/src/cloud/cloud_storage_engine.cpp b/be/src/cloud/cloud_storage_engine.cpp
index 126e9701a79..177c56e2f31 100644
--- a/be/src/cloud/cloud_storage_engine.cpp
+++ b/be/src/cloud/cloud_storage_engine.cpp
@@ -58,6 +58,7 @@
#include "olap/cumulative_compaction_time_series_policy.h"
#include "olap/memtable_flush_executor.h"
#include "olap/storage_policy.h"
+#include "olap/uncommitted_rowset_registry.h"
#include "runtime/memory/cache_manager.h"
#include "util/parse_util.h"
#include "util/time.h"
@@ -231,6 +232,10 @@ Status CloudStorageEngine::open() {
_cloud_snapshot_mgr = std::make_unique<CloudSnapshotMgr>(*this);
+ _uncommitted_rowset_registry = std::make_unique<UncommittedRowsetRegistry>();
+ RETURN_IF_ERROR(_uncommitted_rowset_registry->init(
+ std::max(1, config::calc_delete_bitmap_max_thread / 2)));
+
RETURN_NOT_OK_STATUS_WITH_WARN(
init_stream_load_recorder(ExecEnv::GetInstance()->store_paths()[0].path),
"init StreamLoadRecorder failed");
diff --git a/be/src/cloud/cloud_storage_engine.h b/be/src/cloud/cloud_storage_engine.h
index 9c27e164bba..3a435e10d90 100644
--- a/be/src/cloud/cloud_storage_engine.h
+++ b/be/src/cloud/cloud_storage_engine.h
@@ -51,6 +51,7 @@ class CloudWarmUpManager;
class CloudCompactionStopToken;
class CloudSnapshotMgr;
class CloudIndexChangeCompaction;
+class UncommittedRowsetRegistry;
class CloudStorageEngine final : public BaseStorageEngine {
public:
@@ -91,6 +92,9 @@ public:
CloudTabletMgr& tablet_mgr() const { return *_tablet_mgr; }
CloudSnapshotMgr& cloud_snapshot_mgr() { return *_cloud_snapshot_mgr; }
+ UncommittedRowsetRegistry* uncommitted_rowset_registry() {
+ return _uncommitted_rowset_registry.get();
+ }
CloudTxnDeleteBitmapCache& txn_delete_bitmap_cache() const { return *_txn_delete_bitmap_cache; }
@@ -225,6 +229,7 @@ private:
std::unique_ptr<ThreadPool> _sync_load_for_tablets_thread_pool;
std::unique_ptr<ThreadPool> _warmup_cache_async_thread_pool;
std::unique_ptr<CloudSnapshotMgr> _cloud_snapshot_mgr;
+ std::unique_ptr<UncommittedRowsetRegistry> _uncommitted_rowset_registry;
// FileSystem with latest shared storage info, new data will be written to this fs.
mutable std::mutex _latest_fs_mtx;
diff --git a/be/src/cloud/cloud_txn_delete_bitmap_cache.cpp b/be/src/cloud/cloud_txn_delete_bitmap_cache.cpp
index ab04f4eff16..de3ecc56b10 100644
--- a/be/src/cloud/cloud_txn_delete_bitmap_cache.cpp
+++ b/be/src/cloud/cloud_txn_delete_bitmap_cache.cpp
@@ -29,6 +29,7 @@
#include "olap/olap_common.h"
#include "olap/tablet_meta.h"
#include "olap/txn_manager.h"
+#include "olap/uncommitted_rowset_registry.h"
namespace doris {
@@ -264,6 +265,10 @@ void CloudTxnDeleteBitmapCache::remove_expired_tablet_txn_info() {
.tag("tablet_id", iter->second.tablet_id);
_empty_rowset_markers.erase(marker_iter);
}
+ // Unregister from UncommittedRowsetRegistry on expiration
+ if (auto* registry = get_uncommitted_rowset_registry()) {
+ registry->unregister_rowset(iter->second.tablet_id, iter->second.txn_id);
+ }
_expiration_txn.erase(iter);
}
}
@@ -283,6 +288,11 @@ void CloudTxnDeleteBitmapCache::remove_unused_tablet_txn_info(TTransactionId tra
erase(cache_key);
_txn_map.erase(txn_key);
}
+
+ // Unregister from UncommittedRowsetRegistry
+ if (auto* registry = get_uncommitted_rowset_registry()) {
+ registry->unregister_rowset(tablet_id, transaction_id);
+ }
}
void CloudTxnDeleteBitmapCache::mark_empty_rowset(TTransactionId txn_id, int64_t tablet_id,
diff --git a/be/src/olap/base_tablet.cpp b/be/src/olap/base_tablet.cpp
index 241e531462d..0a6e43506c4 100644
--- a/be/src/olap/base_tablet.cpp
+++ b/be/src/olap/base_tablet.cpp
@@ -48,6 +48,7 @@
#include "olap/rowset/segment_v2/column_reader.h"
#include "olap/tablet_fwd.h"
#include "olap/txn_manager.h"
+#include "olap/uncommitted_rowset_registry.h"
#include "service/point_query_executor.h"
#include "util/bvar_helper.h"
#include "util/debug_points.h"
@@ -169,6 +170,14 @@ Status BaseTablet::set_tablet_state(TabletState state) {
"could not change tablet state from shutdown to {}", state);
}
_tablet_meta->set_tablet_state(state);
+
+ // Notify UncommittedRowsetRegistry when tablet leaves RUNNING state
+ if (state != TABLET_RUNNING) {
+ if (auto* registry = get_uncommitted_rowset_registry()) {
+ registry->on_tablet_state_change(tablet_id(), state);
+ }
+ }
+
return Status::OK();
}
diff --git a/be/src/olap/compaction.cpp b/be/src/olap/compaction.cpp
index 4f3ed34939d..30e7ca86df2 100644
--- a/be/src/olap/compaction.cpp
+++ b/be/src/olap/compaction.cpp
@@ -69,6 +69,7 @@
#include "olap/rowset/segment_v2/inverted_index_desc.h"
#include "olap/rowset/segment_v2/inverted_index_fs_directory.h"
#include "olap/storage_engine.h"
+#include "olap/uncommitted_rowset_registry.h"
#include "olap/storage_policy.h"
#include "olap/tablet.h"
#include "olap/tablet_meta.h"
@@ -1359,6 +1360,13 @@ Status CompactionMixin::modify_rowsets() {
tablet()->merge_delete_bitmap(output_rowset_delete_bitmap);
RETURN_IF_ERROR(tablet()->modify_rowsets(output_rowsets, _input_rowsets, true));
}
+
+ // Notify UncommittedRowsetRegistry that published rowsets changed.
+ // Cross-uncommitted delete bitmaps may reference rows that moved due to
+ // compaction, so they must be re-computed.
+ if (auto* registry = _engine.uncommitted_rowset_registry()) {
+ registry->on_compaction_completed(_tablet->tablet_id());
+ }
} else {
std::lock_guard<std::shared_mutex> wrlock(_tablet->get_header_lock());
SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
diff --git a/be/src/olap/rowset_builder.cpp b/be/src/olap/rowset_builder.cpp
index f250f3bef10..fffa8c49f7a 100644
--- a/be/src/olap/rowset_builder.cpp
+++ b/be/src/olap/rowset_builder.cpp
@@ -34,6 +34,7 @@
#include "io/fs/file_system.h"
#include "io/fs/file_writer.h" // IWYU pragma: keep
#include "olap/calc_delete_bitmap_executor.h"
+#include "olap/uncommitted_rowset_registry.h"
#include "olap/olap_define.h"
#include "olap/partial_update_info.h"
#include "olap/rowset/beta_rowset.h"
@@ -347,6 +348,21 @@ Status RowsetBuilder::commit_txn() {
_delete_bitmap, *_rowset_ids, _partial_update_info);
}
+ // Register uncommitted rowset for READ UNCOMMITTED visibility
+ if (auto* registry = _engine.uncommitted_rowset_registry()) {
+ auto entry = std::make_shared<UncommittedRowsetEntry>();
+ entry->rowset = _rowset;
+ entry->transaction_id = _req.txn_id;
+ entry->partition_id = _req.partition_id;
+ entry->tablet_id = tablet()->tablet_id();
+ entry->unique_key_merge_on_write = _tablet->enable_unique_key_merge_on_write();
+ entry->creation_time = _rowset->creation_time();
+ if (entry->unique_key_merge_on_write && _delete_bitmap) {
+ entry->committed_delete_bitmap = std::make_shared<DeleteBitmap>(*_delete_bitmap);
+ }
+ registry->register_rowset(std::move(entry));
+ }
+
_is_committed = true;
return Status::OK();
}
diff --git a/be/src/olap/storage_engine.cpp b/be/src/olap/storage_engine.cpp
index 3a64a82d1ab..21462f44cc5 100644
--- a/be/src/olap/storage_engine.cpp
+++ b/be/src/olap/storage_engine.cpp
@@ -71,6 +71,7 @@
#include "olap/tablet_meta.h"
#include "olap/tablet_meta_manager.h"
#include "olap/txn_manager.h"
+#include "olap/uncommitted_rowset_registry.h"
#include "runtime/client_cache.h"
#include "runtime/stream_load/stream_load_recorder.h"
#include "util/doris_metrics.h"
@@ -313,6 +314,10 @@ Status StorageEngine::_open() {
? config::calc_delete_bitmap_for_load_max_thread
: std::max(1, CpuInfo::num_cores() / 2));
+ _uncommitted_rowset_registry = std::make_unique<UncommittedRowsetRegistry>();
+ RETURN_IF_ERROR(_uncommitted_rowset_registry->init(
+ std::max(1, config::calc_delete_bitmap_max_thread / 2)));
+
_parse_default_rowset_type();
return Status::OK();
diff --git a/be/src/olap/storage_engine.h b/be/src/olap/storage_engine.h
index 335f4a45f21..05d081d0b69 100644
--- a/be/src/olap/storage_engine.h
+++ b/be/src/olap/storage_engine.h
@@ -73,6 +73,7 @@ class ReportWorker;
class CreateTabletRRIdxCache;
struct DirInfo;
class SnapshotManager;
+class UncommittedRowsetRegistry;
class WorkloadGroup;
using SegCompactionCandidates = std::vector<segment_v2::SegmentSharedPtr>;
@@ -311,6 +312,9 @@ public:
TabletManager* tablet_manager() { return _tablet_manager.get(); }
TxnManager* txn_manager() { return _txn_manager.get(); }
SnapshotManager* snapshot_mgr() { return _snapshot_mgr.get(); }
+ UncommittedRowsetRegistry* uncommitted_rowset_registry() {
+ return _uncommitted_rowset_registry.get();
+ }
// Rowset garbage collection helpers
bool check_rowset_id_in_unused_rowsets(const RowsetId& rowset_id);
PendingRowsetSet& pending_local_rowsets() { return _pending_local_rowsets; }
@@ -533,6 +537,7 @@ private:
std::unique_ptr<TabletManager> _tablet_manager;
std::unique_ptr<TxnManager> _txn_manager;
+ std::unique_ptr<UncommittedRowsetRegistry> _uncommitted_rowset_registry;
// Used to control the migration from segment_v1 to segment_v2, can be deleted in futrue.
// Type of new loaded data
diff --git a/be/src/olap/txn_manager.cpp b/be/src/olap/txn_manager.cpp
index d760e891527..c0b20167e86 100644
--- a/be/src/olap/txn_manager.cpp
+++ b/be/src/olap/txn_manager.cpp
@@ -46,6 +46,7 @@
#include "olap/schema_change.h"
#include "olap/segment_loader.h"
#include "olap/storage_engine.h"
+#include "olap/uncommitted_rowset_registry.h"
#include "olap/tablet_manager.h"
#include "olap/tablet_meta.h"
#include "olap/tablet_meta_manager.h"
@@ -614,6 +615,12 @@ Status TxnManager::publish_txn(OlapMeta* meta, TPartitionId partition_id,
stats->lock_wait_time_us += MonotonicMicros() - t6;
_remove_txn_tablet_info_unlocked(partition_id, transaction_id, tablet_id, tablet_uid, txn_lock,
wrlock);
+
+ // Unregister from UncommittedRowsetRegistry on publish
+ if (auto* registry = _engine.uncommitted_rowset_registry()) {
+ registry->unregister_rowset(tablet_id, transaction_id);
+ }
+
VLOG_NOTICE << "publish txn successfully."
<< " partition_id: " << key.first << ", txn_id: " << key.second
<< ", tablet_id: " << tablet_info.tablet_id << ", rowsetid: " << rowset->rowset_id()
@@ -736,6 +743,12 @@ Status TxnManager::delete_txn(OlapMeta* meta, TPartitionId partition_id,
g_tablet_txn_info_txn_partitions_count << -1;
_clear_txn_partition_map_unlocked(transaction_id, partition_id);
}
+
+ // Unregister from UncommittedRowsetRegistry on delete
+ if (auto* registry = _engine.uncommitted_rowset_registry()) {
+ registry->unregister_rowset(tablet_id, transaction_id);
+ }
+
return st;
}
diff --git a/be/src/olap/uncommitted_rowset_registry.cpp b/be/src/olap/uncommitted_rowset_registry.cpp
new file mode 100644
index 00000000000..b96aedd2af2
--- /dev/null
+++ b/be/src/olap/uncommitted_rowset_registry.cpp
@@ -0,0 +1,333 @@
+// 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 "olap/uncommitted_rowset_registry.h"
+
+#include <algorithm>
+
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/config.h"
+#include "common/logging.h"
+#include "olap/base_tablet.h"
+#include "olap/calc_delete_bitmap_executor.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/segment_v2/segment.h"
+#include "olap/storage_engine.h"
+#include "runtime/exec_env.h"
+
+namespace doris {
+
+UncommittedRowsetRegistry::UncommittedRowsetRegistry() = default;
+
+UncommittedRowsetRegistry::~UncommittedRowsetRegistry() {
+ if (_dedup_thread_pool) {
+ _dedup_thread_pool->shutdown();
+ }
+}
+
+Status UncommittedRowsetRegistry::init(int dedup_threads) {
+ return ThreadPoolBuilder("UncommittedDedupPool")
+ .set_min_threads(1)
+ .set_max_threads(dedup_threads)
+ .build(&_dedup_thread_pool);
+}
+
+std::shared_ptr<std::mutex> UncommittedRowsetRegistry::_get_tablet_dedup_mutex(
+ Shard& shard, int64_t tablet_id) {
+ auto it = shard.tablet_dedup_mutex.find(tablet_id);
+ if (it == shard.tablet_dedup_mutex.end()) {
+ auto mtx = std::make_shared<std::mutex>();
+ shard.tablet_dedup_mutex[tablet_id] = mtx;
+ return mtx;
+ }
+ return it->second;
+}
+
+void UncommittedRowsetRegistry::register_rowset(std::shared_ptr<UncommittedRowsetEntry> entry) {
+ int64_t tablet_id = entry->tablet_id;
+ auto& shard = _get_shard(tablet_id);
+
+ {
+ std::lock_guard wlock(shard.lock);
+ auto& entries = shard.entries[tablet_id];
+ entries.push_back(entry);
+ // Keep entries sorted by creation_time for correct dedup ordering
+ std::sort(entries.begin(), entries.end(),
+ [](const auto& a, const auto& b) {
+ return a->creation_time < b->creation_time;
+ });
+ }
+
+ // For DUP_KEYS tables, no dedup needed — mark ready immediately
+ if (!entry->unique_key_merge_on_write) {
+ entry->dedup_ready.store(true, std::memory_order_release);
+ return;
+ }
+
+ _submit_dedup_task(tablet_id, entry);
+}
+
+void UncommittedRowsetRegistry::unregister_rowset(int64_t tablet_id, int64_t transaction_id) {
+ auto& shard = _get_shard(tablet_id);
+ std::lock_guard wlock(shard.lock);
+
+ auto it = shard.entries.find(tablet_id);
+ if (it == shard.entries.end()) {
+ return;
+ }
+
+ auto& entries = it->second;
+ entries.erase(std::remove_if(entries.begin(), entries.end(),
+ [transaction_id](const auto& e) {
+ return e->transaction_id == transaction_id;
+ }),
+ entries.end());
+
+ if (entries.empty()) {
+ shard.entries.erase(it);
+ shard.tablet_dedup_mutex.erase(tablet_id);
+ }
+}
+
+void UncommittedRowsetRegistry::get_ready_rowsets(
+ int64_t tablet_id, std::vector<std::shared_ptr<UncommittedRowsetEntry>>* result) {
+ auto& shard = _get_shard(tablet_id);
+ std::shared_lock rlock(shard.lock);
+
+ auto it = shard.entries.find(tablet_id);
+ if (it == shard.entries.end()) {
+ return;
+ }
+
+ for (const auto& entry : it->second) {
+ if (entry->dedup_ready.load(std::memory_order_acquire)) {
+ result->push_back(entry);
+ }
+ }
+}
+
+void UncommittedRowsetRegistry::on_compaction_completed(int64_t tablet_id) {
+ auto& shard = _get_shard(tablet_id);
+ bool has_mow_entries = false;
+
+ {
+ std::shared_lock rlock(shard.lock);
+ auto it = shard.entries.find(tablet_id);
+ if (it == shard.entries.end()) {
+ return;
+ }
+
+ // Invalidate all cross-delete bitmaps and mark not ready
+ for (auto& entry : it->second) {
+ if (entry->unique_key_merge_on_write) {
+ entry->dedup_ready.store(false, std::memory_order_release);
+ entry->cross_delete_bitmap.reset();
+ has_mow_entries = true;
+ }
+ }
+ }
+
+ // Re-compute dedup for all MoW entries
+ if (has_mow_entries) {
+ _recompute_all_dedup(tablet_id);
+ }
+}
+
+void UncommittedRowsetRegistry::on_tablet_state_change(int64_t tablet_id,
+ TabletState new_state) {
+ if (new_state == TABLET_RUNNING) {
+ return; // Only clear on non-running states
+ }
+
+ auto& shard = _get_shard(tablet_id);
+ std::lock_guard wlock(shard.lock);
+ shard.entries.erase(tablet_id);
+ shard.tablet_dedup_mutex.erase(tablet_id);
+}
+
+void UncommittedRowsetRegistry::_submit_dedup_task(
+ int64_t tablet_id, std::shared_ptr<UncommittedRowsetEntry> entry) {
+ if (!_dedup_thread_pool) {
+ // No thread pool available, mark ready without cross-bitmap
+ entry->dedup_ready.store(true, std::memory_order_release);
+ return;
+ }
+
+ // Get the per-tablet dedup mutex
+ std::shared_ptr<std::mutex> dedup_mutex;
+ {
+ auto& shard = _get_shard(tablet_id);
+ std::lock_guard wlock(shard.lock);
+ dedup_mutex = _get_tablet_dedup_mutex(shard, tablet_id);
+ }
+
+ // Collect earlier uncommitted rowsets for this tablet
+ std::vector<RowsetSharedPtr> earlier_rowsets;
+ {
+ auto& shard = _get_shard(tablet_id);
+ std::shared_lock rlock(shard.lock);
+ auto it = shard.entries.find(tablet_id);
+ if (it != shard.entries.end()) {
+ for (const auto& other : it->second) {
+ if (other->creation_time < entry->creation_time &&
+ other->unique_key_merge_on_write) {
+ earlier_rowsets.push_back(other->rowset);
+ }
+ }
+ }
+ }
+
+ // If no earlier uncommitted MoW rowsets, no cross-dedup needed
+ if (earlier_rowsets.empty()) {
+ entry->dedup_ready.store(true, std::memory_order_release);
+ return;
+ }
+
+ // Submit async task to compute cross-uncommitted delete bitmap
+ auto st = _dedup_thread_pool->submit_func(
+ [this, tablet_id, entry, earlier_rowsets = std::move(earlier_rowsets),
+ dedup_mutex]() mutable {
+ // Serialize dedup per tablet
+ std::lock_guard tablet_lock(*dedup_mutex);
+
+ // Check if entry was unregistered while waiting
+ {
+ auto& shard = _get_shard(tablet_id);
+ std::shared_lock rlock(shard.lock);
+ auto it = shard.entries.find(tablet_id);
+ if (it == shard.entries.end()) {
+ return;
+ }
+ bool found = false;
+ for (const auto& e : it->second) {
+ if (e->transaction_id == entry->transaction_id) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ return;
+ }
+ }
+
+ // Get tablet for bitmap computation
+ auto tablet_result =
+ ExecEnv::GetInstance()->storage_engine().get_tablet(tablet_id);
+ if (!tablet_result.has_value()) {
+ LOG(WARNING) << "Failed to get tablet " << tablet_id
+ << " for cross-uncommitted dedup, marking ready without "
+ "cross-bitmap";
+ entry->dedup_ready.store(true, std::memory_order_release);
+ return;
+ }
+ auto tablet = tablet_result.value();
+
+ // Load segments from the rowset
+ auto* beta_rowset = dynamic_cast<BetaRowset*>(entry->rowset.get());
+ if (!beta_rowset) {
+ entry->dedup_ready.store(true, std::memory_order_release);
+ return;
+ }
+ std::vector<segment_v2::SegmentSharedPtr> segments;
+ auto load_st = beta_rowset->load_segments(&segments);
+ if (!load_st.ok()) {
+ LOG(WARNING) << "Failed to load segments for cross-uncommitted dedup, "
+ "tablet_id="
+ << tablet_id << " txn_id=" << entry->transaction_id
+ << ": " << load_st;
+ entry->dedup_ready.store(true, std::memory_order_release);
+ return;
+ }
+
+ if (segments.empty()) {
+ entry->dedup_ready.store(true, std::memory_order_release);
+ return;
+ }
+
+ // Use CalcDeleteBitmapExecutor for the actual computation
+ auto* calc_executor =
+ ExecEnv::GetInstance()->storage_engine().calc_delete_bitmap_executor();
+ if (!calc_executor) {
+ entry->dedup_ready.store(true, std::memory_order_release);
+ return;
+ }
+ auto token = calc_executor->create_token();
+
+ // Compute cross-uncommitted delete bitmap
+ auto cross_bitmap = std::make_shared<DeleteBitmap>(tablet_id);
+ auto calc_st = BaseTablet::calc_delete_bitmap(
+ tablet, entry->rowset, segments, earlier_rowsets, cross_bitmap,
+ DeleteBitmap::TEMP_VERSION_COMMON, token.get());
+ if (calc_st.ok()) {
+ calc_st = token->wait();
+ }
+
+ if (calc_st.ok()) {
+ entry->cross_delete_bitmap = cross_bitmap;
+ } else {
+ LOG(WARNING) << "Failed to compute cross-uncommitted delete bitmap, "
+ "tablet_id="
+ << tablet_id << " txn_id=" << entry->transaction_id
+ << ": " << calc_st;
+ }
+
+ entry->dedup_ready.store(true, std::memory_order_release);
+ });
+
+ if (!st.ok()) {
+ LOG(WARNING) << "Failed to submit cross-uncommitted dedup task, tablet_id=" << tablet_id
+ << " txn_id=" << entry->transaction_id << ": " << st;
+ entry->dedup_ready.store(true, std::memory_order_release);
+ }
+}
+
+void UncommittedRowsetRegistry::_recompute_all_dedup(int64_t tablet_id) {
+ std::vector<std::shared_ptr<UncommittedRowsetEntry>> mow_entries;
+
+ {
+ auto& shard = _get_shard(tablet_id);
+ std::shared_lock rlock(shard.lock);
+ auto it = shard.entries.find(tablet_id);
+ if (it == shard.entries.end()) {
+ return;
+ }
+ for (auto& entry : it->second) {
+ if (entry->unique_key_merge_on_write) {
+ mow_entries.push_back(entry);
+ }
+ }
+ }
+
+ // Re-submit dedup tasks in order (they will serialize via tablet dedup mutex)
+ for (auto& entry : mow_entries) {
+ _submit_dedup_task(tablet_id, entry);
+ }
+}
+
+UncommittedRowsetRegistry* get_uncommitted_rowset_registry() {
+ auto* env = ExecEnv::GetInstance();
+ if (!env) {
+ return nullptr;
+ }
+ if (config::is_cloud_mode()) {
+ return env->storage_engine().to_cloud().uncommitted_rowset_registry();
+ } else {
+ return env->storage_engine().to_local().uncommitted_rowset_registry();
+ }
+}
+
+} // namespace doris
diff --git a/be/src/olap/uncommitted_rowset_registry.h b/be/src/olap/uncommitted_rowset_registry.h
new file mode 100644
index 00000000000..e2a07634cc3
--- /dev/null
+++ b/be/src/olap/uncommitted_rowset_registry.h
@@ -0,0 +1,121 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <atomic>
+#include <memory>
+#include <mutex>
+#include <shared_mutex>
+#include <unordered_map>
+#include <vector>
+
+#include "olap/rowset/rowset.h"
+#include "olap/tablet_fwd.h"
+#include "olap/tablet_meta.h"
+#include "util/threadpool.h"
+
+namespace doris {
+
+// Represents a single uncommitted rowset tracked by the registry.
+// The committed_delete_bitmap is computed at commit time (vs published rowsets).
+// The cross_delete_bitmap is computed asynchronously (vs earlier uncommitted rowsets).
+struct UncommittedRowsetEntry {
+ RowsetSharedPtr rowset;
+ int64_t transaction_id;
+ int64_t partition_id;
+ int64_t tablet_id;
+ // Delete bitmap computed at commit phase against published rowsets (layer 2)
+ DeleteBitmapPtr committed_delete_bitmap;
+ // Delete bitmap computed async against earlier uncommitted rowsets (layer 3)
+ DeleteBitmapPtr cross_delete_bitmap;
+ bool unique_key_merge_on_write;
+ int64_t creation_time; // for ordering: later wins in dedup
+ std::atomic<bool> dedup_ready {false}; // only serve to queries when true
+};
+
+// UncommittedRowsetRegistry tracks uncommitted rowsets across all tablets on this BE.
+// It supports the READ UNCOMMITTED isolation level by making uncommitted data visible
+// to queries that opt in via `SET read_uncommitted = true`.
+//
+// Thread safety: Uses sharded locks for the main map and per-tablet mutexes for
+// serializing async dedup computation.
+class UncommittedRowsetRegistry {
+public:
+ UncommittedRowsetRegistry();
+ ~UncommittedRowsetRegistry();
+
+ Status init(int dedup_threads);
+
+ // === Write Path ===
+
+ // Register an uncommitted rowset. For MoW unique key tables, this triggers
+ // async dedup computation against earlier uncommitted rowsets.
+ void register_rowset(std::shared_ptr<UncommittedRowsetEntry> entry);
+
+ // Unregister on publish or rollback. Removes the entry and any associated state.
+ void unregister_rowset(int64_t tablet_id, int64_t transaction_id);
+
+ // === Read Path ===
+
+ // Get all dedup-ready uncommitted rowsets for a tablet.
+ void get_ready_rowsets(int64_t tablet_id,
+ std::vector<std::shared_ptr<UncommittedRowsetEntry>>* result);
+
+ // === Compaction Path ===
+
+ // Called after compaction modifies published rowsets. Invalidates cross-delete
+ // bitmaps for all uncommitted rowsets on this tablet and re-computes them.
+ void on_compaction_completed(int64_t tablet_id);
+
+ // === Schema Change Path ===
+
+ // Called when a tablet transitions to a non-running state. Clears all entries.
+ void on_tablet_state_change(int64_t tablet_id, TabletState new_state);
+
+private:
+ static constexpr int SHARD_COUNT = 16;
+
+ struct Shard {
+ mutable std::shared_mutex lock;
+ // tablet_id -> list of uncommitted entries (ordered by creation_time)
+ std::unordered_map<int64_t, std::vector<std::shared_ptr<UncommittedRowsetEntry>>> entries;
+ // Per-tablet mutex for serializing dedup computation
+ std::unordered_map<int64_t, std::shared_ptr<std::mutex>> tablet_dedup_mutex;
+ };
+
+ Shard& _get_shard(int64_t tablet_id) { return _shards[tablet_id % SHARD_COUNT]; }
+ const Shard& _get_shard(int64_t tablet_id) const { return _shards[tablet_id % SHARD_COUNT]; }
+
+ // Get or create the per-tablet dedup mutex (must hold shard write lock)
+ std::shared_ptr<std::mutex> _get_tablet_dedup_mutex(Shard& shard, int64_t tablet_id);
+
+ // Submit async dedup task for one entry against earlier uncommitted rowsets
+ void _submit_dedup_task(int64_t tablet_id, std::shared_ptr<UncommittedRowsetEntry> entry);
+
+ // Recompute all cross-delete bitmaps for a tablet (after compaction)
+ void _recompute_all_dedup(int64_t tablet_id);
+
+ Shard _shards[SHARD_COUNT];
+ std::unique_ptr<ThreadPool> _dedup_thread_pool;
+};
+
+// Helper function to get the UncommittedRowsetRegistry from the current storage engine.
+// Works for both local and cloud mode. Returns nullptr if not available.
+UncommittedRowsetRegistry* get_uncommitted_rowset_registry();
+
+} // namespace doris
diff --git a/be/src/pipeline/exec/olap_scan_operator.cpp b/be/src/pipeline/exec/olap_scan_operator.cpp
index 44fe060066a..cb8aecaae6d 100644
--- a/be/src/pipeline/exec/olap_scan_operator.cpp
+++ b/be/src/pipeline/exec/olap_scan_operator.cpp
@@ -33,6 +33,7 @@
#include "olap/rowset/segment_v2/ann_index/ann_topn_runtime.h"
#include "olap/storage_engine.h"
#include "olap/tablet_manager.h"
+#include "olap/uncommitted_rowset_registry.h"
#include "pipeline/exec/scan_operator.h"
#include "pipeline/query_cache/query_cache.h"
#include "runtime/runtime_state.h"
@@ -744,6 +745,70 @@ Status OlapScanLocalState::prepare(RuntimeState* state) {
print_id(PipelineXLocalState<>::_state->query_id()));
}
}
+ // Inject uncommitted rowsets for READ UNCOMMITTED isolation level
+ if (_state->read_uncommitted()) {
+ auto* registry = get_uncommitted_rowset_registry();
+ if (registry) {
+ for (size_t i = 0; i < _scan_ranges.size(); i++) {
+ auto& tablet = _tablets[i].tablet;
+ // Only DUP_KEYS and UNIQUE_KEYS supported
+ if (tablet->keys_type() != DUP_KEYS &&
+ tablet->keys_type() != UNIQUE_KEYS) {
+ continue;
+ }
+ // Skip tablets not in RUNNING state
+ if (tablet->tablet_state() != TABLET_RUNNING) {
+ continue;
+ }
+
+ std::vector<std::shared_ptr<UncommittedRowsetEntry>> ready_entries;
+ registry->get_ready_rowsets(tablet->tablet_id(), &ready_entries);
+ if (ready_entries.empty()) {
+ continue;
+ }
+
+ bool bitmap_copied = false;
+ auto ensure_bitmap_copy = [&]() {
+ if (!bitmap_copied) {
+ if (_read_sources[i].delete_bitmap) {
+ _read_sources[i].delete_bitmap =
+ std::make_shared<DeleteBitmap>(*_read_sources[i].delete_bitmap);
+ } else {
+ _read_sources[i].delete_bitmap =
+ std::make_shared<DeleteBitmap>(tablet->tablet_id());
+ }
+ bitmap_copied = true;
+ }
+ };
+
+ for (auto& entry : ready_entries) {
+ RowsetReaderSharedPtr rs_reader;
+ auto st = entry->rowset->create_reader(&rs_reader);
+ if (!st.ok()) {
+ LOG(WARNING) << "Failed to create reader for uncommitted rowset, "
+ "tablet_id="
+ << tablet->tablet_id()
+ << " txn_id=" << entry->transaction_id << ": " << st;
+ continue;
+ }
+ _read_sources[i].rs_splits.emplace_back(std::move(rs_reader));
+
+ bool is_mow = entry->unique_key_merge_on_write;
+ // Merge committed-vs-published bitmap (layer 2)
+ if (is_mow && entry->committed_delete_bitmap) {
+ ensure_bitmap_copy();
+ _read_sources[i].delete_bitmap->merge(*entry->committed_delete_bitmap);
+ }
+ // Merge cross-uncommitted bitmap (layer 3)
+ if (is_mow && entry->cross_delete_bitmap) {
+ ensure_bitmap_copy();
+ _read_sources[i].delete_bitmap->merge(*entry->cross_delete_bitmap);
+ }
+ }
+ }
+ }
+ }
+
timer.stop();
double cost_secs = static_cast<double>(timer.elapsed_time()) / NANOS_PER_SEC;
if (cost_secs > 1) {
diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h
index 70234928c64..3901c6d3eeb 100644
--- a/be/src/runtime/runtime_state.h
+++ b/be/src/runtime/runtime_state.h
@@ -436,6 +436,10 @@ public:
return _query_options.__isset.skip_missing_version && _query_options.skip_missing_version;
}
+ bool read_uncommitted() const {
+ return _query_options.__isset.read_uncommitted && _query_options.read_uncommitted;
+ }
+
int64_t data_queue_max_blocks() const {
return _query_options.__isset.data_queue_max_blocks ? _query_options.data_queue_max_blocks
: 1;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 9468a75ce3b..2c129f680cc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -447,6 +447,8 @@ public class SessionVariable implements Serializable, Writable {
public static final String SKIP_BAD_TABLET = "skip_bad_tablet";
+ public static final String READ_UNCOMMITTED = "read_uncommitted";
+
public static final String ENABLE_PUSH_DOWN_NO_GROUP_AGG = "enable_push_down_no_group_agg";
public static final String ENABLE_CBO_STATISTICS = "enable_cbo_statistics";
@@ -770,7 +772,8 @@ public class SessionVariable implements Serializable, Writable {
SKIP_STORAGE_ENGINE_MERGE,
SKIP_MISSING_VERSION,
SKIP_BAD_TABLET,
- SHOW_HIDDEN_COLUMNS
+ SHOW_HIDDEN_COLUMNS,
+ READ_UNCOMMITTED
);
public static final String ENABLE_STATS = "enable_stats";
@@ -2059,6 +2062,14 @@ public class SessionVariable implements Serializable, Writable {
affectQueryResultInExecution = true)
public boolean skipBadTablet = false;
+ @VariableMgr.VarAttr(name = READ_UNCOMMITTED, needForward = true,
+ varType = VariableAnnotation.EXPERIMENTAL,
+ description = {"启用 READ UNCOMMITTED 隔离级别，允许查询读取未提交的 rowset 数据（实验特性）",
+ "Enable READ UNCOMMITTED isolation level to allow queries to read "
+ + "uncommitted rowset data (experimental)"},
+ affectQueryResultInPlan = true, affectQueryResultInExecution = true)
+ public boolean readUncommitted = false;
+
// This variable is used to avoid FE fallback to the original parser. When we execute SQL in regression tests
// for nereids, fallback will cause the Doris return the correct result although the syntax is unsupported
// in nereids for some mistaken modification. You should set it on the
@@ -3589,7 +3600,7 @@ public class SessionVariable implements Serializable, Writable {
public boolean isInDebugMode() {
return showHiddenColumns || skipDeleteBitmap || skipDeletePredicate || skipDeleteSign || skipStorageEngineMerge
- || skipMissingVersion || skipBadTablet;
+ || skipMissingVersion || skipBadTablet || readUncommitted;
}
public String printDebugModeVariables() {
@@ -5067,6 +5078,8 @@ public class SessionVariable implements Serializable, Writable {
tResult.setSkipDeleteBitmap(skipDeleteBitmap);
+ tResult.setReadUncommitted(readUncommitted);
+
tResult.setEnableFileCache(enableFileCache);
tResult.setEnablePageCache(enablePageCache);
diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift
index 6709520641b..455e7bf3653 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -426,6 +426,8 @@ struct TQueryOptions {
185: optional bool enable_parquet_file_page_cache = true;
+ 186: optional bool read_uncommitted = false;
+
195: optional bool enable_left_semi_direct_return_opt;
// For cloud, to control if the content would be written into file cache
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org