diff --git a/include/paimon/scan_context.h b/include/paimon/scan_context.h index 9c0b9d470..8ff5ba99b 100644 --- a/include/paimon/scan_context.h +++ b/include/paimon/scan_context.h @@ -37,6 +37,7 @@ class ScanFilter; class Executor; class MemoryPool; class Predicate; +class SnapshotReadView; /// `ScanContext` is some configuration for table scan operations. /// @@ -56,6 +57,18 @@ class PAIMON_EXPORT ScanContext { const std::map& options, const std::shared_ptr& cache); + ScanContext(const std::string& path, bool is_streaming_mode, std::optional limit, + const std::shared_ptr& scan_filter, + const std::shared_ptr& global_index_result, + const std::shared_ptr& realtime_context, + const std::shared_ptr& memory_pool, + const std::shared_ptr& executor, + const std::shared_ptr& specific_file_system, + const std::optional& table_schema, + const std::map& options, + const std::shared_ptr& cache, + const std::shared_ptr& snapshot_read_view); + ~ScanContext(); const std::string& GetPath() const { @@ -105,6 +118,10 @@ class PAIMON_EXPORT ScanContext { return cache_; } + std::shared_ptr GetSnapshotReadView() const { + return snapshot_read_view_; + } + private: std::string path_; bool is_streaming_mode_; @@ -118,6 +135,7 @@ class PAIMON_EXPORT ScanContext { std::optional table_schema_; std::map options_; std::shared_ptr cache_; + std::shared_ptr snapshot_read_view_; }; /// Filter configuration for table scan operations @@ -220,6 +238,15 @@ class PAIMON_EXPORT ScanContextBuilder { /// @return Reference to this builder for method chaining. ScanContextBuilder& WithCache(const std::shared_ptr& cache); + /// Reuse the immutable parsed snapshot metadata from an earlier batch scan plan. + /// + /// The view must belong to the same normalized table path and branch. Snapshot read views are + /// supported for non-streaming latest-snapshot scans. + /// @param snapshot_read_view A read view returned by `Plan::GetSnapshotReadView()`. + /// @return Reference to this builder for method chaining. + ScanContextBuilder& WithSnapshotReadView( + const std::shared_ptr& snapshot_read_view); + /// Build and return a `ScanContext` instance with input validation. /// @return Result containing the constructed `ScanContext` or an error status. Result> Finish(); diff --git a/include/paimon/table/source/plan.h b/include/paimon/table/source/plan.h index 1c7174320..b5a9ebf67 100644 --- a/include/paimon/table/source/plan.h +++ b/include/paimon/table/source/plan.h @@ -23,6 +23,7 @@ #include #include +#include "paimon/table/source/snapshot_read_view.h" #include "paimon/table/source/split.h" namespace paimon { @@ -34,5 +35,13 @@ class PAIMON_EXPORT Plan { virtual const std::vector>& Splits() const = 0; /// Snapshot id of this plan, return `std::nullopt` if the table is empty. virtual std::optional SnapshotId() const = 0; + /// Immutable snapshot metadata used to build this plan. + /// + /// The returned view can be injected into a later batch scan to reuse the already resolved and + /// parsed snapshot. Only non-streaming, non-real-time latest-snapshot plans publish a reusable + /// view; other plans return `nullptr`. + virtual std::shared_ptr GetSnapshotReadView() const { + return nullptr; + } }; } // namespace paimon diff --git a/include/paimon/table/source/snapshot_read_view.h b/include/paimon/table/source/snapshot_read_view.h new file mode 100644 index 000000000..9b4e48386 --- /dev/null +++ b/include/paimon/table/source/snapshot_read_view.h @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include +#include + +#include "paimon/visibility.h" + +namespace paimon { + +class SnapshotReadViewImpl; + +/// An immutable, process-local view of the snapshot and table-schema metadata used to create a +/// scan plan. +/// +/// A view is bound to one base table path and branch. Pass a view returned by `Plan` to +/// `ScanContextBuilder::WithSnapshotReadView()` to plan another batch scan against the same +/// parsed planning metadata without resolving the latest snapshot or schema again. Views are +/// published and accepted only for non-streaming, non-real-time latest-snapshot scans. A missing +/// snapshot id represents an empty table at the time the view was created. The base path identity +/// is also used by supported system-table scans such as `$ro`, because they consume the same +/// snapshot and schema metadata. +class PAIMON_EXPORT SnapshotReadView { + public: + virtual ~SnapshotReadView() = default; + + /// Normalized base table path this view belongs to, without a system-table suffix. + virtual const std::string& TablePath() const = 0; + + /// Normalized branch this view belongs to. + virtual const std::string& Branch() const = 0; + + /// Snapshot id captured by this view, or `std::nullopt` if the table was empty. + virtual std::optional SnapshotId() const = 0; + + private: + SnapshotReadView() = default; + + friend class SnapshotReadViewImpl; +}; + +} // namespace paimon diff --git a/src/paimon/core/operation/scan_context.cpp b/src/paimon/core/operation/scan_context.cpp index 16a80731d..25ce6147e 100644 --- a/src/paimon/core/operation/scan_context.cpp +++ b/src/paimon/core/operation/scan_context.cpp @@ -39,6 +39,22 @@ ScanContext::ScanContext(const std::string& path, bool is_streaming_mode, const std::optional& table_schema, const std::map& options, const std::shared_ptr& cache) + : ScanContext(path, is_streaming_mode, limit, scan_filter, global_index_result, + realtime_context, memory_pool, executor, specific_file_system, table_schema, + options, cache, nullptr) {} + +ScanContext::ScanContext(const std::string& path, bool is_streaming_mode, + std::optional limit, + const std::shared_ptr& scan_filter, + const std::shared_ptr& global_index_result, + const std::shared_ptr& realtime_context, + const std::shared_ptr& memory_pool, + const std::shared_ptr& executor, + const std::shared_ptr& specific_file_system, + const std::optional& table_schema, + const std::map& options, + const std::shared_ptr& cache, + const std::shared_ptr& snapshot_read_view) : path_(path), is_streaming_mode_(is_streaming_mode), limit_(limit), @@ -50,7 +66,8 @@ ScanContext::ScanContext(const std::string& path, bool is_streaming_mode, specific_file_system_(specific_file_system), table_schema_(table_schema), options_(options), - cache_(cache) {} + cache_(cache), + snapshot_read_view_(snapshot_read_view) {} ScanContext::~ScanContext() = default; @@ -72,6 +89,7 @@ class ScanContextBuilder::Impl { table_schema_ = std::nullopt; options_.clear(); cache_.reset(); + snapshot_read_view_.reset(); } private: @@ -89,6 +107,7 @@ class ScanContextBuilder::Impl { std::optional table_schema_; std::map options_; std::shared_ptr cache_; + std::shared_ptr snapshot_read_view_; }; ScanContextBuilder::ScanContextBuilder(const std::string& path) @@ -173,6 +192,12 @@ ScanContextBuilder& ScanContextBuilder::WithCache(const std::shared_ptr& return *this; } +ScanContextBuilder& ScanContextBuilder::WithSnapshotReadView( + const std::shared_ptr& snapshot_read_view) { + impl_->snapshot_read_view_ = snapshot_read_view; + return *this; +} + Result> ScanContextBuilder::Finish() { PAIMON_ASSIGN_OR_RAISE(impl_->path_, PathUtil::NormalizePath(impl_->path_)); if (impl_->path_.empty()) { @@ -184,7 +209,7 @@ Result> ScanContextBuilder::Finish() { impl_->bucket_filter_), impl_->global_index_result_, impl_->realtime_context_, impl_->memory_pool_, impl_->executor_, impl_->specific_file_system_, impl_->table_schema_, impl_->options_, - impl_->cache_); + impl_->cache_, impl_->snapshot_read_view_); impl_->Reset(); return ctx; } 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 2f86769ad..791d1b2e8 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.cpp +++ b/src/paimon/core/table/source/data_evolution_batch_scan.cpp @@ -25,6 +25,7 @@ #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/table/source/snapshot_read_view_impl.h" #include "paimon/core/utils/snapshot_manager.h" #include "paimon/global_index/bitmap_global_index_result.h" @@ -44,6 +45,11 @@ DataEvolutionBatchScan::DataEvolutionBatchScan( executor_(executor) {} Result> DataEvolutionBatchScan::CreatePlan() { + std::shared_ptr snapshot_read_view = + snapshot_reader_->GetSnapshotReadView(); + if (snapshot_read_view && !snapshot_read_view->SnapshotId()) { + return PlanImpl::EmptyPlan(snapshot_read_view); + } std::optional> row_ranges; std::shared_ptr final_global_index_result = global_index_result_; if (!final_global_index_result) { @@ -52,6 +58,9 @@ Result> DataEvolutionBatchScan::CreatePlan() { final_global_index_result = index_result; PAIMON_ASSIGN_OR_RAISE(row_ranges, index_result->ToRanges()); } + // EvalGlobalIndex captures the latest snapshot before consulting its index. Refresh the + // local copy so both negative and positive results publish and reuse that exact view. + snapshot_read_view = snapshot_reader_->GetSnapshotReadView(); } else { PAIMON_ASSIGN_OR_RAISE(row_ranges, final_global_index_result->ToRanges()); } @@ -59,7 +68,12 @@ Result> DataEvolutionBatchScan::CreatePlan() { return batch_scan_->CreatePlan(); } if (row_ranges.value().empty()) { - return PlanImpl::EmptyPlan(); + if (snapshot_read_view && snapshot_read_view->SnapshotId()) { + return std::make_shared(snapshot_read_view->SnapshotId(), + std::vector>(), + snapshot_read_view); + } + return PlanImpl::EmptyPlan(snapshot_read_view); } PAIMON_ASSIGN_OR_RAISE(RowRangeIndex row_range_index, RowRangeIndex::Create(row_ranges.value())); @@ -133,7 +147,8 @@ Result> DataEvolutionBatchScan::WrapToIndexedSplits( } indexed_splits.push_back(std::make_shared(data_split, expected, scores)); } - return std::make_shared(data_plan->SnapshotId(), indexed_splits); + return std::make_shared(data_plan->SnapshotId(), indexed_splits, + data_plan->GetSnapshotReadView()); } Result> DataEvolutionBatchScan::EvalGlobalIndex() const { @@ -145,24 +160,40 @@ Result> DataEvolutionBatchScan::EvalGlobalInd return std::shared_ptr(nullptr); } auto partition_filter = batch_scan_->GetPartitionPredicate(); - // TODO(lisizhuo.lsz): support time travel - 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); + StartupMode startup_mode = core_options_.GetStartupMode(); + std::shared_ptr snapshot; + if (!(startup_mode == StartupMode::LatestFull() || startup_mode == StartupMode::Latest())) { + // Snapshot read views deliberately support latest scans only. Preserve main's non-latest + // planning path, including an explicitly selected snapshot and the caller-owned context. + // TODO(lisizhuo.lsz): support tag/timestamp time travel. + std::optional loaded_snapshot; + const std::shared_ptr& snapshot_manager = + snapshot_reader_->GetSnapshotManager(); + if (const std::optional& snapshot_id = core_options_.GetScanSnapshotId()) { + PAIMON_ASSIGN_OR_RAISE(Snapshot selected_snapshot, + snapshot_manager->LoadSnapshot(snapshot_id.value())); + loaded_snapshot = std::move(selected_snapshot); + } else { + PAIMON_ASSIGN_OR_RAISE(loaded_snapshot, snapshot_manager->LatestSnapshot()); + } + if (!loaded_snapshot) { + return Status::Invalid("not found latest snapshot"); + } + snapshot = std::make_shared(std::move(loaded_snapshot).value()); } else { - PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager->LatestSnapshot()); - } - if (!snapshot) { - return Status::Invalid("not found latest snapshot"); + // Capture the latest snapshot through the shared SnapshotReader before evaluating the + // index. The subsequent data scan then consumes the same immutable view, avoiding both a + // second latest-snapshot lookup and a row-id/snapshot race. + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr snapshot_read_view, + snapshot_reader_->CaptureLatestSnapshotReadView()); + PAIMON_ASSIGN_OR_RAISE(snapshot, SnapshotReadViewImpl::GetSnapshot(snapshot_read_view)); + if (!snapshot) { + return std::shared_ptr(nullptr); + } } - PAIMON_ASSIGN_OR_RAISE( std::unique_ptr index_scan, - GlobalIndexScanImpl::Create(table_path_, table_schema_, snapshot.value(), partition_filter, + GlobalIndexScanImpl::Create(table_path_, table_schema_, *snapshot, partition_filter, core_options_, executor_, pool_)); return index_scan->Scan(predicate); } diff --git a/src/paimon/core/table/source/data_table_batch_scan.cpp b/src/paimon/core/table/source/data_table_batch_scan.cpp index 0414045ba..06f76f57b 100644 --- a/src/paimon/core/table/source/data_table_batch_scan.cpp +++ b/src/paimon/core/table/source/data_table_batch_scan.cpp @@ -90,7 +90,7 @@ Result> DataTableBatchScan::ApplyPushDownLimit( std::dynamic_pointer_cast(scan_result); if (!current_scan_result) { // NoSnapshot - return PlanImpl::EmptyPlan(); + return snapshot_reader_->EmptyPlan(); } if (!CanPushDownLimit()) { return current_scan_result->GetPlan(); @@ -126,7 +126,9 @@ Result> DataTableBatchScan::ApplyPushDownLimit( "rows.", limited_data_splits.size(), splits.size(), push_down_limit_.value(), scanned_row_count); - return std::make_shared(snapshot_id, limited_data_splits); + return std::make_shared( + snapshot_id, limited_data_splits, + current_scan_result->GetPlan()->GetSnapshotReadView()); } } return current_scan_result->GetPlan(); diff --git a/src/paimon/core/table/source/plan_impl.cpp b/src/paimon/core/table/source/plan_impl.cpp index 1d418c531..8d3c47ea3 100644 --- a/src/paimon/core/table/source/plan_impl.cpp +++ b/src/paimon/core/table/source/plan_impl.cpp @@ -27,4 +27,13 @@ const std::shared_ptr PlanImpl::EmptyPlan() { return empty_plan; } +const std::shared_ptr PlanImpl::EmptyPlan( + const std::shared_ptr& snapshot_read_view) { + if (!snapshot_read_view) { + return EmptyPlan(); + } + return std::make_shared(std::optional(), + std::vector>(), snapshot_read_view); +} + } // namespace paimon diff --git a/src/paimon/core/table/source/plan_impl.h b/src/paimon/core/table/source/plan_impl.h index e609ccbd8..98808b4d3 100644 --- a/src/paimon/core/table/source/plan_impl.h +++ b/src/paimon/core/table/source/plan_impl.h @@ -33,7 +33,12 @@ class PlanImpl : public Plan { public: PlanImpl(const std::optional& snapshot_id, const std::vector>& splits) - : snapshot_id_(snapshot_id), splits_(splits) {} + : PlanImpl(snapshot_id, splits, nullptr) {} + + PlanImpl(const std::optional& snapshot_id, + const std::vector>& splits, + const std::shared_ptr& snapshot_read_view) + : snapshot_id_(snapshot_id), splits_(splits), snapshot_read_view_(snapshot_read_view) {} std::optional SnapshotId() const override { return snapshot_id_; @@ -43,10 +48,18 @@ class PlanImpl : public Plan { return splits_; } + std::shared_ptr GetSnapshotReadView() const override { + return snapshot_read_view_; + } + static const std::shared_ptr EmptyPlan(); + static const std::shared_ptr EmptyPlan( + const std::shared_ptr& snapshot_read_view); + private: std::optional snapshot_id_; std::vector> splits_; + std::shared_ptr snapshot_read_view_; }; } // namespace paimon diff --git a/src/paimon/core/table/source/primary_key_index_batch_scan.cpp b/src/paimon/core/table/source/primary_key_index_batch_scan.cpp index 7e3872d23..6d084ba83 100644 --- a/src/paimon/core/table/source/primary_key_index_batch_scan.cpp +++ b/src/paimon/core/table/source/primary_key_index_batch_scan.cpp @@ -31,6 +31,7 @@ #include "paimon/core/table/source/primary_key_sorted_index_result.h" #include "paimon/core/table/source/primary_key_sorted_index_scan.h" #include "paimon/core/table/source/snapshot/snapshot_reader.h" +#include "paimon/core/table/source/snapshot_read_view_impl.h" #include "paimon/core/utils/index_file_path_factories.h" #include "paimon/core/utils/snapshot_manager.h" #include "paimon/logging.h" @@ -86,13 +87,23 @@ Result> PrimaryKeyIndexBatchScan::CreatePlan() { int64_t snapshot_id = data_plan->SnapshotId().value(); const std::shared_ptr& snapshot_manager = snapshot_reader_->GetSnapshotManager(); - Result snapshot_result = snapshot_manager->LoadSnapshot(snapshot_id); - if (!snapshot_result.ok()) { - static auto logger = Logger::GetLogger("PrimaryKeyIndexBatchScan"); - PAIMON_LOG_WARN(logger, - "Failed to load snapshot %ld for primary-key sorted-index planning; " - "falling back to the unindexed data plan: %s", - snapshot_id, snapshot_result.status().ToString().c_str()); + std::shared_ptr snapshot; + if (data_plan->GetSnapshotReadView()) { + PAIMON_ASSIGN_OR_RAISE(snapshot, + SnapshotReadViewImpl::GetSnapshot(data_plan->GetSnapshotReadView())); + } else { + Result snapshot_result = snapshot_manager->LoadSnapshot(snapshot_id); + if (!snapshot_result.ok()) { + static auto logger = Logger::GetLogger("PrimaryKeyIndexBatchScan"); + PAIMON_LOG_WARN(logger, + "Failed to load snapshot %ld for primary-key sorted-index planning; " + "falling back to the unindexed data plan: %s", + snapshot_id, snapshot_result.status().ToString().c_str()); + return data_plan; + } + snapshot = std::make_shared(std::move(snapshot_result).value()); + } + if (!snapshot) { return data_plan; } @@ -111,7 +122,7 @@ Result> PrimaryKeyIndexBatchScan::CreatePlan() { indexed_field_ids.count(meta.value().index_field_id) > 0; }; PAIMON_ASSIGN_OR_RAISE(std::vector index_entries, - index_file_handler->Scan(snapshot_result.value(), entry_filter)); + index_file_handler->Scan(*snapshot, entry_filter)); PAIMON_ASSIGN_OR_RAISE(PrimaryKeySortedIndexScan::Plan index_plan, PrimaryKeySortedIndexScan::CreatePlan( @@ -132,7 +143,8 @@ Result> PrimaryKeyIndexBatchScan::CreatePlan() { scalar_definitions_, reader_factory)); PAIMON_ASSIGN_OR_RAISE(std::vector> splits, PrimaryKeySortedIndexResult::ToSplits(evaluated_plan)); - return std::make_shared(data_plan->SnapshotId(), splits); + return std::make_shared(data_plan->SnapshotId(), splits, + data_plan->GetSnapshotReadView()); } } // namespace paimon diff --git a/src/paimon/core/table/source/snapshot/full_starting_scanner.h b/src/paimon/core/table/source/snapshot/full_starting_scanner.h index 28a4872b5..871d38c64 100644 --- a/src/paimon/core/table/source/snapshot/full_starting_scanner.h +++ b/src/paimon/core/table/source/snapshot/full_starting_scanner.h @@ -32,6 +32,14 @@ class FullStartingScanner : public StartingScanner { Result> Scan( const std::shared_ptr& snapshot_reader) override { + PAIMON_ASSIGN_OR_RAISE(std::optional> read_view_plan, + snapshot_reader->ReadFromSnapshotReadView()); + if (read_view_plan) { + if (!read_view_plan.value()->SnapshotId()) { + return std::make_shared(); + } + return std::make_shared(read_view_plan.value()); + } if (starting_snapshot_id_ == std::nullopt) { // try to get first snapshot PAIMON_ASSIGN_OR_RAISE(starting_snapshot_id_, snapshot_manager_->LatestSnapshotId()); diff --git a/src/paimon/core/table/source/snapshot/snapshot_reader.cpp b/src/paimon/core/table/source/snapshot/snapshot_reader.cpp index 0e98b5cf7..89cd1dbf9 100644 --- a/src/paimon/core/table/source/snapshot/snapshot_reader.cpp +++ b/src/paimon/core/table/source/snapshot/snapshot_reader.cpp @@ -36,9 +36,54 @@ #include "paimon/core/snapshot.h" #include "paimon/core/table/source/data_split_impl.h" #include "paimon/core/table/source/plan_impl.h" +#include "paimon/core/table/source/snapshot_read_view_impl.h" #include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" namespace paimon { +Result> SnapshotReader::CaptureLatestSnapshotReadView() { + if (snapshot_read_view_) { + return snapshot_read_view_; + } + if (!table_schema_) { + return Status::Invalid("snapshot reader is not configured to publish a read view"); + } + const std::shared_ptr& snapshot_manager = GetSnapshotManager(); + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, snapshot_manager->LatestSnapshot()); + snapshot_read_view_ = + SnapshotReadViewImpl::Create(snapshot_manager->RootPath(), snapshot_manager->Branch(), + std::move(snapshot), table_schema_); + return snapshot_read_view_; +} + +Result>> SnapshotReader::ReadFromSnapshotReadView() { + if (!snapshot_read_view_) { + return std::optional>(); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr snapshot, + SnapshotReadViewImpl::GetSnapshot(snapshot_read_view_)); + if (!snapshot) { + return std::optional>(EmptyPlan()); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr plan, + WithMode(ScanMode::ALL)->WithSnapshot(*snapshot)->Read()); + return std::optional>(std::move(plan)); +} + +std::shared_ptr SnapshotReader::CreateSnapshotReadView( + std::optional snapshot) const { + if (snapshot_read_view_ || !table_schema_) { + return snapshot_read_view_; + } + const std::shared_ptr& snapshot_manager = GetSnapshotManager(); + return SnapshotReadViewImpl::Create(snapshot_manager->RootPath(), snapshot_manager->Branch(), + std::move(snapshot), table_schema_); +} + +std::shared_ptr SnapshotReader::EmptyPlan() const { + return PlanImpl::EmptyPlan(CreateSnapshotReadView(std::nullopt)); +} + Result> SnapshotReader::Read() const { PAIMON_ASSIGN_OR_RAISE(std::shared_ptr raw_plan, scan_->CreatePlan()); const std::optional& snapshot = raw_plan->GetSnapshot(); @@ -47,7 +92,8 @@ Result> SnapshotReader::Read() const { PAIMON_ASSIGN_OR_RAISE( std::vector> data_splits, GenerateSplits(snapshot, scan_mode_ != ScanMode::ALL, split_generator_, std::move(files))); - return std::make_shared(raw_plan->SnapshotId(), data_splits); + return std::make_shared(raw_plan->SnapshotId(), data_splits, + CreateSnapshotReadView(snapshot)); } Result>> SnapshotReader::GenerateSplits( diff --git a/src/paimon/core/table/source/snapshot/snapshot_reader.h b/src/paimon/core/table/source/snapshot/snapshot_reader.h index 35134d22a..96cb97cad 100644 --- a/src/paimon/core/table/source/snapshot/snapshot_reader.h +++ b/src/paimon/core/table/source/snapshot/snapshot_reader.h @@ -44,6 +44,8 @@ class FileStorePathFactory; class IndexFileMeta; class Snapshot; class SnapshotManager; +class SnapshotReadView; +class TableSchema; struct DataFileMeta; class SnapshotReader { @@ -51,11 +53,15 @@ class SnapshotReader { SnapshotReader(const std::shared_ptr& scan, const std::shared_ptr& path_factory, std::unique_ptr&& split_generator, - std::unique_ptr&& index_file_handler) + std::unique_ptr&& index_file_handler, + const std::shared_ptr& table_schema, + const std::shared_ptr& snapshot_read_view) : scan_(scan), path_factory_(path_factory), split_generator_(std::move(split_generator)), - index_file_handler_(std::move(index_file_handler)) {} + index_file_handler_(std::move(index_file_handler)), + table_schema_(table_schema), + snapshot_read_view_(snapshot_read_view) {} SnapshotReader* WithMode(const ScanMode& scan_mode) { scan_mode_ = scan_mode; @@ -96,6 +102,14 @@ class SnapshotReader { return index_file_handler_; } + const std::shared_ptr& GetSnapshotReadView() const { + return snapshot_read_view_; + } + + /// Resolve and retain the latest snapshot exactly once for callers that need to perform + /// auxiliary planning before the data manifest scan (for example, a global-index lookup). + Result> CaptureLatestSnapshotReadView(); + std::shared_ptr GetNonPartitionPredicate() const { return scan_->GetNonPartitionPredicate(); } @@ -115,7 +129,16 @@ class SnapshotReader { /// Get splits from `FileKind::ADD` files. Result> Read() const; + /// Read the injected snapshot view, if one was provided. + Result>> ReadFromSnapshotReadView(); + + /// Create an empty plan carrying the injected or newly captured empty snapshot view. + std::shared_ptr EmptyPlan() const; + private: + std::shared_ptr CreateSnapshotReadView( + std::optional snapshot) const; + Result>> GenerateSplits( const std::optional& snapshot, bool is_streaming, const std::unique_ptr& split_generator, @@ -131,6 +154,8 @@ class SnapshotReader { std::shared_ptr path_factory_; std::unique_ptr split_generator_; std::unique_ptr index_file_handler_; + std::shared_ptr table_schema_; ScanMode scan_mode_ = ScanMode::ALL; + std::shared_ptr snapshot_read_view_; }; } // namespace paimon diff --git a/src/paimon/core/table/source/snapshot/snapshot_reader_test.cpp b/src/paimon/core/table/source/snapshot/snapshot_reader_test.cpp index 427d68fe2..7fdbdd769 100644 --- a/src/paimon/core/table/source/snapshot/snapshot_reader_test.cpp +++ b/src/paimon/core/table/source/snapshot/snapshot_reader_test.cpp @@ -92,7 +92,9 @@ TEST_F(SnapshotReaderTest, GetDeletionFilesOverwritesDuplicateDataFileName) { ASSERT_OK_AND_ASSIGN(std::unique_ptr index_file_handler, CreateIndexFileHandler()); SnapshotReader snapshot_reader(/*scan=*/nullptr, /*path_factory=*/nullptr, - /*split_generator=*/nullptr, std::move(index_file_handler)); + /*split_generator=*/nullptr, std::move(index_file_handler), + /*table_schema=*/nullptr, + /*snapshot_read_view=*/nullptr); const std::string data_file_name = "data-0.orc"; std::vector> data_files = {CreateDataFileMeta(data_file_name)}; diff --git a/src/paimon/core/table/source/snapshot_read_view_impl.h b/src/paimon/core/table/source/snapshot_read_view_impl.h new file mode 100644 index 000000000..ff376fa3b --- /dev/null +++ b/src/paimon/core/table/source/snapshot_read_view_impl.h @@ -0,0 +1,117 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include +#include +#include + +#include "fmt/format.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/snapshot.h" +#include "paimon/result.h" +#include "paimon/status.h" +#include "paimon/table/source/snapshot_read_view.h" + +namespace paimon { + +class SnapshotReadViewImpl final : public SnapshotReadView { + public: + static std::shared_ptr Create( + const std::string& table_path, const std::string& branch, std::optional snapshot, + std::shared_ptr table_schema) { + std::shared_ptr immutable_snapshot; + if (snapshot) { + immutable_snapshot = std::make_shared(std::move(snapshot).value()); + } + return std::shared_ptr(new SnapshotReadViewImpl( + table_path, branch, std::move(immutable_snapshot), std::move(table_schema))); + } + + static Status ValidateBinding(const std::shared_ptr& read_view, + const std::string& table_path, const std::string& branch) { + if (!read_view) { + return Status::Invalid("snapshot read view is null"); + } + if (read_view->TablePath() != table_path) { + return Status::Invalid( + fmt::format("snapshot read view is bound to table '{}', not '{}'", + read_view->TablePath(), table_path)); + } + if (read_view->Branch() != branch) { + return Status::Invalid( + fmt::format("snapshot read view is bound to branch '{}', not '{}'", + read_view->Branch(), branch)); + } + return Status::OK(); + } + + static Result> GetSnapshot( + const std::shared_ptr& read_view) { + auto impl = std::dynamic_pointer_cast(read_view); + if (!impl) { + return Status::Invalid("unsupported snapshot read view implementation"); + } + return impl->snapshot_; + } + + static Result> GetTableSchema( + const std::shared_ptr& read_view) { + auto impl = std::dynamic_pointer_cast(read_view); + if (!impl) { + return Status::Invalid("unsupported snapshot read view implementation"); + } + if (!impl->table_schema_) { + return Status::Invalid("snapshot read view has no table schema"); + } + return impl->table_schema_; + } + + const std::string& TablePath() const override { + return table_path_; + } + + const std::string& Branch() const override { + return branch_; + } + + std::optional SnapshotId() const override { + return snapshot_ ? std::optional(snapshot_->Id()) : std::nullopt; + } + + private: + SnapshotReadViewImpl(const std::string& table_path, const std::string& branch, + std::shared_ptr snapshot, + std::shared_ptr table_schema) + : table_path_(table_path), + branch_(branch), + snapshot_(std::move(snapshot)), + table_schema_(std::move(table_schema)) {} + + std::string table_path_; + std::string branch_; + std::shared_ptr snapshot_; + // TableSchema exposes no mutators. Keeping the parsed instance alongside the snapshot lets a + // reused view rebuild scan objects without listing and rereading the schema directory. + std::shared_ptr table_schema_; +}; + +} // namespace paimon diff --git a/src/paimon/core/table/source/table_read.cpp b/src/paimon/core/table/source/table_read.cpp index be7c6ee05..b3c25d187 100644 --- a/src/paimon/core/table/source/table_read.cpp +++ b/src/paimon/core/table/source/table_read.cpp @@ -161,10 +161,18 @@ Result> TableRead::Create(std::unique_ptr system_table_path, SystemTableLoader::TryParsePath(context->GetPath())); if (system_table_path) { + std::shared_ptr system_table_schema; + std::string branch = BranchManager::NormalizeBranch( + system_table_path->branch.value_or(tmp_core_options.GetBranch())); + if (branch == BranchManager::DEFAULT_MAIN_BRANCH && context->GetSpecificTableSchema()) { + PAIMON_ASSIGN_OR_RAISE( + system_table_schema, + TableSchema::CreateFromJson(context->GetSpecificTableSchema().value())); + } PAIMON_ASSIGN_OR_RAISE( std::shared_ptr system_table, SystemTableLoader::LoadFromPath(tmp_core_options.GetFileSystem(), context->GetPath(), - context->GetOptions())); + context->GetOptions(), system_table_schema)); return system_table->NewRead(context); } diff --git a/src/paimon/core/table/source/table_scan.cpp b/src/paimon/core/table/source/table_scan.cpp index e4db33fb4..f42bf60ba 100644 --- a/src/paimon/core/table/source/table_scan.cpp +++ b/src/paimon/core/table/source/table_scan.cpp @@ -32,6 +32,7 @@ #include "paimon/common/types/data_field.h" #include "paimon/common/utils/fields_comparator.h" #include "paimon/common/utils/options_utils.h" +#include "paimon/common/utils/string_utils.h" #include "paimon/core/core_options.h" #include "paimon/core/index/index_file_handler.h" #include "paimon/core/index/pk/primary_key_index_definitions.h" @@ -58,7 +59,9 @@ #include "paimon/core/table/source/read_optimized_scan_options.h" #include "paimon/core/table/source/realtime_table_scan.h" #include "paimon/core/table/source/snapshot/snapshot_reader.h" +#include "paimon/core/table/source/snapshot_read_view_impl.h" #include "paimon/core/table/source/split_generator.h" +#include "paimon/core/table/system/read_optimized_system_table.h" #include "paimon/core/table/system/system_table.h" #include "paimon/core/utils/branch_manager.h" #include "paimon/core/utils/field_mapping.h" @@ -212,10 +215,37 @@ Result> TableScan::Create(std::unique_ptr system_table_path, SystemTableLoader::TryParsePath(shared_context->GetPath())); if (system_table_path) { + if (shared_context->GetSnapshotReadView() && shared_context->GetRealtimeContext()) { + return Status::Invalid("snapshot read view does not support real-time union scan"); + } + std::shared_ptr system_table_schema; + if (shared_context->GetSnapshotReadView()) { + if (system_table_path->is_global || + StringUtils::ToLowerCase(system_table_path->system_table_name) != + ReadOptimizedSystemTable::kName) { + return Status::Invalid( + "snapshot read view can only be used with the read-optimized system table"); + } + std::string branch = BranchManager::NormalizeBranch( + system_table_path->branch.value_or(tmp_options.GetBranch())); + PAIMON_RETURN_NOT_OK(SnapshotReadViewImpl::ValidateBinding( + shared_context->GetSnapshotReadView(), system_table_path->table_path, branch)); + PAIMON_ASSIGN_OR_RAISE(system_table_schema, SnapshotReadViewImpl::GetTableSchema( + shared_context->GetSnapshotReadView())); + } else { + std::string branch = BranchManager::NormalizeBranch( + system_table_path->branch.value_or(tmp_options.GetBranch())); + if (branch == BranchManager::DEFAULT_MAIN_BRANCH && + shared_context->GetSpecificTableSchema()) { + PAIMON_ASSIGN_OR_RAISE( + system_table_schema, + TableSchema::CreateFromJson(shared_context->GetSpecificTableSchema().value())); + } + } PAIMON_ASSIGN_OR_RAISE( std::shared_ptr system_table, SystemTableLoader::LoadFromPath(tmp_options.GetFileSystem(), shared_context->GetPath(), - shared_context->GetOptions())); + shared_context->GetOptions(), system_table_schema)); return system_table->NewScan(shared_context); } return NewDataTableScan(shared_context); @@ -266,9 +296,23 @@ Result> NewDataTableScan(const std::shared_ptrGetOptions(), context->GetSpecificFileSystem(), {})); std::string branch = BranchManager::NormalizeBranch(tmp_options.GetBranch()); + std::shared_ptr snapshot_read_view = context->GetSnapshotReadView(); + if (snapshot_read_view) { + PAIMON_RETURN_NOT_OK( + SnapshotReadViewImpl::ValidateBinding(snapshot_read_view, context->GetPath(), branch)); + if (context->IsStreamingMode()) { + return Status::Invalid("snapshot read view does not support streaming scan"); + } + if (context->GetRealtimeContext()) { + return Status::Invalid("snapshot read view does not support real-time union scan"); + } + } std::shared_ptr table_schema; - const auto& specific_table_schema = context->GetSpecificTableSchema(); - if (branch == BranchManager::DEFAULT_MAIN_BRANCH && specific_table_schema) { + if (snapshot_read_view) { + PAIMON_ASSIGN_OR_RAISE(table_schema, + SnapshotReadViewImpl::GetTableSchema(snapshot_read_view)); + } else if (const auto& specific_table_schema = context->GetSpecificTableSchema(); + branch == BranchManager::DEFAULT_MAIN_BRANCH && specific_table_schema) { PAIMON_ASSIGN_OR_RAISE(table_schema, TableSchema::CreateFromJson(specific_table_schema.value())); } else { @@ -292,6 +336,18 @@ Result> NewDataTableScan(const std::shared_ptrGetSpecificFileSystem(), {})); core_options.WithCache(context->GetCache()); + if (snapshot_read_view) { + StartupMode startup_mode = core_options.GetStartupMode(); + if (!(startup_mode == StartupMode::LatestFull() || startup_mode == StartupMode::Latest())) { + return Status::Invalid("snapshot read view requires a latest-snapshot startup mode"); + } + } + + StartupMode startup_mode = core_options.GetStartupMode(); + bool publish_snapshot_read_view = + !context->IsStreamingMode() && !context->GetRealtimeContext() && + (startup_mode == StartupMode::LatestFull() || startup_mode == StartupMode::Latest()); + PAIMON_RETURN_NOT_OK( ValidateRealtimeScan(*table_schema, core_options, *context, read_optimized)); // validate options @@ -340,7 +396,8 @@ Result> NewDataTableScan(const std::shared_ptrGetMemoryPool())); auto snapshot_reader = std::make_shared( - file_store_scan, path_factory, std::move(split_generator), std::move(index_file_handler)); + file_store_scan, path_factory, std::move(split_generator), std::move(index_file_handler), + publish_snapshot_read_view ? table_schema : nullptr, snapshot_read_view); const bool pk_table = !table_schema->PrimaryKeys().empty(); if (read_optimized && pk_table && context->IsStreamingMode()) { return Status::NotImplemented( diff --git a/src/paimon/core/table/source/table_scan_test.cpp b/src/paimon/core/table/source/table_scan_test.cpp index dbc5b9926..331da3cb5 100644 --- a/src/paimon/core/table/source/table_scan_test.cpp +++ b/src/paimon/core/table/source/table_scan_test.cpp @@ -25,10 +25,20 @@ #include #include "gtest/gtest.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/core/schema/schema_manager.h" #include "paimon/defs.h" +#include "paimon/fs/local/local_file_system.h" +#include "paimon/global_index/bitmap_global_index_result.h" #include "paimon/metrics.h" +#include "paimon/predicate/literal.h" +#include "paimon/predicate/predicate_builder.h" +#include "paimon/read_context.h" +#include "paimon/realtime/realtime_context.h" #include "paimon/scan_context.h" #include "paimon/status.h" +#include "paimon/table/source/plan.h" +#include "paimon/table/source/table_read.h" #include "paimon/testing/utils/testharness.h" namespace paimon::test { @@ -41,6 +51,112 @@ class DefaultMetricsTableScan : public TableScan { } }; +class TableMetadataTrackingFileSystem : public FileSystem { + public: + explicit TableMetadataTrackingFileSystem(const std::string& table_path) + : snapshot_directory_(PathUtil::JoinPath(table_path, "snapshot")), + schema_directory_(PathUtil::JoinPath(table_path, "schema")) {} + + Result> Open(const std::string& path) const override { + PAIMON_RETURN_NOT_OK(CheckTableMetadataAccess(path)); + return local_.Open(path); + } + + Result> Create(const std::string& path, + bool overwrite) const override { + return local_.Create(path, overwrite); + } + + Status Mkdirs(const std::string& path) const override { + return local_.Mkdirs(path); + } + + Status Rename(const std::string& src, const std::string& dst) const override { + return local_.Rename(src, dst); + } + + Status Delete(const std::string& path, bool recursive = true) const override { + return local_.Delete(path, recursive); + } + + Result GetFileStatus(const std::string& path) const override { + PAIMON_RETURN_NOT_OK(CheckTableMetadataAccess(path)); + return local_.GetFileStatus(path); + } + + Status ListDir(const std::string& directory, + std::vector* file_status_list) const override { + PAIMON_RETURN_NOT_OK(CheckTableMetadataAccess(directory)); + return local_.ListDir(directory, file_status_list); + } + + Status ListFileStatus(const std::string& path, + std::vector* file_status_list) const override { + PAIMON_RETURN_NOT_OK(CheckTableMetadataAccess(path)); + return local_.ListFileStatus(path, file_status_list); + } + + Result Exists(const std::string& path) const override { + PAIMON_RETURN_NOT_OK(CheckTableMetadataAccess(path)); + return local_.Exists(path); + } + + Status ReadFile(const std::string& path, std::string* content) override { + PAIMON_RETURN_NOT_OK(CheckTableMetadataAccess(path)); + return local_.ReadFile(path, content); + } + + void BlockAndResetTableMetadataAccess() { + snapshot_metadata_access_count_ = 0; + schema_metadata_access_count_ = 0; + block_table_metadata_ = true; + } + + int32_t SnapshotMetadataAccessCount() const { + return snapshot_metadata_access_count_; + } + + int32_t SchemaMetadataAccessCount() const { + return schema_metadata_access_count_; + } + + private: + bool IsSnapshotMetadata(const std::string& path) const { + return path == snapshot_directory_ || + (path.size() > snapshot_directory_.size() && + path.compare(0, snapshot_directory_.size(), snapshot_directory_) == 0 && + path[snapshot_directory_.size()] == '/'); + } + + bool IsSchemaMetadata(const std::string& path) const { + return path == schema_directory_ || + (path.size() > schema_directory_.size() && + path.compare(0, schema_directory_.size(), schema_directory_) == 0 && + path[schema_directory_.size()] == '/'); + } + + Status CheckTableMetadataAccess(const std::string& path) const { + if (IsSnapshotMetadata(path)) { + ++snapshot_metadata_access_count_; + } else if (IsSchemaMetadata(path)) { + ++schema_metadata_access_count_; + } else { + return Status::OK(); + } + if (block_table_metadata_) { + return Status::IOError("table metadata access is blocked by test: ", path); + } + return Status::OK(); + } + + LocalFileSystem local_; + std::string snapshot_directory_; + std::string schema_directory_; + mutable int32_t snapshot_metadata_access_count_ = 0; + mutable int32_t schema_metadata_access_count_ = 0; + bool block_table_metadata_ = false; +}; + } // namespace TEST(TableScanTest, TestDefaultMetricsSnapshot) { @@ -59,13 +175,406 @@ TEST(TableScanTest, TestDefaultMetricsSnapshot) { TEST(TableScanTest, TestNoSnapshot) { std::string path = paimon::test::GetDataDir() + "/orc/append_table_with_nested_type.db/append_table_with_nested_type/"; + ASSERT_OK_AND_ASSIGN(std::string normalized_path, PathUtil::NormalizePath(path)); + auto file_system = std::make_shared(normalized_path); ScanContextBuilder builder(path); builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.WithFileSystem(file_system); ASSERT_OK_AND_ASSIGN(auto context, builder.Finish()); ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(context))); ASSERT_OK_AND_ASSIGN(auto plan, table_scan->CreatePlan()); ASSERT_FALSE(plan->SnapshotId()); ASSERT_TRUE(plan->Splits().empty()); + ASSERT_TRUE(plan->GetSnapshotReadView()); + ASSERT_FALSE(plan->GetSnapshotReadView()->SnapshotId()); + ASSERT_GT(file_system->SnapshotMetadataAccessCount(), 0); + ASSERT_GT(file_system->SchemaMetadataAccessCount(), 0); + + file_system->BlockAndResetTableMetadataAccess(); + ScanContextBuilder reused_builder(path); + reused_builder.AddOption(Options::FILE_FORMAT, "orc"); + reused_builder.WithFileSystem(file_system); + reused_builder.WithSnapshotReadView(plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(auto reused_context, reused_builder.Finish()); + ASSERT_OK_AND_ASSIGN(auto reused_scan, TableScan::Create(std::move(reused_context))); + ASSERT_OK_AND_ASSIGN(auto reused_plan, reused_scan->CreatePlan()); + ASSERT_FALSE(reused_plan->SnapshotId()); + ASSERT_TRUE(reused_plan->Splits().empty()); + ASSERT_EQ(plan->GetSnapshotReadView(), reused_plan->GetSnapshotReadView()); + ASSERT_EQ(0, file_system->SnapshotMetadataAccessCount()); + ASSERT_EQ(0, file_system->SchemaMetadataAccessCount()); +} + +TEST(TableScanTest, TestReuseSnapshotReadViewSkipsSnapshotMetadata) { + std::string path = + paimon::test::GetDataDir() + "/orc/pk_table_with_alter_table.db/pk_table_with_alter_table/"; + ASSERT_OK_AND_ASSIGN(std::string normalized_path, PathUtil::NormalizePath(path)); + auto file_system = std::make_shared(normalized_path); + ScanContextBuilder builder(path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.WithFileSystem(file_system); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, table_scan->CreatePlan()); + ASSERT_TRUE(plan->SnapshotId()); + ASSERT_TRUE(plan->GetSnapshotReadView()); + ASSERT_EQ(plan->SnapshotId(), plan->GetSnapshotReadView()->SnapshotId()); + ASSERT_GT(file_system->SnapshotMetadataAccessCount(), 0); + ASSERT_GT(file_system->SchemaMetadataAccessCount(), 0); + + file_system->BlockAndResetTableMetadataAccess(); + ScanContextBuilder reused_builder(path); + reused_builder.AddOption(Options::FILE_FORMAT, "orc"); + reused_builder.WithFileSystem(file_system); + reused_builder.WithSnapshotReadView(plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reused_context, reused_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reused_scan, + TableScan::Create(std::move(reused_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr reused_plan, reused_scan->CreatePlan()); + ASSERT_EQ(plan->SnapshotId(), reused_plan->SnapshotId()); + ASSERT_EQ(plan->Splits().size(), reused_plan->Splits().size()); + ASSERT_EQ(plan->GetSnapshotReadView(), reused_plan->GetSnapshotReadView()); + ASSERT_EQ(0, file_system->SnapshotMetadataAccessCount()); + ASSERT_EQ(0, file_system->SchemaMetadataAccessCount()); +} + +TEST(TableScanTest, TestDataEvolutionGlobalIndexSnapshotReadViewSkipsTableMetadata) { + std::string path = + paimon::test::GetDataDir() + "/orc/append_with_global_index.db/append_with_global_index"; + ASSERT_OK_AND_ASSIGN(std::string normalized_path, PathUtil::NormalizePath(path)); + auto file_system = std::make_shared(normalized_path); + std::shared_ptr predicate = + PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, + Literal(FieldType::STRING, "Alice", 5)); + ScanContextBuilder builder(path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.AddOption("bitmap-global-index.legacy-format.enabled-for-testing", "true"); + builder.SetPredicate(predicate); + builder.WithFileSystem(file_system); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, table_scan->CreatePlan()); + ASSERT_TRUE(plan->SnapshotId()); + ASSERT_TRUE(plan->GetSnapshotReadView()); + ASSERT_FALSE(plan->Splits().empty()); + ASSERT_GT(file_system->SnapshotMetadataAccessCount(), 0); + ASSERT_GT(file_system->SchemaMetadataAccessCount(), 0); + + file_system->BlockAndResetTableMetadataAccess(); + ScanContextBuilder reused_builder(path); + reused_builder.AddOption(Options::FILE_FORMAT, "orc"); + reused_builder.AddOption("bitmap-global-index.legacy-format.enabled-for-testing", "true"); + reused_builder.SetPredicate(predicate); + reused_builder.WithFileSystem(file_system); + reused_builder.WithSnapshotReadView(plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reused_context, reused_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reused_scan, + TableScan::Create(std::move(reused_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr reused_plan, reused_scan->CreatePlan()); + ASSERT_EQ(plan->SnapshotId(), reused_plan->SnapshotId()); + ASSERT_EQ(plan->Splits().size(), reused_plan->Splits().size()); + ASSERT_EQ(plan->GetSnapshotReadView(), reused_plan->GetSnapshotReadView()); + ASSERT_EQ(0, file_system->SnapshotMetadataAccessCount()); + ASSERT_EQ(0, file_system->SchemaMetadataAccessCount()); +} + +TEST(TableScanTest, TestDataEvolutionEmptyGlobalIndexPublishesReusableSnapshotReadView) { + std::string path = + paimon::test::GetDataDir() + "/orc/append_with_global_index.db/append_with_global_index"; + ASSERT_OK_AND_ASSIGN(std::string normalized_path, PathUtil::NormalizePath(path)); + auto file_system = std::make_shared(normalized_path); + std::shared_ptr predicate = + PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, + Literal(FieldType::STRING, "not-found", 9)); + ScanContextBuilder builder(path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.AddOption("bitmap-global-index.legacy-format.enabled-for-testing", "true"); + builder.SetPredicate(predicate); + builder.WithFileSystem(file_system); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, table_scan->CreatePlan()); + ASSERT_TRUE(plan->SnapshotId()); + ASSERT_TRUE(plan->Splits().empty()); + ASSERT_TRUE(plan->GetSnapshotReadView()); + ASSERT_EQ(plan->SnapshotId(), plan->GetSnapshotReadView()->SnapshotId()); + ASSERT_GT(file_system->SnapshotMetadataAccessCount(), 0); + ASSERT_GT(file_system->SchemaMetadataAccessCount(), 0); + + file_system->BlockAndResetTableMetadataAccess(); + ScanContextBuilder reused_builder(path); + reused_builder.AddOption(Options::FILE_FORMAT, "orc"); + reused_builder.AddOption("bitmap-global-index.legacy-format.enabled-for-testing", "true"); + reused_builder.SetPredicate(predicate); + reused_builder.WithFileSystem(file_system); + reused_builder.WithSnapshotReadView(plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reused_context, reused_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reused_scan, + TableScan::Create(std::move(reused_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr reused_plan, reused_scan->CreatePlan()); + ASSERT_EQ(plan->SnapshotId(), reused_plan->SnapshotId()); + ASSERT_TRUE(reused_plan->Splits().empty()); + ASSERT_EQ(plan->GetSnapshotReadView(), reused_plan->GetSnapshotReadView()); + ASSERT_EQ(0, file_system->SnapshotMetadataAccessCount()); + ASSERT_EQ(0, file_system->SchemaMetadataAccessCount()); +} + +TEST(TableScanTest, TestDataEvolutionSuppliedEmptyGlobalIndexPreservesInjectedSnapshotReadView) { + std::string path = + paimon::test::GetDataDir() + "/orc/append_with_global_index.db/append_with_global_index"; + ScanContextBuilder unbound_builder(path); + unbound_builder.AddOption(Options::FILE_FORMAT, "orc"); + unbound_builder.SetGlobalIndexResult(BitmapGlobalIndexResult::FromRanges({})); + ASSERT_OK_AND_ASSIGN(std::unique_ptr unbound_context, unbound_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr unbound_scan, + TableScan::Create(std::move(unbound_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr unbound_plan, unbound_scan->CreatePlan()); + ASSERT_FALSE(unbound_plan->SnapshotId()); + ASSERT_FALSE(unbound_plan->GetSnapshotReadView()); + + ScanContextBuilder view_builder(path); + view_builder.AddOption(Options::FILE_FORMAT, "orc"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr view_context, view_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr view_scan, + TableScan::Create(std::move(view_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr source_plan, view_scan->CreatePlan()); + ASSERT_TRUE(source_plan->GetSnapshotReadView()); + + ScanContextBuilder builder(path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.SetGlobalIndexResult(BitmapGlobalIndexResult::FromRanges({})); + builder.WithSnapshotReadView(source_plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, table_scan->CreatePlan()); + ASSERT_EQ(source_plan->SnapshotId(), plan->SnapshotId()); + ASSERT_TRUE(plan->Splits().empty()); + ASSERT_EQ(source_plan->GetSnapshotReadView(), plan->GetSnapshotReadView()); +} + +TEST(TableScanTest, TestDataEvolutionExplicitSnapshotEmptyGlobalIndexIsNotReboundToLatest) { + std::string path = + paimon::test::GetDataDir() + "/orc/append_with_global_index.db/append_with_global_index"; + std::shared_ptr predicate = + PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, + Literal(FieldType::STRING, "not-found", 9)); + ScanContextBuilder builder(path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.AddOption(Options::SCAN_MODE, "from-snapshot"); + builder.AddOption(Options::SCAN_SNAPSHOT_ID, "4"); + builder.AddOption("bitmap-global-index.legacy-format.enabled-for-testing", "true"); + builder.SetPredicate(predicate); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, table_scan->CreatePlan()); + ASSERT_TRUE(plan->Splits().empty()); + ASSERT_FALSE(plan->SnapshotId()); + ASSERT_FALSE(plan->GetSnapshotReadView()); +} + +TEST(TableScanTest, TestReadOptimizedSnapshotReadViewSkipsTableMetadata) { + std::string base_path = + paimon::test::GetDataDir() + "/orc/pk_table_with_alter_table.db/pk_table_with_alter_table"; + std::string read_optimized_path = base_path + "$ro"; + ASSERT_OK_AND_ASSIGN(std::string normalized_base_path, PathUtil::NormalizePath(base_path)); + auto file_system = std::make_shared(normalized_base_path); + ScanContextBuilder builder(read_optimized_path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.WithFileSystem(file_system); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, table_scan->CreatePlan()); + ASSERT_TRUE(plan->SnapshotId()); + ASSERT_TRUE(plan->GetSnapshotReadView()); + ASSERT_EQ(normalized_base_path, plan->GetSnapshotReadView()->TablePath()); + ASSERT_EQ("main", plan->GetSnapshotReadView()->Branch()); + ASSERT_OK_AND_ASSIGN(uint64_t scanned_snapshot_id, table_scan->GetMetrics()->GetCounter( + ScanMetrics::LAST_SCANNED_SNAPSHOT_ID)); + ASSERT_EQ(scanned_snapshot_id, static_cast(plan->SnapshotId().value())); + ASSERT_GT(file_system->SnapshotMetadataAccessCount(), 0); + ASSERT_GT(file_system->SchemaMetadataAccessCount(), 0); + + file_system->BlockAndResetTableMetadataAccess(); + ScanContextBuilder reused_builder(read_optimized_path); + reused_builder.AddOption(Options::FILE_FORMAT, "orc"); + reused_builder.WithFileSystem(file_system); + reused_builder.WithSnapshotReadView(plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reused_context, reused_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reused_scan, + TableScan::Create(std::move(reused_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr reused_plan, reused_scan->CreatePlan()); + ASSERT_EQ(plan->SnapshotId(), reused_plan->SnapshotId()); + ASSERT_EQ(plan->Splits().size(), reused_plan->Splits().size()); + ASSERT_TRUE(reused_plan->GetSnapshotReadView()); + ASSERT_EQ(plan->GetSnapshotReadView(), reused_plan->GetSnapshotReadView()); + ASSERT_EQ(normalized_base_path, reused_plan->GetSnapshotReadView()->TablePath()); + ASSERT_EQ("main", reused_plan->GetSnapshotReadView()->Branch()); + ASSERT_EQ(0, file_system->SnapshotMetadataAccessCount()); + ASSERT_EQ(0, file_system->SchemaMetadataAccessCount()); +} + +TEST(TableScanTest, TestBaseSnapshotReadViewFeedsReadOptimizedScan) { + std::string base_path = + paimon::test::GetDataDir() + "/orc/pk_table_with_alter_table.db/pk_table_with_alter_table"; + std::string read_optimized_path = base_path + "$ro"; + ASSERT_OK_AND_ASSIGN(std::string normalized_base_path, PathUtil::NormalizePath(base_path)); + auto file_system = std::make_shared(normalized_base_path); + + ScanContextBuilder base_builder(base_path); + base_builder.AddOption(Options::FILE_FORMAT, "orc"); + base_builder.WithFileSystem(file_system); + ASSERT_OK_AND_ASSIGN(std::unique_ptr base_context, base_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr base_scan, + TableScan::Create(std::move(base_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr base_plan, base_scan->CreatePlan()); + ASSERT_TRUE(base_plan->GetSnapshotReadView()); + ASSERT_EQ(normalized_base_path, base_plan->GetSnapshotReadView()->TablePath()); + + file_system->BlockAndResetTableMetadataAccess(); + ScanContextBuilder read_optimized_builder(read_optimized_path); + read_optimized_builder.AddOption(Options::FILE_FORMAT, "orc"); + read_optimized_builder.WithFileSystem(file_system); + read_optimized_builder.WithSnapshotReadView(base_plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr read_optimized_context, + read_optimized_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr read_optimized_scan, + TableScan::Create(std::move(read_optimized_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr read_optimized_plan, + read_optimized_scan->CreatePlan()); + ASSERT_EQ(base_plan->SnapshotId(), read_optimized_plan->SnapshotId()); + ASSERT_EQ(base_plan->GetSnapshotReadView(), read_optimized_plan->GetSnapshotReadView()); + ASSERT_EQ(0, file_system->SnapshotMetadataAccessCount()); + ASSERT_EQ(0, file_system->SchemaMetadataAccessCount()); +} + +TEST(TableScanTest, TestReadOptimizedSnapshotReadViewRejectsRealtimeContext) { + std::string path = paimon::test::GetDataDir() + + "/orc/append_table_with_nested_type.db/append_table_with_nested_type$ro"; + ScanContextBuilder source_builder(path); + source_builder.AddOption(Options::FILE_FORMAT, "orc"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr source_context, source_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr source_scan, + TableScan::Create(std::move(source_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr source_plan, source_scan->CreatePlan()); + ASSERT_TRUE(source_plan->GetSnapshotReadView()); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr realtime_context, + RealtimeContext::Create()); + ScanContextBuilder builder(path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.WithSnapshotReadView(source_plan->GetSnapshotReadView()); + builder.WithRealtimeContext(realtime_context); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + ASSERT_NOK_WITH_MSG(TableScan::Create(std::move(context)), + "snapshot read view does not support real-time union scan"); +} + +TEST(TableScanTest, TestReadOptimizedTableReadUsesProvidedSchemaWithoutTableMetadata) { + std::string base_path = + paimon::test::GetDataDir() + "/orc/pk_table_with_alter_table.db/pk_table_with_alter_table"; + std::string read_optimized_path = base_path + "$ro"; + ASSERT_OK_AND_ASSIGN(std::string normalized_base_path, PathUtil::NormalizePath(base_path)); + auto file_system = std::make_shared(normalized_base_path); + SchemaManager schema_manager(file_system, normalized_base_path); + ASSERT_OK_AND_ASSIGN(std::optional> latest_schema, + schema_manager.Latest()); + ASSERT_TRUE(latest_schema); + ASSERT_OK_AND_ASSIGN(std::string table_schema_json, latest_schema.value()->ToJsonString()); + + file_system->BlockAndResetTableMetadataAccess(); + ReadContextBuilder builder(read_optimized_path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.WithFileSystem(file_system); + builder.SetTableSchema(table_schema_json); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_read, + TableRead::Create(std::move(context))); + ASSERT_TRUE(table_read); + ASSERT_EQ(0, file_system->SnapshotMetadataAccessCount()); + ASSERT_EQ(0, file_system->SchemaMetadataAccessCount()); +} + +TEST(TableScanTest, TestReadOptimizedBranchSnapshotReadViewSkipsTableMetadata) { + std::string base_path = paimon::test::GetDataDir() + + "/orc/append_table_with_rt_branch.db/append_table_with_rt_branch"; + std::string read_optimized_path = base_path + "$branch_rt$ro"; + ASSERT_OK_AND_ASSIGN(std::string normalized_base_path, PathUtil::NormalizePath(base_path)); + ASSERT_OK_AND_ASSIGN(std::string normalized_branch_path, + PathUtil::NormalizePath(base_path + "/branch/branch-rt")); + auto file_system = std::make_shared(normalized_branch_path); + ScanContextBuilder builder(read_optimized_path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + builder.WithFileSystem(file_system); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, table_scan->CreatePlan()); + ASSERT_TRUE(plan->SnapshotId()); + ASSERT_TRUE(plan->GetSnapshotReadView()); + ASSERT_EQ(normalized_base_path, plan->GetSnapshotReadView()->TablePath()); + ASSERT_EQ("rt", plan->GetSnapshotReadView()->Branch()); + ASSERT_GT(file_system->SnapshotMetadataAccessCount(), 0); + ASSERT_GT(file_system->SchemaMetadataAccessCount(), 0); + + file_system->BlockAndResetTableMetadataAccess(); + ScanContextBuilder reused_builder(read_optimized_path); + reused_builder.AddOption(Options::FILE_FORMAT, "orc"); + reused_builder.WithFileSystem(file_system); + reused_builder.WithSnapshotReadView(plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reused_context, reused_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr reused_scan, + TableScan::Create(std::move(reused_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr reused_plan, reused_scan->CreatePlan()); + ASSERT_EQ(plan->SnapshotId(), reused_plan->SnapshotId()); + ASSERT_EQ(plan->Splits().size(), reused_plan->Splits().size()); + ASSERT_TRUE(reused_plan->GetSnapshotReadView()); + ASSERT_EQ(plan->GetSnapshotReadView(), reused_plan->GetSnapshotReadView()); + ASSERT_EQ(normalized_base_path, reused_plan->GetSnapshotReadView()->TablePath()); + ASSERT_EQ("rt", reused_plan->GetSnapshotReadView()->Branch()); + ASSERT_EQ(0, file_system->SnapshotMetadataAccessCount()); + ASSERT_EQ(0, file_system->SchemaMetadataAccessCount()); +} + +TEST(TableScanTest, TestSnapshotReadViewValidatesTableAndBranchBinding) { + std::string path = + paimon::test::GetDataDir() + "/orc/pk_table_with_alter_table.db/pk_table_with_alter_table/"; + ScanContextBuilder builder(path); + builder.AddOption(Options::FILE_FORMAT, "orc"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, table_scan->CreatePlan()); + ASSERT_TRUE(plan->GetSnapshotReadView()); + + ScanContextBuilder other_table_builder(path + "other"); + other_table_builder.AddOption(Options::FILE_FORMAT, "orc"); + other_table_builder.WithSnapshotReadView(plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr other_table_context, + other_table_builder.Finish()); + ASSERT_NOK_WITH_MSG(TableScan::Create(std::move(other_table_context)), "bound to table"); + + ScanContextBuilder other_branch_builder(path); + other_branch_builder.AddOption(Options::FILE_FORMAT, "orc"); + other_branch_builder.AddOption(Options::BRANCH, "other"); + other_branch_builder.WithSnapshotReadView(plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr other_branch_context, + other_branch_builder.Finish()); + ASSERT_NOK_WITH_MSG(TableScan::Create(std::move(other_branch_context)), "bound to branch"); + + ScanContextBuilder other_system_table_builder(plan->GetSnapshotReadView()->TablePath() + + "$snapshots"); + other_system_table_builder.AddOption(Options::FILE_FORMAT, "orc"); + other_system_table_builder.WithSnapshotReadView(plan->GetSnapshotReadView()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr other_system_table_context, + other_system_table_builder.Finish()); + ASSERT_NOK_WITH_MSG(TableScan::Create(std::move(other_system_table_context)), + "only be used with the read-optimized system table"); } TEST(TableScanTest, TestNonExistTable) { diff --git a/src/paimon/core/table/system/read_optimized_system_table.cpp b/src/paimon/core/table/system/read_optimized_system_table.cpp index d7bfa0912..165edb7d5 100644 --- a/src/paimon/core/table/system/read_optimized_system_table.cpp +++ b/src/paimon/core/table/system/read_optimized_system_table.cpp @@ -34,7 +34,6 @@ #include "paimon/table/source/table_scan.h" namespace paimon { - ReadOptimizedSystemTable::ReadOptimizedSystemTable(std::string table_path, std::shared_ptr table_schema, std::map options) @@ -86,6 +85,9 @@ Result> ReadOptimizedSystemTable::NewScan( if (context->GetSpecificTableSchema().has_value()) { builder.SetTableSchema(context->GetSpecificTableSchema().value()); } + if (context->GetSnapshotReadView()) { + builder.WithSnapshotReadView(context->GetSnapshotReadView()); + } PAIMON_ASSIGN_OR_RAISE(std::unique_ptr base_context, builder.Finish()); return TableScan::Create(std::move(base_context)); } diff --git a/src/paimon/core/table/system/system_table.cpp b/src/paimon/core/table/system/system_table.cpp index 2d67355a8..11d8bc726 100644 --- a/src/paimon/core/table/system/system_table.cpp +++ b/src/paimon/core/table/system/system_table.cpp @@ -223,6 +223,13 @@ Result> SystemTableLoader::TryParsePath(const std Result> SystemTableLoader::LoadFromPath( const std::shared_ptr& fs, const std::string& path, const std::map& dynamic_options) { + return LoadFromPath(fs, path, dynamic_options, nullptr); +} + +Result> SystemTableLoader::LoadFromPath( + const std::shared_ptr& fs, const std::string& path, + const std::map& dynamic_options, + const std::shared_ptr& table_schema) { PAIMON_ASSIGN_OR_RAISE(std::optional system_table_path, TryParsePath(path)); if (!system_table_path) { return Status::Invalid("path is not a system table path: ", path); @@ -243,18 +250,22 @@ Result> SystemTableLoader::LoadFromPath( return GlobalSystemTableLoader::Load(parsed.system_table_name, context); } - SchemaManager schema_manager(fs, parsed.table_path, - parsed.branch.value_or(BranchManager::DEFAULT_MAIN_BRANCH)); - PAIMON_ASSIGN_OR_RAISE(std::optional> latest_schema, - schema_manager.Latest()); - if (!latest_schema) { - return Status::NotExist("base table schema not found for system table path: ", path); + std::shared_ptr resolved_schema = table_schema; + if (!resolved_schema) { + SchemaManager schema_manager(fs, parsed.table_path, + parsed.branch.value_or(BranchManager::DEFAULT_MAIN_BRANCH)); + PAIMON_ASSIGN_OR_RAISE(std::optional> latest_schema, + schema_manager.Latest()); + if (!latest_schema) { + return Status::NotExist("base table schema not found for system table path: ", path); + } + resolved_schema = latest_schema.value(); } auto options = dynamic_options; if (parsed.branch) { options[Options::BRANCH] = parsed.branch.value(); } - return Load(parsed.system_table_name, fs, parsed.table_path, latest_schema.value(), options); + return Load(parsed.system_table_name, fs, parsed.table_path, resolved_schema, options); } } // namespace paimon diff --git a/src/paimon/core/table/system/system_table.h b/src/paimon/core/table/system/system_table.h index 39e1da31b..0fddba574 100644 --- a/src/paimon/core/table/system/system_table.h +++ b/src/paimon/core/table/system/system_table.h @@ -81,6 +81,11 @@ class SystemTableLoader { static Result> LoadFromPath( const std::shared_ptr& fs, const std::string& path, const std::map& dynamic_options); + + static Result> LoadFromPath( + const std::shared_ptr& fs, const std::string& path, + const std::map& dynamic_options, + const std::shared_ptr& table_schema); }; } // namespace paimon diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index e05dc4b23..6170174b0 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -1442,6 +1442,28 @@ TEST_P(GlobalIndexTest, TestDataEvolutionBatchScan) { ASSERT_EQ(cache->SupplierCallCount(CacheKind::MANIFEST), first_supplier_calls); } + { + 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)); + const std::map options = {{Options::SCAN_SNAPSHOT_ID, "2"}}; + ASSERT_OK_AND_ASSIGN(auto first_plan, + ScanGlobalIndexAndData(table_path, predicate, options, + /*index_result=*/nullptr, cache)); + ASSERT_TRUE(first_plan->SnapshotId()); + ASSERT_EQ(first_plan->SnapshotId().value(), 2); + ASSERT_GE(cache->GetCount(CacheKind::MANIFEST), 2); + int64_t first_supplier_calls = cache->SupplierCallCount(CacheKind::MANIFEST); + + ASSERT_OK_AND_ASSIGN(auto second_plan, + ScanGlobalIndexAndData(table_path, predicate, options, + /*index_result=*/nullptr, cache)); + ASSERT_TRUE(second_plan->SnapshotId()); + ASSERT_EQ(second_plan->SnapshotId().value(), 2); + ASSERT_EQ(cache->SupplierCallCount(CacheKind::MANIFEST), first_supplier_calls); + } + // scan and read with global index { auto predicate =