diff --git a/src/paimon/core/table/source/data_evolution_batch_scan.cpp b/src/paimon/core/table/source/data_evolution_batch_scan.cpp index ed09b693..2f86769a 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.cpp +++ b/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& snapshot_reader, std::unique_ptr&& batch_scan, + const std::shared_ptr& table_schema, const std::shared_ptr& global_index_result, const CoreOptions& core_options, const std::shared_ptr& pool, const std::shared_ptr& 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> DataEvolutionBatchScan::EvalGlobalInd } auto partition_filter = batch_scan_->GetPartitionPredicate(); // TODO(lisizhuo.lsz): support time travel - PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr 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(index_scan.get()); - if (!index_scan_impl) { - return Status::Invalid("invalid GlobalIndexScan, cannot cast to GlobalIndexScanImpl"); + std::optional snapshot; + const std::shared_ptr& snapshot_manager = + snapshot_reader_->GetSnapshotManager(); + if (const std::optional& 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 index_scan, + GlobalIndexScanImpl::Create(table_path_, table_schema_, snapshot.value(), partition_filter, + core_options_, executor_, pool_)); + return index_scan->Scan(predicate); } } // namespace paimon diff --git a/src/paimon/core/table/source/data_evolution_batch_scan.h b/src/paimon/core/table/source/data_evolution_batch_scan.h index cfa29785..546ae101 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.h +++ b/src/paimon/core/table/source/data_evolution_batch_scan.h @@ -24,6 +24,7 @@ #include #include +#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& snapshot_reader, std::unique_ptr&& batch_scan, + const std::shared_ptr& table_schema, const std::shared_ptr& global_index_result, const CoreOptions& core_options, const std::shared_ptr& pool, const std::shared_ptr& executor); @@ -55,6 +57,7 @@ class DataEvolutionBatchScan : public AbstractTableScan { std::shared_ptr pool_; std::string table_path_; std::unique_ptr batch_scan_; + std::shared_ptr table_schema_; std::shared_ptr global_index_result_; std::shared_ptr executor_; }; diff --git a/src/paimon/core/table/source/table_scan.cpp b/src/paimon/core/table/source/table_scan.cpp index 95af2a23..e4db33fb 100644 --- a/src/paimon/core/table/source/table_scan.cpp +++ b/src/paimon/core/table/source/table_scan.cpp @@ -361,7 +361,7 @@ Result> NewDataTableScan(const std::shared_ptr( - 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()); } diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index b4ac8b83..e05dc4b2 100644 --- a/test/inte/global_index_test.cpp +++ b/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> ScanGlobalIndexAndData( const std::string& table_path, const std::shared_ptr& predicate, const std::map& options = {}, - const std::shared_ptr& index_result = nullptr) const { + const std::shared_ptr& index_result = nullptr, + const std::shared_ptr& 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(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 =