perf(read): reduce point-lookup metadata I/O

src/paimon/core/utils/snapshot_manager.cpp

@@ -141,10 +141,6 @@ Result<std::optional<int64_t>> SnapshotManager::FindEarliest(
 Result<std::optional<int64_t>> SnapshotManager::FindLatest(
     const std::string& dir, const std::string& prefix,
     const std::function<std::string(int64_t)>& path_func) const {
-    PAIMON_ASSIGN_OR_RAISE(bool is_exist, fs_->Exists(dir));
-    if (!is_exist) {
-        return std::optional<int64_t>();
-    }
     std::optional<int64_t> snapshot_id = ReadHint(LATEST, dir);
     if (snapshot_id != std::nullopt && snapshot_id.value() > 0) {
         int64_t next_snapshot = snapshot_id.value() + 1;
@@ -155,6 +151,8 @@ Result<std::optional<int64_t>> SnapshotManager::FindLatest(
             return snapshot_id;
         }
     }
+    // A valid hint needs no parent-directory probe. The listing fallback checks
+    // directory existence itself, including tables without any snapshots yet.
     return FindByListFiles([](int64_t lhs, int64_t rhs) -> int64_t { return std::max(lhs, rhs); },
                            dir, prefix);
 }

src/paimon/core/utils/snapshot_manager_test.cpp

@@ -18,8 +18,10 @@
 
 #include "paimon/core/utils/snapshot_manager.h"
 
+#include <algorithm>
 #include <filesystem>
 #include <limits>
+#include <vector>
 
 #include "gtest/gtest.h"
 #include "paimon/common/utils/path_util.h"
@@ -139,6 +141,51 @@ TEST(SnapshotManagerTest, TestPathNotExist) {
     ASSERT_EQ(snapshot, std::nullopt);
 }
 
+TEST(SnapshotManagerTest, LatestHintAvoidsDirectoryProbeAndStillFindsNewSnapshots) {
+    class TrackingFileSystem : public LocalFileSystem {
+     public:
+        Result<bool> Exists(const std::string& path) const override {
+            exists_paths.push_back(path);
+            return LocalFileSystem::Exists(path);
+        }
+        mutable std::vector<std::string> exists_paths;
+    };
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    auto fs = std::make_shared<TrackingFileSystem>();
+    SnapshotManager mgr(fs, dir->Str());
+    ASSERT_OK(fs->Mkdirs(mgr.SnapshotDirectory()));
+    ASSERT_OK(fs->WriteFile(mgr.SnapshotPath(1), "{}", true));
+    ASSERT_OK(mgr.CommitLatestHint(1));
+    fs->exists_paths.clear();
+
+    ASSERT_OK_AND_ASSIGN(std::optional<int64_t> latest, mgr.LatestSnapshotId());
+    ASSERT_EQ(latest, 1);
+    ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotDirectory()),
+              0);
+    ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotPath(2)), 1);
+
+    // A commit can publish its snapshot before updating the hint.
+    ASSERT_OK(fs->WriteFile(mgr.SnapshotPath(2), "{}", true));
+    fs->exists_paths.clear();
+    ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId());
+    ASSERT_EQ(latest, 2);
+    ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotDirectory()),
+              1);
+
+    const std::string hint_path = PathUtil::JoinPath(mgr.SnapshotDirectory(), "LATEST");
+    ASSERT_OK(fs->Delete(hint_path));
+    fs->exists_paths.clear();
+    ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId());
+    ASSERT_EQ(latest, 2);
+    ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotDirectory()),
+              1);
+
+    ASSERT_OK(fs->WriteFile(hint_path, "invalid", true));
+    ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId());
+    ASSERT_EQ(latest, 2);
+}
+
 TEST(SnapshotManagerTest, TestEarlierOrEqualTimeMillisExactMatch) {
     std::string test_data_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09";
     auto file_system = std::make_shared<LocalFileSystem>();

src/paimon/format/parquet/file_reader_wrapper.cpp

@@ -29,10 +29,12 @@
 #include "fmt/format.h"
 #include "paimon/common/utils/arrow/arrow_utils.h"
 #include "paimon/common/utils/math.h"
+#include "paimon/common/utils/scope_guard.h"
 #include "paimon/format/parquet/column_index_filter.h"
 #include "paimon/format/parquet/page_filtered_row_group_reader.h"
 #include "paimon/format/parquet/parquet_format_defs.h"
 #include "paimon/macros.h"
+#include "paimon/predicate/predicate_utils.h"
 #include "parquet/arrow/reader.h"
 #include "parquet/arrow/schema.h"
 #include "parquet/file_reader.h"
@@ -602,8 +604,35 @@ Result<RowRanges> FileReaderWrapper::CalculateFilteredRowRanges(
             return RowRanges::CreateSingle(row_count);
         }
 
+        auto page_index_reader = GetPageIndexReader();
+        if (!page_index_reader) {
+            return RowRanges::CreateSingle(row_count);
+        }
+        std::set<std::string> field_names;
+        PAIMON_RETURN_NOT_OK(PredicateUtils::GetAllNames(predicate, &field_names));
+        std::vector<int32_t> predicate_columns;
+        for (const auto& name : field_names) {
+            auto it = column_name_to_index.find(name);
+            if (it == column_name_to_index.end()) {
+                return Status::Invalid(
+                    fmt::format("column '{}' not found in column_name_to_index", name));
+            }
+            predicate_columns.push_back(it->second);
+        }
+        if (predicate_columns.empty()) {
+            return RowRanges::CreateSingle(row_count);
+        }
+
+        // Arrow otherwise reads the column-index envelope of every column, including
+        // potentially large min/max values in payload columns not used by the predicate.
+        const std::vector<int32_t> row_groups = {row_group_index};
+        ScopeGuard clear_hint([&]() { page_index_reader->WillNotNeed(row_groups); });
+        page_index_reader->WillNeed(row_groups, predicate_columns,
+                                    {/*column_index=*/true, /*offset_index=*/true});
+        // Keep this restricted reader separate: projected payload columns still need
+        // their offset indexes when the data reader is initialized later.
         return ColumnIndexFilter::CalculateRowRanges(predicate,
-                                                     GetRowGroupPageIndexReader(row_group_index),
+                                                     page_index_reader->RowGroup(row_group_index),
                                                      column_name_to_index, row_count);
     }
     PAIMON_PARQUET_CATCH_AND_RETURN_STATUS("FileReaderWrapper::CalculateFilteredRowRanges")

src/paimon/format/parquet/file_reader_wrapper_test.cpp

@@ -46,9 +46,14 @@
 #include "paimon/fs/file_system.h"
 #include "paimon/fs/local/local_file_system.h"
 #include "paimon/memory/memory_pool.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/predicate/predicate_builder.h"
 #include "paimon/record_batch.h"
 #include "paimon/testing/utils/testharness.h"
 #include "parquet/arrow/reader.h"
+#include "parquet/file_reader.h"
+#include "parquet/metadata.h"
+#include "parquet/page_index.h"
 #include "parquet/properties.h"
 
 namespace arrow {
@@ -273,6 +278,39 @@ TEST_F(FileReaderWrapperTest, NullFileReader) {
                         "file reader wrapper create failed. file reader is nullptr");
 }
 
+TEST_F(FileReaderWrapperTest, PredicateReadsOnlyItsPageIndexesAndKeepsPayloadReadable) {
+    std::string file_path = PathUtil::JoinPath(dir_->Str(), "predicate-index.parquet");
+    PrepareParquetFile(file_path, /*row_count=*/1000, /*enable_page_index=*/true);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input, fs_->Open(file_path));
+    auto tracking_input = std::make_shared<ReadTrackingInputStream>(input);
+    ASSERT_OK_AND_ASSIGN(auto reader, PrepareReaderWrapperOnStream(tracking_input));
+    auto metadata = reader->GetFileReader()->parquet_reader()->metadata();
+    auto index_ranges =
+        ::parquet::PageIndexReader::DeterminePageIndexRangesInRowGroup(*metadata->RowGroup(0), {1});
+    ASSERT_TRUE(index_ranges.column_index.has_value());
+    ASSERT_TRUE(index_ranges.offset_index.has_value());
+    int64_t bytes_before = tracking_input->GetPositionalReadBytes();
+    auto predicate = PredicateBuilder::Equal(1, "col2", FieldType::INT, Literal(25));
+    ASSERT_OK_AND_ASSIGN(auto ranges, reader->CalculateFilteredRowRanges(
+                                          0, predicate, {{"col1", 0}, {"col2", 1}, {"col3", 2}}));
+    ASSERT_EQ(10, ranges.RowCount());
+    ASSERT_EQ(index_ranges.column_index->length + index_ranges.offset_index->length,
+              tracking_input->GetPositionalReadBytes() - bytes_before);
+
+    // The predicate's read hint must not restrict subsequent payload offset-index reads.
+    ASSERT_OK(reader->PrepareForReading(
+        {TargetRowGroup(/*rg_index=*/0, /*is_partially_matched=*/true, ranges)}, {0, 1, 2}));
+    ASSERT_OK_AND_ASSIGN(auto batch, reader->Next());
+    ASSERT_TRUE(batch);
+    auto expected = PrepareArray(PrepareArrowSchema().second, /*record_batch_size=*/10,
+                                 /*offset=*/20);
+    auto expected_batch = arrow::RecordBatch::FromStructArray(expected);
+    ASSERT_TRUE(expected_batch.ok());
+    ASSERT_TRUE(batch->Equals(*expected_batch.ValueOrDie()));
+    ASSERT_OK_AND_ASSIGN(auto end, reader->Next());
+    ASSERT_FALSE(end);
+}
+
 TEST_F(FileReaderWrapperTest, Simple) {
     std::string file_path = PathUtil::JoinPath(dir_->Str(), "test.parquet");
     PrepareParquetFile(file_path, /*row_count=*/5500);

src/paimon/format/parquet/parquet_file_batch_reader_test.cpp

@@ -343,6 +343,122 @@ TEST_F(ParquetFileBatchReaderTest, TestParquetMetadataCacheBypassesWhenGetUriFai
     ASSERT_EQ(0, cache->Size());
 }
 
+TEST_F(ParquetFileBatchReaderTest, TestPageIndexBytesSurviveReaderClose) {
+    WriteArray(file_path_, struct_array_, schema_, /*write_batch_size=*/1,
+               /*enable_dictionary=*/false, /*max_row_group_length=*/3,
+               /*max_page_size=*/1);
+    auto cache = std::make_shared<paimon::test::CountingRoutingCache>(CacheKind::DATA_FILE_FOOTER,
+                                                                      128 * 1024 * 1024);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> raw_input, fs_->Open(file_path_));
+    ASSERT_OK_AND_ASSIGN(int64_t length, raw_input->Length());
+    auto raw = std::make_shared<ArrowInputStreamAdapter>(raw_input, length, pool_);
+    auto raw_reader = ::parquet::ParquetFileReader::Open(raw);
+    auto metadata = raw_reader->metadata();
+    int64_t expected_index_reads = 0;
+    std::weak_ptr<MemoryPool> cached_pool;
+    for (int32_t round = 0; round < 3; ++round) {
+        if (round == 2) {
+            cache->InvalidateAll();
+            ASSERT_TRUE(cached_pool.expired());
+        }
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input, fs_->Open(file_path_));
+        std::shared_ptr<MemoryPool> query_pool = GetMemoryPool();
+        auto stream = std::make_shared<ParquetInputStream>(input, length, pool_, query_pool, cache,
+                                                           file_path_);
+        stream->SetPageIndexRanges(*metadata);
+        for (int32_t rg = 0; rg < metadata->num_row_groups(); ++rg) {
+            auto ranges = ::parquet::PageIndexReader::DeterminePageIndexRangesInRowGroup(
+                *metadata->RowGroup(rg), {});
+            for (const auto& range : {ranges.column_index, ranges.offset_index}) {
+                ASSERT_TRUE(range.has_value());
+                auto expected = raw->ReadAt(range->offset, range->length);
+                ASSERT_TRUE(expected.ok()) << expected.status().ToString();
+                auto actual = stream->ReadAt(range->offset, range->length);
+                ASSERT_TRUE(actual.ok()) << actual.status().ToString();
+                ASSERT_TRUE(actual.ValueOrDie()->Equals(*expected.ValueOrDie()));
+                if (round != 1) {
+                    ++expected_index_reads;
+                }
+            }
+        }
+        ASSERT_EQ(expected_index_reads, cache->SupplierCallCount());
+        // Reader 2 serves all indexes from the shared cache without storage IO.
+        if (round == 1) {
+            ASSERT_EQ(0, stream->StorageReadBytes()->load());
+        } else {
+            ASSERT_GT(stream->StorageReadBytes()->load(), 0);
+        }
+        // Ordinary file bytes must still use storage, even on a cache hit round.
+        uint64_t bytes_before = stream->StorageReadBytes()->load();
+        auto magic = stream->ReadAt(0, 4);
+        ASSERT_TRUE(magic.ok()) << magic.status().ToString();
+        ASSERT_EQ("PAR1", magic.ValueOrDie()->ToString());
+        ASSERT_EQ(bytes_before + 4, stream->StorageReadBytes()->load());
+        ASSERT_EQ(expected_index_reads, cache->SupplierCallCount());
+        ASSERT_TRUE(stream->Close().ok());
+        if (round != 1) {
+            cached_pool = query_pool;
+        }
+        stream.reset();
+        query_pool.reset();
+        ASSERT_FALSE(cached_pool.expired());
+    }
+    cache->InvalidateAll();
+    ASSERT_TRUE(cached_pool.expired());
+}
+
+TEST_F(ParquetFileBatchReaderTest, TestCachedFooterKeepsAllocatorAliveUntilEviction) {
+    WriteArray(file_path_, struct_array_, schema_, /*write_batch_size=*/1,
+               /*enable_dictionary=*/false, /*max_row_group_length=*/3);
+    auto cache = std::make_shared<paimon::test::CountingRoutingCache>(CacheKind::DATA_FILE_FOOTER,
+                                                                      128 * 1024 * 1024);
+    std::weak_ptr<MemoryPool> cached_pool;
+    {
+        std::shared_ptr<MemoryPool> query_pool = GetMemoryPool();
+        cached_pool = query_pool;
+        ParquetReaderBuilder builder({}, 10);
+        builder.WithMemoryPool(query_pool)->WithCache(cache);
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input, fs_->Open(file_path_));
+        ASSERT_OK_AND_ASSIGN(auto reader, builder.Build(input));
+    }
+    ASSERT_FALSE(cached_pool.expired());
+    cache->InvalidateAll();
+    ASSERT_TRUE(cached_pool.expired());
+}
+
+TEST_F(ParquetFileBatchReaderTest, TestPointReadReusesFooterAndPageIndexes) {
+    WriteArray(file_path_, struct_array_, schema_, /*write_batch_size=*/1,
+               /*enable_dictionary=*/false, /*max_row_group_length=*/3,
+               /*max_page_size=*/1);
+    auto cache = std::make_shared<paimon::test::CountingRoutingCache>(CacheKind::DATA_FILE_FOOTER,
+                                                                      128 * 1024 * 1024);
+    auto projection = arrow::schema({schema_->GetFieldByName("f4"), schema_->GetFieldByName("f8")});
+    auto predicate = PredicateBuilder::Equal(0, "f4", FieldType::INT, Literal(300002));
+    int64_t cold_reads = 0;
+    std::shared_ptr<arrow::ChunkedArray> cold_result;
+    for (int32_t round = 0; round < 2; ++round) {
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input, fs_->Open(file_path_));
+        ParquetReaderBuilder builder({{PARQUET_READ_ENABLE_PAGE_INDEX_FILTER, "true"}}, 10);
+        builder.WithCache(cache);
+        ASSERT_OK_AND_ASSIGN(auto reader, builder.Build(input));
+        ArrowSchema c_schema;
+        ASSERT_TRUE(arrow::ExportSchema(*projection, &c_schema).ok());
+        ASSERT_OK(reader->SetReadSchema(&c_schema, predicate, std::nullopt));
+        ASSERT_OK_AND_ASSIGN(auto result,
+                             paimon::test::ReadResultCollector::CollectResult(reader.get()));
+        ASSERT_EQ(1, result->length());
+        if (round == 0) {
+            cold_reads = cache->SupplierCallCount();
+            ASSERT_GT(cold_reads, 1);  // Footer plus actual page-index ranges.
+            cold_result = result;
+        } else {
+            ASSERT_TRUE(result->Equals(cold_result));
+            ASSERT_EQ(cold_reads, cache->SupplierCallCount());
+            ASSERT_GT(cache->GetCount(), cold_reads);
+        }
+    }
+}
+
 TEST_F(ParquetFileBatchReaderTest, TestReadBinaryWrittenFromBinaryAndLargeBinary) {
     auto check_binary_read_result = [&](const std::shared_ptr<arrow::DataType>& write_type,
                                         const std::string& file_name) {