diff --git a/include/paimon/factories/singleton.h b/include/paimon/factories/singleton.h index 6e12d4561..db0201738 100644 --- a/include/paimon/factories/singleton.h +++ b/include/paimon/factories/singleton.h @@ -30,9 +30,9 @@ class PAIMON_EXPORT LazyInstantiation { protected: template static void Create(T*& ptr) { - T* tmp = new T; - MEMORY_BARRIER(); - ptr = tmp; + // Publication ordering is handled by the release store in + // Singleton::GetInstance(), so no barrier is needed here. + ptr = new T; static std::shared_ptr destroyer(ptr); } }; diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 6aca56d8a..8d07032e8 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -634,7 +634,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 @@ -665,6 +667,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 a0576b469..4b1aee058 100644 --- a/src/paimon/common/factories/io_hook.cpp +++ b/src/paimon/common/factories/io_hook.cpp @@ -19,6 +19,8 @@ #include "paimon/common/factories/io_hook.h" #include +#include +#include #include #include "fmt/format.h" @@ -29,7 +31,8 @@ namespace paimon { class IOHook::Impl { public: Status Try(const std::string& path) { - if (io_count_.fetch_add(1) < pos_.load()) { + std::shared_lock lock(mutex_); + if (io_count_.fetch_add(1) < pos_) { return Status::OK(); } else { switch (mode_) { @@ -37,10 +40,10 @@ class IOHook::Impl { return Status::OK(); case IOHook::Mode::RETURN_ERROR: return Status::IOError(fmt::format( - "io hook triggered io error at position {}, path {}", pos_.load(), path)); + "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_.load(), path)); + "io hook throw io exception at position {}, path {}", pos_, path)); return Status::OK(); default: return Status::OK(); @@ -49,12 +52,14 @@ class IOHook::Impl { } inline void Reset(int64_t pos, IOHook::Mode mode) { + std::unique_lock lock(mutex_); + mode_ = mode; pos_ = pos; io_count_ = 0; - mode_ = mode; } int64_t IOCount() const { + std::shared_lock lock(mutex_); return io_count_.load(); } @@ -63,8 +68,9 @@ class IOHook::Impl { } private: + mutable std::shared_mutex mutex_; std::atomic io_count_ = {0}; - std::atomic pos_ = {-1}; + int64_t pos_ = -1; IOHook::Mode mode_ = IOHook::Mode::SILENT; }; diff --git a/src/paimon/common/factories/io_hook_test.cpp b/src/paimon/common/factories/io_hook_test.cpp index 9bbb1b342..af63f88eb 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 #include +#include +#include #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 OK. +TEST(IOHookTest, TestConcurrentResetAndTry) { + auto hook = IOHook::GetInstance(); + + constexpr int32_t kResetIterations = 200000; + constexpr int32_t kTryIterations = 50000; + constexpr int32_t kNumWorkers = 4; + + std::atomic observed_error{false}; + + std::thread reset_thread([hook]() { + for (int32_t i = 0; i < kResetIterations; i++) { + hook->Reset(INT64_MAX, IOHook::Mode::RETURN_ERROR); + hook->Clear(); + } + }); + + std::vector workers; + workers.reserve(kNumWorkers); + for (int32_t t = 0; t < kNumWorkers; t++) { + workers.emplace_back([hook, &observed_error]() { + 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); + } + } + }); + } + + 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 a97322597..706f97b4b 100644 --- a/src/paimon/common/factories/singleton.cpp +++ b/src/paimon/common/factories/singleton.cpp @@ -19,6 +19,7 @@ #include "paimon/factories/singleton.h" +#include #include #include "paimon/common/factories/io_hook.h" @@ -28,15 +29,20 @@ namespace paimon { template T* Singleton::GetInstance() { - static T* ptr; + static std::atomic 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 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(ptr); + return p; } template class Singleton; diff --git a/src/paimon/common/factories/singleton_test.cpp b/src/paimon/common/factories/singleton_test.cpp new file mode 100644 index 000000000..d9acb8ca0 --- /dev/null +++ b/src/paimon/common/factories/singleton_test.cpp @@ -0,0 +1,88 @@ +/* + * 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 +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/common/factories/io_hook.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 +void RunStorm(const Worker& worker) { + std::atomic start{false}; + std::vector 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 + +// Regression gate for the Singleton double-checked-locking publication race. +// It only exercises the first construction if nothing has touched +// Singleton before, so this must stay the first GetInstance() call in +// this binary (singleton_test.cpp is the first source of common_factories_test +// and this is its first test). A FactoryCreator storm cannot serve as the +// gate: the REGISTER_PAIMON_FACTORY constructors in paimon_shared already +// initialize Singleton before main(). +TEST(SingletonTest, TestConcurrentIOHookGetInstance) { + std::array hooks{}; + std::array try_oks{}; + RunStorm([&hooks, &try_oks](int32_t i) { + hooks[i] = Singleton::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(); +} + +} // namespace paimon::test diff --git a/src/paimon/common/io/cache/cache_manager.h b/src/paimon/common/io/cache/cache_manager.h index f899d46ce..6fafefb5c 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(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(max_memory_bytes * high_priority_pool_ratio); auto data_cache_bytes = - static_cast(max_memory_bytes * (1.0 - high_priority_pool_ratio)); + SaturatingDoubleToInteger(max_memory_bytes * (1.0 - high_priority_pool_ratio)); data_cache_ = std::make_shared(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 000000000..4d7c180a3 --- /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 +#include +#include +#include +#include + +#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 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 DataLru(const CacheManager& manager) const { + return std::dynamic_pointer_cast(manager.DataCache()); + } + + std::shared_ptr IndexLru(const CacheManager& manager) const { + return std::dynamic_pointer_cast(manager.IndexCache()); + } + + private: + std::shared_ptr 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::max(), /*high_priority_pool_ratio=*/0.0); + + std::shared_ptr data_lru = DataLru(manager); + ASSERT_NE(data_lru, nullptr); + ASSERT_GE(data_lru->GetMaxWeight(), 0); + ASSERT_EQ(data_lru->GetMaxWeight(), std::numeric_limits::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 key = MakeKey(0); + auto reader = [&](const std::shared_ptr&) -> Result { + 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 data_lru = DataLru(manager); + std::shared_ptr 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 key = MakeKey(0); + int32_t reader_calls = 0; + auto reader = [&](const std::shared_ptr&) -> Result { + 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 evicted; + auto callback_for = [&evicted](int64_t position) -> CacheCallback { + return + [&evicted, position](const std::shared_ptr&) { evicted.push_back(position); }; + }; + auto reader = [&](const std::shared_ptr&) -> Result { + return MakeSegment(128, 'C'); + }; + + std::shared_ptr key0 = MakeKey(0); + std::shared_ptr key1 = MakeKey(1); + std::shared_ptr 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({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 ec736e33c..f2b3b2dee 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& out, @@ -27,8 +28,10 @@ SstFileWriter::SstFileWriter(const std::shared_ptr& out, const std::shared_ptr& factory, const std::shared_ptr& 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(static_cast(block_size * 1.1), pool); + std::make_unique(SaturatingDoubleToInteger(block_size * 1.1), pool); index_block_writer_ = std::make_unique(BlockHandle::MAX_ENCODED_LENGTH * 1024, pool); compression_type_ = factory->GetCompressionType(); diff --git a/src/paimon/common/utils/saturating_cast.h b/src/paimon/common/utils/saturating_cast.h new file mode 100644 index 000000000..cb9c6039f --- /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 +#include +#include +#include + +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 +inline TargetType SaturatingDoubleToInteger(double value) { + static_assert(std::is_same_v || std::is_same_v, + "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(std::numeric_limits::max())) { + return std::numeric_limits::max(); + } + if (value <= static_cast(std::numeric_limits::lowest())) { + return std::numeric_limits::lowest(); + } + return static_cast(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 000000000..cc6b74cf1 --- /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 +#include + +#include "gtest/gtest.h" + +namespace paimon::test { + +TEST(SaturatingCastTest, TestInt64InRangeTruncatesTowardZero) { + ASSERT_EQ(SaturatingDoubleToInteger(0.0), 0); + ASSERT_EQ(SaturatingDoubleToInteger(1.9), 1); + ASSERT_EQ(SaturatingDoubleToInteger(-1.9), -1); + // 2^63 - 1024 is the largest double below 2^63: it stays on the truncation path. + ASSERT_EQ(SaturatingDoubleToInteger(9223372036854774784.0), 9223372036854774784LL); +} + +TEST(SaturatingCastTest, TestInt64Saturation) { + // (double)INT64_MAX rounds up to 2^63, so the boundary double already saturates. + ASSERT_EQ(SaturatingDoubleToInteger( + static_cast(std::numeric_limits::max())), + std::numeric_limits::max()); + ASSERT_EQ(SaturatingDoubleToInteger(1e300), std::numeric_limits::max()); + ASSERT_EQ(SaturatingDoubleToInteger(-1e300), std::numeric_limits::lowest()); + ASSERT_EQ(SaturatingDoubleToInteger(std::numeric_limits::infinity()), + std::numeric_limits::max()); + ASSERT_EQ(SaturatingDoubleToInteger(-std::numeric_limits::infinity()), + std::numeric_limits::lowest()); + // The lowest bound is exactly representable and must survive as a value. + ASSERT_EQ(SaturatingDoubleToInteger( + static_cast(std::numeric_limits::lowest())), + std::numeric_limits::lowest()); +} + +TEST(SaturatingCastTest, TestInt64NaNBecomesZero) { + // Java's (long)Double.NaN == 0. + ASSERT_EQ(SaturatingDoubleToInteger(std::numeric_limits::quiet_NaN()), 0); +} + +TEST(SaturatingCastTest, TestInt32Path) { + // SstFileWriter converts through the int32_t instantiation. + ASSERT_EQ(SaturatingDoubleToInteger(42.7), 42); + ASSERT_EQ(SaturatingDoubleToInteger(-42.7), -42); + // The int32_t bounds are exactly representable as doubles and saturate inclusively. + ASSERT_EQ(SaturatingDoubleToInteger(2147483647.0), + std::numeric_limits::max()); + ASSERT_EQ(SaturatingDoubleToInteger(2147483648.0), + std::numeric_limits::max()); + ASSERT_EQ(SaturatingDoubleToInteger(-2147483648.0), + std::numeric_limits::lowest()); + ASSERT_EQ(SaturatingDoubleToInteger(-2147483649.0), + std::numeric_limits::lowest()); + ASSERT_EQ(SaturatingDoubleToInteger(std::numeric_limits::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 c1e97d033..449766ca6 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(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 5e612ff2e..56a39a295 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 +#include +#include + #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 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(-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 = SerializationUtils::SerializeBinaryRow(row, memory_pool.get()); + ASSERT_TRUE(bytes); + ASSERT_GE(bytes->size(), 4); + ASSERT_EQ(static_cast(bytes->data()[0]), 0x00); + ASSERT_EQ(static_cast(bytes->data()[1]), 0x00); + ASSERT_EQ(static_cast(bytes->data()[2]), 0x00); + ASSERT_EQ(static_cast(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(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(-9000000000)); + ASSERT_EQ(de_row.GetString(2).ToString(), "hello paimon!"); +} + } // namespace paimon::test