This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new eafe14e0 fix(build): repair singleton double-checked locking race and
harden aarch64 portability (#203)
eafe14e0 is described below
commit eafe14e0208c9d6e815cdebf826298c404949fbc
Author: kid <[email protected]>
AuthorDate: Fri Aug 21 17:48:25 2026 +0800
fix(build): repair singleton double-checked locking race and harden aarch64
portability (#203)
---
include/paimon/factories/singleton.h | 39 +++++-
src/paimon/CMakeLists.txt | 3 +
src/paimon/common/factories/io_hook.cpp | 65 ++++++---
src/paimon/common/factories/io_hook.h | 7 +-
src/paimon/common/factories/io_hook_test.cpp | 81 +++++++++++
src/paimon/common/factories/singleton.cpp | 18 +--
src/paimon/common/factories/singleton_test.cpp | 108 ++++++++++++++
src/paimon/common/io/cache/cache_manager.h | 9 +-
src/paimon/common/io/cache/cache_manager_test.cpp | 155 +++++++++++++++++++++
src/paimon/common/sst/sst_file_writer.cpp | 5 +-
src/paimon/common/utils/read_ahead_cache_test.cpp | 5 +-
src/paimon/common/utils/saturating_cast.h | 53 +++++++
src/paimon/common/utils/saturating_cast_test.cpp | 74 ++++++++++
src/paimon/common/utils/serialization_utils.h | 4 +-
.../common/utils/serialization_utils_test.cpp | 48 +++++++
15 files changed, 629 insertions(+), 45 deletions(-)
diff --git a/include/paimon/factories/singleton.h
b/include/paimon/factories/singleton.h
index 6e12d456..a3030e5e 100644
--- a/include/paimon/factories/singleton.h
+++ b/include/paimon/factories/singleton.h
@@ -19,7 +19,9 @@
#pragma once
+#include <atomic>
#include <memory>
+#include <mutex>
#include "paimon/macros.h"
#include "paimon/visibility.h"
@@ -30,9 +32,9 @@ class PAIMON_EXPORT LazyInstantiation {
protected:
template <typename T>
static void Create(T*& ptr) {
- T* tmp = new T;
- MEMORY_BARRIER();
- ptr = tmp;
+ // Publication ordering is handled by the release store in
+ // Singleton<T, InstPolicy>::GetInstance(), so no barrier is needed
here.
+ ptr = new T;
static std::shared_ptr<T> destroyer(ptr);
}
};
@@ -56,4 +58,35 @@ class PAIMON_EXPORT Singleton : private InstPolicy {
static T* GetInstance();
};
+template <typename T, typename InstPolicy>
+T* Singleton<T, InstPolicy>::GetInstance() {
+ static std::atomic<T*> ptr{nullptr};
+ static std::mutex mutex;
+ T* p = ptr.load(std::memory_order_acquire);
+ if (PAIMON_UNLIKELY(p == nullptr)) {
+ std::lock_guard<std::mutex> lg(mutex);
+ // Re-check under the mutex with a relaxed load; the mutex already
+ // synchronizes with the creating thread.
+ p = ptr.load(std::memory_order_relaxed);
+ if (p == nullptr) {
+ InstPolicy::Create(p);
+ ptr.store(p, std::memory_order_release);
+ }
+ }
+ return p;
+}
+
+// FactoryCreator and IOHook are instantiated exactly once in singleton.cpp,
and the
+// extern declarations below suppress implicit instantiation everywhere else.
The
+// file-format/file-system plugins are separate shared libraries linked with
+// -Bsymbolic, so a per-library copy of GetInstance()'s function-local static
state
+// would never be interposed: factory registrations would land in a different
+// instance than lookups. Do not replace these with implicit instantiation.
Types local to a single
+// translation unit (e.g. test-only types) can still instantiate Singleton<T>
+// implicitly because they cannot span library boundaries.
+class FactoryCreator;
+class IOHook;
+extern template class Singleton<FactoryCreator>;
+extern template class Singleton<IOHook>;
+
} // namespace paimon
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index bdb11005..adfd968d 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -651,7 +651,9 @@ if(PAIMON_BUILD_TESTS)
common/utils/range_helper_test.cpp
common/utils/read_ahead_cache_test.cpp
common/io/cache/lru_cache_test.cpp
+ common/io/cache/cache_manager_test.cpp
common/utils/byte_range_combiner_test.cpp
+ common/utils/saturating_cast_test.cpp
common/utils/scope_guard_test.cpp
common/utils/sensitive_config_utils_test.cpp
common/utils/serialization_utils_test.cpp
@@ -682,6 +684,7 @@ if(PAIMON_BUILD_TESTS)
add_paimon_test(common_factories_test
SOURCES
+ common/factories/singleton_test.cpp
common/factories/factory_creator_test.cpp
common/factories/io_hook_test.cpp
STATIC_LINK_LIBS
diff --git a/src/paimon/common/factories/io_hook.cpp
b/src/paimon/common/factories/io_hook.cpp
index a0576b46..394dc8ea 100644
--- a/src/paimon/common/factories/io_hook.cpp
+++ b/src/paimon/common/factories/io_hook.cpp
@@ -19,9 +19,12 @@
#include "paimon/common/factories/io_hook.h"
#include <atomic>
+#include <mutex>
+#include <shared_mutex>
#include <stdexcept>
#include "fmt/format.h"
+#include "paimon/macros.h"
#include "paimon/status.h"
namespace paimon {
@@ -29,42 +32,66 @@ namespace paimon {
class IOHook::Impl {
public:
Status Try(const std::string& path) {
- if (io_count_.fetch_add(1) < pos_.load()) {
- return Status::OK();
- } else {
- switch (mode_) {
- case IOHook::Mode::SILENT:
- return Status::OK();
- case IOHook::Mode::RETURN_ERROR:
- return Status::IOError(fmt::format(
- "io hook triggered io error at position {}, path {}",
pos_.load(), path));
- case IOHook::Mode::THROW_EXCEPTION:
- throw std::runtime_error(fmt::format(
- "io hook throw io exception at position {}, path {}",
pos_.load(), path));
- return Status::OK();
- default:
- return Status::OK();
- }
+ // Fast path: the hook is disabled, which is always the case in
production;
+ // writers (Reset()/Clear()) only exist in tests. This keeps Try() a
single
+ // atomic load on the IO path instead of a shared_mutex acquisition
per IO.
+ if (PAIMON_UNLIKELY(armed_.load(std::memory_order_acquire))) {
+ return TryArmed(path);
}
+ return Status::OK();
}
inline void Reset(int64_t pos, IOHook::Mode mode) {
+ std::unique_lock<std::shared_mutex> lock(mutex_);
+ mode_ = mode;
pos_ = pos;
io_count_ = 0;
- mode_ = mode;
+ // Arm only after the configuration is complete: TryArmed() reads
mode_/pos_
+ // under mutex_, which synchronizes with this store, so an observed
armed state
+ // always implies a complete configuration.
+ armed_.store(true, std::memory_order_release);
}
int64_t IOCount() const {
+ std::shared_lock<std::shared_mutex> lock(mutex_);
return io_count_.load();
}
void Clear() {
- Reset(-1, IOHook::Mode::SILENT);
+ std::unique_lock<std::shared_mutex> lock(mutex_);
+ // Disarm first so IO threads stop taking the lock as soon as possible.
+ armed_.store(false, std::memory_order_release);
+ mode_ = IOHook::Mode::SILENT;
+ pos_ = -1;
+ io_count_ = 0;
}
private:
+ Status TryArmed(const std::string& path) {
+ std::shared_lock<std::shared_mutex> lock(mutex_);
+ if (io_count_.fetch_add(1) < pos_) {
+ return Status::OK();
+ } else {
+ switch (mode_) {
+ case IOHook::Mode::SILENT:
+ return Status::OK();
+ case IOHook::Mode::RETURN_ERROR:
+ return Status::IOError(fmt::format(
+ "io hook triggered io error at position {}, path {}",
pos_, path));
+ case IOHook::Mode::THROW_EXCEPTION:
+ throw std::runtime_error(fmt::format(
+ "io hook throw io exception at position {}, path {}",
pos_, path));
+ return Status::OK();
+ default:
+ return Status::OK();
+ }
+ }
+ }
+
+ mutable std::shared_mutex mutex_;
+ std::atomic<bool> armed_ = {false};
std::atomic<int64_t> io_count_ = {0};
- std::atomic<int64_t> pos_ = {-1};
+ int64_t pos_ = -1;
IOHook::Mode mode_ = IOHook::Mode::SILENT;
};
diff --git a/src/paimon/common/factories/io_hook.h
b/src/paimon/common/factories/io_hook.h
index e0a2f68b..0c66381b 100644
--- a/src/paimon/common/factories/io_hook.h
+++ b/src/paimon/common/factories/io_hook.h
@@ -45,7 +45,8 @@ class PAIMON_EXPORT IOHook : public Singleton<IOHook> {
};
/// Reset the IO exception position and behavior mode to handle the
exception.
- /// IOCount will be reset to 0.
+ /// IOCount will be reset to 0. Arms the hook: Try() switches from its
lock-free
+ /// disabled fast path to the synchronized armed path.
///
/// @params pos The position where the IO exception occurs.
/// @params mode The mode of behavior for handling the exception.
@@ -56,12 +57,14 @@ class PAIMON_EXPORT IOHook : public Singleton<IOHook> {
Status Try(const std::string& path);
/// Get the count of IO operations that have already occurred.
+ /// IOs are only counted while the hook is armed (after Reset(), before
Clear());
+ /// the disabled fast path does not count.
///
/// @return The number of IO operations executed.
int64_t IOCount() const;
/// Clear the state of the IOHook, including resetting IO count and
- /// any stored exception state.
+ /// any stored exception state. Disarms the hook back to the lock-free
fast path.
void Clear();
private:
diff --git a/src/paimon/common/factories/io_hook_test.cpp
b/src/paimon/common/factories/io_hook_test.cpp
index 9bbb1b34..653dc73b 100644
--- a/src/paimon/common/factories/io_hook_test.cpp
+++ b/src/paimon/common/factories/io_hook_test.cpp
@@ -19,9 +19,13 @@
#include "paimon/common/factories/io_hook.h"
+#include <atomic>
#include <stdexcept>
+#include <thread>
+#include <vector>
#include "gtest/gtest.h"
+#include "paimon/status.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
@@ -64,4 +68,81 @@ TEST(IOHookTest, TestThrowExceptionMode) {
hook->Clear();
}
+// The disabled state is the production default: Try() must take the lock-free
fast
+// path, always return OK, and not count IOs (see IOCount()'s contract).
Clear() first
+// so the test does not depend on execution order.
+TEST(IOHookTest, TestDisabledFastPath) {
+ auto hook = IOHook::GetInstance();
+ hook->Clear();
+ ASSERT_OK(hook->Try("path"));
+ ASSERT_OK(hook->Try("path"));
+ ASSERT_EQ(0, hook->IOCount());
+
+ // Re-arming and disarming must restore the exact disabled behavior.
+ hook->Reset(0, IOHook::Mode::RETURN_ERROR);
+ ASSERT_NOK(hook->Try("path"));
+ ASSERT_EQ(1, hook->IOCount());
+ hook->Clear();
+ ASSERT_OK(hook->Try("path"));
+ ASSERT_OK(hook->Try("path"));
+ ASSERT_EQ(0, hook->IOCount());
+}
+
+// Regression test for torn IOHook configurations: Reset()/Clear() run on one
thread
+// while other threads call Try() concurrently. A shared start barrier
releases all
+// threads together, and the reset thread keeps hammering until every worker
has
+// finished, so overlap is structural rather than timing-dependent. The
continuous
+// arm/disarm cycling also keeps workers switching between the disabled fast
path and
+// the synchronized armed path. Under a ThreadSanitizer build this
deterministically
+// reports any unsynchronized access; functionally every Try() must return OK.
+TEST(IOHookTest, TestConcurrentResetAndTry) {
+ auto hook = IOHook::GetInstance();
+
+ constexpr int32_t kTryIterations = 50000;
+ constexpr int32_t kNumWorkers = 4;
+
+ std::atomic<bool> start{false};
+ std::atomic<int32_t> workers_done{0};
+ std::atomic<bool> observed_error{false};
+
+ std::thread reset_thread([hook, &start, &workers_done]() {
+ while (!start.load(std::memory_order_acquire)) {
+ std::this_thread::yield();
+ }
+ while (workers_done.load(std::memory_order_relaxed) < kNumWorkers) {
+ hook->Reset(INT64_MAX, IOHook::Mode::RETURN_ERROR);
+ hook->Clear();
+ }
+ });
+
+ std::vector<std::thread> workers;
+ workers.reserve(kNumWorkers);
+ for (int32_t t = 0; t < kNumWorkers; t++) {
+ workers.emplace_back([hook, &start, &workers_done, &observed_error]() {
+ while (!start.load(std::memory_order_acquire)) {
+ std::this_thread::yield();
+ }
+ for (int32_t i = 0; i < kTryIterations; i++) {
+ Status status = hook->Try("concurrent_path");
+ // Reset() arms an unreachable position, while Clear() uses
SILENT mode,
+ // so both complete states return OK. An IOError exposes a
torn state.
+ if (!status.ok()) {
+ observed_error.store(true, std::memory_order_relaxed);
+ }
+ }
+ workers_done.fetch_add(1, std::memory_order_relaxed);
+ });
+ }
+
+ start.store(true, std::memory_order_release);
+ reset_thread.join();
+ for (auto& worker : workers) {
+ worker.join();
+ }
+
+ ASSERT_FALSE(observed_error.load(std::memory_order_relaxed));
+ // Leave the process-wide singleton in its default SILENT state for later
tests.
+ hook->Clear();
+}
+
} // namespace paimon::test
diff --git a/src/paimon/common/factories/singleton.cpp
b/src/paimon/common/factories/singleton.cpp
index a9732259..2a5daa89 100644
--- a/src/paimon/common/factories/singleton.cpp
+++ b/src/paimon/common/factories/singleton.cpp
@@ -19,26 +19,14 @@
#include "paimon/factories/singleton.h"
-#include <mutex>
-
#include "paimon/common/factories/io_hook.h"
#include "paimon/factories/factory_creator.h"
namespace paimon {
-template <typename T, typename InstPolicy>
-T* Singleton<T, InstPolicy>::GetInstance() {
- static T* ptr;
- static std::mutex mutex;
- if (PAIMON_UNLIKELY(!ptr)) {
- std::lock_guard<std::mutex> lg(mutex);
- if (!ptr) {
- InstPolicy::Create(ptr);
- }
- }
- return const_cast<T*>(ptr);
-}
-
+// The single definition point for the two cross-library singletons. See the
+// extern template declarations in singleton.h for why implicit instantiation
+// must stay suppressed for these types.
template class Singleton<FactoryCreator>;
template class Singleton<IOHook>;
diff --git a/src/paimon/common/factories/singleton_test.cpp
b/src/paimon/common/factories/singleton_test.cpp
new file mode 100644
index 00000000..efaf921c
--- /dev/null
+++ b/src/paimon/common/factories/singleton_test.cpp
@@ -0,0 +1,108 @@
+/*
+ * 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 "paimon/factories/singleton.h"
+
+#include <array>
+#include <atomic>
+#include <cstdint>
+#include <thread>
+#include <vector>
+
+#include "gtest/gtest.h"
+
+namespace paimon::test {
+
+namespace {
+
+constexpr int32_t kNumThreads = 32;
+
+// Runs `worker(i)` on kNumThreads threads that are all blocked on a shared
start
+// flag and released at (nearly) the same time, so that they race on the first
+// Singleton::GetInstance() publication. Joins all threads before returning.
+template <typename Worker>
+void RunStorm(const Worker& worker) {
+ std::atomic<bool> start{false};
+ std::vector<std::thread> threads;
+ threads.reserve(kNumThreads);
+ for (int32_t i = 0; i < kNumThreads; ++i) {
+ threads.emplace_back([&start, &worker, i]() {
+ while (!start.load(std::memory_order_acquire)) {
+ std::this_thread::yield();
+ }
+ worker(i);
+ });
+ }
+ start.store(true, std::memory_order_release);
+ for (auto& thread : threads) {
+ thread.join();
+ }
+}
+
+// Local to this translation unit, so nothing else in the test binary can have
+// instantiated Singleton<FirstPublicationTarget> before this test runs: the
storm
+// below is guaranteed to race on the *first* publication regardless of link
order,
+// --gtest_shuffle, or --gtest_filter. GetInstance() is defined in the header,
so a
+// translation-unit-local type can instantiate it.
+class FirstPublicationTarget {
+ public:
+ FirstPublicationTarget() {
+ for (size_t i = 0; i < payload_.size(); ++i) {
+ payload_[i] = kMagic ^ (i * 0x9E3779B97F4A7C15ULL);
+ }
+ }
+
+ // The publication race let a reader observe the instance pointer before
the
+ // constructor's stores were visible; this checks every word the ctor
wrote.
+ bool IsFullyConstructed() const {
+ for (size_t i = 0; i < payload_.size(); ++i) {
+ if (payload_[i] != (kMagic ^ (i * 0x9E3779B97F4A7C15ULL))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private:
+ static constexpr uint64_t kMagic = 0xA5A5F00D12345678ULL;
+ std::array<uint64_t, 64> payload_{};
+};
+
+} // namespace
+
+// Regression gate for the Singleton double-checked-locking publication race:
32
+// threads race the first GetInstance() of a type local to this file, so the
gate
+// cannot silently degrade into exercising only the already-published fast
path.
+TEST(SingletonTest, TestConcurrentFirstPublication) {
+ std::array<FirstPublicationTarget*, kNumThreads> instances{};
+ std::array<bool, kNumThreads> fully_constructed{};
+ RunStorm([&instances, &fully_constructed](int32_t i) {
+ instances[i] = Singleton<FirstPublicationTarget>::GetInstance();
+ fully_constructed[i] = instances[i]->IsFullyConstructed();
+ });
+
+ FirstPublicationTarget* expected = instances[0];
+ ASSERT_NE(expected, nullptr);
+ for (int32_t i = 0; i < kNumThreads; ++i) {
+ ASSERT_EQ(expected, instances[i]);
+ ASSERT_TRUE(fully_constructed[i]);
+ }
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/io/cache/cache_manager.h
b/src/paimon/common/io/cache/cache_manager.h
index f899d46c..6fafefb5 100644
--- a/src/paimon/common/io/cache/cache_manager.h
+++ b/src/paimon/common/io/cache/cache_manager.h
@@ -25,6 +25,7 @@
#include "paimon/cache/cache.h"
#include "paimon/common/io/cache/cache_key.h"
#include "paimon/common/io/cache/lru_cache.h"
+#include "paimon/common/utils/saturating_cast.h"
#include "paimon/memory/memory_segment.h"
#include "paimon/result.h"
@@ -59,9 +60,13 @@ class PAIMON_EXPORT CacheManager {
/// @param high_priority_pool_ratio Ratio of capacity reserved for index
cache [0.0, 1.0).
/// If 0, index and data share the same cache.
CacheManager(int64_t max_memory_bytes, double high_priority_pool_ratio) {
- auto index_cache_bytes = static_cast<int64_t>(max_memory_bytes *
high_priority_pool_ratio);
+ // Both factors are config-validated non-negative values, so the
products are finite;
+ // saturation is only a defense against the undefined double->int64_t
conversion when
+ // max_memory_bytes is close enough to INT64_MAX that the product
rounds to 2^63.
+ auto index_cache_bytes =
+ SaturatingDoubleToInteger<int64_t>(max_memory_bytes *
high_priority_pool_ratio);
auto data_cache_bytes =
- static_cast<int64_t>(max_memory_bytes * (1.0 -
high_priority_pool_ratio));
+ SaturatingDoubleToInteger<int64_t>(max_memory_bytes * (1.0 -
high_priority_pool_ratio));
data_cache_ = std::make_shared<LruCache>(data_cache_bytes);
if (high_priority_pool_ratio == 0.0) {
index_cache_ = data_cache_;
diff --git a/src/paimon/common/io/cache/cache_manager_test.cpp
b/src/paimon/common/io/cache/cache_manager_test.cpp
new file mode 100644
index 00000000..4d7c180a
--- /dev/null
+++ b/src/paimon/common/io/cache/cache_manager_test.cpp
@@ -0,0 +1,155 @@
+/*
+ * 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 "paimon/common/io/cache/cache_manager.h"
+
+#include <cstdint>
+#include <cstring>
+#include <limits>
+#include <memory>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/cache/cache.h"
+#include "paimon/common/io/cache/cache_key.h"
+#include "paimon/common/io/cache/lru_cache.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/memory/memory_segment.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class CacheManagerTest : public ::testing::Test {
+ public:
+ void SetUp() override {
+ pool_ = GetDefaultPool();
+ }
+
+ std::shared_ptr<CacheKey> MakeKey(int64_t position, bool is_index = false)
const {
+ return CacheKey::ForPosition("test_file", position, 64, is_index);
+ }
+
+ MemorySegment MakeSegment(int32_t size, char fill_byte) const {
+ auto segment = MemorySegment::AllocateHeapMemory(size, pool_.get());
+ std::memset(segment.MutableData(), fill_byte, size);
+ return segment;
+ }
+
+ std::shared_ptr<LruCache> DataLru(const CacheManager& manager) const {
+ return std::dynamic_pointer_cast<LruCache>(manager.DataCache());
+ }
+
+ std::shared_ptr<LruCache> IndexLru(const CacheManager& manager) const {
+ return std::dynamic_pointer_cast<LruCache>(manager.IndexCache());
+ }
+
+ private:
+ std::shared_ptr<MemoryPool> pool_;
+};
+
+/// Regression test for the double->int64_t conversions in the CacheManager
constructor:
+/// (double)INT64_MAX rounds to 2^63, which is not representable as int64_t,
so casting the
+/// product back is undefined behavior (x86 cvttsd2si yields INT64_MIN,
aarch64 fcvtzs
+/// saturates to INT64_MAX). The conversion must saturate, keeping the
capacity non-negative.
+TEST_F(CacheManagerTest, TestCapacitySaturatesAtInt64Max) {
+ CacheManager manager(std::numeric_limits<int64_t>::max(),
/*high_priority_pool_ratio=*/0.0);
+
+ std::shared_ptr<LruCache> data_lru = DataLru(manager);
+ ASSERT_NE(data_lru, nullptr);
+ ASSERT_GE(data_lru->GetMaxWeight(), 0);
+ ASSERT_EQ(data_lru->GetMaxWeight(), std::numeric_limits<int64_t>::max());
+
+ // A ratio of 0.0 means index and data share the same cache.
+ ASSERT_EQ(manager.DataCache(), manager.IndexCache());
+
+ // The saturated capacity accepts entries instead of rejecting every
insert.
+ std::shared_ptr<CacheKey> key = MakeKey(0);
+ auto reader = [&](const std::shared_ptr<CacheKey>&) ->
Result<MemorySegment> {
+ return MakeSegment(64, 'A');
+ };
+ ASSERT_OK_AND_ASSIGN(MemorySegment segment, manager.GetPage(key, reader,
{}));
+ ASSERT_EQ(segment.Size(), 64);
+ ASSERT_EQ(segment.Get(0), 'A');
+}
+
+/// Verifies the exact capacity split between the data and index caches for a
normal
+/// configuration, plus a Get/Invalidate smoke path through
CacheManager::GetPage.
+TEST_F(CacheManagerTest, TestNormalSplitAndSmokePath) {
+ CacheManager manager(/*max_memory_bytes=*/1024,
/*high_priority_pool_ratio=*/0.5);
+
+ std::shared_ptr<LruCache> data_lru = DataLru(manager);
+ std::shared_ptr<LruCache> index_lru = IndexLru(manager);
+ ASSERT_NE(data_lru, nullptr);
+ ASSERT_NE(index_lru, nullptr);
+ ASSERT_EQ(data_lru->GetMaxWeight(), 512);
+ ASSERT_EQ(index_lru->GetMaxWeight(), 512);
+
+ std::shared_ptr<CacheKey> key = MakeKey(0);
+ int32_t reader_calls = 0;
+ auto reader = [&](const std::shared_ptr<CacheKey>&) ->
Result<MemorySegment> {
+ reader_calls++;
+ return MakeSegment(128, 'B');
+ };
+
+ // The first GetPage is a miss and invokes the reader; the second is a
cache hit.
+ ASSERT_OK_AND_ASSIGN(MemorySegment first, manager.GetPage(key, reader,
{}));
+ ASSERT_EQ(first.Get(0), 'B');
+ ASSERT_EQ(reader_calls, 1);
+ ASSERT_OK_AND_ASSIGN(MemorySegment second, manager.GetPage(key, reader,
{}));
+ ASSERT_EQ(second.Get(0), 'B');
+ ASSERT_EQ(reader_calls, 1);
+
+ // After InvalidPage the reader is invoked again.
+ manager.InvalidPage(key);
+ ASSERT_OK_AND_ASSIGN(MemorySegment third, manager.GetPage(key, reader,
{}));
+ ASSERT_EQ(third.Get(0), 'B');
+ ASSERT_EQ(reader_calls, 2);
+}
+
+/// Verifies weight-based eviction through GetPage: inserting beyond the data
cache capacity
+/// evicts the least recently used page and runs its eviction callback.
+TEST_F(CacheManagerTest, TestGetPageEviction) {
+ // The data cache capacity is 512 * (1.0 - 0.5) = 256 bytes.
+ CacheManager manager(/*max_memory_bytes=*/512,
/*high_priority_pool_ratio=*/0.5);
+
+ std::vector<int64_t> evicted;
+ auto callback_for = [&evicted](int64_t position) -> CacheCallback {
+ return
+ [&evicted, position](const std::shared_ptr<CacheKey>&) {
evicted.push_back(position); };
+ };
+ auto reader = [&](const std::shared_ptr<CacheKey>&) ->
Result<MemorySegment> {
+ return MakeSegment(128, 'C');
+ };
+
+ std::shared_ptr<CacheKey> key0 = MakeKey(0);
+ std::shared_ptr<CacheKey> key1 = MakeKey(1);
+ std::shared_ptr<CacheKey> key2 = MakeKey(2);
+ ASSERT_OK_AND_ASSIGN(MemorySegment segment0, manager.GetPage(key0, reader,
callback_for(0)));
+ ASSERT_EQ(segment0.Get(0), 'C');
+ ASSERT_OK_AND_ASSIGN(MemorySegment segment1, manager.GetPage(key1, reader,
callback_for(1)));
+ ASSERT_EQ(segment1.Get(0), 'C');
+ ASSERT_TRUE(evicted.empty());
+
+ // 128 + 128 + 128 > 256: inserting key2 evicts key0, the least recently
used page.
+ ASSERT_OK_AND_ASSIGN(MemorySegment segment2, manager.GetPage(key2, reader,
callback_for(2)));
+ ASSERT_EQ(segment2.Get(0), 'C');
+ ASSERT_EQ(evicted, std::vector<int64_t>({0}));
+ ASSERT_EQ(manager.DataCache()->Size(), 2);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/sst/sst_file_writer.cpp
b/src/paimon/common/sst/sst_file_writer.cpp
index ec736e33..f2b3b2de 100644
--- a/src/paimon/common/sst/sst_file_writer.cpp
+++ b/src/paimon/common/sst/sst_file_writer.cpp
@@ -20,6 +20,7 @@
#include "paimon/common/utils/crc32c.h"
#include "paimon/common/utils/murmurhash_utils.h"
+#include "paimon/common/utils/saturating_cast.h"
namespace paimon {
SstFileWriter::SstFileWriter(const std::shared_ptr<OutputStream>& out,
@@ -27,8 +28,10 @@ SstFileWriter::SstFileWriter(const
std::shared_ptr<OutputStream>& out,
const std::shared_ptr<BlockCompressionFactory>&
factory,
const std::shared_ptr<MemoryPool>& pool)
: pool_(pool), out_(out), bloom_filter_(bloom_filter),
block_size_(block_size) {
+ // block_size * 1.1 exceeds INT32_MAX for block_size above ~1.9GB;
saturate instead of
+ // relying on the undefined double->int32_t conversion.
data_block_writer_ =
- std::make_unique<BlockWriter>(static_cast<int32_t>(block_size * 1.1),
pool);
+
std::make_unique<BlockWriter>(SaturatingDoubleToInteger<int32_t>(block_size *
1.1), pool);
index_block_writer_ =
std::make_unique<BlockWriter>(BlockHandle::MAX_ENCODED_LENGTH * 1024,
pool);
compression_type_ = factory->GetCompressionType();
diff --git a/src/paimon/common/utils/read_ahead_cache_test.cpp
b/src/paimon/common/utils/read_ahead_cache_test.cpp
index ab1d0ed3..e7900b7b 100644
--- a/src/paimon/common/utils/read_ahead_cache_test.cpp
+++ b/src/paimon/common/utils/read_ahead_cache_test.cpp
@@ -464,7 +464,8 @@ TEST(TestReadAheadCache, TestPreBufferWindowLimit) {
auto io_hook = paimon::IOHook::GetInstance();
paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
- io_hook->Clear();
+ // IOCount() only counts while armed; INT64_MAX never triggers the error
mode.
+ io_hook->Reset(INT64_MAX, paimon::IOHook::Mode::RETURN_ERROR);
AssertReadEquals({0, 10}, "abcdefghij", &cache);
// The second range did not fit into the window: only one prefetch IO.
@@ -475,7 +476,7 @@ TEST(TestReadAheadCache, TestPreBufferWindowLimit) {
ASSERT_EQ(io_hook->IOCount(), 2);
// The range is cached now: re-reading it issues no IO at all.
- io_hook->Clear();
+ io_hook->Reset(INT64_MAX, paimon::IOHook::Mode::RETURN_ERROR);
AssertReadEquals({16, 10}, "qrstuvwxyz", &cache);
ASSERT_EQ(io_hook->IOCount(), 0);
}
diff --git a/src/paimon/common/utils/saturating_cast.h
b/src/paimon/common/utils/saturating_cast.h
new file mode 100644
index 00000000..cb9c6039
--- /dev/null
+++ b/src/paimon/common/utils/saturating_cast.h
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cmath>
+#include <cstdint>
+#include <limits>
+#include <type_traits>
+
+namespace paimon {
+
+/// Converts a double to int32_t or int64_t with Java's float-to-int or
float-to-long saturation
+/// policy: NaN converts to 0 and an out-of-range value saturates at the
bounds of TargetType.
+/// Narrower Java integer conversions require a subsequent narrowing step and
are not supported by
+/// this helper. A bare static_cast of an unrepresentable double is undefined
behavior and diverges
+/// across architectures (x86 cvttsd2si yields the "integer indefinite" value,
while aarch64 fcvtzs
+/// saturates), so doubles that are not provably in range must go through this
helper.
+template <typename TargetType>
+inline TargetType SaturatingDoubleToInteger(double value) {
+ static_assert(std::is_same_v<TargetType, int32_t> ||
std::is_same_v<TargetType, int64_t>,
+ "TargetType must be int32_t or int64_t");
+ if (std::isnan(value)) {
+ return 0;
+ }
+ // Comparing against the bounds converted to double keeps the final
truncation defined:
+ // (double)INT64_MAX rounds up to 2^63, so every value that reaches the
truncation is
+ // representable in TargetType.
+ if (value >= static_cast<double>(std::numeric_limits<TargetType>::max())) {
+ return std::numeric_limits<TargetType>::max();
+ }
+ if (value <=
static_cast<double>(std::numeric_limits<TargetType>::lowest())) {
+ return std::numeric_limits<TargetType>::lowest();
+ }
+ return static_cast<TargetType>(value);
+}
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/saturating_cast_test.cpp
b/src/paimon/common/utils/saturating_cast_test.cpp
new file mode 100644
index 00000000..cc6b74cf
--- /dev/null
+++ b/src/paimon/common/utils/saturating_cast_test.cpp
@@ -0,0 +1,74 @@
+/*
+ * 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 "paimon/common/utils/saturating_cast.h"
+
+#include <cstdint>
+#include <limits>
+
+#include "gtest/gtest.h"
+
+namespace paimon::test {
+
+TEST(SaturatingCastTest, TestInt64InRangeTruncatesTowardZero) {
+ ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(0.0), 0);
+ ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(1.9), 1);
+ ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(-1.9), -1);
+ // 2^63 - 1024 is the largest double below 2^63: it stays on the
truncation path.
+ ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(9223372036854774784.0),
9223372036854774784LL);
+}
+
+TEST(SaturatingCastTest, TestInt64Saturation) {
+ // (double)INT64_MAX rounds up to 2^63, so the boundary double already
saturates.
+ ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(
+ static_cast<double>(std::numeric_limits<int64_t>::max())),
+ std::numeric_limits<int64_t>::max());
+ ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(1e300),
std::numeric_limits<int64_t>::max());
+ ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(-1e300),
std::numeric_limits<int64_t>::lowest());
+
ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(std::numeric_limits<double>::infinity()),
+ std::numeric_limits<int64_t>::max());
+
ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(-std::numeric_limits<double>::infinity()),
+ std::numeric_limits<int64_t>::lowest());
+ // The lowest bound is exactly representable and must survive as a value.
+ ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(
+ static_cast<double>(std::numeric_limits<int64_t>::lowest())),
+ std::numeric_limits<int64_t>::lowest());
+}
+
+TEST(SaturatingCastTest, TestInt64NaNBecomesZero) {
+ // Java's (long)Double.NaN == 0.
+
ASSERT_EQ(SaturatingDoubleToInteger<int64_t>(std::numeric_limits<double>::quiet_NaN()),
0);
+}
+
+TEST(SaturatingCastTest, TestInt32Path) {
+ // SstFileWriter converts through the int32_t instantiation.
+ ASSERT_EQ(SaturatingDoubleToInteger<int32_t>(42.7), 42);
+ ASSERT_EQ(SaturatingDoubleToInteger<int32_t>(-42.7), -42);
+ // The int32_t bounds are exactly representable as doubles and saturate
inclusively.
+ ASSERT_EQ(SaturatingDoubleToInteger<int32_t>(2147483647.0),
+ std::numeric_limits<int32_t>::max());
+ ASSERT_EQ(SaturatingDoubleToInteger<int32_t>(2147483648.0),
+ std::numeric_limits<int32_t>::max());
+ ASSERT_EQ(SaturatingDoubleToInteger<int32_t>(-2147483648.0),
+ std::numeric_limits<int32_t>::lowest());
+ ASSERT_EQ(SaturatingDoubleToInteger<int32_t>(-2147483649.0),
+ std::numeric_limits<int32_t>::lowest());
+
ASSERT_EQ(SaturatingDoubleToInteger<int32_t>(std::numeric_limits<double>::quiet_NaN()),
0);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/utils/serialization_utils.h
b/src/paimon/common/utils/serialization_utils.h
index c1e97d03..449766ca 100644
--- a/src/paimon/common/utils/serialization_utils.h
+++ b/src/paimon/common/utils/serialization_utils.h
@@ -78,7 +78,9 @@ class SerializationUtils {
if (PAIMON_UNLIKELY(bytes->size() < 4)) {
return Status::Invalid(fmt::format("bytes size {} is less than 4",
bytes->size()));
}
- int32_t arity = *(reinterpret_cast<int32_t*>(bytes->data()));
+ // The buffer is byte-filled, so memcpy avoids the strict-aliasing UB
of reinterpret_cast.
+ int32_t arity;
+ memcpy(&arity, bytes->data(), sizeof(int32_t));
if (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN) {
arity = EndianSwapValue(arity);
}
diff --git a/src/paimon/common/utils/serialization_utils_test.cpp
b/src/paimon/common/utils/serialization_utils_test.cpp
index 5e612ff2..56a39a29 100644
--- a/src/paimon/common/utils/serialization_utils_test.cpp
+++ b/src/paimon/common/utils/serialization_utils_test.cpp
@@ -19,7 +19,20 @@
#include "paimon/common/utils/serialization_utils.h"
+#include <cstdint>
+#include <memory>
+#include <string>
+
#include "gtest/gtest.h"
+#include "paimon/common/data/binary_row_writer.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
+#include "paimon/common/memory/memory_segment_utils.h"
+#include "paimon/io/byte_array_input_stream.h"
+#include "paimon/io/data_input_stream.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
@@ -37,4 +50,39 @@ TEST_F(SerializationUtilsTest, TestSerializeBinaryRow) {
ASSERT_TRUE(bytes);
}
+TEST_F(SerializationUtilsTest, TestDeserializeBinaryRowFromStream) {
+ std::shared_ptr<MemoryPool> memory_pool = GetDefaultPool();
+ // a row with mixed field types, including negative integers and a string
+ BinaryRow row(3);
+ BinaryRowWriter writer(&row, 0, memory_pool.get());
+ writer.WriteInt(0, -123456);
+ writer.WriteLong(1, static_cast<int64_t>(-9000000000));
+ writer.WriteString(2, BinaryString::FromString("hello paimon!",
memory_pool.get()));
+ writer.Complete();
+
+ // the first 4 bytes on the wire are the big-endian arity (Java-compatible
format)
+ std::shared_ptr<Bytes> bytes = SerializationUtils::SerializeBinaryRow(row,
memory_pool.get());
+ ASSERT_TRUE(bytes);
+ ASSERT_GE(bytes->size(), 4);
+ ASSERT_EQ(static_cast<uint8_t>(bytes->data()[0]), 0x00);
+ ASSERT_EQ(static_cast<uint8_t>(bytes->data()[1]), 0x00);
+ ASSERT_EQ(static_cast<uint8_t>(bytes->data()[2]), 0x00);
+ ASSERT_EQ(static_cast<uint8_t>(bytes->data()[3]), 0x03);
+
+ // round-trip through the stream overloads, which fill a fresh byte buffer
on deserialize
+ MemorySegmentOutputStream
out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, memory_pool);
+ ASSERT_OK(SerializationUtils::SerializeBinaryRow(row, &out));
+ auto stream_bytes =
+ MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(),
memory_pool.get());
+ auto input_stream =
+ std::make_shared<ByteArrayInputStream>(stream_bytes->data(),
stream_bytes->size());
+ DataInputStream data_input_stream(input_stream);
+ ASSERT_OK_AND_ASSIGN(BinaryRow de_row,
SerializationUtils::DeserializeBinaryRow(
+ &data_input_stream,
memory_pool.get()));
+ ASSERT_EQ(de_row.GetFieldCount(), 3);
+ ASSERT_EQ(de_row.GetInt(0), -123456);
+ ASSERT_EQ(de_row.GetLong(1), static_cast<int64_t>(-9000000000));
+ ASSERT_EQ(de_row.GetString(2).ToString(), "hello paimon!");
+}
+
} // namespace paimon::test
