fix: repair Singleton double-checked locking race and harden aarch64 portability

An aarch64 portability audit (7-category static sweep + on-hardware
validation on a 128-core Kunpeng-920) found one real concurrency bug
hidden by x86 TSO and four latent UB/divergence risks:

- Singleton<T>::GetInstance() published the instance with a plain store
  guarded only by a compiler-only MEMORY_BARRIER, and the fast path read
  it with a plain non-atomic load. On aarch64 this allows readers to
  observe a non-null pointer to a not-yet-constructed object (TSan
  confirms the race; a 128-thread litmus of the exact pattern observed
  torn reads). Use std::atomic<T*> with acquire/release ordering.
- IOHook::Impl::mode_ was a plain enum raced by Reset() and Try()
  (TSan-confirmed). Make it std::atomic<Mode>, stored before the seq_cst
  pos_/io_count_ stores so it is published together with them.
- FieldSumAgg INT8 sum/neg computed on plain char (unsigned on aarch64).
  Bytes were bit-identical to Java today, but any future widening or
  comparison would diverge (the bug class PR #181 fixed in min/max).
  Compute through int8_t like the sibling aggregators.
- SerializationUtils::DeserializeBinaryRow read arity from a byte-filled
  buffer through reinterpret_cast<int32_t*> (strict-aliasing UB). Use
  memcpy like the serialize side; identical codegen.
- CacheManager and SstFileWriter relied on undefined double->int
  conversions (x86 yields INT_MIN, aarch64 saturates) for extreme
  configs. Add common-layer SaturatingDoubleToInteger with the Java
  saturation policy and use it at both sites.

Tests: new SingletonTest concurrent GetInstance storms and
IOHookTest.TestConcurrentResetAndTry are deterministic TSan regression
gates (verified red on the pre-fix code, clean after). FieldSumAgg
INT8 boundary tests lock the Java signed-byte semantics.
SerializationUtilsTest gains a DataInputStream round-trip that also
pins the big-endian wire format. CacheManagerTest locks the saturated
capacity semantics (fails pre-fix on x86).

Generated-by: Claude Code (claude-opus-4-8)

Author: u70b3

include/paimon/factories/singleton.h

@@ -30,9 +30,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);
     }
 };

src/paimon/CMakeLists.txt

@@ -634,6 +634,7 @@ 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/scope_guard_test.cpp
                     common/utils/sensitive_config_utils_test.cpp
@@ -665,6 +666,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

src/paimon/common/factories/io_hook.cpp

@@ -32,7 +32,7 @@ class IOHook::Impl {
         if (io_count_.fetch_add(1) < pos_.load()) {
             return Status::OK();
         } else {
-            switch (mode_) {
+            switch (mode_.load(std::memory_order_relaxed)) {
                 case IOHook::Mode::SILENT:
                     return Status::OK();
                 case IOHook::Mode::RETURN_ERROR:
@@ -49,9 +49,11 @@ class IOHook::Impl {
     }
 
     inline void Reset(int64_t pos, IOHook::Mode mode) {
+        // Store mode_ first: the seq_cst stores below then publish it, so a Try()
+        // that observes the reset pos_ also observes the new mode_.
+        mode_.store(mode, std::memory_order_relaxed);
         pos_ = pos;
         io_count_ = 0;
-        mode_ = mode;
     }
 
     int64_t IOCount() const {
@@ -65,7 +67,7 @@ class IOHook::Impl {
  private:
     std::atomic<int64_t> io_count_ = {0};
     std::atomic<int64_t> pos_ = {-1};
-    IOHook::Mode mode_ = IOHook::Mode::SILENT;
+    std::atomic<IOHook::Mode> mode_{IOHook::Mode::SILENT};
 };
 
 IOHook::IOHook() : impl_(std::make_unique<IOHook::Impl>()) {}

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,49 @@ TEST(IOHookTest, TestThrowExceptionMode) {
     hook->Clear();
 }
 
+// Regression test for the data race on IOHook's mode: Reset()/Clear() run on one
+// thread while other threads call Try() concurrently. Under a ThreadSanitizer build
+// this deterministically reports the unsynchronized mode access; functionally it must
+// never crash and every Try() must return either OK or the injected IOError.
+TEST(IOHookTest, TestConcurrentResetAndTry) {
+    auto hook = IOHook::GetInstance();
+
+    constexpr int32_t kResetIterations = 200000;
+    constexpr int32_t kTryIterations = 50000;
+    constexpr int32_t kNumWorkers = 4;
+
+    std::atomic<bool> unexpected_status{false};
+
+    std::thread reset_thread([hook]() {
+        for (int32_t i = 0; i < kResetIterations; i++) {
+            hook->Reset(i, 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, &unexpected_status]() {
+            for (int32_t i = 0; i < kTryIterations; i++) {
+                Status status = hook->Try("concurrent_path");
+                // Only RETURN_ERROR mode is armed here, so Try() may only return OK or
+                // IOError; anything else means the mode was read as garbage.
+                if (!status.ok() && !status.IsIOError()) {
+                    unexpected_status.store(true, std::memory_order_relaxed);
+                }
+            }
+        });
+    }
+
+    reset_thread.join();
+    for (auto& worker : workers) {
+        worker.join();
+    }
+
+    ASSERT_FALSE(unexpected_status.load(std::memory_order_relaxed));
+    // Leave the process-wide singleton in its default SILENT state for later tests.
+    hook->Clear();
+}
+
 }  // namespace paimon::test

src/paimon/common/factories/singleton.cpp

@@ -19,6 +19,7 @@
 
 #include "paimon/factories/singleton.h"
 
+#include <atomic>
 #include <mutex>
 
 #include "paimon/common/factories/io_hook.h"
@@ -28,15 +29,20 @@ namespace paimon {
 
 template <typename T, typename InstPolicy>
 T* Singleton<T, InstPolicy>::GetInstance() {
-    static T* ptr;
+    static std::atomic<T*> ptr{nullptr};
     static std::mutex mutex;
-    if (PAIMON_UNLIKELY(!ptr)) {
+    T* p = ptr.load(std::memory_order_acquire);
+    if (PAIMON_UNLIKELY(p == nullptr)) {
         std::lock_guard<std::mutex> lg(mutex);
-        if (!ptr) {
-            InstPolicy::Create(ptr);
+        // 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 const_cast<T*>(ptr);
+    return p;
 }
 
 template class Singleton<FactoryCreator>;

src/paimon/common/factories/singleton_test.cpp

@@ -0,0 +1,113 @@
+/*
+ * 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"
+#include "paimon/common/factories/io_hook.h"
+#include "paimon/factories/factory.h"
+#include "paimon/factories/factory_creator.h"
+
+namespace paimon::test {
+
+namespace {
+
+constexpr int32_t kNumThreads = 32;
+
+class StormFactory : public Factory {
+ public:
+    const char* Identifier() const override {
+        return "storm_factory";
+    }
+};
+
+// 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();
+    }
+}
+
+}  // namespace
+
+TEST(SingletonTest, TestConcurrentIOHookGetInstance) {
+    std::array<IOHook*, kNumThreads> hooks{};
+    std::array<bool, kNumThreads> try_oks{};
+    RunStorm([&hooks, &try_oks](int32_t i) {
+        hooks[i] = Singleton<IOHook>::GetInstance();
+        // The default (and cleared) IOHook state is SILENT, so Try() must succeed.
+        try_oks[i] = hooks[i]->Try("singleton_storm_path").ok();
+    });
+
+    IOHook* expected = hooks[0];
+    ASSERT_NE(expected, nullptr);
+    for (int32_t i = 0; i < kNumThreads; ++i) {
+        ASSERT_EQ(expected, hooks[i]);
+        ASSERT_TRUE(try_oks[i]);
+    }
+    ASSERT_GE(expected->IOCount(), kNumThreads);
+    // Leave the process-wide singleton in its default SILENT state for later tests.
+    expected->Clear();
+}
+
+TEST(SingletonTest, TestConcurrentFactoryCreatorGetInstance) {
+    std::array<FactoryCreator*, kNumThreads> creators{};
+    RunStorm([&creators](int32_t i) { creators[i] = Singleton<FactoryCreator>::GetInstance(); });
+
+    FactoryCreator* expected = creators[0];
+    ASSERT_NE(expected, nullptr);
+    for (int32_t i = 0; i < kNumThreads; ++i) {
+        ASSERT_EQ(expected, creators[i]);
+    }
+
+    // Every thread must observe a fully constructed FactoryCreator: a lookup of a
+    // registered factory must succeed from any thread.
+    auto* factory = new StormFactory();
+    expected->Register(factory->Identifier(), factory);
+    std::array<Factory*, kNumThreads> looked_up{};
+    RunStorm([&looked_up](int32_t i) {
+        looked_up[i] = Singleton<FactoryCreator>::GetInstance()->Create("storm_factory");
+    });
+    for (int32_t i = 0; i < kNumThreads; ++i) {
+        ASSERT_EQ(factory, looked_up[i]);
+    }
+}
+
+}  // namespace paimon::test

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_;