fix(read): validate byte ranges and honor disabled prefetch cache

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<std::unique_ptr<FileBatchReader>> Build(
+            const std::shared_ptr<InputStream>& stream) const override {
+            PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileBatchReader> reader,
+                                   MockFormatReaderBuilder::Build(stream));
+            checked_cast<MockFileBatchReader*>(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<MockFileSystem>();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Executor> executor, CreateDefaultExecutor(2));
+    for (WarmupLevel level : {WarmupLevel::NONE, WarmupLevel::RAW, WarmupLevel::DECODED}) {
+        LateMaterializingReaderBuilder builder(
+            std::make_unique<UnavailableRangesBuilder>(data, full_type_, /*batch_size=*/3),
+            GetArrowPool(pool_));
+        ASSERT_OK_AND_ASSIGN(
+            std::unique_ptr<PrefetchFileBatchReaderImpl> 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<arrow::StructArray> result,
+                             CollectStruct(reader.get()));
+        ASSERT_TRUE(result);
+        ASSERT_EQ(result->length(), 6);
+        auto keys = checked_pointer_cast<arrow::Int64Array>(result->GetFieldByName("k"));
+        auto values = checked_pointer_cast<arrow::StringArray>(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) {

src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp

@@ -322,15 +322,20 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
     auto reader = std::unique_ptr<PrefetchFileBatchReaderImpl>(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<std::pair<uint64_t, uint64_t>>&& 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<std::pair<uint64_t, uint64_t>>&& ranges) {
+                    impl->RegisterLatePreBufferRanges(std::move(ranges));
+                });
+        }
     }
     if (initialize_read_ranges) {
         // normally initialize read ranges should be false, as set read schema will refresh read

src/paimon/common/utils/read_ahead_cache.cpp

@@ -30,6 +30,7 @@
 #include <shared_mutex>
 #include <utility>
 
+#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<RangeCacheEntry>& covering, const By
 }  // namespace
 
 Result<std::optional<uint64_t>> ReadAheadCache::AddRanges(std::vector<ByteRange>&& 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<ByteRange> new_ranges,
@@ -216,7 +228,7 @@ void ReadAheadCache::PreBuffer(uint64_t offset) {
 ReadAheadCache::ReadAheadCache(const std::shared_ptr<InputStream>& stream,
                                const CacheConfig& config, uint64_t file_size,
                                const std::shared_ptr<MemoryPool>& 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) {

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<InputStream>& stream, const CacheConfig& config,
                    uint64_t file_size, const std::shared_ptr<MemoryPool>& 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<std::optional<uint64_t>> AddRanges(std::vector<ByteRange>&& ranges);
 
     /// Read a range previously registered through AddRanges(), copying the cached data
@@ -261,6 +262,7 @@ class PAIMON_EXPORT ReadAheadCache {
 
     std::shared_ptr<InputStream> 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.

src/paimon/common/utils/read_ahead_cache_test.cpp

@@ -21,6 +21,7 @@
 
 #include <chrono>
 #include <fstream>
+#include <limits>
 #include <memory>
 #include <optional>
 #include <thread>
@@ -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<uint64_t>::max();
+    for (const ByteRange& range : std::vector<ByteRange>{{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<uint64_t> 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) {

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<DataFileMeta>(
-                    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],

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<uint8_t>& embedded_bytes) {
+            EXPECT_OK_AND_ASSIGN(FileStatus file_status,
+                                 LocalFileSystem().GetFileStatus(path + "bucket-0/" + file_name));
             auto embedded_index = std::make_shared<Bytes>(embedded_bytes.size(), pool_.get());
             memcpy(embedded_index->data(), reinterpret_cast<const void*>(embedded_bytes.data()),
                    embedded_bytes.size());
             return std::make_shared<DataFileMeta>(
-                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<DataFileMeta>(
-            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,