diff --git a/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp b/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp index a95fd91a9..7b62862fa 100644 --- a/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp +++ b/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp @@ -32,6 +32,7 @@ #include "arrow/array/builder_dict.h" #include "arrow/array/builder_nested.h" #include "arrow/c/bridge.h" +#include "fmt/format.h" #include "gtest/gtest.h" #include "paimon/common/reader/late_materializing_reader_builder.h" #include "paimon/common/reader/prefetch_file_batch_reader_impl.h" @@ -942,6 +943,60 @@ TEST_F(LateMaterializingFileBatchReaderTest, PayloadPreBufferRangeErrorFailsRead EXPECT_EQ(calls, 0); } +TEST_F(LateMaterializingFileBatchReaderTest, DisabledPrefetchCacheSkipsRangeQueries) { + class UnavailableRangesBuilder : public MockFormatReaderBuilder { + public: + using MockFormatReaderBuilder::MockFormatReaderBuilder; + + Result> Build( + const std::shared_ptr& stream) const override { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr reader, + MockFormatReaderBuilder::Build(stream)); + checked_cast(reader.get()) + ->SetPreBufferRangeStatus(Status::IOError("pre-buffer range unavailable")); + return reader; + } + }; + + auto data = BuildData({0, 1, 2, 3, 4, 5, 6, 7, 8, 9}); + auto mock_fs = std::make_shared(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr executor, CreateDefaultExecutor(2)); + for (WarmupLevel level : {WarmupLevel::NONE, WarmupLevel::RAW, WarmupLevel::DECODED}) { + LateMaterializingReaderBuilder builder( + std::make_unique(data, full_type_, /*batch_size=*/3), + GetArrowPool(pool_)); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr reader, + PrefetchFileBatchReaderImpl::Create( + /*data_file_path=*/"", /*data_file_size=*/0, &builder, mock_fs, + /*prefetch_max_parallel_num=*/1, /*batch_size=*/3, /*prefetch_batch_count=*/2, + /*enable_adaptive_prefetch_strategy=*/false, executor, + /*initialize_read_ranges=*/false, /*read_ahead_cache_enabled=*/false, CacheConfig(), + /*enable_io_metrics=*/false, level, pool_, GetArrowPool(pool_))); + auto predicate = PredicateBuilder::GreaterOrEqual(/*field_index=*/0, "k", FieldType::BIGINT, + Literal(4l)); + ::ArrowSchema c_schema; + ASSERT_TRUE(arrow::ExportSchema(*arrow::schema(full_fields_), &c_schema).ok()); + ASSERT_OK(reader->SetReadSchema(&c_schema, predicate, std::nullopt)); + reader->Warmup(); + + // Every range query fails, so successful probe/payload reads require skipping them all. + ASSERT_OK_AND_ASSIGN(std::shared_ptr result, + CollectStruct(reader.get())); + ASSERT_TRUE(result); + ASSERT_EQ(result->length(), 6); + auto keys = checked_pointer_cast(result->GetFieldByName("k")); + auto values = checked_pointer_cast(result->GetFieldByName("v")); + ASSERT_TRUE(keys); + ASSERT_TRUE(values); + for (int64_t i = 0; i < result->length(); ++i) { + ASSERT_EQ(keys->Value(i), i + 4); + ASSERT_EQ(values->GetString(i), fmt::format("v_{}", i + 4)); + } + reader->Close(); + } +} + // End to end below the prefetch layer: the ranges the inner LM reader reports mid-read must reach // the shared read-ahead cache and be fetched, on top of the probe ranges registered up front. TEST_F(LateMaterializingFileBatchReaderTest, PrefetchInnerRegistersPayloadPreBufferRanges) { diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp index 4e5e540e4..b3d89b2f7 100644 --- a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp +++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp @@ -322,15 +322,20 @@ Result> PrefetchFileBatchReaderImpl auto reader = std::unique_ptr(new PrefetchFileBatchReaderImpl( readers, batch_size, prefetch_queue_capacity, enable_adaptive_prefetch_strategy, executor, cache, io_metrics, warmup_level, arrow_pool)); - for (const auto& sub_reader : reader->readers_) { - // The sub-readers whose byte ranges only become known once they have read some data (the - // late-materialization payload pass) report them here instead of through PreBufferRange(), - // which is asked once per read-range generation before any read. Capturing the impl raw is - // safe: it owns readers_, and CleanUp() joins the background thread before it goes away. - sub_reader->SetPreBufferRangeCallback( - [impl = reader.get()](std::vector>&& ranges) { - impl->RegisterLatePreBufferRanges(std::move(ranges)); - }); + // LM uses callback presence to decide whether to query payload ranges. A disabled cache + // must not trigger those optional metadata queries. + if (cache) { + for (const auto& sub_reader : reader->readers_) { + // The sub-readers whose byte ranges only become known once they have read some data + // (the late-materialization payload pass) report them here instead of through + // PreBufferRange(), which is asked once per read-range generation before any read. + // Capturing the impl raw is safe: it owns readers_, and CleanUp() joins the background + // thread before it goes away. + sub_reader->SetPreBufferRangeCallback( + [impl = reader.get()](std::vector>&& ranges) { + impl->RegisterLatePreBufferRanges(std::move(ranges)); + }); + } } if (initialize_read_ranges) { // normally initialize read ranges should be false, as set read schema will refresh read diff --git a/src/paimon/common/utils/read_ahead_cache.cpp b/src/paimon/common/utils/read_ahead_cache.cpp index 68dc18034..a321c2e94 100644 --- a/src/paimon/common/utils/read_ahead_cache.cpp +++ b/src/paimon/common/utils/read_ahead_cache.cpp @@ -30,6 +30,7 @@ #include #include +#include "fmt/format.h" #include "paimon/common/memory/bytes_utils.h" #include "paimon/common/metrics/atomic_counter_pair.h" #include "paimon/common/utils/byte_range_combiner.h" @@ -69,6 +70,17 @@ void CopyRangeFromEntries(const std::vector& covering, const By } // namespace Result> ReadAheadCache::AddRanges(std::vector&& ranges) { + // Validate before coalescing can split an invalid range into many entries. Zero retains + // the existing unknown-size convention; the combiner still validates integer bounds. + if (file_size_ > 0) { + for (const auto& range : ranges) { + if (range.offset > file_size_ || range.length > file_size_ - range.offset) { + return Status::Invalid( + fmt::format("byte range (offset={}, length={}) exceeds file size {}", + range.offset, range.length, file_size_)); + } + } + } // Both an up-front registration and a mid-read one are cut at the plain size limit. PAIMON_ASSIGN_OR_RAISE( std::vector new_ranges, @@ -216,7 +228,7 @@ void ReadAheadCache::PreBuffer(uint64_t offset) { ReadAheadCache::ReadAheadCache(const std::shared_ptr& stream, const CacheConfig& config, uint64_t file_size, const std::shared_ptr& memory_pool) - : stream_(stream), config_(config), memory_pool_(memory_pool) { + : stream_(stream), config_(config), file_size_(file_size), memory_pool_(memory_pool) { // An unknown file size cannot be aligned to, and a zero limit or block size // means the block cache is turned off: leave it null in those cases. if (file_size > 0 && config_.GetBlockSize() > 0 && config_.GetBlockCacheLimit() > 0) { diff --git a/src/paimon/common/utils/read_ahead_cache.h b/src/paimon/common/utils/read_ahead_cache.h index 4eec73836..cf84ca09d 100644 --- a/src/paimon/common/utils/read_ahead_cache.h +++ b/src/paimon/common/utils/read_ahead_cache.h @@ -152,9 +152,9 @@ class PAIMON_EXPORT ReadAheadCache { /// Construct a read cache with given options /// @param stream The stream the cache fetches from. /// @param config The cache configuration. - /// @param file_size Size of the file behind `stream`, used to align the - /// block cache to the end of the file. Zero means unknown and disables the - /// block cache. + /// @param file_size Size of the file behind `stream`, used to validate registered ranges + /// and align the block cache to the end of the file. Zero means unknown, skipping file-size + /// validation and disabling the block cache. /// @param memory_pool The pool the cached buffers are allocated from. ReadAheadCache(const std::shared_ptr& stream, const CacheConfig& config, uint64_t file_size, const std::shared_ptr& memory_pool); @@ -174,6 +174,7 @@ class PAIMON_EXPORT ReadAheadCache { /// /// @param ranges The byte ranges to register. /// @return The offset of the first newly registered range, or nullopt when nothing was added. + /// Returns an error if a range exceeds the known file size or integer bounds. Result> AddRanges(std::vector&& ranges); /// Read a range previously registered through AddRanges(), copying the cached data @@ -261,6 +262,7 @@ class PAIMON_EXPORT ReadAheadCache { std::shared_ptr stream_; CacheConfig config_; + const uint64_t file_size_; // Every registered range, ordered by offset and disjoint (so a matching // region is found by binary search). An entry whose buffer is still null is // registered but not fetched yet; publishing fills buffer/future in place. diff --git a/src/paimon/common/utils/read_ahead_cache_test.cpp b/src/paimon/common/utils/read_ahead_cache_test.cpp index c9f7454a1..121db3469 100644 --- a/src/paimon/common/utils/read_ahead_cache_test.cpp +++ b/src/paimon/common/utils/read_ahead_cache_test.cpp @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -503,6 +504,57 @@ TEST(TestReadAheadCache, TestReinitAfterReset) { AssertReadMiss({20, 2}, &cache); } +TEST(TestReadAheadCache, TestAddRangesRejectsOutOfFileRanges) { + CacheConfig config = TestCacheConfig(/*range_size_limit=*/1024, + /*hole_size_limit=*/0, /*pre_buffer_limit=*/2048); + auto cache = CreateTestFileAndCache("data_file", std::string(1024, 'x'), config, {}, + /*file_size=*/1024); + constexpr uint64_t kMaxValue = std::numeric_limits::max(); + for (const ByteRange& range : std::vector{{512, 32 * 1024 * 1024}, + {1024, 1}, + {1025, 0}, + {1023, 2}, + {kMaxValue, 1}, + {1, kMaxValue}}) { + ASSERT_NOK_WITH_MSG(cache->AddRanges({range}), "exceeds file size 1024"); + } +} + +TEST(TestReadAheadCache, TestInvalidLateRangesPreserveRegisteredRanges) { + CacheConfig config = TestCacheConfig(/*range_size_limit=*/8, + /*hole_size_limit=*/0, /*pre_buffer_limit=*/32); + config.SetBlockCacheLimit(0); + auto cache = CreateTestFileAndCache("data_file", std::string(32, 'x'), config, {{0, 8}}, + /*file_size=*/32); + AssertReadEquals({0, 8}, std::string(8, 'x'), cache.get()); + + ASSERT_NOK_WITH_MSG(cache->AddRanges({{16, 8}, {31, 2}}), "exceeds file size 32"); + + AssertReadEquals({0, 8}, std::string(8, 'x'), cache.get()); + AssertReadMiss({16, 8}, cache.get()); +} + +TEST(TestReadAheadCache, TestAddRangesAcceptsFileEndBoundary) { + CacheConfig config = TestCacheConfig(/*range_size_limit=*/4, + /*hole_size_limit=*/0, /*pre_buffer_limit=*/16); + auto cache = CreateTestFileAndCache("data_file", "abcdefgh", config, {{4, 4}}, + /*file_size=*/8); + AssertReadEquals({4, 4}, "efgh", cache.get()); + ASSERT_OK_AND_ASSIGN(std::optional first_added, cache->AddRanges({{8, 0}})); + ASSERT_FALSE(first_added.has_value()); + ASSERT_OK(cache->AddRanges({{0, 8}}).status()); + AssertReadEquals({0, 8}, "abcdefgh", cache.get()); +} + +TEST(TestReadAheadCache, TestAddRangesPreservesUnknownFileSize) { + CacheConfig config = TestCacheConfig(/*range_size_limit=*/1024 * 1024, + /*hole_size_limit=*/0, /*pre_buffer_limit=*/2048); + auto cache = CreateTestFileAndCache("data_file", std::string(1024, 'x'), config, {}, + /*file_size=*/0); + // Zero retains its existing unknown-size meaning. Registration performs no I/O. + ASSERT_OK(cache->AddRanges({{512, 32 * 1024 * 1024}}).status()); +} + // Test that AddRanges() registers ranges that only become known after the cache was // initialized, so the reads that follow are served from the cache. TEST(TestReadAheadCache, TestAddRangesRegistersNewRanges) { diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index 6f3aa0bea..ca889342c 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -172,8 +172,12 @@ class ReadInteTest : public testing::Test, public ::testing::WithParamInterface< EXPECT_TRUE(schema_ids.empty() || schema_ids.size() == data_files.size()); for (uint32_t j = 0; j < data_files.size(); j++) { const auto& data_file = data_files[j]; + EXPECT_OK_AND_ASSIGN( + FileStatus file_status, + LocalFileSystem().GetFileStatus(PathUtil::JoinPath(bucket_path, data_file))); data_file_metas.push_back(std::make_shared( - data_file, /*file_size=*/1, /*row_count=*/1, /*min_key=*/BinaryRow::EmptyRow(), + data_file, /*file_size=*/file_status.GetLen(), /*row_count=*/1, + /*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/BinaryRow::EmptyRow(), /*key_stats=*/SimpleStats::EmptyStats(), /*value_stats=*/SimpleStats::EmptyStats(), /*min_sequence_number=*/0, /*max_sequence_number=*/0, /*schema_id=*/schema_ids.empty() ? 0 : schema_ids[j], diff --git a/test/inte/write_and_read_inte_with_index_test.cpp b/test/inte/write_and_read_inte_with_index_test.cpp index 99bd90e2e..8e7ba94fa 100644 --- a/test/inte/write_and_read_inte_with_index_test.cpp +++ b/test/inte/write_and_read_inte_with_index_test.cpp @@ -56,6 +56,8 @@ #include "paimon/data/timestamp.h" #include "paimon/defs.h" #include "paimon/factories/factory_creator.h" +#include "paimon/fs/file_system.h" +#include "paimon/fs/local/local_file_system.h" #include "paimon/memory/bytes.h" #include "paimon/memory/memory_pool.h" #include "paimon/metrics.h" @@ -1945,11 +1947,13 @@ TEST_P(WriteAndReadInteWithIndexTest, TestWithAlterTable) { auto create_data_file_meta = [&](const std::string& file_name, int32_t schema_id, const std::vector& embedded_bytes) { + EXPECT_OK_AND_ASSIGN(FileStatus file_status, + LocalFileSystem().GetFileStatus(path + "bucket-0/" + file_name)); auto embedded_index = std::make_shared(embedded_bytes.size(), pool_.get()); memcpy(embedded_index->data(), reinterpret_cast(embedded_bytes.data()), embedded_bytes.size()); return std::make_shared( - file_name, /*file_size=*/10, + file_name, /*file_size=*/file_status.GetLen(), /*row_count=*/10, /*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/BinaryRow::EmptyRow(), /*key_stats=*/SimpleStats::EmptyStats(), @@ -2663,8 +2667,10 @@ TEST_P(WriteAndReadInteWithIndexTest, TestBitmapPushDownWithMultiStripes) { } // each record is a stripe + ASSERT_OK_AND_ASSIGN(FileStatus file_status, + LocalFileSystem().GetFileStatus(path + "bucket-0/" + file_name)); auto data_file_meta = std::make_shared( - file_name, /*file_size=*/2637, + file_name, /*file_size=*/file_status.GetLen(), /*row_count=*/8, /*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/BinaryRow::EmptyRow(), /*key_stats=*/SimpleStats::EmptyStats(), /*value_stats=*/SimpleStats::EmptyStats(), /*min_sequence_number=*/0,