This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 b1a0598 feat: add read-ahead cache, serialization, stream and object
utils (#75)
b1a0598 is described below
commit b1a05987803b8ebd2081c7eb285b8ccb03397171
Author: dalingmeng <[email protected]>
AuthorDate: Tue Jun 16 09:12:02 2026 +0800
feat: add read-ahead cache, serialization, stream and object utils (#75)
---
LICENSE | 3 +
include/paimon/utils/read_ahead_cache.h | 183 +++++++++++++++
src/paimon/common/utils/object_utils.h | 153 ++++++++++++
src/paimon/common/utils/object_utils_test.cpp | 116 +++++++++
src/paimon/common/utils/read_ahead_cache.cpp | 258 +++++++++++++++++++++
src/paimon/common/utils/read_ahead_cache_test.cpp | 159 +++++++++++++
src/paimon/common/utils/serialization_utils.h | 103 ++++++++
.../common/utils/serialization_utils_test.cpp | 40 ++++
src/paimon/common/utils/stream_utils.h | 102 ++++++++
src/paimon/common/utils/stream_utils_test.cpp | 146 ++++++++++++
10 files changed, 1263 insertions(+)
diff --git a/LICENSE b/LICENSE
index c69084f..175f5e7 100644
--- a/LICENSE
+++ b/LICENSE
@@ -383,6 +383,9 @@ This product includes code from Apache ORC.
* ORC patch file:
* cmake_modules/orc.diff
+* Adapted read-ahead cache:
+ * include/paimon/utils/read_ahead_cache.h (adapted from Cache.hh)
+ * src/paimon/common/utils/read_ahead_cache.cpp (adapted from Cache.cc)
Copyright: 2013 and onwards The Apache Software Foundation.
Home page: https://orc.apache.org/
diff --git a/include/paimon/utils/read_ahead_cache.h
b/include/paimon/utils/read_ahead_cache.h
new file mode 100644
index 0000000..196045b
--- /dev/null
+++ b/include/paimon/utils/read_ahead_cache.h
@@ -0,0 +1,183 @@
+/*
+ * 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.
+ */
+
+// Adapted from Apache ORC
+// https://github.com/apache/orc/blob/main/c%2B%2B/src/io/Cache.hh
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+#include "paimon/fs/file_system.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// PrefetchCacheMode
+/// Cache prefetch switch modes.
+/// Controls whether to enable cache prefetching under different
circumstances, such as queries with
+/// predicates or bitmap indexes.
+///
+/// - ALWAYS: Enable cache in all scenarios.
+/// - EXCLUDE_PREDICATE: Disable cache when query has predicates.
+/// - EXCLUDE_BITMAP: Disable cache when using bitmap index.
+/// - EXCLUDE_BITMAP_OR_PREDICATE: Disable cache if query has predicates or
bitmap index.
+/// - NEVER: Always disable cache.
+enum class PAIMON_EXPORT PrefetchCacheMode {
+ ALWAYS = 1,
+ EXCLUDE_PREDICATE = 2,
+ EXCLUDE_BITMAP = 3,
+ EXCLUDE_BITMAP_OR_PREDICATE = 4,
+ NEVER = 5
+};
+
+/// Configuration parameters for the read-ahead cache behavior.
+///
+/// This struct controls various limits and prefetching strategies used by
+/// ReadAheadCache to balance memory usage, I/O efficiency, and latency hiding.
+class PAIMON_EXPORT CacheConfig {
+ public:
+ CacheConfig();
+ CacheConfig(uint64_t buffer_size_limit, uint64_t range_size_limit,
uint64_t hole_size_limit,
+ uint64_t pre_buffer_limit);
+
+ /// Returns the maximum total size (in bytes) of cached data.
+ uint64_t GetBufferSizeLimit() const {
+ return buffer_size_limit_;
+ }
+
+ /// Sets the maximum total size (in bytes) of cached data.
+ void SetBufferSizeLimit(uint64_t buffer_size_limit) {
+ buffer_size_limit_ = buffer_size_limit;
+ }
+
+ /// Returns the maximum allowed size (in bytes) for a single cached range.
+ uint64_t GetRangeSizeLimit() const {
+ return range_size_limit_;
+ }
+
+ /// Sets the maximum allowed size (in bytes) for a single cached range.
+ void SetRangeSizeLimit(uint64_t range_size_limit) {
+ range_size_limit_ = range_size_limit;
+ }
+
+ /// Returns the maximum gap size (in bytes) considered mergeable between
adjacent ranges.
+ uint64_t GetHoleSizeLimit() const {
+ return hole_size_limit_;
+ }
+
+ /// Sets the maximum gap size (in bytes) considered mergeable between
adjacent ranges.
+ void SetHoleSizeLimit(uint64_t hole_size_limit) {
+ hole_size_limit_ = hole_size_limit;
+ }
+
+ /// Returns the maximum size to pre-buffer ahead of the current read
position.
+ uint64_t GetPreBufferLimit() const {
+ return pre_buffer_limit_;
+ }
+
+ /// Sets the maximum size to pre-buffer ahead of the current read position.
+ void SetPreBufferLimit(uint64_t pre_buffer_limit) {
+ pre_buffer_limit_ = pre_buffer_limit;
+ }
+
+ private:
+ uint64_t buffer_size_limit_;
+ uint64_t range_size_limit_;
+ uint64_t hole_size_limit_;
+ uint64_t pre_buffer_limit_;
+};
+
+/// A byte range with offset and length.
+struct PAIMON_EXPORT ByteRange {
+ uint64_t offset;
+ uint64_t length;
+
+ ByteRange() = default;
+ ByteRange(uint64_t offset, uint64_t length) : offset(offset),
length(length) {}
+
+ friend bool operator==(const ByteRange& left, const ByteRange& right) {
+ return (left.offset == right.offset && left.length == right.length);
+ }
+ friend bool operator!=(const ByteRange& left, const ByteRange& right) {
+ return !(left == right);
+ }
+
+ /// @param other The other byte range to check.
+ /// @return true if this range contains the other range
+ bool Contains(const ByteRange& other) const {
+ return (offset <= other.offset && offset + length >= other.offset +
other.length);
+ }
+};
+
+/// A byte slice with buffer, offset and length.
+struct PAIMON_EXPORT ByteSlice {
+ std::shared_ptr<Bytes> buffer = nullptr;
+ uint64_t offset = 0;
+ uint64_t length = 0;
+};
+
+/// A read cache designed to hide IO latencies when reading.
+/// Prefetching strategy: When a range is read, the cache will prefetch up to
+/// `pre_buffer_range_count` additional adjacent ranges ahead of the requested
offset. This helps
+/// hide I/O latency for sequential access. Example: If you read range [0,
100), and
+/// pre_buffer_range_count=2, the next two configured ranges will also be
prefetched.
+///
+/// Eviction policy: The cache uses a simple FIFO eviction policy based on
total cached byte size.
+/// When adding new ranges would exceed `buffer_size_limit`, the oldest cached
ranges are evicted
+/// first until there is enough space for the new data.
+class PAIMON_EXPORT ReadAheadCache {
+ public:
+ /// Construct a read cache with given options
+ ReadAheadCache(const std::shared_ptr<InputStream>& stream, const
CacheConfig& config,
+ const std::shared_ptr<MemoryPool>& memory_pool);
+ ~ReadAheadCache();
+
+ /// Initialize the cache with given byte ranges to be cached.
+ /// @param ranges The byte ranges to be cached.
+ /// @return Status of the operation.
+ /// @note This method must be called before any Read() calls. Ranges will
be coalesced based
+ /// on the cache configuration.
+ Status Init(std::vector<ByteRange>&& ranges);
+
+ /// Read a range previously provided to Init().
+ /// @param range The byte range to read.
+ /// @return The byte slice containing the requested data. If the data is
not yet cached
+ /// (cache miss), the returned `ByteSlice` will have a null buffer
(`buffer == nullptr`)
+ Result<ByteSlice> Read(const ByteRange& range);
+
+ /// Reset the cache to its initial state, clearing all cached data and
configuration.
+ ///
+ /// This method waits for all ongoing asynchronous read operations to
complete,
+ /// clears all cached entries, and resets the internal state so that
Init() can be called again.
+ /// After calling Reset, the cache can be safely re-initialized with new
ranges.
+ void Reset();
+
+ private:
+ class Impl;
+ std::unique_ptr<Impl> impl_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/object_utils.h
b/src/paimon/common/utils/object_utils.h
new file mode 100644
index 0000000..66b6c54
--- /dev/null
+++ b/src/paimon/common/utils/object_utils.h
@@ -0,0 +1,153 @@
+/*
+ * 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 <cstddef>
+#include <cstdint>
+#include <functional>
+#include <map>
+#include <set>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
+#include "paimon/result.h"
+#include "paimon/traits.h"
+namespace paimon {
+/// Utils for objects.
+class ObjectUtils {
+ public:
+ ObjectUtils() = delete;
+ ~ObjectUtils() = delete;
+
+ template <typename T, typename U>
+ static bool ContainsAll(const T& all, const U& contains) {
+ using V = typename T::value_type;
+ std::unordered_set<V> all_set(all.begin(), all.end());
+ for (const auto& element : contains) {
+ if (all_set.find(element) == all_set.end()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ template <typename T>
+ static bool Contains(const std::vector<T>& all, const T& target) {
+ for (const auto& v : all) {
+ if (v == target) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ template <typename T>
+ static std::unordered_set<T> DuplicateItems(const std::vector<T>& entries)
{
+ std::unordered_map<T, int32_t> counts;
+ for (const auto& entry : entries) {
+ counts[entry]++;
+ }
+
+ std::unordered_set<T> duplicates;
+ for (const auto& entry : counts) {
+ if (entry.second > 1) {
+ duplicates.insert(entry.first);
+ }
+ }
+ return duplicates;
+ }
+
+ template <typename T>
+ static bool TEST_Equal(const std::vector<T>& lhs, const std::vector<T>&
rhs) {
+ if (lhs.size() != rhs.size()) {
+ return false;
+ }
+ for (size_t i = 0; i < lhs.size(); i++) {
+ if constexpr (is_pointer<T>::value) {
+ if (!lhs[i]->TEST_Equal(*rhs[i])) {
+ return false;
+ }
+ } else {
+ if (!lhs[i].TEST_Equal(rhs[i])) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ template <typename T>
+ static bool Equal(const std::vector<T>& lhs, const std::vector<T>& rhs) {
+ if (lhs.size() != rhs.size()) {
+ return false;
+ }
+ for (size_t i = 0; i < lhs.size(); i++) {
+ if constexpr (is_pointer<T>::value) {
+ if (!(*(lhs[i]) == *(rhs[i]))) {
+ return false;
+ }
+ } else {
+ if (!(lhs[i] == rhs[i])) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ // used to create an identifier to index map, param func is used to
convert T to identifier I
+ // e.g., T = DataField, I = std::string, return std::map<std::string,
int32_t>
+ template <typename T, typename Func>
+ static auto CreateIdentifierToIndexMap(const std::vector<T>& vec, const
Func& func)
+ -> std::map<decltype(func(std::declval<T>())), int32_t> {
+ using KeyType = decltype(func(std::declval<T>()));
+ std::map<KeyType, int32_t> result;
+ for (int32_t i = 0; i < static_cast<int32_t>(vec.size()); ++i) {
+ result[func(vec[i])] = i;
+ }
+ return result;
+ }
+
+ template <typename T>
+ static std::map<T, int32_t> CreateIdentifierToIndexMap(const
std::vector<T>& vec) {
+ std::map<T, int32_t> index_map;
+ for (int32_t i = 0; i < static_cast<int32_t>(vec.size()); i++) {
+ index_map[vec[i]] = i;
+ }
+ return index_map;
+ }
+
+ /// Precondition: U and T must be pointer and U::value can move to T::value
+ template <typename T, typename U>
+ static std::vector<T> MoveVector(std::vector<U>&& input) {
+ static_assert(is_pointer<U>::value && is_pointer<T>::value &&
+ std::is_convertible_v<value_type_traits_t<U>,
value_type_traits_t<T>>,
+ "U and T must be pointer and U::value can move to
T::value");
+ std::vector<T> result;
+ result.reserve(input.size());
+ for (auto& item : input) {
+ result.push_back(std::move(item));
+ }
+ input.clear();
+ return result;
+ }
+};
+} // namespace paimon
diff --git a/src/paimon/common/utils/object_utils_test.cpp
b/src/paimon/common/utils/object_utils_test.cpp
new file mode 100644
index 0000000..6fdee78
--- /dev/null
+++ b/src/paimon/common/utils/object_utils_test.cpp
@@ -0,0 +1,116 @@
+/*
+ * 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/object_utils.h"
+
+#include <string>
+
+#include "gtest/gtest.h"
+#include "paimon/common/types/data_field.h"
+
+namespace paimon::test {
+
+TEST(ObjectUtilsTest, TestContainsAll) {
+ std::vector<int32_t> all = {0, 1, 2, 3};
+ std::vector<int32_t> contains1 = {1, 2};
+ std::vector<int32_t> contains2 = {1, 10};
+ ASSERT_TRUE(ObjectUtils::ContainsAll(all, contains1));
+ ASSERT_FALSE(ObjectUtils::ContainsAll(all, contains2));
+}
+
+TEST(ObjectUtilsTest, TestContains) {
+ std::vector<int32_t> all = {0, 2, 4, 6};
+ ASSERT_TRUE(ObjectUtils::Contains(all, 4));
+ ASSERT_FALSE(ObjectUtils::Contains(all, 8));
+}
+
+TEST(ObjectUtilsTest, TestDuplicateItems) {
+ std::vector<std::string> entries = {"apple", "banana", "cherry", "apple",
+ "banana", "date", "fig",
"cherry"};
+ std::unordered_set<std::string> expected = {"banana", "apple", "cherry"};
+ ASSERT_EQ(ObjectUtils::DuplicateItems(entries), expected);
+}
+
+TEST(ObjectUtilsTest, TestCreateIdentifierToIndexMap) {
+ std::vector<DataField> fields = {
+ DataField(0, arrow::field("f11", arrow::boolean())),
+ DataField(1, arrow::field("f10", arrow::int8())),
+ DataField(2, arrow::field("f9", arrow::int16())),
+ DataField(3, arrow::field("f8", arrow::int32())),
+ DataField(4, arrow::field("f7", arrow::int64())),
+ DataField(5, arrow::field("f6", arrow::float32())),
+ DataField(6, arrow::field("f5", arrow::float64())),
+ DataField(7, arrow::field("f4", arrow::utf8())),
+ DataField(8, arrow::field("f3", arrow::binary())),
+ DataField(9, arrow::field("f2",
arrow::timestamp(arrow::TimeUnit::NANO))),
+ DataField(10, arrow::field("f1", arrow::date32())),
+ DataField(11, arrow::field("f0", arrow::decimal128(2, 2))),
+ };
+
+ {
+ auto result_map = ObjectUtils::CreateIdentifierToIndexMap(
+ fields, [](const DataField& field) { return field.Name(); });
+ ASSERT_EQ(result_map.size(), 12);
+ std::map<std::string, int32_t> expected_map = {
+ {"f0", 11}, {"f1", 10}, {"f2", 9}, {"f3", 8}, {"f4", 7}, {"f5",
6},
+ {"f6", 5}, {"f7", 4}, {"f8", 3}, {"f9", 2}, {"f10", 1}, {"f11",
0},
+ };
+ ASSERT_EQ(expected_map, result_map);
+ }
+ {
+ auto result_map = ObjectUtils::CreateIdentifierToIndexMap(
+ std::vector<DataField>({}), [](const DataField& field) { return
field.Name(); });
+ ASSERT_TRUE(result_map.empty());
+ }
+ {
+ std::vector<std::string> vec = {"f0", "f1", "f2"};
+ std::map<std::string, int32_t> result_map =
ObjectUtils::CreateIdentifierToIndexMap(vec);
+ std::map<std::string, int32_t> expected_map = {
+ {"f0", 0},
+ {"f1", 1},
+ {"f2", 2},
+ };
+ ASSERT_EQ(expected_map, result_map);
+ }
+}
+TEST(ObjectUtilsTest, TestMoveVector) {
+ struct Base {
+ virtual ~Base() = default;
+ virtual int32_t Value() const = 0;
+ };
+
+ struct Derived : Base {
+ explicit Derived(int32_t v) : val(v) {}
+ int32_t Value() const override {
+ return val;
+ }
+ int32_t val;
+ };
+ std::vector<std::unique_ptr<Derived>> derived_vec;
+ derived_vec.push_back(std::make_unique<Derived>(10));
+ derived_vec.push_back(std::make_unique<Derived>(20));
+ derived_vec.push_back(std::make_unique<Derived>(30));
+
+ auto base_vec =
paimon::ObjectUtils::MoveVector<std::unique_ptr<Base>>(std::move(derived_vec));
+
+ ASSERT_EQ(base_vec[0]->Value(), 10);
+ ASSERT_EQ(base_vec[1]->Value(), 20);
+ ASSERT_EQ(base_vec[2]->Value(), 30);
+}
+} // namespace paimon::test
diff --git a/src/paimon/common/utils/read_ahead_cache.cpp
b/src/paimon/common/utils/read_ahead_cache.cpp
new file mode 100644
index 0000000..aa7e5cb
--- /dev/null
+++ b/src/paimon/common/utils/read_ahead_cache.cpp
@@ -0,0 +1,258 @@
+/*
+ * 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.
+ */
+
+// Adapted from Apache ORC
+// https://github.com/apache/orc/blob/main/c%2B%2B/src/io/Cache.cc
+
+#include "paimon/utils/read_ahead_cache.h"
+
+#include <algorithm>
+#include <cassert>
+#include <future>
+#include <shared_mutex>
+
+#include "paimon/common/utils/byte_range_combiner.h"
+
+namespace paimon {
+
+struct RangeCacheEntry {
+ ByteRange range;
+ std::shared_ptr<Bytes> buffer;
+ std::shared_future<Status> future; // use shared_future in case of
multiple get calls
+
+ RangeCacheEntry() = default;
+ RangeCacheEntry(const ByteRange& range, std::shared_ptr<Bytes> buffer,
+ std::future<Status> future)
+ : range(range), buffer(std::move(buffer)),
future(std::move(future).share()) {}
+
+ friend bool operator<(const RangeCacheEntry& left, const RangeCacheEntry&
right) {
+ return left.range.offset < right.range.offset;
+ }
+};
+
+CacheConfig::CacheConfig(uint64_t buffer_size_limit, uint64_t range_size_limit,
+ uint64_t hole_size_limit, uint64_t pre_buffer_limit)
+ : buffer_size_limit_(buffer_size_limit),
+ range_size_limit_(range_size_limit),
+ hole_size_limit_(hole_size_limit),
+ pre_buffer_limit_(pre_buffer_limit) {}
+
+CacheConfig::CacheConfig()
+ : CacheConfig(/*buffer_size_limit=*/512 * 1024 * 1024,
+ /*range_size_limit=*/16 * 1024 * 1024,
+ /*hole_size_limit=*/8 * 1024,
+ /*pre_buffer_limit=*/128 * 1024 * 1024) {}
+
+class ReadAheadCache::Impl {
+ public:
+ Impl(const std::shared_ptr<InputStream>& stream, const CacheConfig& config,
+ const std::shared_ptr<MemoryPool>& memory_pool);
+ ~Impl();
+
+ Status Init(std::vector<ByteRange>&& ranges);
+ Result<ByteSlice> Read(const ByteRange& range);
+ void Reset();
+
+ private:
+ std::vector<RangeCacheEntry> MakeCacheEntries(const
std::vector<ByteRange>& ranges) const;
+ void PreBuffer(uint64_t offset);
+
+ /// Cache the given ranges in the background.
+ ///
+ /// The caller must ensure that the ranges do not overlap with each other,
+ /// nor with previously cached ranges. Otherwise, behaviour will be
undefined.
+ void Cache(std::vector<ByteRange> ranges);
+
+ std::shared_ptr<InputStream> stream_;
+ CacheConfig config_;
+ // Ordered by offset (so as to find a matching region by binary search)
+ std::vector<RangeCacheEntry> entries_;
+ std::shared_ptr<MemoryPool> memory_pool_;
+ std::shared_mutex rw_mutex_;
+ std::vector<std::atomic<bool>> is_cached_;
+ std::vector<ByteRange> pending_ranges_;
+ bool is_initialized_ = false;
+};
+
+void ReadAheadCache::Impl::Cache(std::vector<ByteRange> ranges) {
+ std::sort(ranges.begin(), ranges.end(),
+ [](const ByteRange& a, const ByteRange& b) { return a.offset <
b.offset; });
+ std::vector<RangeCacheEntry> new_entries = MakeCacheEntries(ranges);
+ // Add new entries, themselves ordered by offset
+ std::unique_lock<std::shared_mutex> lock(rw_mutex_);
+ if (entries_.size() > 0) {
+ size_t new_entries_size = 0;
+ for (const auto& e : new_entries) {
+ new_entries_size += e.range.length;
+ }
+
+ size_t total_size = 0;
+ for (const auto& e : entries_) {
+ total_size += e.range.length;
+ }
+ size_t limit = config_.GetBufferSizeLimit();
+ while (!entries_.empty() && total_size + new_entries_size > limit) {
+ auto iter = entries_.begin();
+ total_size -= entries_.front().range.length;
+ entries_.erase(iter);
+ }
+
+ std::vector<RangeCacheEntry> merged(entries_.size() +
new_entries.size());
+ std::merge(entries_.begin(), entries_.end(), new_entries.begin(),
new_entries.end(),
+ merged.begin());
+ entries_ = std::move(merged);
+ } else {
+ entries_ = std::move(new_entries);
+ }
+}
+
+Status ReadAheadCache::Impl::Init(std::vector<ByteRange>&& ranges) {
+ if (is_initialized_) {
+ return Status::Invalid("Cache has already been initialized");
+ }
+ if (config_.GetRangeSizeLimit() >
static_cast<uint64_t>(std::numeric_limits<uint32_t>::max())) {
+ return Status::Invalid("CacheConfig range_size_limit exceeds uint32_t
max");
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<ByteRange> pending_ranges,
+ ByteRangeCombiner::CoalesceByteRanges(std::move(ranges),
config_.GetHoleSizeLimit(),
+ config_.GetRangeSizeLimit()));
+ for (const auto& pending_range : pending_ranges) {
+ if (pending_range.length >
static_cast<uint64_t>(std::numeric_limits<uint32_t>::max())) {
+ return Status::Invalid("range length should not be larger than
uint32_t max");
+ }
+ }
+ pending_ranges_ = pending_ranges;
+ is_cached_ = std::vector<std::atomic<bool>>(pending_ranges_.size());
+ for (auto& is_cached : is_cached_) {
+ is_cached.store(false);
+ }
+ is_initialized_ = true;
+ return Status::OK();
+}
+
+void ReadAheadCache::Impl::PreBuffer(uint64_t offset) {
+ auto it = std::lower_bound(pending_ranges_.begin(), pending_ranges_.end(),
offset,
+ [](const ByteRange& range, uint64_t offset) {
+ return range.offset + range.length <=
offset;
+ });
+ if (it == pending_ranges_.end() || it->offset > offset) {
+ return;
+ }
+
+ size_t start_idx = std::distance(pending_ranges_.begin(), it);
+ std::vector<ByteRange> ranges;
+ size_t total_bytes = 0;
+ for (size_t i = start_idx; i < pending_ranges_.size(); ++i) {
+ size_t range_size = pending_ranges_[i].length;
+ total_bytes += range_size;
+ if (total_bytes > config_.GetPreBufferLimit()) {
+ break;
+ }
+ if (is_cached_[i].exchange(true)) {
+ continue;
+ }
+ ranges.emplace_back(pending_ranges_[i]);
+ }
+
+ if (!ranges.empty()) {
+ Cache(std::move(ranges));
+ }
+}
+
+ReadAheadCache::Impl::Impl(const std::shared_ptr<InputStream>& stream, const
CacheConfig& config,
+ const std::shared_ptr<MemoryPool>& memory_pool)
+ : stream_(stream), config_(config), memory_pool_(memory_pool) {}
+
+ReadAheadCache::Impl::~Impl() {
+ std::unique_lock<std::shared_mutex> lock(rw_mutex_);
+ for (auto& entry : entries_) {
+ entry.future.wait();
+ }
+}
+
+void ReadAheadCache::Impl::Reset() {
+ std::unique_lock<std::shared_mutex> lock(rw_mutex_);
+ for (auto& entry : entries_) {
+ entry.future.wait();
+ }
+ entries_.clear();
+ is_cached_.clear();
+ pending_ranges_.clear();
+ is_initialized_ = false;
+}
+
+Result<ByteSlice> ReadAheadCache::Impl::Read(const ByteRange& range) {
+ if (range.length == 0) {
+ return ByteSlice{std::make_shared<Bytes>(0, memory_pool_.get()), 0, 0};
+ }
+ PreBuffer(range.offset);
+ ByteSlice result{};
+ {
+ std::shared_lock<std::shared_mutex> lock(rw_mutex_);
+ auto it = std::lower_bound(entries_.begin(), entries_.end(),
range.offset,
+ [](const RangeCacheEntry& e, uint64_t
offset) {
+ return e.range.offset + e.range.length
<= offset;
+ });
+ if (it != entries_.end() && it->range.Contains(range)) {
+ PAIMON_RETURN_NOT_OK(it->future.get());
+ result = ByteSlice{it->buffer, range.offset - it->range.offset,
range.length};
+ return result;
+ }
+ }
+ return result;
+}
+
+std::vector<RangeCacheEntry> ReadAheadCache::Impl::MakeCacheEntries(
+ const std::vector<ByteRange>& ranges) const {
+ std::vector<RangeCacheEntry> new_entries;
+ new_entries.reserve(ranges.size());
+ for (const auto& range : ranges) {
+ auto promise = std::make_shared<std::promise<Status>>();
+ auto future = promise->get_future();
+ auto buffer = std::make_shared<Bytes>(range.length,
memory_pool_.get());
+ stream_->ReadAsync(
+ buffer->data(), static_cast<uint32_t>(buffer->size()),
range.offset,
+ [promise, buffer](Status status) mutable {
promise->set_value(status); });
+ new_entries.emplace_back(range, std::move(buffer), std::move(future));
+ }
+ return new_entries;
+}
+
+ReadAheadCache::ReadAheadCache(const std::shared_ptr<InputStream>& stream,
+ const CacheConfig& config,
+ const std::shared_ptr<MemoryPool>& memory_pool)
+ : impl_(std::make_unique<Impl>(stream, config, memory_pool)) {}
+
+ReadAheadCache::~ReadAheadCache() = default;
+
+Status ReadAheadCache::Init(std::vector<ByteRange>&& ranges) {
+ return impl_->Init(std::move(ranges));
+}
+
+Result<ByteSlice> ReadAheadCache::Read(const ByteRange& range) {
+ return impl_->Read(range);
+}
+
+void ReadAheadCache::Reset() {
+ return impl_->Reset();
+}
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/read_ahead_cache_test.cpp
b/src/paimon/common/utils/read_ahead_cache_test.cpp
new file mode 100644
index 0000000..6762542
--- /dev/null
+++ b/src/paimon/common/utils/read_ahead_cache_test.cpp
@@ -0,0 +1,159 @@
+/*
+ * 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/utils/read_ahead_cache.h"
+
+#include <fstream>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/fs/file_system_factory.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+// Helper to create a test file, write content, and return a ready
ReadAheadCache.
+struct TestCacheEnv {
+ std::string path;
+ std::shared_ptr<paimon::ReadAheadCache> cache;
+ std::shared_ptr<paimon::MemoryPool> pool;
+};
+
+TestCacheEnv CreateTestFileAndCache(const std::string& filename, const
std::string& content,
+ const paimon::CacheConfig& config,
+ std::vector<paimon::ByteRange> ranges) {
+ auto dir = UniqueTestDirectory::Create();
+ EXPECT_TRUE(dir);
+ std::string path = dir->Str() + "/" + filename;
+ std::ofstream file(path, std::ios::binary);
+ EXPECT_TRUE(file.is_open());
+ file.write(content.data(), content.size());
+ EXPECT_FALSE(file.fail());
+ file.close();
+
+ auto fs_result = FileSystemFactory::Get("local", path, {});
+ EXPECT_TRUE(fs_result.ok());
+ auto fs = std::move(fs_result).value();
+ auto in_result = fs->Open(path);
+ EXPECT_TRUE(in_result.ok());
+ auto in = std::move(in_result).value();
+
+ auto pool = GetDefaultPool();
+ auto cache = std::make_shared<ReadAheadCache>(std::move(in), config, pool);
+ EXPECT_OK(cache->Init(std::move(ranges)));
+ return {path, cache, pool};
+}
+
+TEST(TestReadAheadCache, TestBasics) {
+ CacheConfig config(/*buffer_size_limit=*/256 * 1024 * 1024,
/*range_size_limit=*/10,
+ /*hole_size_limit=*/2, /*pre_buffer_limit=*/128 * 1024
* 1024);
+ std::string content = "abcdefghijklmnopqrstuvwxyz";
+ auto env = CreateTestFileAndCache(
+ "data_file", content, config,
+ {{1, 2}, {3, 2}, {8, 2}, {10, 4}, {14, 0}, {15, 4}, {20, 2}, {25, 0}});
+ auto& cache = *env.cache;
+
+ auto assert_slice_equal = [](const ByteSlice& slice, const std::string&
expected) {
+ ASSERT_TRUE(slice.buffer) << expected;
+ EXPECT_EQ(expected, std::string_view(slice.buffer->data() +
slice.offset, slice.length));
+ };
+
+ ByteSlice slice;
+
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({20, 2}));
+ assert_slice_equal(slice, "uv");
+
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({1, 2}));
+ assert_slice_equal(slice, "bc");
+
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({3, 2}));
+ assert_slice_equal(slice, "de");
+
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({8, 2}));
+ assert_slice_equal(slice, "ij");
+
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({10, 4}));
+ assert_slice_equal(slice, "klmn");
+
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({15, 4}));
+ assert_slice_equal(slice, "pqrs");
+
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({19, 3}));
+ assert_slice_equal(slice, "tuv");
+
+ // Zero-sized
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({14, 0}));
+ assert_slice_equal(slice, "");
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({25, 0}));
+ assert_slice_equal(slice, "");
+
+ // Non-cached ranges
+
+ ASSERT_FALSE(cache.Read({20, 3}).value().buffer);
+ ASSERT_FALSE(cache.Read({0, 3}).value().buffer);
+ ASSERT_FALSE(cache.Read({25, 2}).value().buffer);
+}
+
+// Test repeated reads to the same range to ensure cache reuse.
+TEST(TestReadAheadCache, TestRepeatedReadCacheReuse) {
+ CacheConfig config(/*buffer_size_limit=*/64, /*range_size_limit=*/10,
+ /*hole_size_limit=*/2, /*pre_buffer_limit=*/64);
+ std::string content = "abcdefghijklmnopqrstuvwxyz";
+ auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5},
{7, 5}});
+ auto& cache = *env.cache;
+
+ ByteSlice slice;
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({0, 5}));
+ ASSERT_TRUE(slice.buffer);
+ std::string first_read(slice.buffer->data() + slice.offset, slice.length);
+ ASSERT_EQ(first_read, "abcde");
+
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({0, 5}));
+ ASSERT_TRUE(slice.buffer);
+ std::string second_read(slice.buffer->data() + slice.offset, slice.length);
+ ASSERT_EQ(second_read, "abcde");
+}
+
+// Test cache eviction when buffer size is limited.
+TEST(TestReadAheadCache, TestCacheEviction) {
+ CacheConfig config(/*buffer_size_limit=*/10, /*range_size_limit=*/5,
+ /*hole_size_limit=*/2, /*pre_buffer_limit=*/10);
+ std::string content = "abcdefghijklmnopqrstuvwxyz";
+ auto env = CreateTestFileAndCache("data_file", content, config, {{0, 5},
{8, 5}, {16, 5}});
+ auto& cache = *env.cache;
+
+ ByteSlice slice;
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({0, 5}));
+ ASSERT_TRUE(slice.buffer);
+ std::string first_read(slice.buffer->data() + slice.offset, slice.length);
+ ASSERT_EQ(first_read, "abcde");
+
+ // Reading another range should evict the first one due to buffer size
limit
+ ASSERT_OK_AND_ASSIGN(slice, cache.Read({8, 5}));
+ ASSERT_TRUE(slice.buffer);
+ std::string second_read(slice.buffer->data() + slice.offset, slice.length);
+ ASSERT_EQ(second_read, "ijklm");
+
+ // The first range should now be a cache miss (buffer is nullptr)
+ auto miss = cache.Read({0, 5});
+ ASSERT_FALSE(miss.value().buffer);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/utils/serialization_utils.h
b/src/paimon/common/utils/serialization_utils.h
new file mode 100644
index 0000000..4718ea9
--- /dev/null
+++ b/src/paimon/common/utils/serialization_utils.h
@@ -0,0 +1,103 @@
+/*
+ * 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 <cstdint>
+#include <cstring>
+#include <memory>
+#include <utility>
+
+#include "fmt/format.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
+#include "paimon/common/memory/memory_segment.h"
+#include "paimon/common/memory/memory_segment_utils.h"
+#include "paimon/common/utils/math.h"
+#include "paimon/io/byte_order.h"
+#include "paimon/io/data_input_stream.h"
+#include "paimon/macros.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/type_fwd.h"
+
+namespace paimon {
+
+/// Utils for serialization.
+class SerializationUtils {
+ public:
+ SerializationUtils() = delete;
+ ~SerializationUtils() = delete;
+
+ /// Serialize `BinaryRow`, the difference between this and
`BinaryRowSerializer` is
+ /// that arity is also serialized here, so the deserialization is
schemaless.
+ static std::shared_ptr<Bytes> SerializeBinaryRow(const BinaryRow& row,
MemoryPool* pool) {
+ int32_t size_in_row = row.GetSizeInBytes();
+ auto bytes = Bytes::AllocateBytes(size_in_row + sizeof(int32_t), pool);
+ int32_t arity = row.GetFieldCount();
+ if (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN) {
+ arity = EndianSwapValue(arity);
+ }
+ memcpy(bytes->data(), &arity, sizeof(int32_t));
+ MemorySegmentUtils::CopyToBytes({row.GetSegment()}, row.GetOffset(),
bytes.get(),
+ sizeof(int32_t), size_in_row);
+ return bytes;
+ }
+
+ static Status SerializeBinaryRow(const BinaryRow& row,
MemorySegmentOutputStream* out) {
+ if (row.GetSizeInBytes() < 0) {
+ return Status::Invalid(
+ fmt::format("bytes size {} is less than 0",
row.GetSizeInBytes()));
+ }
+ out->WriteValue<int32_t>(4 + row.GetSizeInBytes());
+ out->WriteValue<int32_t>(row.GetFieldCount());
+ return MemorySegmentUtils::CopyToStream({row.GetSegment()},
row.GetOffset(),
+ row.GetSizeInBytes(), out);
+ }
+
+ /// Schemaless deserialization for `BinaryRow`.
+ static Result<BinaryRow> DeserializeBinaryRow(const
std::shared_ptr<Bytes>& bytes) {
+ 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()));
+ if (SystemByteOrder() == ByteOrder::PAIMON_LITTLE_ENDIAN) {
+ arity = EndianSwapValue(arity);
+ }
+ if (PAIMON_UNLIKELY(arity < 0)) {
+ return Status::Invalid("arity is less than 0");
+ }
+ BinaryRow row(arity);
+ row.PointTo(MemorySegment::Wrap(bytes), 4, bytes->size() - 4);
+ return row;
+ }
+
+ /// Schemaless deserialization for `BinaryRow` from a `DataInputStream`.
+ static Result<BinaryRow> DeserializeBinaryRow(DataInputStream* input,
MemoryPool* pool) {
+ int32_t read_length = -1;
+ PAIMON_ASSIGN_OR_RAISE(read_length, input->ReadValue<int32_t>());
+ std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(read_length, pool);
+ PAIMON_RETURN_NOT_OK(input->ReadBytes(bytes.get()));
+ return DeserializeBinaryRow(bytes);
+ }
+};
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/serialization_utils_test.cpp
b/src/paimon/common/utils/serialization_utils_test.cpp
new file mode 100644
index 0000000..5e612ff
--- /dev/null
+++ b/src/paimon/common/utils/serialization_utils_test.cpp
@@ -0,0 +1,40 @@
+/*
+ * 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/serialization_utils.h"
+
+#include "gtest/gtest.h"
+
+namespace paimon::test {
+
+class SerializationUtilsTest : public ::testing::Test {
+ public:
+ void SetUp() override {}
+ void TearDown() override {}
+};
+
+TEST_F(SerializationUtilsTest, TestSerializeBinaryRow) {
+ std::shared_ptr<MemoryPool> memory_pool = GetDefaultPool();
+ MemorySegmentOutputStream
out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, memory_pool);
+ BinaryRow row = BinaryRow::EmptyRow();
+ std::shared_ptr<Bytes> bytes = SerializationUtils::SerializeBinaryRow(row,
memory_pool.get());
+ ASSERT_TRUE(bytes);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/utils/stream_utils.h
b/src/paimon/common/utils/stream_utils.h
new file mode 100644
index 0000000..5537c83
--- /dev/null
+++ b/src/paimon/common/utils/stream_utils.h
@@ -0,0 +1,102 @@
+/*
+ * 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 <algorithm>
+#include <cstdint>
+#include <functional>
+#include <future>
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "fmt/format.h"
+#include "paimon/common/executor/future.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/macros.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+class StreamUtils {
+ public:
+ StreamUtils() = default;
+ ~StreamUtils() = default;
+
+ static Result<PAIMON_UNIQUE_PTR<Bytes>>
ReadFully(std::unique_ptr<InputStream> input_stream,
+ const
std::shared_ptr<MemoryPool>& pool) {
+ PAIMON_RETURN_NOT_OK(input_stream->Seek(0, FS_SEEK_SET));
+ PAIMON_ASSIGN_OR_RAISE(uint64_t file_length, input_stream->Length());
+ PAIMON_UNIQUE_PTR<Bytes> content = Bytes::AllocateBytes(file_length,
pool.get());
+ PAIMON_ASSIGN_OR_RAISE(int32_t actual_read_len,
+ input_stream->Read(content->data(),
content->size()));
+ if (static_cast<uint32_t>(actual_read_len) != file_length) {
+ return Status::Invalid("actual read length {}, not match with
expect length {}",
+ actual_read_len, file_length);
+ }
+ return content;
+ }
+
+ static Result<PAIMON_UNIQUE_PTR<Bytes>> ReadAsyncFully(
+ std::unique_ptr<InputStream> input_stream, const
std::shared_ptr<MemoryPool>& pool) {
+ PAIMON_ASSIGN_OR_RAISE(uint64_t file_length, input_stream->Length());
+ PAIMON_UNIQUE_PTR<Bytes> content = Bytes::AllocateBytes(file_length,
pool.get());
+ PAIMON_RETURN_NOT_OK(ReadAsyncFully(std::move(input_stream),
content->data()));
+ return content;
+ }
+
+ static Status ReadAsyncFully(std::unique_ptr<InputStream> input_stream,
char* content) {
+ PAIMON_RETURN_NOT_OK(input_stream->Seek(0, FS_SEEK_SET));
+ PAIMON_ASSIGN_OR_RAISE(uint64_t file_length, input_stream->Length());
+
+ uint64_t read_offset = 0;
+ uint32_t read_len = std::min(file_length, kDefaultReadChunkSize);
+ std::vector<std::future<Status>> futures;
+ futures.reserve(file_length / kDefaultReadChunkSize + 1);
+ while (read_len > 0) {
+ auto promise = std::make_shared<std::promise<Status>>();
+ futures.push_back(promise->get_future());
+ input_stream->ReadAsync(content, read_len, read_offset,
+ [promise](Status status) {
promise->set_value(status); });
+ read_offset += read_len;
+ content += read_len;
+ read_len = std::min(file_length - read_offset,
kDefaultReadChunkSize);
+ }
+ for (const auto& status : CollectAll(futures)) {
+ if (!status.ok()) {
+ return status;
+ }
+ }
+ if (PAIMON_UNLIKELY(read_offset != file_length)) {
+ return Status::IOError(
+ fmt::format("Total read length {} does not match expected
length {}.", read_offset,
+ file_length));
+ }
+ return Status::OK();
+ }
+
+ private:
+ static constexpr uint64_t kDefaultReadChunkSize = 1024 * 1024;
+};
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/stream_utils_test.cpp
b/src/paimon/common/utils/stream_utils_test.cpp
new file mode 100644
index 0000000..9786a3e
--- /dev/null
+++ b/src/paimon/common/utils/stream_utils_test.cpp
@@ -0,0 +1,146 @@
+/*
+ * 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/stream_utils.h"
+
+#include <cstring>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon {
+
+class StreamUtilsTest : public ::testing::Test {
+ public:
+ void SetUp() override {
+ pool_ = GetDefaultPool();
+ dir_ = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir_);
+ file_system_ = std::make_shared<LocalFileSystem>();
+ }
+
+ void TearDown() override {
+ dir_.reset();
+ file_system_.reset();
+ pool_.reset();
+ }
+
+ std::string CreateTestFile(const std::string& content) {
+ std::string file_path = dir_->Str() + "/test_file.txt";
+ EXPECT_OK_AND_ASSIGN(auto output_stream,
file_system_->Create(file_path, true));
+ EXPECT_OK_AND_ASSIGN(int32_t length,
output_stream->Write(content.data(), content.size()));
+ EXPECT_EQ(length, static_cast<int32_t>(content.size()));
+ EXPECT_OK(output_stream->Close());
+ return file_path;
+ }
+
+ private:
+ std::shared_ptr<MemoryPool> pool_;
+ std::unique_ptr<paimon::test::UniqueTestDirectory> dir_;
+ std::shared_ptr<FileSystem> file_system_;
+};
+
+TEST_F(StreamUtilsTest, ReadBasicTest) {
+ std::string test_content = "Hello, World! This is a test file for
StreamUtils.";
+ std::string file_path = CreateTestFile(test_content);
+
+ {
+ // ReadFully
+ ASSERT_OK_AND_ASSIGN(auto input_stream, file_system_->Open(file_path));
+ ASSERT_OK(input_stream->Seek(10, FS_SEEK_SET)); // ReadFully ignore
any seek
+ ASSERT_OK_AND_ASSIGN(auto result,
StreamUtils::ReadFully(std::move(input_stream), pool_));
+ ASSERT_EQ(result->size(), test_content.size());
+ ASSERT_EQ(std::string(result->data(), result->size()), test_content);
+ }
+ {
+ // ReadAsyncFully
+ ASSERT_OK_AND_ASSIGN(auto input_stream, file_system_->Open(file_path));
+ ASSERT_OK(input_stream->Seek(10, FS_SEEK_SET)); // ReadAsyncFully
ignore any seek
+ ASSERT_OK_AND_ASSIGN(auto result,
+
StreamUtils::ReadAsyncFully(std::move(input_stream), pool_));
+ ASSERT_EQ(result->size(), test_content.size());
+ ASSERT_EQ(std::string(result->data(), result->size()), test_content);
+ }
+ {
+ // ReadAsyncFully with buffer
+ ASSERT_OK_AND_ASSIGN(auto input_stream, file_system_->Open(file_path));
+ ASSERT_OK(input_stream->Seek(10, FS_SEEK_SET)); // ReadAsyncFully
ignore any seek
+ std::vector<char> buffer(test_content.size());
+ ASSERT_OK(StreamUtils::ReadAsyncFully(std::move(input_stream),
buffer.data()));
+ ASSERT_EQ(std::string(buffer.data(), buffer.size()), test_content);
+ }
+}
+
+TEST_F(StreamUtilsTest, ReadEmptyFile) {
+ std::string file_path = CreateTestFile("");
+
+ {
+ // ReadFully
+ ASSERT_OK_AND_ASSIGN(auto input_stream, file_system_->Open(file_path));
+ ASSERT_OK_AND_ASSIGN(auto result,
StreamUtils::ReadFully(std::move(input_stream), pool_));
+ ASSERT_EQ(result->size(), 0);
+ }
+ {
+ // ReadAsyncFully
+ ASSERT_OK_AND_ASSIGN(auto input_stream, file_system_->Open(file_path));
+ ASSERT_OK_AND_ASSIGN(auto result,
+
StreamUtils::ReadAsyncFully(std::move(input_stream), pool_));
+ ASSERT_EQ(result->size(), 0);
+ }
+}
+
+TEST_F(StreamUtilsTest, ReadLargeFile) {
+ std::string large_content;
+ const size_t file_size = 3 * 1024 * 1024 + 512; // 3.5MB
+ large_content.reserve(file_size);
+ for (size_t i = 0; i < file_size; ++i) {
+ large_content += static_cast<char>('A' + (i % 26));
+ }
+ std::string file_path = CreateTestFile(large_content);
+
+ {
+ // ReadFully
+ ASSERT_OK_AND_ASSIGN(auto input_stream, file_system_->Open(file_path));
+ ASSERT_OK_AND_ASSIGN(auto result,
StreamUtils::ReadFully(std::move(input_stream), pool_));
+ ASSERT_EQ(result->size(), large_content.size());
+ ASSERT_EQ(std::string(result->data(), result->size()), large_content);
+ }
+ {
+ // ReadAsyncFully
+ ASSERT_OK_AND_ASSIGN(auto input_stream, file_system_->Open(file_path));
+ ASSERT_OK_AND_ASSIGN(auto result,
+
StreamUtils::ReadAsyncFully(std::move(input_stream), pool_));
+ ASSERT_EQ(result->size(), large_content.size());
+ ASSERT_EQ(std::string(result->data(), result->size()), large_content);
+ }
+ {
+ // ReadAsyncFully with buffer
+ ASSERT_OK_AND_ASSIGN(auto input_stream, file_system_->Open(file_path));
+ std::vector<char> buffer(file_size);
+ ASSERT_OK(StreamUtils::ReadAsyncFully(std::move(input_stream),
buffer.data()));
+ ASSERT_EQ(std::string(buffer.data(), buffer.size()), large_content);
+ }
+}
+
+} // namespace paimon
