This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 648697d1414 [fix](be) Avoid remote IO under cache shard locks (#69324)
648697d1414 is described below
commit 648697d14141972e617f086b689464674c34cd7c
Author: daidai <[email protected]>
AuthorDate: Thu Oct 22 10:01:34 2026 +0800
[fix](be) Avoid remote IO under cache shard locks (#69324)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary:
`KVCache` executed value factories while holding the shard mutex.
Iceberg and Paimon deletion-vector loading could therefore block
unrelated cache keys during remote I/O and decoding, and different value
types could share the same string key.
This PR moves factory execution to a per-key lock, keeps one in-flight
load per key, isolates entries by value type, and simplifies Fluss tail
loading to use the same single-flight path.
### Release note
None
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- 55 targeted ASAN tests passed: KVCache concurrency, Iceberg V1/V2
deletion vectors, Paimon deletion vectors, and Fluss cache paths.
- clang-format and build hygiene checks passed.
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [x] No.
- [ ] Yes.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/format/format_common.h | 77 ++++++---
be/src/format_v2/table/fluss_union_lake_reader.cpp | 34 ++--
be/test/format/format_common_test.cpp | 174 +++++++++++++++++++++
3 files changed, 247 insertions(+), 38 deletions(-)
diff --git a/be/src/format/format_common.h b/be/src/format/format_common.h
index 2d65ed721ef..eb3eed7d148 100644
--- a/be/src/format/format_common.h
+++ b/be/src/format/format_common.h
@@ -17,6 +17,10 @@
#pragma once
+#include <memory>
+#include <typeindex>
+
+#include "core/data_type/primitive_type.h"
#include "core/types.h"
#include "exec/common/int_exp.h"
@@ -71,40 +75,47 @@ template <typename KType>
class KVCache {
public:
KVCache() = default;
+ ~KVCache() = default;
- ~KVCache() {
- for (auto& kv : _storage) {
- _delete_fn[kv.first](kv.second);
- }
- }
-
+ template <class T>
void erase(const KType& key) {
- std::lock_guard<std::mutex> lock(_lock);
- auto it = _storage.find(key);
- if (it != _storage.end()) {
- _delete_fn[key](_storage[key]);
- _storage.erase(key);
- _delete_fn.erase(key);
+ std::shared_ptr<Entry> erased_entry;
+ {
+ std::lock_guard<std::mutex> lock(_lock);
+ auto key_it = _entries.find(key);
+ if (key_it == _entries.end()) {
+ return;
+ }
+ auto type_it = key_it->second.find(typeid(T));
+ if (type_it == key_it->second.end()) {
+ return;
+ }
+ erased_entry = std::move(type_it->second);
+ key_it->second.erase(type_it);
+ if (key_it->second.empty()) {
+ _entries.erase(key_it);
+ }
}
+ // Destroy the cached value outside the shard lock.
}
template <class T>
T* get(const KType& key, const std::function<T*()> create_func, bool*
cache_hit = nullptr) {
- std::lock_guard<std::mutex> lock(_lock);
- auto it = _storage.find(key);
- if (it != _storage.end()) {
+ std::shared_ptr<Entry> entry = _get_or_create_entry<T>(key);
+ std::lock_guard<std::mutex> entry_lock(entry->_lock);
+ if (entry->_value != nullptr) {
if (cache_hit != nullptr) {
*cache_hit = true;
}
- return reinterpret_cast<T*>(it->second);
+ return static_cast<T*>(entry->_value);
} else {
if (cache_hit != nullptr) {
*cache_hit = false;
}
T* rawPtr = create_func();
if (rawPtr != nullptr) {
- _delete_fn[key] = [](void* obj) { delete
reinterpret_cast<T*>(obj); };
- _storage[key] = rawPtr;
+ entry->_delete_fn = [](void* obj) { delete
static_cast<T*>(obj); };
+ entry->_value = rawPtr;
}
return rawPtr;
}
@@ -113,9 +124,35 @@ public:
private:
using DeleteFn = void (*)(void*);
+ struct Entry {
+ ~Entry() {
+ if (_value != nullptr) {
+ _delete_fn(_value);
+ }
+ }
+
+ // The shard lock only protects the map. Factory calls hold this entry
lock, so an
+ // unrelated key can keep loading while another key is doing remote
I/O.
+ std::mutex _lock;
+ void* _value = nullptr;
+ DeleteFn _delete_fn = nullptr;
+ };
+
+ template <class T>
+ std::shared_ptr<Entry> _get_or_create_entry(const KType& key) {
+ std::lock_guard<std::mutex> lock(_lock);
+ auto key_it = _entries.try_emplace(key).first;
+ auto type_it = key_it->second.find(typeid(T));
+ if (type_it != key_it->second.end()) {
+ return type_it->second;
+ }
+ auto entry = std::make_shared<Entry>();
+ key_it->second.emplace(typeid(T), entry);
+ return entry;
+ }
+
std::mutex _lock;
- std::unordered_map<KType, DeleteFn> _delete_fn;
- std::unordered_map<KType, void*> _storage;
+ std::unordered_map<KType, std::unordered_map<std::type_index,
std::shared_ptr<Entry>>> _entries;
};
class ShardedKVCache {
diff --git a/be/src/format_v2/table/fluss_union_lake_reader.cpp
b/be/src/format_v2/table/fluss_union_lake_reader.cpp
index d49803943f3..b381ac67684 100644
--- a/be/src/format_v2/table/fluss_union_lake_reader.cpp
+++ b/be/src/format_v2/table/fluss_union_lake_reader.cpp
@@ -354,27 +354,25 @@ Status FlussUnionLakeReader::_load_suppression_keys(const
format::SplitReadOptio
// Length-prefixed so that no boundary between the fixed prefix and the
tail can be reinterpreted
// as part of the tail itself. One scan node reads one table, so the tail
alone identifies it.
const auto cache_key = fmt::format("fluss_union_tail:{}:{}",
tail.spec.size(), tail.spec);
+ Status read_status = Status::OK();
bool cache_hit = false;
auto* cached = options.cache->get<SuppressionKeys>(
- cache_key, []() -> SuppressionKeys* { return nullptr; },
&cache_hit);
- if (cached == nullptr) {
- // A tail read is network I/O. KVCache invokes its create callback
while holding a shard
- // mutex, so do that I/O before the insertion attempt; otherwise one
slow bucket blocks every
- // unrelated tail that hashes to the same shard.
- auto loaded = std::make_unique<SuppressionKeys>();
- RETURN_IF_ERROR(_read_tail_keys(tail, loaded.get()));
- update_counter(_tail_keys_read_counter, loaded->records);
-
- bool raced_cache_hit = false;
- cached = options.cache->get<SuppressionKeys>(
- cache_key, [&]() -> SuppressionKeys* { return
loaded.release(); },
- &raced_cache_hit);
- cache_hit = raced_cache_hit;
- if (!raced_cache_hit) {
- update_counter(_tail_keys_retained_counter,
cast_set<int64_t>(cached->keys.rows()));
- }
- }
+ cache_key,
+ [&]() -> SuppressionKeys* {
+ auto loaded = std::make_unique<SuppressionKeys>();
+ read_status = _read_tail_keys(tail, loaded.get());
+ if (!read_status.ok()) {
+ return nullptr;
+ }
+ update_counter(_tail_keys_read_counter, loaded->records);
+ return loaded.release();
+ },
+ &cache_hit);
+ RETURN_IF_ERROR(read_status);
DORIS_CHECK(cached != nullptr);
+ if (!cache_hit) {
+ update_counter(_tail_keys_retained_counter,
cast_set<int64_t>(cached->keys.rows()));
+ }
if (cache_hit) {
update_counter(_tail_cache_hit_counter, 1);
}
diff --git a/be/test/format/format_common_test.cpp
b/be/test/format/format_common_test.cpp
new file mode 100644
index 00000000000..051b5b36b55
--- /dev/null
+++ b/be/test/format/format_common_test.cpp
@@ -0,0 +1,174 @@
+// 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 "format/format_common.h"
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <memory>
+#include <thread>
+#include <vector>
+
+namespace doris {
+
+TEST(KVCacheTest, DifferentKeysOnSameShardLoadConcurrently) {
+ ShardedKVCache cache(1);
+ std::promise<void> first_entered;
+ std::promise<void> release_first;
+ std::promise<void> second_entered;
+ auto first_entered_future = first_entered.get_future();
+ auto release_first_future = release_first.get_future().share();
+ auto second_entered_future = second_entered.get_future();
+
+ std::thread first([&]() {
+ cache.get<int>("first", [&]() -> int* {
+ first_entered.set_value();
+ release_first_future.wait();
+ return new int(1);
+ });
+ });
+ first_entered_future.wait();
+
+ std::thread second([&]() {
+ cache.get<int>("second", [&]() -> int* {
+ second_entered.set_value();
+ return new int(2);
+ });
+ });
+
+ bool second_loaded_before_first =
+ second_entered_future.wait_for(std::chrono::seconds(1)) ==
std::future_status::ready;
+ release_first.set_value();
+ first.join();
+ second.join();
+
+ EXPECT_TRUE(second_loaded_before_first);
+}
+
+TEST(KVCacheTest, SameKeyLoadsOnceForConcurrentCallers) {
+ constexpr int num_callers = 8;
+ ShardedKVCache cache(1);
+ std::atomic<int> factory_calls {0};
+ std::promise<void> factory_entered;
+ std::promise<void> release_factory;
+ auto factory_entered_future = factory_entered.get_future();
+ auto release_factory_future = release_factory.get_future().share();
+ std::vector<int*> values(num_callers);
+ std::vector<std::thread> callers;
+
+ for (int i = 0; i < num_callers; ++i) {
+ callers.emplace_back([&, i]() {
+ values[i] = cache.get<int>("shared", [&]() -> int* {
+ if (++factory_calls == 1) {
+ factory_entered.set_value();
+ }
+ release_factory_future.wait();
+ return new int(7);
+ });
+ });
+ }
+
+ factory_entered_future.wait();
+ release_factory.set_value();
+ for (auto& caller : callers) {
+ caller.join();
+ }
+
+ EXPECT_EQ(factory_calls.load(), 1);
+ ASSERT_NE(values.front(), nullptr);
+ for (int* value : values) {
+ EXPECT_EQ(value, values.front());
+ }
+}
+
+TEST(KVCacheTest, FailedFactoryDoesNotPoisonEntry) {
+ ShardedKVCache cache(1);
+ std::atomic<int> factory_calls {0};
+ std::promise<void> failed_factory_entered;
+ std::promise<void> release_failed_factory;
+ auto failed_factory_entered_future = failed_factory_entered.get_future();
+ auto release_failed_factory_future =
release_failed_factory.get_future().share();
+ bool first_cache_hit = true;
+ bool second_cache_hit = true;
+ int* second_value = nullptr;
+
+ std::thread first([&]() {
+ EXPECT_EQ(cache.get<int>(
+ "retry",
+ [&]() -> int* {
+ if (++factory_calls == 1) {
+ failed_factory_entered.set_value();
+ }
+ release_failed_factory_future.wait();
+ return nullptr;
+ },
+ &first_cache_hit),
+ nullptr);
+ });
+ failed_factory_entered_future.wait();
+
+ std::thread second([&]() {
+ second_value = cache.get<int>(
+ "retry",
+ [&]() -> int* {
+ ++factory_calls;
+ return new int(9);
+ },
+ &second_cache_hit);
+ });
+
+ release_failed_factory.set_value();
+ first.join();
+ second.join();
+
+ EXPECT_EQ(factory_calls.load(), 2);
+ EXPECT_FALSE(first_cache_hit);
+ EXPECT_FALSE(second_cache_hit);
+ ASSERT_NE(second_value, nullptr);
+ EXPECT_EQ(*second_value, 9);
+}
+
+TEST(KVCacheTest, SameStringKeyIsIsolatedByValueType) {
+ struct FirstValue {
+ int value = 1;
+ };
+ struct SecondValue {
+ int value = 2;
+ };
+
+ ShardedKVCache cache(1);
+ auto* first =
+ cache.get<FirstValue>("same-key", []() -> FirstValue* { return new
FirstValue(); });
+ auto* second =
+ cache.get<SecondValue>("same-key", []() -> SecondValue* { return
new SecondValue(); });
+
+ ASSERT_NE(first, nullptr);
+ ASSERT_NE(second, nullptr);
+ EXPECT_EQ(first->value, 1);
+ EXPECT_EQ(second->value, 2);
+ EXPECT_EQ(cache.get<FirstValue>("same-key",
+ []() -> FirstValue* {
+ ADD_FAILURE() << "cached FirstValue
was rebuilt";
+ return nullptr;
+ }),
+ first);
+}
+
+} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
