From 430ec654acc8a1c9eca4f7cf85963ac79aecd368 Mon Sep 17 00:00:00 2001 From: fastio Date: Mon, 2 Nov 2026 14:31:18 +0800 Subject: [PATCH] fix(system): honor scan.snapshot-id in metadata system tables --- .../table/system/metadata_system_tables.cpp | 32 ++++++++--- test/inte/read_inte_test.cpp | 55 +++++++++++++++++++ 2 files changed, 78 insertions(+), 9 deletions(-) diff --git a/src/paimon/core/table/system/metadata_system_tables.cpp b/src/paimon/core/table/system/metadata_system_tables.cpp index 5279e6871..8db1c41cf 100644 --- a/src/paimon/core/table/system/metadata_system_tables.cpp +++ b/src/paimon/core/table/system/metadata_system_tables.cpp @@ -219,9 +219,21 @@ Result> CreatePathFactory( return path_factory; } -Result> LatestSnapshot(const MetadataSystemTableContext& context) { +// Reads the snapshot pinned by `scan.snapshot-id`, or the latest snapshot when the option is unset. +// A pinned snapshot that does not exist is an error instead of a fallback to the latest snapshot. +Result> SelectedSnapshot(const MetadataSystemTableContext& context, + const CoreOptions& core_options) { SnapshotManager snapshot_manager(context.fs, context.table_path, context.branch); - return snapshot_manager.LatestSnapshot(); + std::optional snapshot_id = core_options.GetScanSnapshotId(); + if (snapshot_id == std::nullopt) { + return snapshot_manager.LatestSnapshot(); + } + PAIMON_ASSIGN_OR_RAISE(bool exists, snapshot_manager.SnapshotExists(snapshot_id.value())); + if (!exists) { + return Status::NotExist(fmt::format("snapshot {} does not exist", snapshot_id.value())); + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager.LoadSnapshot(snapshot_id.value())); + return std::optional(std::move(snapshot)); } Result> ReadDataManifests( @@ -255,11 +267,12 @@ Result> CreateManifestFile( partition_schema); } -Result> ReadLatestManifestEntries( +Result> ReadSelectedManifestEntries( const MetadataSystemTableContext& context, const std::shared_ptr& path_factory, const CoreOptions& core_options, const std::shared_ptr& pool) { - PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, LatestSnapshot(context)); + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, + SelectedSnapshot(context, core_options)); if (!snapshot) { return std::vector(); } @@ -276,12 +289,12 @@ Result> ReadLatestManifestEntries( return entries; } -Result> ReadLatestDataFiles( +Result> ReadSelectedDataFiles( const MetadataSystemTableContext& context, const std::shared_ptr& path_factory, const CoreOptions& core_options, const std::shared_ptr& pool) { PAIMON_ASSIGN_OR_RAISE(std::vector entries, - ReadLatestManifestEntries(context, path_factory, core_options, pool)); + ReadSelectedManifestEntries(context, path_factory, core_options, pool)); std::vector merged_entries; PAIMON_RETURN_NOT_OK(FileEntry::MergeEntries(entries, &merged_entries)); return merged_entries; @@ -798,13 +811,14 @@ Result> ManifestsSystemTable::ArrowSchema() const Result> ManifestsSystemTable::BuildRows() const { PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, ArrowSchema()); - PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, LatestSnapshot(context_)); + PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, CreateCoreOptions(context_)); + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, + SelectedSnapshot(context_, core_options)); if (!snapshot) { return std::vector(); } std::shared_ptr pool = GetDefaultPool(); - PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, CreateCoreOptions(context_)); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr path_factory, CreatePathFactory(context_, core_options, pool)); PAIMON_ASSIGN_OR_RAISE( @@ -884,7 +898,7 @@ Result> FilesSystemTable::BuildRows() const { PAIMON_ASSIGN_OR_RAISE(std::shared_ptr path_factory, CreatePathFactory(context_, core_options, pool)); PAIMON_ASSIGN_OR_RAISE(std::vector entries, - ReadLatestDataFiles(context_, path_factory, core_options, pool)); + ReadSelectedDataFiles(context_, path_factory, core_options, pool)); std::shared_ptr arrow_schema = DataField::ConvertDataFieldsToArrowSchema(context_.table_schema->Fields()); PAIMON_ASSIGN_OR_RAISE( diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index 284cd362f..6f3aa0bea 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -867,6 +867,61 @@ TEST(SystemTableReadInteTest, TestReadMetadataSystemTables) { ASSERT_FALSE(creation_time_array->IsNull(0)); } +TEST(SystemTableReadInteTest, TestReadMetadataSystemTablesAtPinnedSnapshot) { + arrow::FieldVector fields = { + arrow::field("pk", arrow::utf8()), + arrow::field("v", arrow::int32()), + }; + auto schema = arrow::schema(fields); + std::map options = { + {Options::FILE_SYSTEM, "local"}, {Options::FILE_FORMAT, "parquet"}, {Options::BUCKET, "1"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + ASSERT_OK_AND_ASSIGN(auto helper, TestHelper::Create(dir->Str(), schema, + /*partition_keys=*/{}, + /*primary_keys=*/{"pk"}, options, + /*is_streaming_mode=*/true)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_1, + TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([["a", 1]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch_1), /*commit_identifier=*/0, + /*expected_commit_messages=*/std::nullopt)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch_2, + TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([["b", 2]])", + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch_2), /*commit_identifier=*/1, + /*expected_commit_messages=*/std::nullopt)); + std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); + + ASSERT_OK_AND_ASSIGN(auto latest_files, ReadSystemTable(table_path + "$files", options)); + ASSERT_EQ(SingleStructChunk(latest_files)->length(), 2); + ASSERT_OK_AND_ASSIGN(auto latest_manifests, + ReadSystemTable(table_path + "$manifests", options)); + const int64_t latest_manifest_count = SingleStructChunk(latest_manifests)->length(); + + std::map pinned_options = options; + pinned_options[Options::SCAN_SNAPSHOT_ID] = "1"; + ASSERT_OK_AND_ASSIGN(auto pinned_files, ReadSystemTable(table_path + "$files", pinned_options)); + auto pinned_files_array = SingleStructChunk(pinned_files); + ASSERT_EQ(pinned_files_array->length(), 1); + auto record_count_array = + std::dynamic_pointer_cast(pinned_files_array->field(6)); + ASSERT_TRUE(record_count_array); + ASSERT_EQ(record_count_array->Value(0), 1); + ASSERT_OK_AND_ASSIGN(auto pinned_manifests, + ReadSystemTable(table_path + "$manifests", pinned_options)); + ASSERT_EQ(SingleStructChunk(pinned_manifests)->length(), 1); + ASSERT_LT(SingleStructChunk(pinned_manifests)->length(), latest_manifest_count); + + // A pinned snapshot that does not exist must not fall back to the latest snapshot. + std::map missing_options = options; + missing_options[Options::SCAN_SNAPSHOT_ID] = "3"; + ASSERT_NOK_WITH_MSG(ReadSystemTable(table_path + "$files", missing_options), + "snapshot 3 does not exist"); + ASSERT_NOK_WITH_MSG(ReadSystemTable(table_path + "$manifests", missing_options), + "snapshot 3 does not exist"); +} + TEST(SystemTableReadInteTest, TestReadOptimizedSystemTable) { arrow::FieldVector fields = { arrow::field("k", arrow::int32()),