perf(scan): reuse loaded global index planning context

src/paimon/core/table/source/data_evolution_batch_scan.cpp

@@ -25,19 +25,21 @@
 #include "paimon/core/global_index/global_index_scan_impl.h"
 #include "paimon/core/global_index/indexed_split_impl.h"
 #include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/core/utils/snapshot_manager.h"
 #include "paimon/global_index/bitmap_global_index_result.h"
-#include "paimon/global_index/global_index_scan.h"
 
 namespace paimon {
 DataEvolutionBatchScan::DataEvolutionBatchScan(
     const std::string& table_path, const std::shared_ptr<SnapshotReader>& snapshot_reader,
     std::unique_ptr<DataTableBatchScan>&& batch_scan,
+    const std::shared_ptr<TableSchema>& table_schema,
     const std::shared_ptr<GlobalIndexResult>& global_index_result, const CoreOptions& core_options,
     const std::shared_ptr<MemoryPool>& pool, const std::shared_ptr<Executor>& executor)
     : AbstractTableScan(core_options, snapshot_reader),
       pool_(pool),
       table_path_(table_path),
       batch_scan_(std::move(batch_scan)),
+      table_schema_(table_schema),
       global_index_result_(global_index_result),
       executor_(executor) {}
 
@@ -144,17 +146,25 @@ Result<std::shared_ptr<GlobalIndexResult>> DataEvolutionBatchScan::EvalGlobalInd
     }
     auto partition_filter = batch_scan_->GetPartitionPredicate();
     // TODO(lisizhuo.lsz): support time travel
-    PAIMON_ASSIGN_OR_RAISE(
-        std::unique_ptr<GlobalIndexScan> index_scan,
-        GlobalIndexScan::Create(table_path_, core_options_.GetScanSnapshotId(), partition_filter,
-                                core_options_.ToMap(), core_options_.GetFileSystem(), executor_,
-                                pool_));
-    auto index_scan_impl = dynamic_cast<GlobalIndexScanImpl*>(index_scan.get());
-    if (!index_scan_impl) {
-        return Status::Invalid("invalid GlobalIndexScan, cannot cast to GlobalIndexScanImpl");
+    std::optional<Snapshot> snapshot;
+    const std::shared_ptr<SnapshotManager>& snapshot_manager =
+        snapshot_reader_->GetSnapshotManager();
+    if (const std::optional<int64_t>& snapshot_id = core_options_.GetScanSnapshotId()) {
+        PAIMON_ASSIGN_OR_RAISE(Snapshot loaded_snapshot,
+                               snapshot_manager->LoadSnapshot(snapshot_id.value()));
+        snapshot = std::move(loaded_snapshot);
+    } else {
+        PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager->LatestSnapshot());
+    }
+    if (!snapshot) {
+        return Status::Invalid("not found latest snapshot");
     }
 
-    return index_scan_impl->Scan(predicate);
+    PAIMON_ASSIGN_OR_RAISE(
+        std::unique_ptr<GlobalIndexScanImpl> index_scan,
+        GlobalIndexScanImpl::Create(table_path_, table_schema_, snapshot.value(), partition_filter,
+                                    core_options_, executor_, pool_));
+    return index_scan->Scan(predicate);
 }
 
 }  // namespace paimon

src/paimon/core/table/source/data_evolution_batch_scan.h

@@ -24,6 +24,7 @@
 #include <utility>
 #include <vector>
 
+#include "paimon/core/schema/table_schema.h"
 #include "paimon/core/table/source/abstract_table_scan.h"
 #include "paimon/core/table/source/data_table_batch_scan.h"
 #include "paimon/result.h"
@@ -35,6 +36,7 @@ class DataEvolutionBatchScan : public AbstractTableScan {
     DataEvolutionBatchScan(const std::string& table_path,
                            const std::shared_ptr<SnapshotReader>& snapshot_reader,
                            std::unique_ptr<DataTableBatchScan>&& batch_scan,
+                           const std::shared_ptr<TableSchema>& table_schema,
                            const std::shared_ptr<GlobalIndexResult>& global_index_result,
                            const CoreOptions& core_options, const std::shared_ptr<MemoryPool>& pool,
                            const std::shared_ptr<Executor>& executor);
@@ -55,6 +57,7 @@ class DataEvolutionBatchScan : public AbstractTableScan {
     std::shared_ptr<MemoryPool> pool_;
     std::string table_path_;
     std::unique_ptr<DataTableBatchScan> batch_scan_;
+    std::shared_ptr<TableSchema> table_schema_;
     std::shared_ptr<GlobalIndexResult> global_index_result_;
     std::shared_ptr<Executor> executor_;
 };

src/paimon/core/table/source/table_scan.cpp

@@ -361,7 +361,7 @@ Result<std::unique_ptr<TableScan>> NewDataTableScan(const std::shared_ptr<ScanCo
     }
     if (core_options.DataEvolutionEnabled()) {
         return std::make_unique<DataEvolutionBatchScan>(
-            context->GetPath(), snapshot_reader, std::move(batch_scan),
+            context->GetPath(), snapshot_reader, std::move(batch_scan), table_schema,
             context->GetGlobalIndexResult(), core_options, context->GetMemoryPool(),
             context->GetExecutor());
     }

test/inte/global_index_test.cpp

@@ -39,6 +39,7 @@
 #include "paimon/result.h"
 #include "paimon/status.h"
 #include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/counting_cache_test_utils.h"
 #include "paimon/testing/utils/io_exception_helper.h"
 #include "paimon/testing/utils/test_helper.h"
 #include "paimon/testing/utils/testharness.h"
@@ -167,12 +168,16 @@ class GlobalIndexTest : public ::testing::Test, public ::testing::WithParamInter
     Result<std::shared_ptr<Plan>> ScanGlobalIndexAndData(
         const std::string& table_path, const std::shared_ptr<Predicate>& predicate,
         const std::map<std::string, std::string>& options = {},
-        const std::shared_ptr<GlobalIndexResult>& index_result = nullptr) const {
+        const std::shared_ptr<GlobalIndexResult>& index_result = nullptr,
+        const std::shared_ptr<Cache>& cache = nullptr) const {
         ScanContextBuilder scan_context_builder(table_path);
         scan_context_builder.SetPredicate(predicate)
             .SetOptions(options)
             .SetGlobalIndexResult(index_result)
             .WithFileSystem(fs_);
+        if (cache) {
+            scan_context_builder.WithCache(cache);
+        }
         PAIMON_ASSIGN_OR_RAISE(auto scan_context, scan_context_builder.Finish());
         PAIMON_ASSIGN_OR_RAISE(auto table_scan, TableScan::Create(std::move(scan_context)));
         PAIMON_ASSIGN_OR_RAISE(auto result_plan, table_scan->CreatePlan());
@@ -1422,6 +1427,21 @@ TEST_P(GlobalIndexTest, TestDataEvolutionBatchScan) {
     ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0", "bitmap", /*options=*/{},
                          Range(0, 7)));
 
+    {
+        auto cache = std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST, 64 * 1024 * 1024);
+        auto predicate =
+            PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING,
+                                    Literal(FieldType::STRING, "Alice", 5));
+        ASSERT_OK(ScanGlobalIndexAndData(table_path, predicate, /*options=*/{},
+                                         /*index_result=*/nullptr, cache));
+        ASSERT_GE(cache->GetCount(CacheKind::MANIFEST), 2);
+        int64_t first_supplier_calls = cache->SupplierCallCount(CacheKind::MANIFEST);
+
+        ASSERT_OK(ScanGlobalIndexAndData(table_path, predicate, /*options=*/{},
+                                         /*index_result=*/nullptr, cache));
+        ASSERT_EQ(cache->SupplierCallCount(CacheKind::MANIFEST), first_supplier_calls);
+    }
+
     // scan and read with global index
     {
         auto predicate =