From 1956f5719761e6915779bf5a37a2cada972045ec Mon Sep 17 00:00:00 2001 From: gripleaf <425797155@qq.com> Date: Wed, 16 Sep 2026 17:53:46 +0800 Subject: [PATCH] feat(scan): share table metadata resources across scans Reuse schema-derived metadata and optional snapshot caches across scans of the same table and branch. Keep scan state independent, bypass cached snapshots for existence-sensitive reads, and validate bound file system identity. --- docs/source/api/scan.rst | 69 +++ include/paimon/scan_context.h | 20 +- .../table/source/table_scan_resources.h | 68 +++ src/paimon/CMakeLists.txt | 2 + .../core/operation/expire_snapshots.cpp | 6 +- .../core/operation/expire_snapshots_test.cpp | 26 ++ src/paimon/core/operation/scan_context.cpp | 33 +- .../core/operation/scan_context_test.cpp | 87 ++++ src/paimon/core/schema/schema_manager.cpp | 19 +- src/paimon/core/schema/schema_manager.h | 7 +- .../core/schema/schema_manager_test.cpp | 38 ++ .../core/table/format/format_table_loader.cpp | 7 +- src/paimon/core/table/source/table_read.cpp | 2 +- src/paimon/core/table/source/table_scan.cpp | 106 +++-- .../table/source/table_scan_resources.cpp | 112 +++++ .../table/source/table_scan_resources_impl.h | 74 +++ .../source/table_scan_resources_test.cpp | 442 ++++++++++++++++++ .../table/system/audit_log_system_table.cpp | 3 +- .../system/read_optimized_system_table.cpp | 3 +- src/paimon/core/table/system/system_table.cpp | 14 +- src/paimon/core/table/system/system_table.h | 4 +- .../core/table/system/system_table_test.cpp | 5 +- src/paimon/core/utils/snapshot_manager.cpp | 84 +++- src/paimon/core/utils/snapshot_manager.h | 39 ++ .../core/utils/snapshot_manager_test.cpp | 385 +++++++++++++++ 25 files changed, 1585 insertions(+), 70 deletions(-) create mode 100644 include/paimon/table/source/table_scan_resources.h create mode 100644 src/paimon/core/table/source/table_scan_resources.cpp create mode 100644 src/paimon/core/table/source/table_scan_resources_impl.h create mode 100644 src/paimon/core/table/source/table_scan_resources_test.cpp diff --git a/docs/source/api/scan.rst b/docs/source/api/scan.rst index 0194aa6d3..01a84cfdf 100644 --- a/docs/source/api/scan.rst +++ b/docs/source/api/scan.rst @@ -47,6 +47,72 @@ Decimal literals are rescaled to the bucket field's type only when the conversio is exact. NaN literals and decimals that cannot be represented exactly disable inferred bucket pruning. +Sharing table metadata +====================== + +``TableScanResources`` retains schema metadata across scans of the same managed table and +branch. Create it once with a file system and pass it to each ``ScanContextBuilder``: + +.. code-block:: cpp + + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr resources, + paimon::TableScanResources::Create(table_path, file_system, "main")); + + // Repeat for each query, reusing resources. + paimon::ScanContextBuilder builder(table_path); + builder.WithTableResources(resources).SetPredicate(predicate); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr context, builder.Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr scan, + paimon::TableScan::Create(std::move(context))); + +Include ``paimon/table/source/table_scan_resources.h`` to create the resource object. +At ``Finish()``, the resources supply the context's file system when none was explicitly set. +An explicit ``WithFileSystem()`` must point to the same instance as the resources' file system; +a different instance is rejected regardless of builder call order. The resources also supply +the default branch; conflicting explicit branches are rejected. The physical table path must match, +including when scanning its ``$ro`` or ``$audit_log`` system table. Format tables and global system tables do not use these resources. +``Finish()`` resets the builder's resource setting, like ``WithCache()`` and ``WithExecutor()``. + +Every new scan still checks for the latest schema ID. Previously loaded schema versions, Arrow +schemas, and partition and primary-key field information are reused. Existing scans retain their +original schema. Snapshot selection, filters, streaming progress, executors and +scan metrics remain independent. ``SetTableSchema()`` keeps its existing +behavior: on main it bypasses the shared schema cache; on other branches it is ignored. + +Successfully loaded snapshots are cached by their full file paths, with an LRU limit of 20 entries +per resource object. The whole snapshot cache is replaced on the first access after it reaches +30 minutes of age. Reads and writes do not extend this deadline; recently added entries are +discarded along with older ones. There is no per-entry TTL or background refresh thread. +Snapshot caching is enabled by these resources; ordinary snapshot managers have no cache unless +one is explicitly supplied. +Latest and earliest snapshot discovery and existence checks still use the catalog or file system +as appropriate. Read or parse failures are not cached. + +Snapshot deletion through a manager replaces its entire injected snapshot cache before and after +the deletion attempt. Explicit invalidation also discards the whole cache. In-flight loads can +finish against the old cache but cannot populate the replacement used by subsequent callers. +Historical commit lookup, timestamp searches, and retained-snapshot publication checks read files +directly so cached metadata cannot hide missing or replaced files. Other clients' deletions do not +immediately invalidate this process's cache. A cached snapshot can remain available after its +metadata file expires; a cache hit does not establish that the snapshot or its data is still readable. + +The resources can be shared by concurrent scans when the supplied file system supports concurrent +use. Schema versions and schema-derived resources have no entry-count limit and are retained for +the resource object's lifetime. Eviction releases the cache's references; active scans retain +the metadata they need. +The snapshot cache limit bounds entry counts, not bytes or metadata held by active scans. +There is no background refresh. +The caller must recreate the resources after fast-forward, deleting and recreating a table or +branch, or changing file system access configuration. +Fast-forward can replace schema and snapshot contents under existing IDs; discovering the latest ID +does not refresh their cached contents. After fast-forward completes, create a new +``TableScanResources`` and use it for subsequent scans; cache hits in existing resources may still +return the old contents, and existing scans retain their original metadata. This follows +Paimon's `fast-forward cache refresh requirement +`_. +A metadata cache does not pin snapshots or prevent their data files from expiring. + Interface ========= @@ -54,6 +120,9 @@ Interface :members: :undoc-members: +.. doxygenclass:: paimon::TableScanResources + :members: + .. doxygenclass:: paimon::ScanContextBuilder :members: :undoc-members: diff --git a/include/paimon/scan_context.h b/include/paimon/scan_context.h index 28906253a..737a75cee 100644 --- a/include/paimon/scan_context.h +++ b/include/paimon/scan_context.h @@ -33,6 +33,7 @@ #include "paimon/visibility.h" namespace paimon { class ScanContextBuilder; +class TableScanResources; class ScanFilter; class Executor; class FormatTable; @@ -56,7 +57,8 @@ class PAIMON_EXPORT ScanContext { const std::optional& table_schema, const std::map& options, const std::shared_ptr& cache, - const std::shared_ptr& format_table); + const std::shared_ptr& format_table, + const std::shared_ptr& table_resources); ~ScanContext(); @@ -99,6 +101,10 @@ class PAIMON_EXPORT ScanContext { return specific_file_system_; } + const std::shared_ptr& GetTableResources() const { + return table_resources_; + } + const std::optional& GetSpecificTableSchema() const { return table_schema_; } @@ -127,6 +133,7 @@ class PAIMON_EXPORT ScanContext { std::map options_; std::shared_ptr cache_; std::shared_ptr format_table_; + std::shared_ptr table_resources_; }; /// Filter configuration for table scan operations @@ -219,7 +226,8 @@ class PAIMON_EXPORT ScanContextBuilder { /// This bypasses the global file system registry and uses the provided implementation directly. /// @param file_system The file system to use. /// @return Reference to this builder for method chaining. - /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`) + /// @note If not set, use the table resources' file system when provided, otherwise the + /// default file system (configured in `Options::FILE_SYSTEM`). ScanContextBuilder& WithFileSystem(const std::shared_ptr& file_system); /// Set the table schema as a string to avoid schema loading I/O operations. @@ -238,6 +246,14 @@ class PAIMON_EXPORT ScanContextBuilder { /// @return Reference to this builder for method chaining. ScanContextBuilder& WithCache(const std::shared_ptr& cache); + /// Share metadata resources with other scans of the same managed table and branch. + /// Finish() fills the context's file system from the resources when none was explicitly set. + /// An explicit file system must point to the same instance as the resources' file system. + /// The resources also supply the default branch; conflicting explicit branches are rejected. + /// Passing nullptr disables sharing. Finish() resets this setting. + /// On main, SetTableSchema() continues to bypass the shared schema cache. + ScanContextBuilder& WithTableResources(const std::shared_ptr& resources); + /// 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/table_scan_resources.h b/include/paimon/table/source/table_scan_resources.h new file mode 100644 index 000000000..b8dc58e2c --- /dev/null +++ b/include/paimon/table/source/table_scan_resources.h @@ -0,0 +1,68 @@ +/* + * 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 "paimon/result.h" +#include "paimon/visibility.h" + +namespace paimon { +class FileSystem; +class TableScanResourcesAccess; + +/// Metadata resources shared by scans of one managed table and branch. +/// +/// Share this object through ScanContextBuilder::WithTableResources(). Each scan still discovers +/// the latest schema ID and chooses its own snapshot. Schema versions and schema-derived resources +/// are retained for this object's lifetime without an entry limit. +/// Snapshots use a separate 20-entry LRU cache. On access, the whole snapshot +/// cache is replaced once it is 30 minutes old; reads and writes do not extend that deadline. +/// Eviction releases cache references; active scans retain the metadata they use. The snapshot +/// cache limit counts entries, not bytes. Latest snapshot discovery and existence checks query the +/// file system. A cached snapshot does not pin its metadata or data files. The caller must recreate +/// this object after fast-forward, dropping and recreating a table or branch, or changing the file +/// system's access configuration. Fast-forward can replace schema and snapshot contents under +/// existing IDs; there is no immediate cross-client invalidation. Use the new resources for +/// subsequent scans. +/// +/// Concurrent scans may share these resources. The supplied file system must support concurrent +/// use. Filters, executors, scan progress and scan metrics are not shared. +class PAIMON_EXPORT TableScanResources { + public: + /// Creates resources without loading table metadata. + /// @param table_path Physical table root, without a system table suffix. + /// @param file_system Non-null file system used by scans sharing these resources. + /// @param branch Branch to scan; an empty branch is normalized to main. + static Result> Create( + const std::string& table_path, const std::shared_ptr& file_system, + const std::string& branch); + + ~TableScanResources(); + + private: + friend class TableScanResourcesAccess; + class Impl; + explicit TableScanResources(std::unique_ptr&& impl); + + std::unique_ptr impl_; +}; +} // namespace paimon diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index e32b3a9e4..80fb59368 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -449,6 +449,7 @@ set(PAIMON_CORE_SRCS core/table/source/startup_mode.cpp core/table/source/table_read.cpp core/table/source/table_scan.cpp + core/table/source/table_scan_resources.cpp core/table/source/data_evolution_batch_scan.cpp core/table/source/primary_key_sorted_index_scan.cpp core/table/source/primary_key_sorted_index_result.cpp @@ -957,6 +958,7 @@ if(PAIMON_BUILD_TESTS) core/table/source/snapshot/snapshot_reader_test.cpp core/table/source/startup_mode_test.cpp core/table/source/table_scan_test.cpp + core/table/source/table_scan_resources_test.cpp core/table/system/system_table_test.cpp core/tag/tag_test.cpp core/utils/blob_view_lookup_test.cpp diff --git a/src/paimon/core/operation/expire_snapshots.cpp b/src/paimon/core/operation/expire_snapshots.cpp index 9915efd46..0fd0db894 100644 --- a/src/paimon/core/operation/expire_snapshots.cpp +++ b/src/paimon/core/operation/expire_snapshots.cpp @@ -151,7 +151,9 @@ Result ExpireSnapshots::ExpireUntil(int64_t earliest_snapshot_id, int64 } std::vector retained_snapshots; for (int64_t id = end_exclusive_id; id <= latest_snapshot_id; ++id) { - PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id)); + // Cached metadata cannot prove that the current file has been published. + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, + snapshot_manager_->LoadSnapshotFromFileSystem(id)); retained_snapshots.push_back(std::move(snapshot)); } if (latest.from_catalog && !(retained_snapshots.back() == latest.snapshot.value())) { @@ -226,7 +228,7 @@ Result ExpireSnapshots::ExpireUntil(int64_t earliest_snapshot_id, int64 expired_offset_files.insert(offsets_path.value()); } } - auto status = fs_->Delete(snapshot_manager_->SnapshotPath(id)); + auto status = snapshot_manager_->DeleteSnapshot(id); // delete quietly will ignore any status error (void)status; } diff --git a/src/paimon/core/operation/expire_snapshots_test.cpp b/src/paimon/core/operation/expire_snapshots_test.cpp index 778ee469f..7490e7d91 100644 --- a/src/paimon/core/operation/expire_snapshots_test.cpp +++ b/src/paimon/core/operation/expire_snapshots_test.cpp @@ -43,6 +43,7 @@ #include "paimon/format/file_format.h" #include "paimon/fs/local/local_file_system.h" #include "paimon/memory/memory_pool.h" +#include "paimon/testing/utils/snapshot_test_helper.h" #include "paimon/testing/utils/testharness.h" namespace paimon::test { @@ -228,6 +229,31 @@ TEST_F(ExpireSnapshotsTest, TestInvalidInput) { } } +TEST_F(ExpireSnapshotsTest, CachedRetainedSnapshotDoesNotProvePublication) { + auto cache = std::make_shared(); + auto manager = std::make_shared(fs_, test_data_path_, "main", cache); + ASSERT_OK(fs_->Mkdirs(manager->SnapshotDirectory())); + for (int64_t id : {1, 2, 3}) { + ASSERT_OK_AND_ASSIGN(std::string json, BuildTestSnapshot(id).ToJsonString()); + ASSERT_OK(fs_->WriteFile(manager->SnapshotPath(id), json, false)); + ASSERT_OK(manager->LoadSnapshot(id)); + } + ASSERT_OK(manager->CommitEarliestHint(1)); + ASSERT_OK(manager->CommitLatestHint(3)); + ASSERT_OK(fs_->Delete(manager->SnapshotPath(3))); + manager->SetSnapshotLoader([]() -> Result> { + return std::optional(BuildTestSnapshot(3)); + }); + ASSERT_OK_AND_ASSIGN(CoreOptions options, + CoreOptions::FromMap({{Options::SNAPSHOT_NUM_RETAINED_MIN, "2"}, + {Options::SNAPSHOT_NUM_RETAINED_MAX, "2"}})); + ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_, fs_, + options.GetExpireConfig(), options.RealtimeEnabled(), executor_); + ASSERT_TRUE(expire.Expire().status().IsNotExist()); + ASSERT_OK_AND_ASSIGN(bool exists, manager->SnapshotExists(1)); + ASSERT_TRUE(exists); +} + TEST_F(ExpireSnapshotsTest, TestGetDataFileToDelete) { auto mgr = std::make_shared(fs_, test_data_path_); ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({})); diff --git a/src/paimon/core/operation/scan_context.cpp b/src/paimon/core/operation/scan_context.cpp index 6bc80fe71..51e9f27e0 100644 --- a/src/paimon/core/operation/scan_context.cpp +++ b/src/paimon/core/operation/scan_context.cpp @@ -21,6 +21,8 @@ #include #include "paimon/common/utils/path_util.h" +#include "paimon/core/table/source/table_scan_resources_impl.h" +#include "paimon/defs.h" #include "paimon/executor.h" #include "paimon/memory/memory_pool.h" #include "paimon/status.h" @@ -40,7 +42,8 @@ ScanContext::ScanContext(const std::string& path, bool is_streaming_mode, const std::optional& table_schema, const std::map& options, const std::shared_ptr& cache, - const std::shared_ptr& format_table) + const std::shared_ptr& format_table, + const std::shared_ptr& table_resources) : path_(path), is_streaming_mode_(is_streaming_mode), limit_(limit), @@ -53,7 +56,8 @@ ScanContext::ScanContext(const std::string& path, bool is_streaming_mode, table_schema_(table_schema), options_(options), cache_(cache), - format_table_(format_table) {} + format_table_(format_table), + table_resources_(table_resources) {} ScanContext::~ScanContext() = default; @@ -75,6 +79,7 @@ class ScanContextBuilder::Impl { table_schema_ = std::nullopt; options_.clear(); cache_.reset(); + table_resources_.reset(); } private: @@ -97,6 +102,7 @@ class ScanContextBuilder::Impl { std::optional table_schema_; std::map options_; std::shared_ptr cache_; + std::shared_ptr table_resources_; }; ScanContextBuilder::ScanContextBuilder(const std::string& path) @@ -191,11 +197,20 @@ ScanContextBuilder& ScanContextBuilder::WithCache(const std::shared_ptr& return *this; } +ScanContextBuilder& ScanContextBuilder::WithTableResources( + const std::shared_ptr& resources) { + impl_->table_resources_ = resources; + return *this; +} + Result> ScanContextBuilder::Finish() { if (impl_->built_from_format_table_ && impl_->format_table_ == nullptr) { return Status::Invalid("cannot scan with null format table"); } if (impl_->format_table_ != nullptr) { + if (impl_->table_resources_) { + return Status::Invalid("table scan resources cannot be used with a format table"); + } // The table already answers both, and from a source this cannot see behind, so a second // answer is refused rather than silently dropped. if (impl_->table_schema_) { @@ -213,6 +228,16 @@ Result> ScanContextBuilder::Finish() { if (impl_->path_.empty()) { return Status::Invalid("cannot scan with empty table path"); } + auto options = impl_->options_; + std::shared_ptr file_system = impl_->specific_file_system_; + if (impl_->table_resources_) { + auto& resources = TableScanResourcesAccess::Get(*impl_->table_resources_); + auto branch = options.emplace(Options::BRANCH, resources.branch_).first; + if (!file_system) { + file_system = resources.file_system_; + } + PAIMON_RETURN_NOT_OK(resources.Validate(impl_->path_, branch->second, file_system)); + } std::shared_ptr executor = impl_->executor_ ? impl_->executor_ : CreateDefaultExecutor(); auto ctx = std::make_unique( @@ -220,8 +245,8 @@ Result> ScanContextBuilder::Finish() { std::make_shared(impl_->predicates_, impl_->partition_filters_, impl_->bucket_filter_), impl_->global_index_result_, impl_->realtime_context_, impl_->memory_pool_, executor, - impl_->specific_file_system_, impl_->table_schema_, impl_->options_, impl_->cache_, - impl_->format_table_); + file_system, impl_->table_schema_, options, impl_->cache_, impl_->format_table_, + impl_->table_resources_); impl_->Reset(); return ctx; } diff --git a/src/paimon/core/operation/scan_context_test.cpp b/src/paimon/core/operation/scan_context_test.cpp index 3b9658892..4dcd11c59 100644 --- a/src/paimon/core/operation/scan_context_test.cpp +++ b/src/paimon/core/operation/scan_context_test.cpp @@ -26,6 +26,7 @@ #include "paimon/memory/memory_pool.h" #include "paimon/predicate/predicate_builder.h" #include "paimon/status.h" +#include "paimon/table/source/table_scan_resources.h" #include "paimon/testing/mock/mock_file_system.h" #include "paimon/testing/utils/testharness.h" @@ -126,4 +127,90 @@ TEST(ScanContextTest, TestDefaultExecutorIsCreatedPerContext) { ASSERT_NE(executor, reset_ctx->GetExecutor()); } +TEST(ScanContextTest, TestTableResourcesValidationAndReset) { + auto fs = std::make_shared(); + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create("table/", fs, "dev")); + ScanContextBuilder builder("table"); + builder.WithTableResources(resources); + ASSERT_OK_AND_ASSIGN(auto context, builder.Finish()); + ASSERT_EQ(context->GetTableResources(), resources); + ASSERT_EQ(context->GetSpecificFileSystem(), fs); + ASSERT_EQ(context->GetOptions().at(Options::BRANCH), "dev"); + ASSERT_OK_AND_ASSIGN(auto reset_context, builder.Finish()); + ASSERT_FALSE(reset_context->GetTableResources()); + ASSERT_FALSE(reset_context->GetSpecificFileSystem()); + ASSERT_TRUE(reset_context->GetOptions().empty()); + + builder.WithTableResources(resources).WithTableResources(nullptr); + ASSERT_OK_AND_ASSIGN(auto disabled_context, builder.Finish()); + ASSERT_FALSE(disabled_context->GetTableResources()); + ASSERT_FALSE(disabled_context->GetSpecificFileSystem()); + ASSERT_TRUE(disabled_context->GetOptions().empty()); + + builder.WithTableResources(resources).WithFileSystem(fs); + ASSERT_OK(builder.Finish()); + builder.WithTableResources(resources).WithFileSystem(std::make_shared()); + ASSERT_NOK_WITH_MSG(builder.Finish(), "file system does not match"); + builder.WithFileSystem(nullptr).AddOption(Options::BRANCH, "main"); + ASSERT_NOK_WITH_MSG(builder.Finish(), "branch does not match"); + + ScanContextBuilder other("other_table"); + other.WithTableResources(resources); + ASSERT_NOK_WITH_MSG(other.Finish(), "path does not match"); + ScanContextBuilder system("table$branch_dev$ro"); + system.WithTableResources(resources); + ASSERT_OK(system.Finish()); + ScanContextBuilder wrong_branch("table$branch_other$ro"); + wrong_branch.WithTableResources(resources); + ASSERT_NOK_WITH_MSG(wrong_branch.Finish(), "branch does not match system table path"); + ScanContextBuilder global("warehouse/sys/catalog_options"); + global.WithTableResources(resources); + ASSERT_NOK_WITH_MSG(global.Finish(), "global system table"); +} + +TEST(ScanContextTest, TestTableResourcesFileSystemBindingIsOrderIndependent) { + auto fs = std::make_shared(); + auto other_fs = std::make_shared(); + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create("table", fs, "main")); + for (bool resources_first : {false, true}) { + for (const auto& explicit_fs : {fs, other_fs}) { + ScanContextBuilder builder("table"); + if (resources_first) { + builder.WithTableResources(resources).WithFileSystem(explicit_fs); + } else { + builder.WithFileSystem(explicit_fs).WithTableResources(resources); + } + if (explicit_fs != fs) { + ASSERT_NOK_WITH_MSG(builder.Finish(), "file system does not match"); + // A failed build must allow correcting the explicit file system and retrying. + builder.WithFileSystem(fs); + } + ASSERT_OK_AND_ASSIGN(auto context, builder.Finish()); + ASSERT_EQ(context->GetSpecificFileSystem(), fs); + ASSERT_EQ(context->GetTableResources(), resources); + } + } + + // Clearing an explicit override lets Finish() use the resources' instance. + ScanContextBuilder builder("table"); + builder.WithFileSystem(other_fs).WithTableResources(resources).WithFileSystem(nullptr); + ASSERT_OK_AND_ASSIGN(auto context, builder.Finish()); + ASSERT_EQ(context->GetSpecificFileSystem(), fs); + ASSERT_OK_AND_ASSIGN(auto reset_context, builder.Finish()); + ASSERT_FALSE(reset_context->GetSpecificFileSystem()); + ASSERT_FALSE(reset_context->GetTableResources()); +} + +TEST(ScanContextTest, TestInvalidTableResources) { + auto fs = std::make_shared(); + ASSERT_NOK(TableScanResources::Create("", fs, "main")); + ASSERT_NOK(TableScanResources::Create("table", nullptr, "main")); + ASSERT_NOK(TableScanResources::Create("table", fs, "../other")); + ASSERT_NOK_WITH_MSG(TableScanResources::Create("table$ro", fs, "main"), "physical table path"); + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create("table", fs, "")); + ScanContextBuilder builder("table"); + builder.WithTableResources(resources).AddOption(Options::BRANCH, "main"); + ASSERT_OK(builder.Finish()); +} + } // namespace paimon::test diff --git a/src/paimon/core/schema/schema_manager.cpp b/src/paimon/core/schema/schema_manager.cpp index 4ec1f8ec2..2979cd6de 100644 --- a/src/paimon/core/schema/schema_manager.cpp +++ b/src/paimon/core/schema/schema_manager.cpp @@ -71,17 +71,14 @@ Result>> SchemaManager::Latest() cons } Result> SchemaManager::ReadSchema(int64_t schema_id) const { - auto cached = schema_cache_.Find(schema_id); - if (cached) { - return cached.value(); - } - auto path = ToSchemaPath(schema_id); - std::string content; - PAIMON_RETURN_NOT_OK(file_system_->ReadFile(path, &content)); - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, - TableSchema::CreateFromJson(content)); - schema_cache_.Insert(schema_id, schema); - return schema; + return schema_cache_.Get( + schema_id, [this](const int64_t& id) -> Result> { + std::string content; + PAIMON_RETURN_NOT_OK(file_system_->ReadFile(ToSchemaPath(id), &content)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, + TableSchema::CreateFromJson(content)); + return schema; + }); } std::string SchemaManager::SchemaDirectory() const { diff --git a/src/paimon/core/schema/schema_manager.h b/src/paimon/core/schema/schema_manager.h index da81bcb81..209089014 100644 --- a/src/paimon/core/schema/schema_manager.h +++ b/src/paimon/core/schema/schema_manager.h @@ -26,7 +26,7 @@ #include #include -#include "paimon/common/utils/concurrent_hash_map.h" +#include "paimon/common/utils/generic_lru_cache.h" #include "paimon/core/schema/table_schema.h" #include "paimon/fs/file_system.h" #include "paimon/result.h" @@ -46,7 +46,7 @@ class SchemaManager { SchemaManager(const std::shared_ptr& file_system, const std::string& table_root, const std::string& branch); - /// Read schema for schema id. Find schema in cache first. + /// Read schema by ID, retaining loaded versions for the lifetime of this manager. /// Safe to call concurrently. Result> ReadSchema(int64_t schema_id) const; Result>> Latest() const; @@ -71,7 +71,8 @@ class SchemaManager { std::shared_ptr file_system_; std::string table_root_; const std::string branch_; - mutable ConcurrentHashMap> schema_cache_; + using SchemaCache = GenericLruCache>; + mutable SchemaCache schema_cache_{SchemaCache::Options{}}; }; } // namespace paimon diff --git a/src/paimon/core/schema/schema_manager_test.cpp b/src/paimon/core/schema/schema_manager_test.cpp index 0b078462b..994a6978f 100644 --- a/src/paimon/core/schema/schema_manager_test.cpp +++ b/src/paimon/core/schema/schema_manager_test.cpp @@ -47,6 +47,44 @@ TEST(SchemaManagerTest, ConcurrentHistoricalSchemaReads) { ASSERT_EQ(manager.schema_cache_.Size(), 2); } +TEST(SchemaManagerTest, RetainsMoreThan64SchemasAndRetriesFailures) { + auto directory = UniqueTestDirectory::Create(); + ASSERT_TRUE(directory); + auto fs = std::make_shared(); + SchemaManager manager(fs, directory->Str()); + ASSERT_OK(fs->Mkdirs(manager.SchemaDirectory())); + auto write_schema = [&](int64_t id) -> Status { + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr schema, + TableSchema::Create(id, arrow::schema({arrow::field("value", arrow::int32())}), {}, {}, + {})); + PAIMON_ASSIGN_OR_RAISE(std::string json, schema->GetJsonSchema()); + return fs->WriteFile(manager.ToSchemaPath(id), json, true); + }; + constexpr int64_t kSchemaCount = 65; + std::vector> schemas; + for (int64_t id = 0; id < kSchemaCount; ++id) { + ASSERT_OK(write_schema(id)); + ASSERT_OK_AND_ASSIGN(auto schema, manager.ReadSchema(id)); + schemas.push_back(schema); + // Loaded schemas remain cached after their files are removed. + ASSERT_OK(fs->Delete(manager.ToSchemaPath(id))); + } + for (int32_t round = 0; round < 3; ++round) { + for (int64_t id = 0; id < kSchemaCount; ++id) { + ASSERT_OK_AND_ASSIGN(auto cached, manager.ReadSchema(id)); + ASSERT_EQ(cached, schemas[id]); + } + } + + ASSERT_NOK(manager.ReadSchema(kSchemaCount)); + ASSERT_OK(fs->WriteFile(manager.ToSchemaPath(kSchemaCount), "invalid JSON", true)); + ASSERT_NOK(manager.ReadSchema(kSchemaCount)); + ASSERT_OK(write_schema(kSchemaCount)); + ASSERT_OK_AND_ASSIGN(auto reloaded, manager.ReadSchema(kSchemaCount)); + ASSERT_EQ(reloaded->Id(), kSchemaCount); +} + TEST(SchemaManagerTest, TestSimple) { auto fs = std::make_shared(); std::string table_root = diff --git a/src/paimon/core/table/format/format_table_loader.cpp b/src/paimon/core/table/format/format_table_loader.cpp index c50b85e4f..adc733e4d 100644 --- a/src/paimon/core/table/format/format_table_loader.cpp +++ b/src/paimon/core/table/format/format_table_loader.cpp @@ -51,11 +51,10 @@ Result> FormatTableLoader::TryLoad( } else { // Through the caller's manager when it has one, so that the read warms the cache it goes // on to use rather than a cache that dies with this call. - SchemaManager own_schema_manager(file_system, table_path, branch); - const SchemaManager& reader = - schema_manager != nullptr ? *schema_manager : own_schema_manager; PAIMON_ASSIGN_OR_RAISE(std::optional> latest_schema, - reader.Latest()); + schema_manager + ? schema_manager->Latest() + : SchemaManager(file_system, table_path, branch).Latest()); if (!latest_schema) { return std::shared_ptr(); } diff --git a/src/paimon/core/table/source/table_read.cpp b/src/paimon/core/table/source/table_read.cpp index 557f2fd5c..a0ef9a734 100644 --- a/src/paimon/core/table/source/table_read.cpp +++ b/src/paimon/core/table/source/table_read.cpp @@ -196,7 +196,7 @@ Result> TableRead::Create(std::unique_ptr system_table, SystemTableLoader::LoadFromPath(tmp_core_options.GetFileSystem(), context->GetPath(), - context->GetOptions())); + context->GetOptions(), nullptr)); 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 7ede285fe..c4f517975 100644 --- a/src/paimon/core/table/source/table_scan.cpp +++ b/src/paimon/core/table/source/table_scan.cpp @@ -61,6 +61,7 @@ #include "paimon/core/table/source/realtime_table_scan.h" #include "paimon/core/table/source/snapshot/snapshot_reader.h" #include "paimon/core/table/source/split_generator.h" +#include "paimon/core/table/source/table_scan_resources_impl.h" #include "paimon/core/table/system/system_table.h" #include "paimon/core/utils/branch_manager.h" #include "paimon/core/utils/field_mapping.h" @@ -91,21 +92,35 @@ class TableScanImpl { const std::shared_ptr& path_factory, const std::shared_ptr& arrow_schema, const std::shared_ptr& table_schema, const CoreOptions& core_options, - const std::shared_ptr& executor, const std::shared_ptr& memory_pool, - const ScanContext* context) { + const std::shared_ptr& executor, const ScanContext* context, + const std::shared_ptr& schema_resources, + const std::shared_ptr& memory_pool) { auto fs = core_options.GetFileSystem(); auto manifest_file_format = core_options.GetManifestFormat(); std::string branch = BranchManager::NormalizeBranch(core_options.GetBranch()); - auto snapshot_manager = std::make_shared(fs, context->GetPath(), branch); - // TODO(liancheng.lsz): support fallback branch in scan - auto schema_manager = std::make_shared(fs, context->GetPath(), branch); + std::shared_ptr snapshot_manager; + std::shared_ptr schema_manager; + if (context->GetTableResources()) { + auto& resources = TableScanResourcesAccess::Get(*context->GetTableResources()); + snapshot_manager = resources.snapshot_manager_; + schema_manager = resources.schema_manager_; + } else { + snapshot_manager = std::make_shared(fs, context->GetPath(), branch); + // TODO(liancheng.lsz): support fallback branch in scan + schema_manager = std::make_shared(fs, context->GetPath(), branch); + } PAIMON_ASSIGN_OR_RAISE( std::shared_ptr manifest_list, ManifestList::Create(fs, manifest_file_format, core_options.GetManifestCompression(), path_factory, core_options.GetCache(), memory_pool)); - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr partition_schema, - FieldMapping::GetPartitionSchema(arrow_schema, table_schema->PartitionKeys())); + std::shared_ptr partition_schema; + if (schema_resources) { + partition_schema = schema_resources->partition_schema; + } else { + PAIMON_ASSIGN_OR_RAISE( + partition_schema, + FieldMapping::GetPartitionSchema(arrow_schema, table_schema->PartitionKeys())); + } PAIMON_ASSIGN_OR_RAISE( std::shared_ptr manifest_file, ManifestFile::Create(fs, manifest_file_format, core_options.GetManifestCompression(), @@ -144,7 +159,8 @@ class TableScanImpl { static Result> CreateSplitGenerator( const std::shared_ptr& table_schema, const CoreOptions& core_options, - const ScanContext* context) { + const ScanContext* context, + const std::shared_ptr& schema_resources) { auto source_split_target_size = core_options.GetSourceSplitTargetSize(); auto source_split_open_file_cost = core_options.GetSourceSplitOpenFileCost(); if (table_schema->PrimaryKeys().empty()) { @@ -159,10 +175,12 @@ class TableScanImpl { source_split_target_size, source_split_open_file_cost, bucket_mode); } else { // TODO(liancheng.lsz): support evolution - PAIMON_ASSIGN_OR_RAISE(std::vector trimmed_primary_keys, - table_schema->TrimmedPrimaryKeys()); - PAIMON_ASSIGN_OR_RAISE(std::vector trimmed_pk_fields, - table_schema->GetFields(trimmed_primary_keys)); + std::vector trimmed_pk_fields; + if (schema_resources) { + trimmed_pk_fields = schema_resources->primary_key_fields; + } else { + PAIMON_ASSIGN_OR_RAISE(trimmed_pk_fields, table_schema->TrimmedPrimaryKeyFields()); + } PAIMON_ASSIGN_OR_RAISE( std::shared_ptr key_comparator, FieldsComparator::Create(trimmed_pk_fields, /*is_ascending_order=*/true)); @@ -266,6 +284,9 @@ Result> TableScan::Create(std::unique_ptr shared_context = std::move(context); // A table the caller already loaded says what it is, so nothing is read to find out. if (shared_context->GetFormatTable() != nullptr) { + if (shared_context->GetTableResources()) { + return Status::Invalid("table scan resources cannot be used with a format table"); + } PAIMON_ASSIGN_OR_RAISE( std::shared_ptr format_table, FormatTable::Copy(shared_context->GetFormatTable(), shared_context->GetOptions())); @@ -275,13 +296,20 @@ Result> TableScan::Create(std::unique_ptrGetOptions(), shared_context->GetSpecificFileSystem(), {})); + SchemaManager* shared_schema_manager = nullptr; + if (shared_context->GetTableResources()) { + auto& resources = TableScanResourcesAccess::Get(*shared_context->GetTableResources()); + PAIMON_RETURN_NOT_OK(resources.Validate(shared_context->GetPath(), tmp_options.GetBranch(), + shared_context->GetSpecificFileSystem())); + shared_schema_manager = resources.schema_manager_.get(); + } PAIMON_ASSIGN_OR_RAISE(std::optional system_table_path, SystemTableLoader::TryParsePath(shared_context->GetPath())); if (system_table_path) { PAIMON_ASSIGN_OR_RAISE( std::shared_ptr system_table, SystemTableLoader::LoadFromPath(tmp_options.GetFileSystem(), shared_context->GetPath(), - shared_context->GetOptions())); + shared_context->GetOptions(), shared_schema_manager)); return system_table->NewScan(shared_context); } // A format table is planned by listing directories, so it never reaches the manifest path @@ -290,12 +318,14 @@ Result> TableScan::Create(std::unique_ptr latest_schema; PAIMON_ASSIGN_OR_RAISE( std::shared_ptr format_table, - FormatTableLoader::TryLoad(tmp_options.GetFileSystem(), shared_context->GetPath(), - BranchManager::NormalizeBranch(tmp_options.GetBranch()), - shared_context->GetOptions(), - shared_context->GetSpecificTableSchema(), - /*schema_manager=*/nullptr, &latest_schema)); + FormatTableLoader::TryLoad( + tmp_options.GetFileSystem(), shared_context->GetPath(), + BranchManager::NormalizeBranch(tmp_options.GetBranch()), shared_context->GetOptions(), + shared_context->GetSpecificTableSchema(), shared_schema_manager, &latest_schema)); if (format_table != nullptr) { + if (shared_context->GetTableResources()) { + return Status::Invalid("table scan resources cannot be used with a format table"); + } return NewFormatTableScan(format_table, shared_context); } // With the schema the dispatch already read, so the managed path does not read it again. @@ -361,9 +391,15 @@ Result> NewDataTableScan( PAIMON_ASSIGN_OR_RAISE(table_schema, TableSchema::CreateFromJson(specific_table_schema.value())); } else { - SchemaManager schema_manager(tmp_options.GetFileSystem(), context->GetPath(), branch); - PAIMON_ASSIGN_OR_RAISE(std::optional> latest_table_schema, - schema_manager.Latest()); + const SchemaManager* schema_manager = + context->GetTableResources() + ? TableScanResourcesAccess::Get(*context->GetTableResources()).schema_manager_.get() + : nullptr; + PAIMON_ASSIGN_OR_RAISE( + std::optional> latest_table_schema, + schema_manager + ? schema_manager->Latest() + : SchemaManager(tmp_options.GetFileSystem(), context->GetPath(), branch).Latest()); if (latest_table_schema == std::nullopt) { return Status::Invalid("not found latest schema"); } @@ -396,7 +432,18 @@ Result> NewDataTableScan( } // validate schema and scan filter - auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields()); + std::shared_ptr schema_resources; + // A caller-supplied schema may reuse an ID with different content. Keep its derived metadata + // private to this scan, just as its JSON is kept out of SchemaManager's cache. + if (context->GetTableResources() && + !(branch == BranchManager::DEFAULT_MAIN_BRANCH && specific_table_schema)) { + PAIMON_ASSIGN_OR_RAISE(schema_resources, + TableScanResourcesAccess::Get(*context->GetTableResources()) + .GetSchemaResources(table_schema)); + } + auto arrow_schema = schema_resources + ? schema_resources->arrow_schema + : DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields()); if (context->GetScanFilters() && context->GetScanFilters()->GetPredicate()) { PAIMON_RETURN_NOT_OK(PredicateValidator::ValidatePredicateWithSchema( *arrow_schema, context->GetScanFilters()->GetPredicate(), @@ -416,13 +463,14 @@ Result> NewDataTableScan( external_paths, global_index_external_path, core_options.IndexFileInDataFileDir(), context->GetMemoryPool())); - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr file_store_scan, - TableScanImpl::CreateFileStoreScan( - path_factory, arrow_schema, table_schema, core_options, - context->GetExecutor(), context->GetMemoryPool(), context.get())); PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr split_generator, - TableScanImpl::CreateSplitGenerator(table_schema, core_options, context.get())); + std::shared_ptr file_store_scan, + TableScanImpl::CreateFileStoreScan(path_factory, arrow_schema, table_schema, core_options, + context->GetExecutor(), context.get(), schema_resources, + context->GetMemoryPool())); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr split_generator, + TableScanImpl::CreateSplitGenerator(table_schema, core_options, + context.get(), schema_resources)); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr index_file_handler, TableScanImpl::CreateIndexFileHandler(core_options, path_factory, context->GetMemoryPool())); diff --git a/src/paimon/core/table/source/table_scan_resources.cpp b/src/paimon/core/table/source/table_scan_resources.cpp new file mode 100644 index 000000000..9688b62a0 --- /dev/null +++ b/src/paimon/core/table/source/table_scan_resources.cpp @@ -0,0 +1,112 @@ +/* + * 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. + */ + +#include "paimon/table/source/table_scan_resources.h" + +#include + +#include "paimon/common/utils/path_util.h" +#include "paimon/core/table/source/table_scan_resources_impl.h" +#include "paimon/core/table/system/system_table.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/core/utils/field_mapping.h" + +namespace paimon { +Result> TableScanResources::Create( + const std::string& table_path, const std::shared_ptr& file_system, + const std::string& branch) { + if (!file_system) { + return Status::Invalid("table scan resources require a file system"); + } + PAIMON_ASSIGN_OR_RAISE(std::string path, PathUtil::NormalizePath(table_path)); + if (path.empty()) { + return Status::Invalid("table scan resources require a non-empty table path"); + } + PAIMON_RETURN_NOT_OK(BranchManager::CheckValidBranch(branch)); + PAIMON_ASSIGN_OR_RAISE(std::optional system_path, + SystemTableLoader::TryParsePath(path)); + if (system_path) { + return Status::Invalid("table scan resources require a physical table path"); + } + return std::shared_ptr(new TableScanResources( + std::make_unique(path, BranchManager::NormalizeBranch(branch), file_system))); +} + +TableScanResources::TableScanResources(std::unique_ptr&& impl) : impl_(std::move(impl)) {} + +TableScanResources::~TableScanResources() = default; + +TableScanResources::Impl::Impl(const std::string& path, const std::string& branch, + const std::shared_ptr& file_system) + : path_(path), + branch_(branch), + file_system_(file_system), + schema_manager_(std::make_shared(file_system, path, branch)), + snapshot_manager_(std::make_shared( + file_system, path, branch, std::make_shared())) {} + +Status TableScanResources::Impl::Validate( + const std::string& scan_path, const std::string& scan_branch, + const std::shared_ptr& specific_file_system) const { + PAIMON_ASSIGN_OR_RAISE(std::string physical_path, PathUtil::NormalizePath(scan_path)); + PAIMON_ASSIGN_OR_RAISE(std::optional system_path, + SystemTableLoader::TryParsePath(physical_path)); + if (system_path) { + if (system_path->is_global) { + return Status::Invalid( + "table scan resources cannot be used with a global system table"); + } + PAIMON_ASSIGN_OR_RAISE(physical_path, PathUtil::NormalizePath(system_path->table_path)); + if (system_path->branch && + BranchManager::NormalizeBranch(*system_path->branch) != branch_) { + return Status::Invalid("table scan resources branch does not match system table path"); + } + } + if (physical_path != path_) { + return Status::Invalid("table scan resources path does not match scan path"); + } + if (BranchManager::NormalizeBranch(scan_branch) != branch_) { + return Status::Invalid("table scan resources branch does not match scan branch"); + } + if (specific_file_system != file_system_) { + return Status::Invalid("table scan resources file system does not match scan file system"); + } + return Status::OK(); +} + +Result> TableScanResources::Impl::GetSchemaResources( + const std::shared_ptr& table_schema) { + // External schemas supplied through SetTableSchema() never enter this cache. + return schemas_.Get( + table_schema->Id(), + [&table_schema](const int64_t&) -> Result> { + auto resources = std::make_shared(); + resources->arrow_schema = + DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields()); + PAIMON_ASSIGN_OR_RAISE(resources->partition_schema, + FieldMapping::GetPartitionSchema(resources->arrow_schema, + table_schema->PartitionKeys())); + if (!table_schema->PrimaryKeys().empty()) { + PAIMON_ASSIGN_OR_RAISE(resources->primary_key_fields, + table_schema->TrimmedPrimaryKeyFields()); + } + return std::shared_ptr(std::move(resources)); + }); +} +} // namespace paimon diff --git a/src/paimon/core/table/source/table_scan_resources_impl.h b/src/paimon/core/table/source/table_scan_resources_impl.h new file mode 100644 index 000000000..f0a2be9e2 --- /dev/null +++ b/src/paimon/core/table/source/table_scan_resources_impl.h @@ -0,0 +1,74 @@ +/* + * 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 "paimon/common/types/data_field.h" +#include "paimon/common/utils/generic_lru_cache.h" +#include "paimon/core/schema/schema_manager.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/table/source/table_scan_resources.h" + +namespace paimon { + +/// Schema-derived metadata shared across scans. +struct ScanSchemaResources { + std::shared_ptr arrow_schema; + std::shared_ptr partition_schema; + std::vector primary_key_fields; +}; + +class TableScanResources::Impl { + public: + Impl(const std::string& path, const std::string& branch, + const std::shared_ptr& file_system); + + Status Validate(const std::string& scan_path, const std::string& scan_branch, + const std::shared_ptr& specific_file_system) const; + + Result> GetSchemaResources( + const std::shared_ptr& table_schema); + + const std::string path_; + const std::string branch_; + const std::shared_ptr file_system_; + const std::shared_ptr schema_manager_; + const std::shared_ptr snapshot_manager_; + + private: + using SchemaCache = GenericLruCache>; + SchemaCache schemas_{SchemaCache::Options{}}; +}; + +/// Keeps implementation types out of the public resource interface. +class TableScanResourcesAccess { + public: + TableScanResourcesAccess() = delete; + ~TableScanResourcesAccess() = delete; + + static TableScanResources::Impl& Get(const TableScanResources& resources) { + return *resources.impl_; + } +}; +} // namespace paimon diff --git a/src/paimon/core/table/source/table_scan_resources_test.cpp b/src/paimon/core/table/source/table_scan_resources_test.cpp new file mode 100644 index 000000000..25df2cce3 --- /dev/null +++ b/src/paimon/core/table/source/table_scan_resources_test.cpp @@ -0,0 +1,442 @@ +/* + * 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. + */ + +#include "paimon/table/source/table_scan_resources.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "arrow/api.h" +#include "gtest/gtest.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/table/source/data_split_impl.h" +#include "paimon/core/table/source/table_scan_resources_impl.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/defs.h" +#include "paimon/executor.h" +#include "paimon/fs/local/local_file_system.h" +#include "paimon/predicate/predicate_builder.h" +#include "paimon/scan_context.h" +#include "paimon/table/format/format_table.h" +#include "paimon/table/source/table_scan.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { +namespace { + +class SchemaCountingFileSystem : public LocalFileSystem { + public: + Status ReadFile(const std::string& path, std::string* content) override { + if (path.find("/schema/schema-") != std::string::npos) { + schema_reads.fetch_add(1); + if (fail_schema_read.exchange(false)) { + return Status::IOError("injected schema read failure"); + } + } + if (path.find("/snapshot/snapshot-") != std::string::npos) { + snapshot_reads.fetch_add(1); + } + return LocalFileSystem::ReadFile(path, content); + } + + std::atomic schema_reads{0}; + std::atomic snapshot_reads{0}; + std::atomic fail_schema_read{false}; +}; + +void CheckPlans(const std::shared_ptr& expected, const std::shared_ptr& actual) { + ASSERT_EQ(expected->SnapshotId(), actual->SnapshotId()); + ASSERT_EQ(expected->Splits().size(), actual->Splits().size()); + for (size_t i = 0; i < expected->Splits().size(); ++i) { + auto expected_split = std::dynamic_pointer_cast(expected->Splits()[i]); + auto actual_split = std::dynamic_pointer_cast(actual->Splits()[i]); + ASSERT_TRUE(expected_split); + ASSERT_TRUE(actual_split); + ASSERT_EQ(*expected_split, *actual_split); + } +} + +} // namespace + +class TableScanResourcesTest : public testing::Test { + protected: + Result> NewScan(const std::string& path, + const std::shared_ptr& resources, + const std::shared_ptr& predicate, + const std::map& options, + bool streaming) { + ScanContextBuilder builder(path); + builder.WithFileSystem(fs_) + .WithExecutor(executor_) + .WithTableResources(resources) + .SetPredicate(predicate) + .SetOptions(options) + .WithStreamingMode(streaming); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr context, builder.Finish()); + return TableScan::Create(std::move(context)); + } + + const std::string append_path_ = GetDataDir() + "/orc/append_09.db/append_09"; + const std::string pk_path_ = + GetDataDir() + "/orc/pk_table_with_alter_table.db/pk_table_with_alter_table"; + std::shared_ptr fs_ = std::make_shared(); + std::shared_ptr executor_ = CreateDefaultExecutor(); +}; + +TEST_F(TableScanResourcesTest, RepeatedScansReuseSchemasAndKeepFiltersIndependent) { + for (const auto& path : {append_path_, pk_path_}) { + const bool pk = path == pk_path_; + auto predicate = PredicateBuilder::IsNotNull(pk ? 3 : 0, pk ? "c" : "f0", + pk ? FieldType::INT : FieldType::STRING); + ASSERT_OK_AND_ASSIGN(auto expected_scan, NewScan(path, nullptr, predicate, {}, false)); + ASSERT_OK_AND_ASSIGN(auto expected, expected_scan->CreatePlan()); + ASSERT_FALSE(expected->Splits().empty()); + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(path, fs_, "main")); + ASSERT_OK_AND_ASSIGN(auto first, NewScan(path, resources, predicate, {}, false)); + ASSERT_OK_AND_ASSIGN(auto first_plan, first->CreatePlan()); + CheckPlans(expected, first_plan); + int32_t reads = fs_->schema_reads.load(); + ASSERT_OK_AND_ASSIGN(auto second, NewScan(path, resources, predicate, {}, false)); + ASSERT_OK_AND_ASSIGN(auto second_plan, second->CreatePlan()); + CheckPlans(expected, second_plan); + ASSERT_EQ(fs_->schema_reads.load(), reads); + + ScanContextBuilder missing_partition(path); + missing_partition.WithTableResources(resources).WithExecutor(executor_).SetPartitionFilter( + {{{pk ? "key0" : "f1", "-999"}}}); + ASSERT_OK_AND_ASSIGN(auto context, missing_partition.Finish()); + ASSERT_OK_AND_ASSIGN(auto filtered, TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(auto filtered_plan, filtered->CreatePlan()); + ASSERT_TRUE(filtered_plan->Splits().empty()); + ASSERT_EQ(fs_->schema_reads.load(), reads); + ASSERT_NOK_WITH_MSG(first->CreatePlan(), "end of scan"); + } +} + +TEST_F(TableScanResourcesTest, DirectContextRequiresResourcesFileSystem) { + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(append_path_, fs_, "main")); + ScanContextBuilder builder(append_path_); + ASSERT_OK_AND_ASSIGN(auto defaults, builder.Finish()); + const std::vector> file_systems = { + nullptr, std::make_shared(), fs_}; + for (const auto& file_system : file_systems) { + auto context = std::make_unique( + append_path_, false, std::nullopt, defaults->GetScanFilters(), nullptr, nullptr, + GetDefaultPool(), executor_, file_system, std::nullopt, defaults->GetOptions(), nullptr, + nullptr, resources); + if (file_system != fs_) { + ASSERT_NOK_WITH_MSG(TableScan::Create(std::move(context)), + "file system does not match"); + } else { + ASSERT_OK_AND_ASSIGN(auto scan, TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan()); + ASSERT_FALSE(plan->Splits().empty()); + } + } +} + +TEST_F(TableScanResourcesTest, SharedSnapshotCacheDiscoversNewSnapshot) { + auto directory = UniqueTestDirectory::Create(); + ASSERT_TRUE(directory); + const std::string path = directory->Str() + "/table"; + ASSERT_TRUE(TestUtil::CopyDirectory(append_path_, path)); + const std::string next_snapshot_path = path + "/snapshot/snapshot-5"; + std::string next_snapshot_json; + ASSERT_OK(fs_->ReadFile(next_snapshot_path, &next_snapshot_json)); + ASSERT_OK(fs_->Delete(next_snapshot_path)); + ASSERT_OK(fs_->WriteFile(path + "/snapshot/LATEST", "4", true)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr resources, + TableScanResources::Create(path, fs_, "main")); + ASSERT_OK_AND_ASSIGN(std::unique_ptr first, + NewScan(path, resources, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr first_plan, first->CreatePlan()); + ASSERT_EQ(first_plan->SnapshotId(), 4); + int32_t reads = fs_->snapshot_reads.load(); + ASSERT_OK_AND_ASSIGN(std::unique_ptr repeated, + NewScan(path, resources, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr repeated_plan, repeated->CreatePlan()); + CheckPlans(first_plan, repeated_plan); + ASSERT_EQ(fs_->snapshot_reads.load(), reads); + + // Publish the next snapshot while LATEST still points at the previous one. + ASSERT_OK(fs_->WriteFile(next_snapshot_path, next_snapshot_json, false)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr next, + NewScan(path, resources, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr next_plan, next->CreatePlan()); + ASSERT_EQ(next_plan->SnapshotId(), 5); + ASSERT_EQ(fs_->snapshot_reads.load(), reads + 1); + ASSERT_OK_AND_ASSIGN(std::unique_ptr baseline, + NewScan(path, nullptr, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr baseline_plan, baseline->CreatePlan()); + CheckPlans(baseline_plan, next_plan); +} + +TEST_F(TableScanResourcesTest, NewSchemaDoesNotChangeExistingScans) { + auto directory = UniqueTestDirectory::Create(); + ASSERT_TRUE(directory); + std::string path = directory->Str() + "/table"; + ASSERT_TRUE(TestUtil::CopyDirectory(append_path_, path)); + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(path, fs_, "main")); + auto old_predicate = PredicateBuilder::IsNotNull(0, "f0", FieldType::STRING); + ASSERT_OK_AND_ASSIGN(auto baseline_scan, NewScan(path, nullptr, old_predicate, {}, false)); + ASSERT_OK_AND_ASSIGN(auto baseline, baseline_scan->CreatePlan()); + ASSERT_FALSE(baseline->Splits().empty()); + ASSERT_OK_AND_ASSIGN(auto old_scan, NewScan(path, resources, old_predicate, {}, false)); + auto new_predicate = PredicateBuilder::IsNotNull(4, "added", FieldType::INT); + ASSERT_NOK(NewScan(path, resources, new_predicate, {}, false)); + + std::string schema_json; + ASSERT_OK(fs_->ReadFile(path + "/schema/schema-0", &schema_json)); + ASSERT_OK_AND_ASSIGN(auto old_schema, TableSchema::CreateFromJson(schema_json)); + auto schema = + arrow::schema({arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()), + arrow::field("f2", arrow::int32()), arrow::field("f3", arrow::float64()), + arrow::field("added", arrow::int32())}); + ASSERT_OK_AND_ASSIGN(auto new_schema, + TableSchema::Create(1, schema, {"f1"}, {}, old_schema->Options())); + ASSERT_OK_AND_ASSIGN(auto new_json, new_schema->GetJsonSchema()); + ASSERT_OK(fs_->WriteFile(path + "/schema/schema-1", new_json, false)); + ASSERT_OK_AND_ASSIGN(auto new_scan, NewScan(path, resources, new_predicate, {}, false)); + ASSERT_OK_AND_ASSIGN(auto new_plan, new_scan->CreatePlan()); + ASSERT_TRUE(new_plan->Splits().empty()); + ASSERT_OK_AND_ASSIGN(auto old_plan, old_scan->CreatePlan()); + CheckPlans(baseline, old_plan); + + int32_t reads = fs_->schema_reads.load(); + ASSERT_OK_AND_ASSIGN(auto again, NewScan(path, resources, new_predicate, {}, false)); + ASSERT_OK_AND_ASSIGN(auto again_plan, again->CreatePlan()); + ASSERT_TRUE(again_plan->Splits().empty()); + ASSERT_EQ(fs_->schema_reads.load(), reads); +} + +TEST_F(TableScanResourcesTest, SchemaResourcesRetainedBeyond64Versions) { + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(append_path_, fs_, "main")); + auto predicate = PredicateBuilder::IsNotNull(0, "f0", FieldType::STRING); + ASSERT_OK_AND_ASSIGN(auto baseline_scan, NewScan(append_path_, nullptr, predicate, {}, false)); + ASSERT_OK_AND_ASSIGN(auto baseline, baseline_scan->CreatePlan()); + ASSERT_OK_AND_ASSIGN(auto active_scan, NewScan(append_path_, resources, predicate, {}, false)); + + auto& impl = TableScanResourcesAccess::Get(*resources); + ASSERT_OK_AND_ASSIGN(auto original_schema, impl.schema_manager_->ReadSchema(0)); + ASSERT_OK_AND_ASSIGN(auto original, impl.GetSchemaResources(original_schema)); + std::weak_ptr retained = original; + auto original_arrow_schema = original->arrow_schema; + original.reset(); + auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(original_schema->Fields()); + for (int64_t id = 1; id <= 64; ++id) { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr schema, + TableSchema::Create(id, arrow_schema, original_schema->PartitionKeys(), + original_schema->PrimaryKeys(), original_schema->Options())); + ASSERT_OK(impl.GetSchemaResources(schema)); + } + ASSERT_FALSE(retained.expired()); + ASSERT_OK_AND_ASSIGN(auto actual, active_scan->CreatePlan()); + CheckPlans(baseline, actual); + ASSERT_OK_AND_ASSIGN(auto reloaded, impl.GetSchemaResources(original_schema)); + ASSERT_EQ(reloaded, retained.lock()); + ASSERT_EQ(reloaded->arrow_schema, original_arrow_schema); + ASSERT_TRUE(reloaded->arrow_schema->Equals(*original_arrow_schema)); + ASSERT_OK_AND_ASSIGN(auto repeated, impl.GetSchemaResources(original_schema)); + ASSERT_EQ(repeated, reloaded); +} + +TEST_F(TableScanResourcesTest, SuppliedSchemaDoesNotPopulateSharedSchemaCache) { + std::string json; + ASSERT_OK(fs_->ReadFile(append_path_ + "/schema/schema-0", &json)); + ASSERT_OK_AND_ASSIGN(auto schema, TableSchema::CreateFromJson(json)); + auto external_arrow_schema = arrow::schema( + {arrow::field("external_f0", arrow::utf8()), arrow::field("f1", arrow::int32()), + arrow::field("f2", arrow::int32()), arrow::field("f3", arrow::float64())}); + ASSERT_OK_AND_ASSIGN(auto external_schema, TableSchema::Create(0, external_arrow_schema, {"f1"}, + {}, schema->Options())); + ASSERT_OK_AND_ASSIGN(auto external_json, external_schema->GetJsonSchema()); + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(append_path_, fs_, "main")); + int32_t reads = fs_->schema_reads.load(); + ScanContextBuilder builder(append_path_); + builder.WithTableResources(resources) + .WithExecutor(executor_) + .SetTableSchema(external_json) + .SetPredicate(PredicateBuilder::IsNotNull(0, "external_f0", FieldType::STRING)); + ASSERT_OK_AND_ASSIGN(auto context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(auto external_scan, TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(auto external_plan, external_scan->CreatePlan()); + ASSERT_FALSE(external_plan->Splits().empty()); + ASSERT_EQ(fs_->schema_reads.load(), reads); + + auto predicate = PredicateBuilder::IsNotNull(0, "f0", FieldType::STRING); + ASSERT_OK_AND_ASSIGN(auto normal_scan, NewScan(append_path_, resources, predicate, {}, false)); + ASSERT_OK_AND_ASSIGN(auto normal_plan, normal_scan->CreatePlan()); + ASSERT_FALSE(normal_plan->Splits().empty()); + ASSERT_EQ(fs_->schema_reads.load(), reads + 1); + ASSERT_NOK(NewScan(append_path_, resources, + PredicateBuilder::IsNotNull(0, "external_f0", FieldType::STRING), {}, + false)); +} + +TEST_F(TableScanResourcesTest, ConcurrentScansAndSchemaReadRetry) { + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(append_path_, fs_, "main")); + fs_->fail_schema_read = true; + ASSERT_NOK_WITH_MSG(NewScan(append_path_, resources, nullptr, {}, false), + "injected schema read failure"); + std::vector>>> futures; + for (int32_t i = 0; i < 8; ++i) { + futures.push_back( + std::async(std::launch::async, [this, resources]() -> Result> { + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr scan, + NewScan(append_path_, resources, + PredicateBuilder::IsNotNull(0, "f0", FieldType::STRING), {}, false)); + return scan->CreatePlan(); + })); + } + ASSERT_OK_AND_ASSIGN(auto expected, futures[0].get()); + ASSERT_FALSE(expected->Splits().empty()); + for (size_t i = 1; i < futures.size(); ++i) { + ASSERT_OK_AND_ASSIGN(auto actual, futures[i].get()); + CheckPlans(expected, actual); + } + int32_t reads = fs_->schema_reads.load(); + ASSERT_OK_AND_ASSIGN(auto scan, NewScan(append_path_, resources, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan()); + ASSERT_EQ(fs_->schema_reads.load(), reads); +} + +TEST_F(TableScanResourcesTest, SnapshotsAndStreamingProgressStayIndependent) { + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(append_path_, fs_, "main")); + for (const std::string snapshot : {"1", "5"}) { + std::map options = {{Options::SCAN_SNAPSHOT_ID, snapshot}}; + ASSERT_OK_AND_ASSIGN(auto expected_scan, + NewScan(append_path_, nullptr, nullptr, options, false)); + ASSERT_OK_AND_ASSIGN(auto expected, expected_scan->CreatePlan()); + ASSERT_OK_AND_ASSIGN(auto actual_scan, + NewScan(append_path_, resources, nullptr, options, false)); + ASSERT_OK_AND_ASSIGN(auto actual, actual_scan->CreatePlan()); + CheckPlans(expected, actual); + } + std::map options = {{Options::SCAN_SNAPSHOT_ID, "1"}}; + ASSERT_OK_AND_ASSIGN(auto first, NewScan(append_path_, resources, nullptr, options, true)); + ASSERT_OK_AND_ASSIGN(auto second, NewScan(append_path_, resources, nullptr, options, true)); + ASSERT_OK_AND_ASSIGN(auto first_plan, first->CreatePlan()); + ASSERT_OK_AND_ASSIGN(auto next_plan, first->CreatePlan()); + ASSERT_OK_AND_ASSIGN(auto second_plan, second->CreatePlan()); + CheckPlans(first_plan, second_plan); + ASSERT_NE(first_plan->SnapshotId(), next_plan->SnapshotId()); +} + +TEST_F(TableScanResourcesTest, SystemTableScansForwardResources) { + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(pk_path_, fs_, "main")); + for (const std::string suffix : {"$ro", "$audit_log"}) { + ASSERT_OK_AND_ASSIGN(auto expected_scan, + NewScan(pk_path_ + suffix, nullptr, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(auto expected, expected_scan->CreatePlan()); + ASSERT_OK_AND_ASSIGN(auto first, NewScan(pk_path_ + suffix, resources, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(auto first_plan, first->CreatePlan()); + CheckPlans(expected, first_plan); + int32_t reads = fs_->schema_reads.load(); + ASSERT_OK_AND_ASSIGN(auto second, + NewScan(pk_path_ + suffix, resources, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(auto second_plan, second->CreatePlan()); + CheckPlans(expected, second_plan); + ASSERT_EQ(fs_->schema_reads.load(), reads); + } +} + +TEST_F(TableScanResourcesTest, BranchResourcesUseBranchSchemaAndSnapshot) { + auto directory = UniqueTestDirectory::Create(); + ASSERT_TRUE(directory); + std::string path = directory->Str() + "/table"; + ASSERT_TRUE(TestUtil::CopyDirectory(append_path_, path)); + std::string branch_path = BranchManager::BranchPath(path, "dev"); + ASSERT_OK(fs_->Mkdirs(branch_path)); + ASSERT_TRUE(TestUtil::CopyDirectory(append_path_ + "/schema", branch_path + "/schema")); + ASSERT_TRUE(TestUtil::CopyDirectory(append_path_ + "/snapshot", branch_path + "/snapshot")); + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(path, fs_, "dev")); + std::map options = {{Options::BRANCH, "dev"}}; + ASSERT_OK_AND_ASSIGN(auto expected_scan, NewScan(path, nullptr, nullptr, options, false)); + ASSERT_OK_AND_ASSIGN(auto expected, expected_scan->CreatePlan()); + ASSERT_FALSE(expected->Splits().empty()); + ScanContextBuilder builder(path); + // A supplied schema is only used on main; it must not enter the branch's schema cache. + builder.WithTableResources(resources).WithExecutor(executor_).SetTableSchema("not JSON"); + ASSERT_OK_AND_ASSIGN(auto context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(auto scan, TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(auto actual, scan->CreatePlan()); + CheckPlans(expected, actual); + int32_t reads = fs_->schema_reads.load(); + ASSERT_OK_AND_ASSIGN(auto second, NewScan(path, resources, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(auto second_plan, second->CreatePlan()); + CheckPlans(expected, second_plan); + ASSERT_EQ(fs_->schema_reads.load(), reads); +} + +TEST_F(TableScanResourcesTest, LimitDoesNotAffectOtherScans) { + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(append_path_, fs_, "main")); + ScanContextBuilder builder(append_path_); + builder.WithTableResources(resources).WithExecutor(executor_).SetLimit(1); + ASSERT_OK_AND_ASSIGN(auto context, builder.Finish()); + ASSERT_OK_AND_ASSIGN(auto limited, TableScan::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(auto limited_plan, limited->CreatePlan()); + ASSERT_OK_AND_ASSIGN(auto full, NewScan(append_path_, resources, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(auto full_plan, full->CreatePlan()); + ASSERT_FALSE(limited_plan->Splits().empty()); + ASSERT_LT(limited_plan->Splits().size(), full_plan->Splits().size()); + ASSERT_OK_AND_ASSIGN(auto baseline, NewScan(append_path_, nullptr, nullptr, {}, false)); + ASSERT_OK_AND_ASSIGN(auto baseline_plan, baseline->CreatePlan()); + CheckPlans(baseline_plan, full_plan); +} + +TEST_F(TableScanResourcesTest, FormatTableRejectsManagedResources) { + auto directory = UniqueTestDirectory::Create(); + ASSERT_TRUE(directory); + std::string path = directory->Str(); + ASSERT_OK_AND_ASSIGN( + auto schema, + TableSchema::Create(0, arrow::schema({arrow::field("value", arrow::int32())}), {}, {}, + {{Options::TYPE, "format-table"}, {Options::FILE_FORMAT, "parquet"}})); + ASSERT_OK_AND_ASSIGN(auto json, schema->GetJsonSchema()); + ASSERT_OK(fs_->Mkdirs(path + "/schema")); + ASSERT_OK(fs_->WriteFile(path + "/schema/schema-0", json, false)); + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(path, fs_, "main")); + ASSERT_NOK_WITH_MSG(NewScan(path, resources, nullptr, {}, false), + "cannot be used with a format table"); + ASSERT_OK_AND_ASSIGN(auto table, FormatTable::Create(fs_, path, Identifier("db", "table"), {})); + ScanContextBuilder builder(table); + builder.WithTableResources(resources); + ASSERT_NOK_WITH_MSG(builder.Finish(), "cannot be used with a format table"); +} + +TEST_F(TableScanResourcesTest, ActiveScanOutlivesResources) { + ASSERT_OK_AND_ASSIGN(auto resources, TableScanResources::Create(append_path_, fs_, "main")); + std::weak_ptr weak_resources = resources; + ASSERT_OK_AND_ASSIGN( + auto scan, NewScan(append_path_, resources, + PredicateBuilder::IsNotNull(0, "f0", FieldType::STRING), {}, false)); + resources.reset(); + ASSERT_TRUE(weak_resources.expired()); + ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan()); + ASSERT_FALSE(plan->Splits().empty()); +} + +} // namespace paimon::test diff --git a/src/paimon/core/table/system/audit_log_system_table.cpp b/src/paimon/core/table/system/audit_log_system_table.cpp index 8de9742e1..1b0cfb970 100644 --- a/src/paimon/core/table/system/audit_log_system_table.cpp +++ b/src/paimon/core/table/system/audit_log_system_table.cpp @@ -394,7 +394,8 @@ Result> AuditLogSystemTable::NewScan( .WithMemoryPool(context->GetMemoryPool()) .WithExecutor(context->GetExecutor()) .WithFileSystem(context->GetSpecificFileSystem()) - .WithCache(context->GetCache()); + .WithCache(context->GetCache()) + .WithTableResources(context->GetTableResources()); if (scan_filter) { if (scan_filter->GetBucketFilter()) { builder.SetBucketFilter(scan_filter->GetBucketFilter().value()); 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 8d5696447..c59df0af7 100644 --- a/src/paimon/core/table/system/read_optimized_system_table.cpp +++ b/src/paimon/core/table/system/read_optimized_system_table.cpp @@ -69,7 +69,8 @@ Result> ReadOptimizedSystemTable::NewScan( .WithMemoryPool(context->GetMemoryPool()) .WithExecutor(context->GetExecutor()) .WithFileSystem(context->GetSpecificFileSystem()) - .WithCache(context->GetCache()); + .WithCache(context->GetCache()) + .WithTableResources(context->GetTableResources()); if (context->GetLimit().has_value()) { builder.SetLimit(context->GetLimit().value()); } diff --git a/src/paimon/core/table/system/system_table.cpp b/src/paimon/core/table/system/system_table.cpp index 2d67355a8..4b91bf907 100644 --- a/src/paimon/core/table/system/system_table.cpp +++ b/src/paimon/core/table/system/system_table.cpp @@ -222,7 +222,8 @@ Result> SystemTableLoader::TryParsePath(const std Result> SystemTableLoader::LoadFromPath( const std::shared_ptr& fs, const std::string& path, - const std::map& dynamic_options) { + const std::map& dynamic_options, + const SchemaManager* shared_schema_manager) { 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,10 +244,13 @@ 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()); + PAIMON_ASSIGN_OR_RAISE( + std::optional> latest_schema, + shared_schema_manager + ? shared_schema_manager->Latest() + : SchemaManager(fs, parsed.table_path, + parsed.branch.value_or(BranchManager::DEFAULT_MAIN_BRANCH)) + .Latest()); if (!latest_schema) { return Status::NotExist("base table schema not found for system table path: ", path); } diff --git a/src/paimon/core/table/system/system_table.h b/src/paimon/core/table/system/system_table.h index 39e1da31b..5421bd1ab 100644 --- a/src/paimon/core/table/system/system_table.h +++ b/src/paimon/core/table/system/system_table.h @@ -33,6 +33,7 @@ namespace paimon { class FileSystem; class ReadContext; class ScanContext; +class SchemaManager; class TableScan; class TableRead; class TableSchema; @@ -80,7 +81,8 @@ class SystemTableLoader { static Result> LoadFromPath( const std::shared_ptr& fs, const std::string& path, - const std::map& dynamic_options); + const std::map& dynamic_options, + const SchemaManager* schema_manager); }; } // namespace paimon diff --git a/src/paimon/core/table/system/system_table_test.cpp b/src/paimon/core/table/system/system_table_test.cpp index 00a538f3b..ca3699540 100644 --- a/src/paimon/core/table/system/system_table_test.cpp +++ b/src/paimon/core/table/system/system_table_test.cpp @@ -278,8 +278,9 @@ TEST(SystemTableTest, TestNewReadPropagatesWarmupLevel) { TEST(SystemTableTest, TestGlobalSystemTableWithoutCatalogReturnsNotImplemented) { ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", "/tmp", {})); std::shared_ptr shared_fs(std::move(fs)); - ASSERT_NOK_WITH_MSG(SystemTableLoader::LoadFromPath(shared_fs, "/tmp/warehouse/sys/tables", {}), - "global system table requires catalog context: tables"); + ASSERT_NOK_WITH_MSG( + SystemTableLoader::LoadFromPath(shared_fs, "/tmp/warehouse/sys/tables", {}, nullptr), + "global system table requires catalog context: tables"); } TEST(SystemTableTest, TestScanMetricsAreSnapshots) { diff --git a/src/paimon/core/utils/snapshot_manager.cpp b/src/paimon/core/utils/snapshot_manager.cpp index 3e6c85eaa..e93af09a9 100644 --- a/src/paimon/core/utils/snapshot_manager.cpp +++ b/src/paimon/core/utils/snapshot_manager.cpp @@ -35,13 +35,52 @@ namespace paimon { +SnapshotManager::SnapshotCache::SnapshotCache() : SnapshotCache(std::chrono::steady_clock::now) {} + +SnapshotManager::SnapshotCache::SnapshotCache(Clock clock) : clock_(std::move(clock)) {} + +std::shared_ptr SnapshotManager::SnapshotCache::GetCache() { + std::lock_guard lock(mutex_); + auto now = clock_(); + if (!cache_ || now - created_at_ >= std::chrono::minutes(30)) { + cache_ = std::make_shared(Cache::Options{/*max_weight=*/20}); + created_at_ = now; + } + return cache_; +} + +Result SnapshotManager::SnapshotCache::Get( + const std::string& path, std::function(const std::string&)> supplier) { + // Hold this cache through the I/O. Invalidation replaces it instead of allowing an + // overlapping load to repopulate the cache used by subsequent callers. + auto cache = GetCache(); + return cache->Get(path, std::move(supplier)); +} + +Status SnapshotManager::SnapshotCache::Put(const std::string& path, const Snapshot& snapshot) { + return GetCache()->Put(path, snapshot); +} + +void SnapshotManager::SnapshotCache::InvalidateAll() { + std::lock_guard lock(mutex_); + cache_.reset(); +} + SnapshotManager::SnapshotManager(const std::shared_ptr& fs, const std::string& root_path) : SnapshotManager(fs, root_path, BranchManager::DEFAULT_MAIN_BRANCH) {} SnapshotManager::SnapshotManager(const std::shared_ptr& fs, const std::string& root_path, const std::string& branch) - : fs_(fs), root_path_(root_path), branch_(BranchManager::NormalizeBranch(branch)) {} + : SnapshotManager(fs, root_path, branch, nullptr) {} + +SnapshotManager::SnapshotManager(const std::shared_ptr& fs, + const std::string& root_path, const std::string& branch, + const std::shared_ptr& snapshot_cache) + : fs_(fs), + root_path_(root_path), + branch_(BranchManager::NormalizeBranch(branch)), + snapshot_cache_(snapshot_cache) {} SnapshotManager::~SnapshotManager() = default; @@ -60,6 +99,9 @@ const std::string& SnapshotManager::Branch() const { Result> SnapshotManager::LatestSnapshotOfUser(const std::string& user) { // Catalog snapshots may have no corresponding file in the table directory. PAIMON_ASSIGN_OR_RAISE(LatestSnapshotResult latest, LatestSnapshotWithSource()); + if (snapshot_cache_ && !latest.from_catalog && latest.snapshot) { + PAIMON_ASSIGN_OR_RAISE(latest.snapshot, LoadSnapshotFromFileSystem(latest.snapshot->Id())); + } return LatestSnapshotOfUserAtOrBefore(user, latest.snapshot, latest.from_catalog); } @@ -84,7 +126,7 @@ Result> SnapshotManager::LatestSnapshotOfUserAtOrBefore( } search_end = std::max(search_end, Snapshot::FIRST_SNAPSHOT_ID); for (int64_t id = latest.value().Id() - 1; id >= search_end; id--) { - Result snapshot = LoadSnapshot(id); + Result snapshot = LoadSnapshotFromFileSystem(id); if (!snapshot.ok()) { if (snapshot.status().IsNotExist()) { if (latest_from_catalog) { @@ -124,9 +166,38 @@ bool SnapshotManager::ExpiredSinceBoundaryWasRead(int64_t id) const { } Result SnapshotManager::LoadSnapshot(int64_t snapshot_id) const { + if (!snapshot_cache_) { + return LoadSnapshotFromFileSystem(snapshot_id); + } + return snapshot_cache_->Get(SnapshotPath(snapshot_id), [this](const std::string& path) { + return Snapshot::FromPath(fs_, path); + }); +} + +Result SnapshotManager::LoadSnapshotFromFileSystem(int64_t snapshot_id) const { return Snapshot::FromPath(fs_, SnapshotPath(snapshot_id)); } +Status SnapshotManager::DeleteSnapshot(int64_t snapshot_id) { + const std::string path = SnapshotPath(snapshot_id); + if (snapshot_cache_) { + snapshot_cache_->InvalidateAll(); + } + Status status = fs_->Delete(path); + // A read may start between the first invalidation and deletion. Invalidate again even on + // failure so neither cached entries nor in-flight loads from that window survive the call. + if (snapshot_cache_) { + snapshot_cache_->InvalidateAll(); + } + return status; +} + +void SnapshotManager::InvalidateCache() { + if (snapshot_cache_) { + snapshot_cache_->InvalidateAll(); + } +} + void SnapshotManager::SetSnapshotLoader(SnapshotLoader loader) { snapshot_loader_ = std::move(loader); } @@ -140,6 +211,10 @@ Result SnapshotManager::LatestSnapshotWit if (snapshot_loader_) { Result> loaded = snapshot_loader_(); if (loaded.ok()) { + if (snapshot_cache_ && loaded.value()) { + // Cache admission must not turn a successful catalog read into an error. + (void)snapshot_cache_->Put(SnapshotPath(loaded.value()->Id()), *loaded.value()); + } return LatestSnapshotResult{loaded.value(), /*from_catalog=*/true}; } if (!loaded.status().IsNotImplemented()) { @@ -347,7 +422,8 @@ Result> SnapshotManager::FindSnapshotBeforeTimestamp( return std::optional(); } - PAIMON_ASSIGN_OR_RAISE(Snapshot earliest_snapshot, LoadSnapshot(earliest_id.value())); + PAIMON_ASSIGN_OR_RAISE(Snapshot earliest_snapshot, + LoadSnapshotFromFileSystem(earliest_id.value())); if (!compare(earliest_snapshot.TimeMillis(), timestamp_millis)) { return std::optional(); } @@ -358,7 +434,7 @@ Result> SnapshotManager::FindSnapshotBeforeTimestamp( while (lo <= hi) { int64_t mid = lo + (hi - lo) / 2; - PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, LoadSnapshot(mid)); + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, LoadSnapshotFromFileSystem(mid)); if (compare(snapshot.TimeMillis(), timestamp_millis)) { lo = mid + 1; result = std::move(snapshot); diff --git a/src/paimon/core/utils/snapshot_manager.h b/src/paimon/core/utils/snapshot_manager.h index 5b346fe91..4429f0c84 100644 --- a/src/paimon/core/utils/snapshot_manager.h +++ b/src/paimon/core/utils/snapshot_manager.h @@ -18,14 +18,17 @@ #pragma once +#include #include #include #include +#include #include #include #include #include +#include "paimon/common/utils/generic_lru_cache.h" #include "paimon/core/snapshot.h" #include "paimon/result.h" #include "paimon/status.h" @@ -45,10 +48,36 @@ class SnapshotManager { /// Loads the catalog's latest snapshot. Null means no snapshot; only `NotImplemented` /// permits file-system fallback. using SnapshotLoader = std::function>()>; + /// Shared snapshot metadata cache. Replace the whole cache after 30 minutes or on + /// invalidation; in-flight loads may finish against the discarded cache only. + class SnapshotCache { + public: + using Clock = std::function; + + SnapshotCache(); + explicit SnapshotCache(Clock clock); + + Result Get(const std::string& path, + std::function(const std::string&)> supplier); + Status Put(const std::string& path, const Snapshot& snapshot); + void InvalidateAll(); + + private: + using Cache = GenericLruCache; + std::shared_ptr GetCache(); + + Clock clock_; + std::mutex mutex_; + std::chrono::steady_clock::time_point created_at_; + std::shared_ptr cache_; + }; SnapshotManager(const std::shared_ptr& fs, const std::string& root_path); SnapshotManager(const std::shared_ptr& fs, const std::string& root_path, const std::string& branch); + SnapshotManager(const std::shared_ptr& fs, const std::string& root_path, + const std::string& branch, + const std::shared_ptr& snapshot_cache); ~SnapshotManager(); /// Sets the loader for `LatestSnapshot()` and `LatestSnapshotId()`. @@ -86,7 +115,16 @@ class SnapshotManager { bool latest_from_catalog) const; Status CommitLatestHint(int64_t snapshot_id); Status CommitEarliestHint(int64_t snapshot_id); + /// Returns snapshot metadata, using the optional cache supplied at construction. + /// Managers without a cache always read the file system. + /// A cache hit does not guarantee that the snapshot or its data files still exist. Result LoadSnapshot(int64_t snapshot_id) const; + /// Bypasses the cache when file existence or current contents must be observed. + Result LoadSnapshotFromFileSystem(int64_t snapshot_id) const; + /// Deletes a snapshot and invalidates the entire injected cache, including concurrent loads. + Status DeleteSnapshot(int64_t snapshot_id); + /// Clears the injected cache. Does not invalidate metadata already held by active scans. + void InvalidateCache(); Result> EarliestSnapshotId() const; Result> LatestSnapshotId() const; /// Finds the latest snapshot published to the file system, without consulting the loader. @@ -126,6 +164,7 @@ class SnapshotManager { std::string root_path_; std::string branch_; SnapshotLoader snapshot_loader_; + const std::shared_ptr snapshot_cache_; }; } // namespace paimon diff --git a/src/paimon/core/utils/snapshot_manager_test.cpp b/src/paimon/core/utils/snapshot_manager_test.cpp index 984bbb137..8b5fcb7d7 100644 --- a/src/paimon/core/utils/snapshot_manager_test.cpp +++ b/src/paimon/core/utils/snapshot_manager_test.cpp @@ -18,8 +18,12 @@ #include "paimon/core/utils/snapshot_manager.h" +#include +#include #include #include +#include +#include #include #include #include @@ -37,6 +41,387 @@ namespace paimon::test { +namespace { + +class SnapshotCountingFileSystem : public LocalFileSystem { + public: + Status ReadFile(const std::string& path, std::string* content) override { + if (path.find("/snapshot/snapshot-") != std::string::npos) { + snapshot_reads.fetch_add(1); + if (fail_snapshot_read.exchange(false)) { + return Status::IOError("injected snapshot read failure"); + } + } + PAIMON_RETURN_NOT_OK(LocalFileSystem::ReadFile(path, content)); + if (path.find("/snapshot/snapshot-") != std::string::npos && after_snapshot_read) { + after_snapshot_read(); + } + return Status::OK(); + } + + Status Delete(const std::string& path, bool recursive = true) const override { + if (before_snapshot_delete) { + before_snapshot_delete(); + } + if (fail_snapshot_delete) { + return Status::IOError("injected snapshot delete failure"); + } + return LocalFileSystem::Delete(path, recursive); + } + + std::atomic snapshot_reads{0}; + std::atomic fail_snapshot_read{false}; + bool fail_snapshot_delete = false; + std::function after_snapshot_read; + std::function before_snapshot_delete; +}; + +} // namespace + +class SnapshotManagerCacheTest : public testing::Test { + protected: + void SetUp() override { + directory_ = UniqueTestDirectory::Create(); + ASSERT_TRUE(directory_); + cache_ = std::make_shared(); + manager_ = std::make_unique(fs_, directory_->Str(), "main", cache_); + ASSERT_OK(fs_->Mkdirs(manager_->SnapshotDirectory())); + } + + Status WriteSnapshot(const SnapshotManager& manager, int64_t id, + const std::string& user = "user") { + Snapshot snapshot(id, /*schema_id=*/0, /*base_manifest_list=*/"base", + /*base_manifest_list_size=*/std::nullopt, /*delta_manifest_list=*/"delta", + /*delta_manifest_list_size=*/std::nullopt, + /*changelog_manifest_list=*/std::nullopt, + /*changelog_manifest_list_size=*/std::nullopt, + /*index_manifest=*/std::nullopt, user, /*commit_identifier=*/id, + Snapshot::CommitKind::Append(), /*time_millis=*/id, + /*total_record_count=*/id, /*delta_record_count=*/1, + /*changelog_record_count=*/std::nullopt, /*watermark=*/std::nullopt, + /*statistics=*/std::nullopt, /*properties=*/std::nullopt, + /*next_row_id=*/std::nullopt); + PAIMON_ASSIGN_OR_RAISE(std::string json, snapshot.ToJsonString()); + return fs_->WriteFile(manager.SnapshotPath(id), json, /*overwrite=*/true); + } + + std::unique_ptr directory_; + std::shared_ptr fs_ = + std::make_shared(); + std::unique_ptr manager_; + std::shared_ptr cache_; +}; + +TEST_F(SnapshotManagerCacheTest, LatestSnapshotAdvancesWithCachedHistory) { + ASSERT_OK(WriteSnapshot(*manager_, 1)); + ASSERT_OK(manager_->CommitLatestHint(1)); + ASSERT_OK_AND_ASSIGN(std::optional first, manager_->LatestSnapshot()); + ASSERT_TRUE(first); + ASSERT_EQ(first->Id(), 1); + ASSERT_OK_AND_ASSIGN(std::optional repeated, manager_->LatestSnapshot()); + ASSERT_TRUE(repeated); + ASSERT_EQ(*first, *repeated); + ASSERT_EQ(fs_->snapshot_reads.load(), 1); + + // The new snapshot is visible before its LATEST hint is updated. + ASSERT_OK(WriteSnapshot(*manager_, 2)); + ASSERT_OK_AND_ASSIGN(std::optional latest, manager_->LatestSnapshot()); + ASSERT_TRUE(latest); + ASSERT_EQ(latest->Id(), 2); + ASSERT_EQ(fs_->snapshot_reads.load(), 2); + ASSERT_OK(manager_->CommitLatestHint(2)); + ASSERT_OK_AND_ASSIGN(Snapshot historical, manager_->LoadSnapshot(1)); + ASSERT_EQ(historical, *first); + ASSERT_OK_AND_ASSIGN(latest, manager_->LatestSnapshot()); + ASSERT_TRUE(latest); + ASSERT_EQ(latest->Id(), 2); + ASSERT_EQ(fs_->snapshot_reads.load(), 2); +} + +TEST_F(SnapshotManagerCacheTest, MissingMalformedAndFailedReadsAreRetried) { + ASSERT_NOK(manager_->LoadSnapshot(1)); + ASSERT_OK(fs_->WriteFile(manager_->SnapshotPath(1), "invalid JSON", true)); + ASSERT_NOK(manager_->LoadSnapshot(1)); + ASSERT_OK(WriteSnapshot(*manager_, 1)); + fs_->fail_snapshot_read = true; + ASSERT_NOK_WITH_MSG(manager_->LoadSnapshot(1), "injected snapshot read failure"); + ASSERT_OK_AND_ASSIGN(Snapshot snapshot, manager_->LoadSnapshot(1)); + ASSERT_EQ(snapshot.Id(), 1); + ASSERT_EQ(fs_->snapshot_reads.load(), 4); + ASSERT_OK(manager_->LoadSnapshot(1)); + ASSERT_EQ(fs_->snapshot_reads.load(), 4); +} + +TEST_F(SnapshotManagerCacheTest, EvictsLeastRecentlyUsedSnapshot) { + constexpr int64_t kCapacity = 20; + for (int64_t id = 1; id <= kCapacity; ++id) { + ASSERT_OK(WriteSnapshot(*manager_, id)); + ASSERT_OK(manager_->LoadSnapshot(id)); + } + ASSERT_EQ(fs_->snapshot_reads.load(), kCapacity); + ASSERT_OK(manager_->LoadSnapshot(1)); + ASSERT_EQ(fs_->snapshot_reads.load(), kCapacity); + ASSERT_OK(WriteSnapshot(*manager_, kCapacity + 1)); + ASSERT_OK(manager_->LoadSnapshot(kCapacity + 1)); + ASSERT_OK(manager_->LoadSnapshot(1)); + ASSERT_EQ(fs_->snapshot_reads.load(), kCapacity + 1); + ASSERT_OK_AND_ASSIGN(Snapshot reloaded, manager_->LoadSnapshot(2)); + ASSERT_EQ(reloaded.Id(), 2); + ASSERT_EQ(fs_->snapshot_reads.load(), kCapacity + 2); +} + +TEST_F(SnapshotManagerCacheTest, CachedMetadataDoesNotMakeExpiredSnapshotExist) { + ASSERT_OK(WriteSnapshot(*manager_, 1)); + ASSERT_OK(WriteSnapshot(*manager_, 2)); + ASSERT_OK(manager_->CommitEarliestHint(1)); + ASSERT_OK(manager_->CommitLatestHint(2)); + ASSERT_OK_AND_ASSIGN(Snapshot cached, manager_->LoadSnapshot(1)); + ASSERT_OK(fs_->Delete(manager_->SnapshotPath(1))); + ASSERT_OK(manager_->CommitEarliestHint(2)); + ASSERT_OK_AND_ASSIGN(bool exists, manager_->SnapshotExists(1)); + ASSERT_FALSE(exists); + ASSERT_OK_AND_ASSIGN(std::optional earliest, manager_->EarliestSnapshotId()); + ASSERT_EQ(earliest, 2); + ASSERT_OK_AND_ASSIGN(Snapshot historical, manager_->LoadSnapshot(1)); + ASSERT_EQ(historical, cached); + ASSERT_EQ(fs_->snapshot_reads.load(), 1); + SnapshotManager fresh(fs_, directory_->Str()); + ASSERT_NOK(fresh.LoadSnapshot(1)); +} + +TEST_F(SnapshotManagerCacheTest, RecreatedManagerLoadsReplacedSnapshot) { + ASSERT_OK(WriteSnapshot(*manager_, 1, "before")); + ASSERT_OK_AND_ASSIGN(Snapshot first, manager_->LoadSnapshot(1)); + ASSERT_EQ(first.CommitUser(), "before"); + // Fast-forward can replace contents under the same snapshot ID. + ASSERT_OK(WriteSnapshot(*manager_, 1, "after")); + ASSERT_OK_AND_ASSIGN(Snapshot cached, manager_->LoadSnapshot(1)); + ASSERT_EQ(cached.CommitUser(), "before"); + SnapshotManager fresh(fs_, directory_->Str()); + ASSERT_OK_AND_ASSIGN(Snapshot reloaded, fresh.LoadSnapshot(1)); + ASSERT_EQ(reloaded.CommitUser(), "after"); +} + +TEST_F(SnapshotManagerCacheTest, TablesAndBranchesKeepSeparateCaches) { + SnapshotManager branch(fs_, directory_->Str(), "dev", cache_); + SnapshotManager other(fs_, PathUtil::JoinPath(directory_->Str(), "other"), "main", cache_); + ASSERT_OK(fs_->Mkdirs(branch.SnapshotDirectory())); + ASSERT_OK(fs_->Mkdirs(other.SnapshotDirectory())); + ASSERT_OK(WriteSnapshot(*manager_, 1, "main")); + ASSERT_OK(WriteSnapshot(branch, 1, "dev")); + ASSERT_OK(WriteSnapshot(other, 1, "other")); + for (int32_t i = 0; i < 2; ++i) { + ASSERT_OK_AND_ASSIGN(Snapshot main_snapshot, manager_->LoadSnapshot(1)); + ASSERT_EQ(main_snapshot.CommitUser(), "main"); + ASSERT_OK_AND_ASSIGN(Snapshot branch_snapshot, branch.LoadSnapshot(1)); + ASSERT_EQ(branch_snapshot.CommitUser(), "dev"); + ASSERT_OK_AND_ASSIGN(Snapshot other_snapshot, other.LoadSnapshot(1)); + ASSERT_EQ(other_snapshot.CommitUser(), "other"); + } + ASSERT_EQ(fs_->snapshot_reads.load(), 3); +} + +TEST_F(SnapshotManagerCacheTest, ConcurrentLoadsReturnCompleteSnapshots) { + ASSERT_OK(WriteSnapshot(*manager_, 1)); + std::promise start; + std::shared_future ready = start.get_future().share(); + std::vector>> futures; + for (int32_t i = 0; i < 8; ++i) { + futures.push_back(std::async(std::launch::async, [this, ready]() { + ready.wait(); + return manager_->LoadSnapshot(1); + })); + } + start.set_value(); + ASSERT_OK_AND_ASSIGN(Snapshot expected, futures.front().get()); + ASSERT_EQ(expected.Id(), 1); + for (size_t i = 1; i < futures.size(); ++i) { + ASSERT_OK_AND_ASSIGN(Snapshot actual, futures[i].get()); + ASSERT_EQ(actual, expected); + } + int32_t reads = fs_->snapshot_reads.load(); + ASSERT_GE(reads, 1); + ASSERT_OK(manager_->LoadSnapshot(1)); + ASSERT_EQ(fs_->snapshot_reads.load(), reads); +} + +TEST_F(SnapshotManagerCacheTest, DefaultManagerDoesNotCacheSnapshots) { + SnapshotManager manager(fs_, directory_->Str()); + ASSERT_OK(WriteSnapshot(manager, 1, "before")); + ASSERT_OK(manager.LoadSnapshot(1)); + ASSERT_OK(WriteSnapshot(manager, 1, "after")); + ASSERT_OK_AND_ASSIGN(Snapshot reloaded, manager.LoadSnapshot(1)); + ASSERT_EQ(reloaded.CommitUser(), "after"); + ASSERT_OK(fs_->Delete(manager.SnapshotPath(1))); + ASSERT_TRUE(manager.LoadSnapshot(1).status().IsNotExist()); +} + +TEST_F(SnapshotManagerCacheTest, DeleteAndInvalidateClearSharedCache) { + SnapshotManager other(fs_, directory_->Str(), "main", cache_); + ASSERT_OK(WriteSnapshot(*manager_, 1, "before")); + ASSERT_OK(other.LoadSnapshot(1)); + ASSERT_OK(WriteSnapshot(*manager_, 2, "before")); + ASSERT_OK(other.LoadSnapshot(2)); + ASSERT_OK(WriteSnapshot(*manager_, 2, "after")); + ASSERT_OK(manager_->DeleteSnapshot(1)); + ASSERT_TRUE(other.LoadSnapshot(1).status().IsNotExist()); + // Deleting one snapshot also discards the other entries in this shared cache. + ASSERT_OK_AND_ASSIGN(Snapshot retained, other.LoadSnapshot(2)); + ASSERT_EQ(retained.CommitUser(), "after"); + ASSERT_OK(WriteSnapshot(*manager_, 1, "after")); + ASSERT_OK_AND_ASSIGN(Snapshot reloaded, other.LoadSnapshot(1)); + ASSERT_EQ(reloaded.CommitUser(), "after"); + ASSERT_OK(WriteSnapshot(*manager_, 1, "replaced")); + manager_->InvalidateCache(); + ASSERT_OK_AND_ASSIGN(reloaded, other.LoadSnapshot(1)); + ASSERT_EQ(reloaded.CommitUser(), "replaced"); +} + +TEST_F(SnapshotManagerCacheTest, DeletePreventsRefillByOverlappingReads) { + ASSERT_OK(WriteSnapshot(*manager_, 1)); + bool deleted = false; + bool read_during_delete = false; + Status delete_status; + fs_->after_snapshot_read = [&]() { + if (!deleted) { + deleted = true; + delete_status = manager_->DeleteSnapshot(1); + } + }; + fs_->before_snapshot_delete = [&]() { + // Interleave a second read between invalidation and file deletion. + read_during_delete = manager_->LoadSnapshot(1).ok(); + }; + // This read already obtained its contents before deletion, but must not refill the cache. + ASSERT_OK(manager_->LoadSnapshot(1)); + ASSERT_TRUE(deleted); + ASSERT_TRUE(read_during_delete); + ASSERT_OK(delete_status); + ASSERT_TRUE(manager_->LoadSnapshot(1).status().IsNotExist()); +} + +TEST_F(SnapshotManagerCacheTest, FailedDeleteStillInvalidatesCachedContents) { + ASSERT_OK(WriteSnapshot(*manager_, 1, "before")); + ASSERT_OK(manager_->LoadSnapshot(1)); + ASSERT_OK(WriteSnapshot(*manager_, 1, "after")); + fs_->fail_snapshot_delete = true; + ASSERT_NOK_WITH_MSG(manager_->DeleteSnapshot(1), "injected snapshot delete failure"); + ASSERT_OK_AND_ASSIGN(Snapshot reloaded, manager_->LoadSnapshot(1)); + ASSERT_EQ(reloaded.CommitUser(), "after"); +} + +TEST_F(SnapshotManagerCacheTest, InvalidationPreventsRefillByOverlappingRead) { + ASSERT_OK(WriteSnapshot(*manager_, 1, "before")); + fs_->after_snapshot_read = [&]() { manager_->InvalidateCache(); }; + ASSERT_OK(manager_->LoadSnapshot(1)); + fs_->after_snapshot_read = nullptr; + ASSERT_OK(WriteSnapshot(*manager_, 1, "after")); + ASSERT_OK_AND_ASSIGN(Snapshot reloaded, manager_->LoadSnapshot(1)); + ASSERT_EQ(reloaded.CommitUser(), "after"); +} + +TEST_F(SnapshotManagerCacheTest, UserLookupDoesNotReturnDeletedCachedLatest) { + ASSERT_OK(WriteSnapshot(*manager_, 1)); + ASSERT_OK(manager_->CommitLatestHint(1)); + ASSERT_OK(manager_->LoadSnapshot(1)); + ASSERT_OK(fs_->Delete(manager_->SnapshotPath(1))); + ASSERT_TRUE(manager_->LatestSnapshotOfUser("user").status().IsNotExist()); +} + +TEST_F(SnapshotManagerCacheTest, WarmCacheDoesNotHideCatalogHistoryGap) { + ASSERT_OK(WriteSnapshot(*manager_, 1, "target-user")); + ASSERT_OK(WriteSnapshot(*manager_, 2, "other-user")); + ASSERT_OK(WriteSnapshot(*manager_, 3, "other-user")); + ASSERT_OK(manager_->CommitEarliestHint(1)); + ASSERT_OK_AND_ASSIGN(Snapshot latest, manager_->LoadSnapshot(3)); + ASSERT_OK(manager_->LoadSnapshot(2)); + ASSERT_OK(fs_->Delete(manager_->SnapshotPath(2))); + manager_->SetSnapshotLoader( + [latest]() -> Result> { return std::optional(latest); }); + ASSERT_NOK_WITH_MSG(manager_->LatestSnapshotOfUser("target-user"), + "is not under the table directory"); + ASSERT_OK(manager_->CommitEarliestHint(3)); + ASSERT_OK_AND_ASSIGN(std::optional expired, + manager_->LatestSnapshotOfUser("target-user")); + ASSERT_FALSE(expired); +} + +TEST_F(SnapshotManagerCacheTest, TimestampSearchDoesNotUseMissingCachedSnapshot) { + for (int64_t id : {1, 2, 3}) { + ASSERT_OK(WriteSnapshot(*manager_, id)); + ASSERT_OK(manager_->LoadSnapshot(id)); + } + ASSERT_OK(manager_->CommitEarliestHint(1)); + ASSERT_OK(manager_->CommitLatestHint(3)); + ASSERT_OK(fs_->Delete(manager_->SnapshotPath(2))); + ASSERT_TRUE(manager_->EarlierOrEqualTimeMillis(2).status().IsNotExist()); + ASSERT_TRUE(manager_->EarlierThanTimeMillis(3).status().IsNotExist()); +} + +TEST_F(SnapshotManagerCacheTest, CatalogReadRefreshesCachedContents) { + ASSERT_OK(WriteSnapshot(*manager_, 1, "before")); + ASSERT_OK(manager_->LoadSnapshot(1)); + ASSERT_OK(WriteSnapshot(*manager_, 1, "after")); + ASSERT_OK_AND_ASSIGN(Snapshot latest, manager_->LoadSnapshotFromFileSystem(1)); + manager_->SetSnapshotLoader( + [latest]() -> Result> { return std::optional(latest); }); + ASSERT_OK(manager_->LatestSnapshot()); + ASSERT_OK_AND_ASSIGN(Snapshot cached, manager_->LoadSnapshot(1)); + ASSERT_EQ(cached.CommitUser(), "after"); +} + +TEST_F(SnapshotManagerCacheTest, ExpirationReplacesWholeCacheDespiteReadsAndWrites) { + std::chrono::steady_clock::time_point now{}; + auto cache = std::make_shared([&now]() { return now; }); + SnapshotManager manager(fs_, directory_->Str(), "main", cache); + SnapshotManager other(fs_, directory_->Str(), "main", cache); + ASSERT_OK(WriteSnapshot(manager, 1, "before")); + ASSERT_OK(manager.LoadSnapshot(1)); + ASSERT_OK(WriteSnapshot(manager, 1, "after")); + for (int32_t i = 0; i < 5; ++i) { + now += std::chrono::minutes(5); + ASSERT_OK_AND_ASSIGN(Snapshot cached, manager.LoadSnapshot(1)); + ASSERT_EQ(cached.CommitUser(), "before"); + // Neither reads nor catalog cache writes may postpone the whole-cache deadline. + ASSERT_OK(cache->Put(manager.SnapshotPath(1), cached)); + } + // Even a recently added entry is discarded with the rest of this cache. + ASSERT_OK(WriteSnapshot(manager, 2, "before")); + ASSERT_OK(other.LoadSnapshot(2)); + ASSERT_OK(WriteSnapshot(manager, 2, "after")); + now += std::chrono::minutes(5); + ASSERT_OK_AND_ASSIGN(Snapshot reloaded, manager.LoadSnapshot(1)); + ASSERT_EQ(reloaded.CommitUser(), "after"); + ASSERT_OK_AND_ASSIGN(reloaded, other.LoadSnapshot(2)); + ASSERT_EQ(reloaded.CommitUser(), "after"); + ASSERT_OK(fs_->Delete(manager.SnapshotPath(1))); + now += std::chrono::minutes(30); + ASSERT_TRUE(manager.LoadSnapshot(1).status().IsNotExist()); +} + +TEST_F(SnapshotManagerCacheTest, ExpirationPreventsRefillByOverlappingRead) { + std::chrono::steady_clock::time_point now{}; + auto cache = std::make_shared([&now]() { return now; }); + SnapshotManager manager(fs_, directory_->Str(), "main", cache); + ASSERT_OK(WriteSnapshot(manager, 1, "before")); + ASSERT_OK(WriteSnapshot(manager, 2)); + bool expired = false; + fs_->after_snapshot_read = [&]() { + if (!expired) { + expired = true; + now += std::chrono::minutes(30); + // A different load starts a new cache while the first load still holds the old one. + ASSERT_OK(manager.LoadSnapshot(2)); + } + }; + ASSERT_OK_AND_ASSIGN(Snapshot old, manager.LoadSnapshot(1)); + ASSERT_EQ(old.CommitUser(), "before"); + fs_->after_snapshot_read = nullptr; + ASSERT_OK(WriteSnapshot(manager, 1, "after")); + ASSERT_OK_AND_ASSIGN(Snapshot reloaded, manager.LoadSnapshot(1)); + ASSERT_EQ(reloaded.CommitUser(), "after"); +} + TEST(SnapshotManagerTest, TestSnapshotDirectory) { auto fs = std::make_shared(); SnapshotManager manager(fs, paimon::test::GetDataDir() + "/append_09.db/append_09");