Copilot commented on code in PR #60586:
URL: https://github.com/apache/doris/pull/60586#discussion_r2778315768
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -3589,7 +3600,7 @@ public String printFuzzyVariables() {
public boolean isInDebugMode() {
return showHiddenColumns || skipDeleteBitmap || skipDeletePredicate ||
skipDeleteSign || skipStorageEngineMerge
- || skipMissingVersion || skipBadTablet;
+ || skipMissingVersion || skipBadTablet || readUncommitted;
Review Comment:
`read_uncommitted` is added to `DEBUG_VARIABLES`/`isInDebugMode()`, which
will unexpectedly put the session into “debug mode” and therefore disables MV
rewrite and forbids UPDATE/DELETE (see callers of `isInDebugMode()`). This
looks like a functional behavior change unrelated to READ UNCOMMITTED; consider
removing it from `DEBUG_VARIABLES` and `isInDebugMode()` (or introduce a
separate flag/check) so enabling READ UNCOMMITTED doesn’t block DML or planner
features.
```suggestion
|| skipMissingVersion || skipBadTablet;
```
##########
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;
+ }
+ }
Review Comment:
Compaction changes published rowsets/row-ids, so `committed_delete_bitmap`
(layer 2) computed at commit time against published rowsets can become stale
after compaction. `on_compaction_completed()` currently only clears/recomputes
`cross_delete_bitmap` (layer 3), which risks incorrect MoW dedup under READ
UNCOMMITTED after compaction. Consider also invalidating and recomputing the
committed-vs-published bitmap (or otherwise ensuring it is mapped to the
post-compaction rowsets).
##########
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;
+ }
Review Comment:
`on_compaction_completed()` mutates `UncommittedRowsetEntry` state
(`dedup_ready`, `cross_delete_bitmap`) while only holding a shard
`shared_lock`, and readers access these fields without any locking (e.g. scan
injection reads `entry->cross_delete_bitmap`). This can cause a real data race
on the `std::shared_ptr` assignment/reset. Consider making entry updates
copy-on-write (replace the shared_ptr entry object in the vector), or guard
bitmap pointers with synchronization (e.g. per-entry mutex or
`std::atomic<std::shared_ptr<...>>`), and use an exclusive shard lock for
updates.
```suggestion
std::lock_guard wlock(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;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
