diff --git a/include/paimon/file_store_commit.h b/include/paimon/file_store_commit.h index 9d31688d1..d3c699d0f 100644 --- a/include/paimon/file_store_commit.h +++ b/include/paimon/file_store_commit.h @@ -135,6 +135,30 @@ class PAIMON_EXPORT FileStoreCommit { virtual Status DropPartition(const std::vector>& partitions, int64_t commit_identifier) = 0; + /// Truncate the whole table by overwriting all partitions with empty data. The generated + /// snapshot has commit kind OVERWRITE. + /// + /// @param commit_identifier An identifier for the commit operation. + /// @return Status indicating the success or failure of the truncate operation. + virtual Status TruncateTable(int64_t commit_identifier) = 0; + + /// Abort an unsuccessful commit. The data and index files described by the given commit + /// messages will be deleted on a best-effort basis (delete failures are ignored). + /// + /// @param commit_messages A vector of commit messages whose files should be cleaned up. + /// @return Status indicating the success or failure of the abort operation. + virtual Status Abort(const std::vector>& commit_messages) = 0; + + /// Roll back to the target snapshot and materialize it as the latest snapshot. + /// + /// Reads the surviving files of both the current latest snapshot and the target + /// snapshot, then commits an OVERWRITE snapshot whose visible state equals the target. + /// + /// @param target_snapshot_id The snapshot id to roll back to. + /// @return Result; true if the atomic commit succeeded. Returns an error status if + /// there is no latest snapshot or the target snapshot does not exist. + virtual Result RollbackToAsLatest(int64_t target_snapshot_id) = 0; + /// Configure row-id conflict checking from a specific snapshot id. /// /// If set to a snapshot id, commit conflict detection will additionally validate row-id diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 7219310d6..457970681 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -720,6 +720,7 @@ if(PAIMON_BUILD_TESTS) core/operation/commit/overwrite_changes_provider_test.cpp core/operation/commit/row_id_column_conflict_checker_test.cpp core/operation/commit/row_tracking_commit_utils_test.cpp + core/operation/commit/sequence_snapshot_properties_test.cpp core/operation/commit/retry_waiter_test.cpp core/operation/key_value_file_store_write_test.cpp core/operation/internal_read_context_test.cpp diff --git a/src/paimon/common/utils/vector_store_utils.h b/src/paimon/common/utils/vector_store_utils.h new file mode 100644 index 000000000..9f20893ac --- /dev/null +++ b/src/paimon/common/utils/vector_store_utils.h @@ -0,0 +1,37 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed 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 + +namespace paimon { + +/// Utils for vector-store files. +/// +/// A vector-store file is identified by the `.vector.` marker in its name. +class VectorStoreUtils { + public: + VectorStoreUtils() = delete; + ~VectorStoreUtils() = delete; + + /// Returns true if `file_name` is a vector-store file (contains the `.vector.` marker). + static bool IsVectorStoreFile(const std::string& file_name) { + return file_name.find(".vector.") != std::string::npos; + } +}; + +} // namespace paimon diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp index 03a67afc1..615a606c2 100644 --- a/src/paimon/core/core_options.cpp +++ b/src/paimon/core/core_options.cpp @@ -686,7 +686,7 @@ struct CoreOptions::Impl { Options::SEQUENCE_FIELD, Options::FIELDS_SEPARATOR, &sequence_field)); // Parse sequence.field.sort-order - order of sequence field, default "ascending" PAIMON_RETURN_NOT_OK(parser.ParseSortOrder(&sequence_field_sort_order)); - // Parse write-sequence-number-init-mode - sequence init mode for write path + // Parse write.sequence-number-init-mode - sequence init mode for write path std::string write_sequence_init_mode_str = "scan"; PAIMON_RETURN_NOT_OK( parser.Parse(Options::WRITE_SEQUENCE_NUMBER_INIT_MODE, &write_sequence_init_mode_str)); diff --git a/src/paimon/core/io/append_data_file_writer_factory.cpp b/src/paimon/core/io/append_data_file_writer_factory.cpp index a486216c0..81baff3f8 100644 --- a/src/paimon/core/io/append_data_file_writer_factory.cpp +++ b/src/paimon/core/io/append_data_file_writer_factory.cpp @@ -40,10 +40,13 @@ AppendDataFileWriterFactory::AppendDataFileWriterFactory( file_source_(file_source), path_factory_(path_factory) {} +std::shared_ptr AppendDataFileWriterFactory::ResolveSeqNumCounter() const { + return options_.DataEvolutionEnabled() ? std::make_shared(0) : seq_num_counter_; +} + Result>>> AppendDataFileWriterFactory::CreateWriter() const { - std::shared_ptr seq_num_counter = - options_.DataEvolutionEnabled() ? std::make_shared(0) : seq_num_counter_; + std::shared_ptr seq_num_counter = ResolveSeqNumCounter(); PAIMON_ASSIGN_OR_RAISE(WriterResources resources, CreateWriterResources(*options_.GetFileFormat(), write_schema_, /*create_stats_extractor=*/true)); diff --git a/src/paimon/core/io/append_data_file_writer_factory.h b/src/paimon/core/io/append_data_file_writer_factory.h index 9776d4933..57b4e32c2 100644 --- a/src/paimon/core/io/append_data_file_writer_factory.h +++ b/src/paimon/core/io/append_data_file_writer_factory.h @@ -57,6 +57,11 @@ class AppendDataFileWriterFactory CreateWriter() const override; protected: + // Resolves the sequence-number counter for a newly created writer. When data evolution is + // enabled each file gets a fresh counter starting at 0; otherwise the factory's shared counter + // is reused. + std::shared_ptr ResolveSeqNumCounter() const; + std::shared_ptr write_schema_; std::optional> write_cols_; std::shared_ptr seq_num_counter_; diff --git a/src/paimon/core/io/shredding_append_data_file_writer_factory.cpp b/src/paimon/core/io/shredding_append_data_file_writer_factory.cpp index ef330af3f..ae72a8085 100644 --- a/src/paimon/core/io/shredding_append_data_file_writer_factory.cpp +++ b/src/paimon/core/io/shredding_append_data_file_writer_factory.cpp @@ -47,8 +47,7 @@ ShreddingAppendDataFileWriterFactory::CreateWriter() const { if (!shredding_context_) { return Status::Invalid("Shared-shredding append writer requires a shredding context."); } - std::shared_ptr seq_num_counter = - options_.DataEvolutionEnabled() ? std::make_shared(0) : seq_num_counter_; + std::shared_ptr seq_num_counter = ResolveSeqNumCounter(); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr converter, MapSharedShreddingBatchConverter::Create( write_schema_, shredding_context_, options_, pool_)); diff --git a/src/paimon/core/operation/commit/commit_changes_provider_test.cpp b/src/paimon/core/operation/commit/commit_changes_provider_test.cpp index 9ff458a3c..8c6daf99c 100644 --- a/src/paimon/core/operation/commit/commit_changes_provider_test.cpp +++ b/src/paimon/core/operation/commit/commit_changes_provider_test.cpp @@ -99,9 +99,9 @@ TEST(CommitChangesProviderTest, TestProvideReturnsGivenEntries) { ASSERT_EQ(delta_files.size(), provided_delta.size()); ASSERT_EQ(changelog_files.size(), provided_changelog.size()); ASSERT_EQ(index_entries.size(), provided_index.size()); - EXPECT_EQ("delta-1", provided_delta[0].FileName()); - EXPECT_EQ("changelog-1", provided_changelog[0].FileName()); - EXPECT_EQ("index-1", provided_index[0].index_file->FileName()); + ASSERT_EQ("delta-1", provided_delta[0].FileName()); + ASSERT_EQ("changelog-1", provided_changelog[0].FileName()); + ASSERT_EQ("index-1", provided_index[0].index_file->FileName()); } TEST(CommitChangesProviderTest, TestProvideUsesCopiedInputs) { @@ -123,7 +123,7 @@ TEST(CommitChangesProviderTest, TestProvideUsesCopiedInputs) { ASSERT_EQ(1u, provided->delta_files.size()); ASSERT_EQ(0u, provided->changelog_files.size()); ASSERT_EQ(0u, provided->index_entries.size()); - EXPECT_EQ("delta-1", provided->delta_files[0].FileName()); + ASSERT_EQ("delta-1", provided->delta_files[0].FileName()); } } // namespace paimon::test diff --git a/src/paimon/core/operation/commit/commit_scanner.h b/src/paimon/core/operation/commit/commit_scanner.h index daada81f1..e0fe08d21 100644 --- a/src/paimon/core/operation/commit/commit_scanner.h +++ b/src/paimon/core/operation/commit/commit_scanner.h @@ -50,7 +50,7 @@ class Snapshot; class SnapshotManager; class TableSchema; -// Manifest entries scanner for commit operations. +/// Manifest entries scanner for commit operations. class CommitScanner { public: using ScanSupplier = diff --git a/src/paimon/core/operation/commit/commit_scanner_test.cpp b/src/paimon/core/operation/commit/commit_scanner_test.cpp index 5eb73939f..a46364002 100644 --- a/src/paimon/core/operation/commit/commit_scanner_test.cpp +++ b/src/paimon/core/operation/commit/commit_scanner_test.cpp @@ -112,8 +112,8 @@ TEST_F(CommitScannerTest, TestReadAllEntriesFromChangedPartitionsEmptyFastExit) ASSERT_OK_AND_ASSIGN(std::vector entries, scanner.ReadAllEntriesFromChangedPartitions(MakeSnapshot(), /*changed_partitions=*/{})); - EXPECT_TRUE(entries.empty()); - EXPECT_FALSE(supplier_called); + ASSERT_TRUE(entries.empty()); + ASSERT_FALSE(supplier_called); } TEST_F(CommitScannerTest, TestReadAllEntriesFromPartitionsRequiresSupplier) { @@ -144,7 +144,7 @@ TEST_F(CommitScannerTest, TestReadAllEntriesFromChangedPartitionsBuildsScanFilte ASSERT_TRUE(supplier_called); ASSERT_EQ(1u, captured_partition_filters.size()); ASSERT_EQ(1u, captured_partition_filters[0].size()); - EXPECT_EQ("42", captured_partition_filters[0]["pt"]); + ASSERT_EQ("42", captured_partition_filters[0]["pt"]); } } // namespace paimon::test diff --git a/src/paimon/core/operation/commit/compacted_changelog_path_resolver.h b/src/paimon/core/operation/commit/compacted_changelog_path_resolver.h index 4ecb9dd7c..d42fe63e5 100644 --- a/src/paimon/core/operation/commit/compacted_changelog_path_resolver.h +++ b/src/paimon/core/operation/commit/compacted_changelog_path_resolver.h @@ -20,11 +20,41 @@ namespace paimon { +/// Utility class for resolving compacted changelog file paths. +/// +/// This class provides functionality to resolve fake compacted changelog file paths to their real +/// file paths. +/// +/// File Name Protocol +/// +/// There are two kinds of file name. In the following description, `bid1` and `bid2` are bucket +/// id, `off` is offset, `len1` and `len2` are lengths. +/// +/// - `bucket-bid1/compacted-changelog-xxx$bid1-len1`: This is the real file name. If this file +/// name is recorded in manifest file meta, reader should read the bytes of this file starting +/// from offset `0` with length `len1`. +/// - `bucket-bid2/compacted-changelog-xxx$bid1-len1-off-len2`: This is the fake file name. Reader +/// should read the bytes of file `bucket-bid1/compacted-changelog-xxx$bid1-len1` starting from +/// offset `off` with length `len2`. class CompactedChangelogPathResolver { public: - static bool IsCompactedChangelogPath(const std::string& path); - + /// Resolves a file path, handling compacted changelog file path resolution if applicable. + /// + /// For compacted changelog files, resolves fake file paths to their real file paths as + /// described in the protocol above. For non-compacted changelog files, returns the path + /// unchanged. + /// + /// @param path The file path to resolve. + /// @return The resolved real file path for compacted changelog files, or the original path + /// unchanged for other files. static std::string Resolve(const std::string& path); + + private: + /// Checks if the given path is a compacted changelog file path. + /// + /// @param path The file path to check. + /// @return true if the path is a compacted changelog file, false otherwise. + static bool IsCompactedChangelogPath(const std::string& path); }; } // namespace paimon diff --git a/src/paimon/core/operation/commit/conflict_detection.cpp b/src/paimon/core/operation/commit/conflict_detection.cpp index 63f302e78..ef900d2f6 100644 --- a/src/paimon/core/operation/commit/conflict_detection.cpp +++ b/src/paimon/core/operation/commit/conflict_detection.cpp @@ -32,6 +32,7 @@ #include "paimon/common/utils/binary_row_partition_computer.h" #include "paimon/common/utils/fields_comparator.h" #include "paimon/common/utils/range_helper.h" +#include "paimon/common/utils/vector_store_utils.h" #include "paimon/core/deletionvectors/deletion_vectors_index_file.h" #include "paimon/core/manifest/file_entry.h" #include "paimon/core/manifest/file_kind.h" @@ -55,12 +56,8 @@ namespace paimon { namespace { -bool IsVectorStoreFile(const std::string& file_name) { - return file_name.find(".vector.") != std::string::npos; -} - bool IsDedicatedStorageFile(const std::string& file_name) { - return BlobUtils::IsBlobFile(file_name) || IsVectorStoreFile(file_name); + return BlobUtils::IsBlobFile(file_name) || VectorStoreUtils::IsVectorStoreFile(file_name); } struct PartitionBucketKey { diff --git a/src/paimon/core/operation/commit/manifest_entry_changes_test.cpp b/src/paimon/core/operation/commit/manifest_entry_changes_test.cpp index 6fae6cf96..10449eca4 100644 --- a/src/paimon/core/operation/commit/manifest_entry_changes_test.cpp +++ b/src/paimon/core/operation/commit/manifest_entry_changes_test.cpp @@ -109,18 +109,18 @@ TEST_F(ManifestEntryChangesTest, TestCollectAndSummary) { ASSERT_EQ(1u, changes.compact_changelog.size()); ASSERT_EQ(2u, changes.compact_index_files.size()); - EXPECT_TRUE(changes.HasAppendChanges()); - EXPECT_FALSE(changes.HasGlobalIndexFileAdditions()); - EXPECT_TRUE(changes.HasCompactChanges()); + ASSERT_TRUE(changes.HasAppendChanges()); + ASSERT_FALSE(changes.HasGlobalIndexFileAdditions()); + ASSERT_TRUE(changes.HasCompactChanges()); - EXPECT_EQ(FileKind::Add(), changes.append_table_files[0].Kind()); - EXPECT_EQ(FileKind::Delete(), changes.append_table_files[1].Kind()); - EXPECT_EQ(4, changes.append_table_files[0].TotalBuckets()); + ASSERT_EQ(FileKind::Add(), changes.append_table_files[0].Kind()); + ASSERT_EQ(FileKind::Delete(), changes.append_table_files[1].Kind()); + ASSERT_EQ(4, changes.append_table_files[0].TotalBuckets()); std::string summary = changes.ToString(); - EXPECT_NE(std::string::npos, summary.find("2 append table files")); - EXPECT_NE(std::string::npos, summary.find("1 append Changelogs")); - EXPECT_NE(std::string::npos, summary.find("2 compact index files")); + ASSERT_NE(std::string::npos, summary.find("2 append table files")); + ASSERT_NE(std::string::npos, summary.find("1 append Changelogs")); + ASSERT_NE(std::string::npos, summary.find("2 compact index files")); } TEST_F(ManifestEntryChangesTest, TestHasGlobalIndexFileAdditions) { @@ -141,7 +141,7 @@ TEST_F(ManifestEntryChangesTest, TestHasGlobalIndexFileAdditions) { ManifestEntryChanges changes(/*default_num_bucket=*/8); ASSERT_OK(changes.Collect(message)); - EXPECT_TRUE(changes.HasGlobalIndexFileAdditions()); + ASSERT_TRUE(changes.HasGlobalIndexFileAdditions()); } TEST_F(ManifestEntryChangesTest, TestCollectInvalidCommitMessageType) { @@ -176,10 +176,10 @@ TEST_F(ManifestEntryChangesTest, TestChangedPartitionsIncludesDvAndGlobalIndex) return std::find(changed.begin(), changed.end(), target) != changed.end(); }; - EXPECT_TRUE(contains(partition_data)); - EXPECT_TRUE(contains(partition_dv)); - EXPECT_TRUE(contains(partition_global)); - EXPECT_FALSE(contains(partition_plain_index)); + ASSERT_TRUE(contains(partition_data)); + ASSERT_TRUE(contains(partition_dv)); + ASSERT_TRUE(contains(partition_global)); + ASSERT_FALSE(contains(partition_plain_index)); } } // namespace paimon::test diff --git a/src/paimon/core/operation/commit/row_id_column_conflict_checker_test.cpp b/src/paimon/core/operation/commit/row_id_column_conflict_checker_test.cpp index b4cd439e2..4240f19f9 100644 --- a/src/paimon/core/operation/commit/row_id_column_conflict_checker_test.cpp +++ b/src/paimon/core/operation/commit/row_id_column_conflict_checker_test.cpp @@ -74,7 +74,7 @@ TEST_F(RowIdColumnConflictCheckerTest, TestAllowsDisjointWriteColumns) { auto historical = CreateFile("historical", /*first_row_id=*/0, /*row_count=*/10, /*schema_id=*/0, std::vector{"c"}); ASSERT_OK_AND_ASSIGN(bool conflicts, checker->ConflictsWith(historical)); - EXPECT_FALSE(conflicts); + ASSERT_FALSE(conflicts); } TEST_F(RowIdColumnConflictCheckerTest, TestDetectsSameWriteColumns) { @@ -85,7 +85,7 @@ TEST_F(RowIdColumnConflictCheckerTest, TestDetectsSameWriteColumns) { auto historical = CreateFile("historical", /*first_row_id=*/0, /*row_count=*/10, /*schema_id=*/0, std::vector{"b"}); ASSERT_OK_AND_ASSIGN(bool conflicts, checker->ConflictsWith(historical)); - EXPECT_TRUE(conflicts); + ASSERT_TRUE(conflicts); } TEST_F(RowIdColumnConflictCheckerTest, TestUsesFieldIdAcrossRename) { @@ -96,7 +96,7 @@ TEST_F(RowIdColumnConflictCheckerTest, TestUsesFieldIdAcrossRename) { auto historical = CreateFile("historical", /*first_row_id=*/0, /*row_count=*/10, /*schema_id=*/0, std::vector{"b"}); ASSERT_OK_AND_ASSIGN(bool conflicts, checker->ConflictsWith(historical)); - EXPECT_TRUE(conflicts); + ASSERT_TRUE(conflicts); } TEST_F(RowIdColumnConflictCheckerTest, TestTreatsNullWriteColumnsAsFullSchemaWrite) { @@ -107,7 +107,7 @@ TEST_F(RowIdColumnConflictCheckerTest, TestTreatsNullWriteColumnsAsFullSchemaWri auto historical = CreateFile("historical", /*first_row_id=*/0, /*row_count=*/10, /*schema_id=*/0, std::vector{"b"}); ASSERT_OK_AND_ASSIGN(bool conflicts, checker->ConflictsWith(historical)); - EXPECT_TRUE(conflicts); + ASSERT_TRUE(conflicts); } TEST_F(RowIdColumnConflictCheckerTest, TestMergesOverlappedDeltaRangesAndWriteColumns) { @@ -123,8 +123,8 @@ TEST_F(RowIdColumnConflictCheckerTest, TestMergesOverlappedDeltaRangesAndWriteCo /*schema_id=*/0, std::vector{"c"}); ASSERT_OK_AND_ASSIGN(bool conflicts_b, checker->ConflictsWith(historical_b)); ASSERT_OK_AND_ASSIGN(bool conflicts_c, checker->ConflictsWith(historical_c)); - EXPECT_TRUE(conflicts_b); - EXPECT_TRUE(conflicts_c); + ASSERT_TRUE(conflicts_b); + ASSERT_TRUE(conflicts_c); } TEST_F(RowIdColumnConflictCheckerTest, TestScansAllOverlappedRangesAfterBinarySearch) { @@ -137,7 +137,7 @@ TEST_F(RowIdColumnConflictCheckerTest, TestScansAllOverlappedRangesAfterBinarySe auto historical = CreateFile("historical", /*first_row_id=*/3, /*row_count=*/10, /*schema_id=*/0, std::vector{"c"}); ASSERT_OK_AND_ASSIGN(bool conflicts, checker->ConflictsWith(historical)); - EXPECT_TRUE(conflicts); + ASSERT_TRUE(conflicts); } TEST_F(RowIdColumnConflictCheckerTest, TestIgnoreUnknownNonSystemWriteColumn) { @@ -148,7 +148,7 @@ TEST_F(RowIdColumnConflictCheckerTest, TestIgnoreUnknownNonSystemWriteColumn) { auto historical = CreateFile("historical", /*first_row_id=*/0, /*row_count=*/10, /*schema_id=*/0, std::vector{"missing"}); auto conflicts = checker->ConflictsWith(historical); - EXPECT_FALSE(conflicts.ok()); + ASSERT_FALSE(conflicts.ok()); } } // namespace paimon::test diff --git a/src/paimon/core/operation/commit/row_tracking_commit_utils.cpp b/src/paimon/core/operation/commit/row_tracking_commit_utils.cpp index 4e36fba57..02969cd10 100644 --- a/src/paimon/core/operation/commit/row_tracking_commit_utils.cpp +++ b/src/paimon/core/operation/commit/row_tracking_commit_utils.cpp @@ -23,6 +23,7 @@ #include "fmt/format.h" #include "paimon/common/data/blob_utils.h" #include "paimon/common/table/special_fields.h" +#include "paimon/common/utils/vector_store_utils.h" #include "paimon/core/manifest/file_source.h" #include "paimon/status.h" @@ -30,10 +31,6 @@ namespace paimon { namespace { -bool IsVectorStoreFile(const std::string& file_name) { - return file_name.find(".vector.") != std::string::npos; -} - ManifestEntry CloneEntryWithClonedFileMeta(const ManifestEntry& entry) { auto cloned_file = std::make_shared(*entry.File()); return ManifestEntry(entry.Kind(), entry.Partition(), entry.Bucket(), entry.TotalBuckets(), @@ -124,7 +121,7 @@ Result RowTrackingCommitUtils::AssignRowTrackingMeta( } assigned_entry.AssignFirstRowId(blob_start); blob_starts[blob_field_name] = blob_start + row_count; - } else if (IsVectorStoreFile(entry.File()->file_name)) { + } else if (VectorStoreUtils::IsVectorStoreFile(entry.File()->file_name)) { if (vector_store_start >= start) { return Status::Invalid(fmt::format( "This is a bug, vectorStoreStart {} should be less than start {} " diff --git a/src/paimon/core/operation/commit/sequence_snapshot_properties_test.cpp b/src/paimon/core/operation/commit/sequence_snapshot_properties_test.cpp new file mode 100644 index 000000000..35d4c1310 --- /dev/null +++ b/src/paimon/core/operation/commit/sequence_snapshot_properties_test.cpp @@ -0,0 +1,176 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed 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/core/operation/commit/sequence_snapshot_properties.h" + +#include +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/common/data/binary_row.h" +#include "paimon/core/io/data_file_meta.h" +#include "paimon/core/manifest/file_kind.h" +#include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/stats/simple_stats.h" +#include "paimon/data/timestamp.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +class SequenceSnapshotPropertiesTest : public testing::Test { + protected: + Snapshot MakeSnapshot( + const std::optional>& properties) const { + return Snapshot( + /*id=*/1, + /*schema_id=*/1, + /*base_manifest_list=*/"base-manifest-list", + /*base_manifest_list_size=*/std::nullopt, + /*delta_manifest_list=*/"delta-manifest-list", + /*delta_manifest_list_size=*/std::nullopt, + /*changelog_manifest_list=*/std::nullopt, + /*changelog_manifest_list_size=*/std::nullopt, + /*index_manifest=*/std::nullopt, + /*commit_user=*/"test-user", + /*commit_identifier=*/1, Snapshot::CommitKind::Append(), + /*time_millis=*/0, + /*total_record_count=*/0, + /*delta_record_count=*/0, + /*changelog_record_count=*/std::nullopt, + /*watermark=*/std::nullopt, + /*statistics=*/std::nullopt, properties, + /*next_row_id=*/std::nullopt); + } + + std::shared_ptr CreateDataFileMeta(int64_t max_sequence_number) const { + return std::make_shared( + "data-file", 1024, 8, DataFileMeta::EmptyMinKey(), DataFileMeta::EmptyMaxKey(), + SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), /*min_seq_no=*/0, + /*max_seq_no=*/max_sequence_number, + /*schema_id=*/1, /*level=*/0, + /*extra_files=*/std::vector>(), + /*creation_time=*/Timestamp(0, 0), + /*delete_row_count=*/std::nullopt, + /*embedded_index=*/nullptr, /*file_source=*/std::nullopt, + /*external_path=*/std::nullopt, + /*value_stats_cols=*/std::nullopt, /*first_row_id=*/std::nullopt, + /*write_cols=*/std::nullopt); + } + + ManifestEntry CreateEntry(const FileKind& kind, int64_t max_sequence_number) const { + return ManifestEntry(kind, BinaryRow(0), /*bucket=*/0, /*total_buckets=*/1, + CreateDataFileMeta(max_sequence_number)); + } +}; + +TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberEmptySnapshot) { + ASSERT_OK_AND_ASSIGN(std::optional result, + SequenceSnapshotProperties::MaxSequenceNumber(std::nullopt)); + ASSERT_FALSE(result.has_value()); +} + +TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberSnapshotWithoutProperties) { + ASSERT_OK_AND_ASSIGN(std::optional result, + SequenceSnapshotProperties::MaxSequenceNumber(MakeSnapshot(std::nullopt))); + ASSERT_FALSE(result.has_value()); +} + +TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberKeyMissing) { + std::map properties{{"other-key", "42"}}; + ASSERT_OK_AND_ASSIGN(std::optional result, + SequenceSnapshotProperties::MaxSequenceNumber(MakeSnapshot(properties))); + ASSERT_FALSE(result.has_value()); +} + +TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberValid) { + std::map properties{ + {SequenceSnapshotProperties::kMaxSequenceNumberKey, "123"}}; + ASSERT_OK_AND_ASSIGN(std::optional result, + SequenceSnapshotProperties::MaxSequenceNumber(MakeSnapshot(properties))); + ASSERT_TRUE(result.has_value()); + ASSERT_EQ(123, result.value()); +} + +TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberTrailingCharacters) { + std::map properties{ + {SequenceSnapshotProperties::kMaxSequenceNumberKey, "123abc"}}; + ASSERT_NOK_WITH_MSG(SequenceSnapshotProperties::MaxSequenceNumber(MakeSnapshot(properties)), + "trailing characters are not allowed"); +} + +TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberNotANumber) { + std::map properties{ + {SequenceSnapshotProperties::kMaxSequenceNumberKey, "not-a-number"}}; + ASSERT_NOK_WITH_MSG(SequenceSnapshotProperties::MaxSequenceNumber(MakeSnapshot(properties)), + "Invalid"); +} + +TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberFromFilesEmpty) { + ASSERT_FALSE(SequenceSnapshotProperties::MaxSequenceNumberFromFiles({}).has_value()); +} + +TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberFromFilesOnlyDelete) { + std::vector files{CreateEntry(FileKind::Delete(), 100)}; + ASSERT_FALSE(SequenceSnapshotProperties::MaxSequenceNumberFromFiles(files).has_value()); +} + +TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberFromFilesSkipsDelete) { + std::vector files{CreateEntry(FileKind::Add(), 10), + CreateEntry(FileKind::Delete(), 999), + CreateEntry(FileKind::Add(), 42)}; + std::optional result = SequenceSnapshotProperties::MaxSequenceNumberFromFiles(files); + ASSERT_TRUE(result.has_value()); + ASSERT_EQ(42, result.value()); +} + +TEST_F(SequenceSnapshotPropertiesTest, MergeMaxSequenceNumberNoInput) { + std::map properties{{"existing", "value"}}; + std::map merged = SequenceSnapshotProperties::MergeMaxSequenceNumber( + properties, /*latest_max_sequence_number=*/std::nullopt, /*delta_files=*/{}); + ASSERT_EQ(properties, merged); + ASSERT_EQ(0u, merged.count(SequenceSnapshotProperties::kMaxSequenceNumberKey)); +} + +TEST_F(SequenceSnapshotPropertiesTest, MergeMaxSequenceNumberLatestOnly) { + std::map merged = SequenceSnapshotProperties::MergeMaxSequenceNumber( + /*properties=*/{}, /*latest_max_sequence_number=*/50, /*delta_files=*/{}); + ASSERT_EQ("50", merged.at(SequenceSnapshotProperties::kMaxSequenceNumberKey)); +} + +TEST_F(SequenceSnapshotPropertiesTest, MergeMaxSequenceNumberDeltaOnly) { + std::vector delta_files{CreateEntry(FileKind::Add(), 77)}; + std::map merged = SequenceSnapshotProperties::MergeMaxSequenceNumber( + /*properties=*/{}, /*latest_max_sequence_number=*/std::nullopt, delta_files); + ASSERT_EQ("77", merged.at(SequenceSnapshotProperties::kMaxSequenceNumberKey)); +} + +TEST_F(SequenceSnapshotPropertiesTest, MergeMaxSequenceNumberTakesMaximum) { + std::vector delta_files{CreateEntry(FileKind::Add(), 30)}; + std::map merged = SequenceSnapshotProperties::MergeMaxSequenceNumber( + /*properties=*/{}, /*latest_max_sequence_number=*/90, delta_files); + ASSERT_EQ("90", merged.at(SequenceSnapshotProperties::kMaxSequenceNumberKey)); + + std::vector larger_delta{CreateEntry(FileKind::Add(), 150)}; + std::map merged2 = SequenceSnapshotProperties::MergeMaxSequenceNumber( + /*properties=*/{}, /*latest_max_sequence_number=*/90, larger_delta); + ASSERT_EQ("150", merged2.at(SequenceSnapshotProperties::kMaxSequenceNumberKey)); +} + +} // namespace paimon::test diff --git a/src/paimon/core/operation/file_store_commit_impl.cpp b/src/paimon/core/operation/file_store_commit_impl.cpp index 9e6da0634..c7e2d247f 100644 --- a/src/paimon/core/operation/file_store_commit_impl.cpp +++ b/src/paimon/core/operation/file_store_commit_impl.cpp @@ -202,6 +202,172 @@ Status FileStoreCommitImpl::DropPartition( return Status::OK(); } +Status FileStoreCommitImpl::TruncateTable(int64_t commit_identifier) { + // An empty partition list means "all partitions", so this overwrites the whole table with + // no new files, effectively truncating it. Mirrors Java tryOverwritePartition(null, ...). + PAIMON_ASSIGN_OR_RAISE([[maybe_unused]] int32_t attempt, + TryOverwrite(/*partition=*/{}, /*changes=*/{}, /*index_entries=*/{}, + commit_identifier, std::nullopt, /*properties=*/{})); + return Status::OK(); +} + +Status FileStoreCommitImpl::Abort( + const std::vector>& commit_messages) { + for (const auto& message : commit_messages) { + auto* msg = dynamic_cast(message.get()); + if (msg == nullptr) { + return Status::Invalid("fail to cast commit message to impl"); + } + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr data_file_path_factory, + path_factory_->CreateDataFilePathFactory(msg->Partition(), msg->Bucket())); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr index_file_path_factory, + path_factory_->CreateIndexFileFactory(msg->Partition(), msg->Bucket())); + + const DataIncrement& new_files_increment = msg->GetNewFilesIncrement(); + const CompactIncrement& compact_increment = msg->GetCompactIncrement(); + + std::vector> data_files_to_delete; + auto append_data_files = + [&data_files_to_delete](const std::vector>& files) { + data_files_to_delete.insert(data_files_to_delete.end(), files.begin(), files.end()); + }; + append_data_files(new_files_increment.NewFiles()); + append_data_files(new_files_increment.ChangelogFiles()); + append_data_files(compact_increment.CompactAfter()); + append_data_files(compact_increment.ChangelogFiles()); + for (const auto& file : data_files_to_delete) { + // Best-effort cleanup: ignore delete failures, aligning with Java deleteQuietly. + [[maybe_unused]] Status status = + fs_->Delete(data_file_path_factory->ToPath(file), /*recursive=*/false); + } + + std::vector> index_files_to_delete; + auto append_index_files = [&index_files_to_delete]( + const std::vector>& files) { + index_files_to_delete.insert(index_files_to_delete.end(), files.begin(), files.end()); + }; + append_index_files(new_files_increment.NewIndexFiles()); + append_index_files(compact_increment.NewIndexFiles()); + for (const auto& file : index_files_to_delete) { + [[maybe_unused]] Status status = + fs_->Delete(index_file_path_factory->ToPath(file), /*recursive=*/false); + } + } + return Status::OK(); +} + +Result> FileStoreCommitImpl::ReadAddManifestEntries( + const Snapshot& snapshot) const { + std::vector data_manifests; + PAIMON_RETURN_NOT_OK(manifest_list_->ReadDataManifests(snapshot, &data_manifests)); + std::vector unmerged_entries; + for (const auto& meta : data_manifests) { + std::vector entries; + PAIMON_RETURN_NOT_OK(manifest_file_->Read( + meta.FileName(), [](const ManifestEntry&) -> Result { return true; }, &entries)); + unmerged_entries.insert(unmerged_entries.end(), std::make_move_iterator(entries.begin()), + std::make_move_iterator(entries.end())); + } + std::vector merged_entries; + PAIMON_RETURN_NOT_OK(FileEntry::MergeEntries(unmerged_entries, &merged_entries)); + std::vector add_entries; + for (auto& entry : merged_entries) { + if (entry.Kind() == FileKind::Add()) { + add_entries.push_back(std::move(entry)); + } + } + return add_entries; +} + +Result FileStoreCommitImpl::RollbackToAsLatest(int64_t target_snapshot_id) { + PAIMON_ASSIGN_OR_RAISE(std::optional latest_opt, snapshot_manager_->LatestSnapshot()); + if (!latest_opt) { + return Status::Invalid("Latest snapshot is null, can not roll back."); + } + const Snapshot& latest = latest_opt.value(); + PAIMON_ASSIGN_OR_RAISE(Snapshot target_snapshot, + snapshot_manager_->LoadSnapshot(target_snapshot_id)); + + PAIMON_ASSIGN_OR_RAISE(std::vector latest_entries, + ReadAddManifestEntries(latest)); + PAIMON_ASSIGN_OR_RAISE(std::vector target_entries, + ReadAddManifestEntries(target_snapshot)); + + std::unordered_set latest_identifiers; + latest_identifiers.reserve(latest_entries.size()); + for (const auto& entry : latest_entries) { + latest_identifiers.insert(entry.CreateIdentifier()); + } + std::unordered_set target_identifiers; + target_identifiers.reserve(target_entries.size()); + for (const auto& entry : target_entries) { + target_identifiers.insert(entry.CreateIdentifier()); + } + + std::vector delta_files; + for (const auto& entry : latest_entries) { + if (target_identifiers.find(entry.CreateIdentifier()) == target_identifiers.end()) { + delta_files.emplace_back(FileKind::Delete(), entry.Partition(), entry.Bucket(), + entry.TotalBuckets(), entry.File()); + } + } + for (const auto& entry : target_entries) { + if (latest_identifiers.find(entry.CreateIdentifier()) == latest_identifiers.end()) { + delta_files.emplace_back(FileKind::Add(), entry.Partition(), entry.Bucket(), + entry.TotalBuckets(), entry.File()); + } + } + + std::pair base_manifest_list; + std::pair delta_manifest_list; + PAIMON_ASSIGN_OR_RAISE(std::vector base_manifests, + manifest_file_->Write(latest_entries)); + PAIMON_ASSIGN_OR_RAISE(base_manifest_list, manifest_list_->Write(base_manifests)); + PAIMON_ASSIGN_OR_RAISE(std::vector delta_manifests, + manifest_file_->Write(delta_files)); + PAIMON_ASSIGN_OR_RAISE(delta_manifest_list, manifest_list_->Write(delta_manifests)); + + // For row-tracking tables nextRowId must stay monotonic: a rollback to an older snapshot must + // not move it backwards, otherwise new appends would reuse row ids already assigned by the + // snapshots between the target and the previous latest, breaking the global uniqueness of + // _ROW_ID. Keep the larger of the previous latest and the target nextRowId. + std::optional next_row_id = std::max(latest.NextRowId(), target_snapshot.NextRowId()); + + int64_t delta_record_count = + ManifestEntry::RecordCountAdd(delta_files) - ManifestEntry::RecordCountDelete(delta_files); + Snapshot new_snapshot( + latest.Id() + 1, target_snapshot.SchemaId(), base_manifest_list.first, + base_manifest_list.second, delta_manifest_list.first, delta_manifest_list.second, + /*changelog_manifest_list=*/std::nullopt, /*changelog_manifest_list_size=*/std::nullopt, + target_snapshot.IndexManifest(), commit_user_, + /*commit_identifier=*/std::numeric_limits::max(), + Snapshot::CommitKind::Overwrite(), DateTimeUtils::GetCurrentUTCTimeUs() / 1000, + target_snapshot.TotalRecordCount(), delta_record_count, + /*changelog_record_count=*/std::nullopt, target_snapshot.Watermark(), + target_snapshot.Statistics(), target_snapshot.Properties(), next_row_id); + + std::unordered_map partition_entry_map; + PAIMON_RETURN_NOT_OK(PartitionEntry::Merge(delta_files, &partition_entry_map)); + std::vector delta_statistics; + delta_statistics.reserve(partition_entry_map.size()); + for (const auto& [_, partition_entry] : partition_entry_map) { + delta_statistics.push_back(partition_entry); + } + + PAIMON_ASSIGN_OR_RAISE(bool success, CommitSnapshotImpl(new_snapshot, delta_statistics)); + if (success) { + PAIMON_LOG_INFO(logger_, + "Successfully rolled back table %s to snapshot %ld as new snapshot %ld by " + "user %s.", + root_path_.c_str(), target_snapshot_id, new_snapshot.Id(), + commit_user_.c_str()); + last_committed_snapshot_id_ = new_snapshot.Id(); + } + return success; +} + FileStoreCommit& FileStoreCommitImpl::RowIdCheckConflict( std::optional row_id_check_from_snapshot) { conflict_detection_.SetRowIdCheckFromSnapshot(row_id_check_from_snapshot); diff --git a/src/paimon/core/operation/file_store_commit_impl.h b/src/paimon/core/operation/file_store_commit_impl.h index fb59826b1..b171e2150 100644 --- a/src/paimon/core/operation/file_store_commit_impl.h +++ b/src/paimon/core/operation/file_store_commit_impl.h @@ -125,6 +125,12 @@ class FileStoreCommitImpl : public FileStoreCommit { Status DropPartition(const std::vector>& partitions, int64_t commit_identifier) override; + Status TruncateTable(int64_t commit_identifier) override; + + Status Abort(const std::vector>& commit_messages) override; + + Result RollbackToAsLatest(int64_t target_snapshot_id) override; + FileStoreCommit& RowIdCheckConflict(std::optional row_id_check_from_snapshot) override; std::shared_ptr GetCommitMetrics() const override { @@ -196,6 +202,8 @@ class FileStoreCommitImpl : public FileStoreCommit { Result CommitSnapshotImpl(const Snapshot& new_snapshot, const std::vector& delta_statistics); + Result> ReadAddManifestEntries(const Snapshot& snapshot) const; + void CleanUpTmpManifests(const std::string& previous_changes_list_name, const std::string& new_changes_list_name, const std::vector& old_metas, diff --git a/src/paimon/core/operation/file_store_commit_impl_test.cpp b/src/paimon/core/operation/file_store_commit_impl_test.cpp index 3163f2a68..96e3363f1 100644 --- a/src/paimon/core/operation/file_store_commit_impl_test.cpp +++ b/src/paimon/core/operation/file_store_commit_impl_test.cpp @@ -37,11 +37,20 @@ #include "paimon/common/data/binary_row.h" #include "paimon/common/data/binary_row_writer.h" #include "paimon/common/factories/io_hook.h" +#include "paimon/common/utils/linked_hash_map.h" #include "paimon/common/utils/path_util.h" #include "paimon/common/utils/scope_guard.h" #include "paimon/core/catalog/commit_table_request.h" +#include "paimon/core/core_options.h" +#include "paimon/core/deletionvectors/deletion_vectors_index_file.h" +#include "paimon/core/index/deletion_vector_meta.h" #include "paimon/core/index/global_index_meta.h" +#include "paimon/core/index/index_file_meta.h" +#include "paimon/core/index/index_path_factory.h" +#include "paimon/core/io/compact_increment.h" #include "paimon/core/io/data_file_meta.h" +#include "paimon/core/io/data_file_path_factory.h" +#include "paimon/core/io/data_increment.h" #include "paimon/core/manifest/file_kind.h" #include "paimon/core/manifest/file_source.h" #include "paimon/core/manifest/index_manifest_entry.h" @@ -55,6 +64,7 @@ #include "paimon/core/schema/table_schema.h" #include "paimon/core/stats/simple_stats.h" #include "paimon/core/table/sink/commit_message_impl.h" +#include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/file_utils.h" #include "paimon/core/utils/snapshot_manager.h" #include "paimon/data/timestamp.h" @@ -82,6 +92,7 @@ class GmockFileSystem : public LocalFileSystem { (const, override)); MOCK_METHOD(Status, AtomicStore, (const std::string& path, const std::string& content), (override)); + MOCK_METHOD(Result, Exists, (const std::string& path), (const, override)); }; class GmockFileSystemFactory : public LocalFileSystemFactory { @@ -115,6 +126,11 @@ class GmockFileSystemFactory : public LocalFileSystemFactory { return fs_ptr->FileSystem::AtomicStore(path, content); })); + ON_CALL(*fs, Exists(A())) + .WillByDefault(Invoke([fs_ptr](const std::string& path) { + return fs_ptr->LocalFileSystem::Exists(path); + })); + return fs; } }; @@ -219,6 +235,17 @@ class FileStoreCommitImplTest : public testing::Test { return result; } + size_t CountFiles(const std::string& dir) const { + size_t count = 0; + std::error_code ec; + for (std::filesystem::directory_iterator it(dir, ec), end; it != end; it.increment(ec)) { + if (it->is_regular_file(ec)) { + ++count; + } + } + return count; + } + std::shared_ptr CreateIndexFileMeta(const std::string& file_name, const std::string& index_type = "bitmap") { return std::make_shared(index_type, file_name, /*file_size=*/100, @@ -238,6 +265,22 @@ class FileStoreCommitImplTest : public testing::Test { /*dv_ranges=*/std::nullopt, /*external_path=*/std::nullopt, global_index); } + std::shared_ptr CreateLeveledDataFileMeta(const std::string& file_name, + const BinaryRow& min_key, + const BinaryRow& max_key, int32_t level, + int32_t schema_id = 0) { + return std::make_shared( + file_name, 1024, 8, min_key, max_key, SimpleStats::EmptyStats(), + SimpleStats::EmptyStats(), /*min_seq_no=*/16, /*max_seq_no=*/32, schema_id, level, + /*extra_files=*/std::vector>(), + /*creation_time=*/Timestamp(0, 0), + /*delete_row_count=*/std::nullopt, + /*embedded_index=*/nullptr, FileSource::Append(), + /*external_path=*/std::nullopt, + /*value_stats_cols=*/std::nullopt, /*first_row_id=*/std::nullopt, + /*write_cols=*/std::nullopt); + } + std::shared_ptr CreateAppendDataFileMeta(const std::string& file_name, int64_t row_count) { return std::make_shared( @@ -763,6 +806,224 @@ TEST_F(FileStoreCommitImplTest, TestCommitMultipleTimes) { } } +TEST_F(FileStoreCommitImplTest, TestRollbackToAsLatest) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = dynamic_cast(commit.get()); + ASSERT_TRUE(commit_impl); + + // Append three times so the latest snapshot (3) is a superset of snapshot 1. + const std::string base_path = paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-0"; + for (int32_t i = 1; i <= 3; ++i) { + std::vector> msgs = + GetCommitMessages(base_path + std::to_string(i), /*version=*/3); + ASSERT_GT(msgs.size(), 0); + ASSERT_OK(commit->Commit(msgs, /*commit_identifier=*/i)); + } + + ASSERT_OK_AND_ASSIGN(Snapshot snapshot1, commit_impl->snapshot_manager_->LoadSnapshot(1)); + ASSERT_OK_AND_ASSIGN(Snapshot snapshot3, commit_impl->snapshot_manager_->LoadSnapshot(3)); + ASSERT_OK_AND_ASSIGN(std::vector snapshot1_entries, + commit_impl->ReadAddManifestEntries(snapshot1)); + ASSERT_OK_AND_ASSIGN(std::vector snapshot3_entries, + commit_impl->ReadAddManifestEntries(snapshot3)); + + // Roll back to snapshot 1: the delta only removes files (DELETE branch). + ASSERT_OK_AND_ASSIGN(bool rolled_back, commit->RollbackToAsLatest(/*target_snapshot_id=*/1)); + ASSERT_TRUE(rolled_back); + ASSERT_OK_AND_ASSIGN(bool snapshot4_exist, file_system_->Exists(PathUtil::JoinPath( + table_path_, "snapshot/snapshot-4"))); + ASSERT_TRUE(snapshot4_exist); + ASSERT_OK_AND_ASSIGN(Snapshot snapshot4, commit_impl->snapshot_manager_->LoadSnapshot(4)); + ASSERT_EQ(snapshot4.Id(), 4); + ASSERT_TRUE(snapshot4.GetCommitKind() == Snapshot::CommitKind::Overwrite()); + ASSERT_EQ(snapshot4.SchemaId(), snapshot1.SchemaId()); + ASSERT_EQ(snapshot4.TotalRecordCount(), snapshot1.TotalRecordCount()); + ASSERT_EQ(snapshot4.NextRowId(), std::max(snapshot3.NextRowId(), snapshot1.NextRowId())); + ASSERT_EQ(snapshot4.IndexManifest(), snapshot1.IndexManifest()); + ASSERT_OK_AND_ASSIGN(std::vector snapshot4_entries, + commit_impl->ReadAddManifestEntries(snapshot4)); + ASSERT_EQ(CollectFileNames(snapshot4_entries), CollectFileNames(snapshot1_entries)); + + // Roll back forward to snapshot 3: the delta only adds files (ADD branch). + ASSERT_OK_AND_ASSIGN(bool rolled_forward, commit->RollbackToAsLatest(/*target_snapshot_id=*/3)); + ASSERT_TRUE(rolled_forward); + ASSERT_OK_AND_ASSIGN(Snapshot snapshot5, commit_impl->snapshot_manager_->LoadSnapshot(5)); + ASSERT_EQ(snapshot5.Id(), 5); + ASSERT_TRUE(snapshot5.GetCommitKind() == Snapshot::CommitKind::Overwrite()); + ASSERT_EQ(snapshot5.TotalRecordCount(), snapshot3.TotalRecordCount()); + ASSERT_EQ(snapshot5.IndexManifest(), snapshot3.IndexManifest()); + ASSERT_OK_AND_ASSIGN(std::vector snapshot5_entries, + commit_impl->ReadAddManifestEntries(snapshot5)); + ASSERT_EQ(CollectFileNames(snapshot5_entries), CollectFileNames(snapshot3_entries)); +} + +TEST_F(FileStoreCommitImplTest, TestRollbackToAsLatestNoLatestSnapshotReturnsError) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + ASSERT_NOK_WITH_MSG(commit->RollbackToAsLatest(/*target_snapshot_id=*/1), + "Latest snapshot is null, can not roll back."); +} + +TEST_F(FileStoreCommitImplTest, TestRollbackToAsLatestTargetNotExistReturnsError) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + std::vector> msgs = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-01", + /*version=*/3); + ASSERT_GT(msgs.size(), 0); + ASSERT_OK(commit->Commit(msgs, /*commit_identifier=*/1)); + ASSERT_FALSE(commit->RollbackToAsLatest(/*target_snapshot_id=*/999).ok()); +} + +TEST_F(FileStoreCommitImplTest, TestRollbackToAsLatestDeletionVectorOnlyChange) { + // Mirror Java testRollbackToAsLatestDeletionVectorChangeIsInvisibleToStreaming: when the target + // and the latest snapshot share identical data files and differ only in the index (deletion + // vector) manifest, the rollback produces an empty data delta and the new snapshot inherits the + // target's index manifest. + std::map table_options = {{Options::FILE_FORMAT, "orc"}, + {Options::FILE_SYSTEM, "local"}, + {Options::BUCKET, "2"}, + {Options::BUCKET_KEY, "f2"}}; + ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(test_root_, table_options)); + arrow::Schema typed_schema(fields_); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + ASSERT_OK(catalog->CreateTable(Identifier("foo", "dv_bar"), &schema, + /*partition_keys=*/{"f1"}, + /*primary_keys=*/{}, table_options, + /*ignore_if_exists=*/false)); + std::string dv_table_path = PathUtil::JoinPath(test_root_, "foo.db/dv_bar"); + + CommitContextBuilder context_builder(dv_table_path, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = dynamic_cast(commit.get()); + ASSERT_TRUE(commit_impl); + + BinaryRow partition = BinaryRowGenerator::GenerateRow({10}, GetDefaultPool().get()); + + // snapshot 1 (target): a single data file, no deletion vectors. + std::vector> new_files; + new_files.push_back(CreateAppendDataFileMeta("data-file-1", /*row_count=*/10)); + DataIncrement data_increment(std::move(new_files), {}, {}); + std::shared_ptr data_msg = std::make_shared( + partition, /*bucket=*/0, /*total_buckets=*/2, data_increment, CompactIncrement({}, {}, {})); + ASSERT_OK(commit->Commit({data_msg}, /*commit_identifier=*/1)); + ASSERT_OK_AND_ASSIGN(Snapshot target_snapshot, commit_impl->snapshot_manager_->LoadSnapshot(1)); + + // snapshot 2 (latest): a deletion-vector-only change on the same data file. The data file is + // unchanged, only the index manifest gains a deletion vector. + LinkedHashMap dv_ranges; + dv_ranges.insert_or_assign("data-file-1", + DeletionVectorMeta(/*data_file_name=*/"data-file-1", /*offset=*/0, + /*length=*/10, /*cardinality=*/1)); + std::vector> new_index_files; + new_index_files.push_back(std::make_shared( + DeletionVectorsIndexFile::DELETION_VECTORS_INDEX, "dv-index-1", /*file_size=*/100, + /*row_count=*/1, /*dv_ranges=*/dv_ranges, /*external_path=*/std::nullopt)); + DataIncrement dv_increment({}, {}, {}, std::move(new_index_files), {}); + std::shared_ptr dv_msg = std::make_shared( + partition, /*bucket=*/0, /*total_buckets=*/2, dv_increment, CompactIncrement({}, {}, {})); + ASSERT_OK(commit->Commit({dv_msg}, /*commit_identifier=*/2)); + ASSERT_OK_AND_ASSIGN(Snapshot latest_snapshot, commit_impl->snapshot_manager_->LoadSnapshot(2)); + ASSERT_TRUE(latest_snapshot.IndexManifest().has_value()); + + // Roll back to snapshot 1: the data files are identical, so the delta carries no data change + // and the new snapshot inherits the target's (empty) index manifest, dropping the deletion + // vector. + ASSERT_OK_AND_ASSIGN(bool rolled_back, commit->RollbackToAsLatest(/*target_snapshot_id=*/1)); + ASSERT_TRUE(rolled_back); + ASSERT_OK_AND_ASSIGN(Snapshot rolled_back_snapshot, + commit_impl->snapshot_manager_->LoadSnapshot(3)); + ASSERT_EQ(rolled_back_snapshot.Id(), 3); + ASSERT_TRUE(rolled_back_snapshot.GetCommitKind() == Snapshot::CommitKind::Overwrite()); + // Empty data delta: no records are added or removed by the rollback. + ASSERT_EQ(rolled_back_snapshot.DeltaRecordCount(), 0); + // The new snapshot points back to the target's index manifest. + ASSERT_EQ(rolled_back_snapshot.IndexManifest(), target_snapshot.IndexManifest()); + // The surviving data files are unchanged relative to the target. + ASSERT_OK_AND_ASSIGN(std::vector target_entries, + commit_impl->ReadAddManifestEntries(target_snapshot)); + ASSERT_OK_AND_ASSIGN(std::vector rolled_back_entries, + commit_impl->ReadAddManifestEntries(rolled_back_snapshot)); + ASSERT_EQ(CollectFileNames(rolled_back_entries), CollectFileNames(target_entries)); +} + +TEST_F(FileStoreCommitImplTest, TestRollbackToAsLatestConcurrentConflictReturnsFalse) { + // When a concurrent writer wins the race for the next snapshot id, the atomic snapshot commit + // reports failure (returns false); RollbackToAsLatest surfaces that without error and without + // advancing the committed snapshot id. The temporary base/delta manifest lists written before + // the failed commit are intentionally left behind, matching Java (no cleanup on this path). + // + // The conflict is a TOCTOU race: snapshot-2 does not exist when LatestSnapshot() resolves the + // current latest (so it stays snapshot-1), but exists by the time RenamingSnapshotCommit checks + // before writing. A stateful Exists() mock reproduces exactly that ordering. + ASSERT_OK_AND_ASSIGN(std::shared_ptr fs, + FileSystemFactory::Get("gmock_fs", table_path_, {})); + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .WithFileSystem(fs) + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = dynamic_cast(commit.get()); + ASSERT_TRUE(commit_impl); + + std::vector> msgs = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-01", + /*version=*/3); + ASSERT_GT(msgs.size(), 0); + ASSERT_OK(commit->Commit(msgs, /*commit_identifier=*/1)); + ASSERT_EQ(commit_impl->last_committed_snapshot_id_, 1); + + // snapshot-2 is invisible the first time it is probed (LatestSnapshot keeps snapshot-1 as the + // latest) and visible afterwards (RenamingSnapshotCommit sees the conflicting file). + const std::string next_snapshot = PathUtil::JoinPath(table_path_, "snapshot/snapshot-2"); + auto* mock_fs = dynamic_cast(fs.get()); + ASSERT_TRUE(mock_fs); + EXPECT_CALL(*mock_fs, Exists(testing::_)) + .Times(testing::AnyNumber()) + .WillRepeatedly(testing::Invoke( + [mock_fs](const std::string& path) { return mock_fs->LocalFileSystem::Exists(path); })); + EXPECT_CALL(*mock_fs, Exists(testing::StrEq(next_snapshot))) + .WillOnce(testing::Return(Result(false))) + .WillRepeatedly(testing::Return(Result(true))); + + const std::string manifest_dir = PathUtil::JoinPath(table_path_, "manifest"); + const size_t manifest_count_before = CountFiles(manifest_dir); + + ASSERT_OK_AND_ASSIGN(bool rolled_back, commit->RollbackToAsLatest(/*target_snapshot_id=*/1)); + ASSERT_FALSE(rolled_back); + // The committed snapshot id must not advance on the failed rollback. + ASSERT_EQ(commit_impl->last_committed_snapshot_id_, 1); + // The temporary base/delta manifest lists are leaked (not cleaned up) on the failure path. + ASSERT_GT(CountFiles(manifest_dir), manifest_count_before); +} + TEST_F(FileStoreCommitImplTest, TestCommitAndOverwriteWithNoPartitionKey) { CommitContextBuilder context_builder(table_path_, "commit_user_1"); ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, @@ -1246,6 +1507,182 @@ TEST_F(FileStoreCommitImplTest, TestDropMultiPartitionAndExpireSnapshot) { ASSERT_EQ(3, manifests[0].NumDeletedFiles()); } +TEST_F(FileStoreCommitImplTest, TestTruncateTable) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .IgnoreEmptyCommit(true) + .Finish()); + + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + std::vector> msgs = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/" + "commit_messages-01", + /*version=*/3); + ASSERT_OK(commit->Commit(msgs, /*commit_identifier=*/0)); + + // Truncate overwrites all partitions with no new files, deleting every existing file. + ASSERT_OK(commit->TruncateTable(/*commit_identifier=*/1)); + + ASSERT_OK_AND_ASSIGN(bool exist, file_system_->Exists(table_path_ + "/snapshot/snapshot-2")); + ASSERT_TRUE(exist); + auto commit_impl = dynamic_cast(commit.get()); + ASSERT_TRUE(commit_impl); + ASSERT_OK_AND_ASSIGN(Snapshot snapshot, commit_impl->snapshot_manager_->LoadSnapshot(2)); + ASSERT_EQ(Snapshot::CommitKind::Overwrite(), snapshot.GetCommitKind()); + std::vector manifests; + ASSERT_OK(commit_impl->manifest_list_->ReadDeltaManifests(snapshot, &manifests)); + ASSERT_EQ(1, manifests.size()); + ASSERT_EQ(0, manifests[0].NumAddedFiles()); + // The append_09 fixture contains 3 data files spread across partitions f1=10 and f1=20. + ASSERT_EQ(3, manifests[0].NumDeletedFiles()); +} + +TEST_F(FileStoreCommitImplTest, TestAbortDeletesDataAndIndexFiles) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + const BinaryRow partition = CreateIntRow(10); + const int32_t bucket = 0; + auto new_data_file = CreateAppendDataFileMeta("abort-new-data", 1); + auto compact_data_file = CreateAppendDataFileMeta("abort-compact-data", 1); + auto new_index_file = CreateIndexFileMeta("abort-new-index"); + auto compact_index_file = CreateIndexFileMeta("abort-compact-index"); + + DataIncrement data_increment(/*new_files=*/{new_data_file}, /*deleted_files=*/{}, + /*changelog_files=*/{}, /*new_index_files=*/{new_index_file}, + /*deleted_index_files=*/{}); + CompactIncrement compact_increment(/*compact_before=*/{}, /*compact_after=*/{compact_data_file}, + /*changelog_files=*/{}, + /*new_index_files=*/{compact_index_file}, + /*deleted_index_files=*/{}); + std::shared_ptr message = std::make_shared( + partition, bucket, /*total_buckets=*/2, data_increment, compact_increment); + + // Materialize the referenced files so we can observe them being cleaned up. + ASSERT_OK_AND_ASSIGN(std::shared_ptr data_pf, + commit_impl->path_factory_->CreateDataFilePathFactory(partition, bucket)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr index_pf, + commit_impl->path_factory_->CreateIndexFileFactory(partition, bucket)); + std::vector paths = { + data_pf->ToPath(new_data_file), data_pf->ToPath(compact_data_file), + index_pf->ToPath(new_index_file), index_pf->ToPath(compact_index_file)}; + for (const auto& path : paths) { + ASSERT_OK(file_system_->WriteFile(path, /*content=*/"", /*overwrite=*/false)); + ASSERT_OK_AND_ASSIGN(bool exist, file_system_->Exists(path)); + ASSERT_TRUE(exist); + } + + ASSERT_OK(commit_impl->Abort({message})); + + for (const auto& path : paths) { + ASSERT_OK_AND_ASSIGN(bool exist, file_system_->Exists(path)); + ASSERT_FALSE(exist); + } +} + +TEST_F(FileStoreCommitImplTest, AbortIgnoresMissingFilesAndFailsForNonImplMessage) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + // Deleting files that were never written is a best-effort no-op, not an error. + DataIncrement data_increment(/*new_files=*/{CreateAppendDataFileMeta("abort-missing", 1)}, + /*deleted_files=*/{}, /*changelog_files=*/{}); + std::shared_ptr message = + std::make_shared(CreateIntRow(10), /*bucket=*/0, /*total_buckets=*/2, + data_increment, CompactIncrement({}, {}, {})); + ASSERT_OK(commit_impl->Abort({message})); + + // A commit message that is not a CommitMessageImpl is rejected. + ASSERT_NOK_WITH_MSG(commit_impl->Abort({std::make_shared()}), + "fail to cast commit message to impl"); +} + +TEST_F(FileStoreCommitImplTest, AbortIgnoresDeleteFailures) { + // A delete that fails with an IO error (e.g. permission denied) must be swallowed by + // best-effort cleanup, mirroring Java deleteQuietly: Abort still returns OK and does not + // propagate the failure. + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = dynamic_cast(commit.get()); + ASSERT_TRUE(commit_impl); + + const BinaryRow partition = CreateIntRow(10); + const int32_t bucket = 0; + auto new_data_file = CreateAppendDataFileMeta("abort-io-fail-data", 1); + DataIncrement data_increment(/*new_files=*/{new_data_file}, /*deleted_files=*/{}, + /*changelog_files=*/{}); + std::shared_ptr message = std::make_shared( + partition, bucket, /*total_buckets=*/2, data_increment, CompactIncrement({}, {}, {})); + + // Materialize the file so we can observe that a failed delete leaves it untouched. + ASSERT_OK_AND_ASSIGN(std::shared_ptr data_pf, + commit_impl->path_factory_->CreateDataFilePathFactory(partition, bucket)); + const std::string data_path = data_pf->ToPath(new_data_file); + ASSERT_OK(file_system_->WriteFile(data_path, /*content=*/"", /*overwrite=*/false)); + + // Fault-inject an IO error on the delete. Abort performs no local file IO before its delete + // loop, so position 0 lands on the delete itself. The error must be swallowed. Clearing on + // scope exit keeps the process-global hook from leaking into other tests. + auto io_hook = IOHook::GetInstance(); + ScopeGuard guard([&io_hook]() { io_hook->Clear(); }); + io_hook->Reset(/*pos=*/0, IOHook::Mode::RETURN_ERROR); + ASSERT_OK(commit_impl->Abort({message})); + io_hook->Clear(); + + // Because the delete failed, the file remains on disk. + ASSERT_OK_AND_ASSIGN(bool exist, file_system_->Exists(data_path)); + ASSERT_TRUE(exist); +} + +TEST_F(FileStoreCommitImplTest, TestTruncateEmptyTable) { + // Truncating a table that has never been committed succeeds and produces an OVERWRITE snapshot + // with no added and no deleted files: there is nothing to overwrite, but the overwrite is still + // materialized. + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .IgnoreEmptyCommit(true) + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + + ASSERT_OK(commit->TruncateTable(/*commit_identifier=*/1)); + + ASSERT_OK_AND_ASSIGN(bool exist, file_system_->Exists(table_path_ + "/snapshot/snapshot-1")); + ASSERT_TRUE(exist); + auto commit_impl = dynamic_cast(commit.get()); + ASSERT_TRUE(commit_impl); + ASSERT_OK_AND_ASSIGN(Snapshot snapshot, commit_impl->snapshot_manager_->LoadSnapshot(1)); + ASSERT_EQ(Snapshot::CommitKind::Overwrite(), snapshot.GetCommitKind()); + std::vector manifests; + ASSERT_OK(commit_impl->manifest_list_->ReadDeltaManifests(snapshot, &manifests)); + for (const auto& manifest : manifests) { + ASSERT_EQ(0, manifest.NumAddedFiles()); + ASSERT_EQ(0, manifest.NumDeletedFiles()); + } +} + TEST_F(FileStoreCommitImplTest, TestCreateManifestCommittable) { CommitContextBuilder context_builder(table_path_, "commit_user_1"); ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, @@ -2098,4 +2535,450 @@ TEST_F(FileStoreCommitImplTest, TestFixedBucketPKTableCommitAllowed) { ASSERT_TRUE(committer != nullptr); } +TEST_F(FileStoreCommitImplTest, ValidateCommitOptionsRejectsUnsupportedOptions) { + const std::vector unsupported_keys = { + "commit.strict-mode.last-safe-snapshot", "manifest.delete-file-drop-stats", + "sequence.snapshot-ordering", "pk-clustering-override"}; + for (const auto& key : unsupported_keys) { + ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({{key, "true"}})); + ASSERT_NOK_WITH_MSG(FileStoreCommitImpl::ValidateCommitOptions(options), + "not supported by C++ commit path"); + } + + // Supported options should validate successfully. + ASSERT_OK_AND_ASSIGN(CoreOptions ok_options, + CoreOptions::FromMap({{Options::FILE_SYSTEM, "local"}})); + ASSERT_OK(FileStoreCommitImpl::ValidateCommitOptions(ok_options)); +} + +TEST_F(FileStoreCommitImplTest, DropPartitionWithEmptyPartitionsFails) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + ASSERT_NOK_WITH_MSG(commit->DropPartition({}, /*commit_identifier=*/1), + "partitions list cannot be empty"); +} + +TEST_F(FileStoreCommitImplTest, FilterAndCommitMultipleIdentifiersAndEmptyInput) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + // Empty input map takes the FilterCommitted fast-exit path (nothing to commit). + ASSERT_OK_AND_ASSIGN(int32_t committed_empty, commit_impl->FilterAndCommit({}, 1)); + ASSERT_EQ(0, committed_empty); + + std::vector data_files = { + "/f1=10/bucket-0/data-51a45441-6037-4af3-b67b-5cefd75dc6f2-0.orc", + "/f1=10/bucket-1/data-6828284c-e707-49b5-af6b-69be79af120c-0.orc", + "/f1=20/bucket-0/data-8dc7f04c-3c98-48b2-9d56-834d746c4a40-0.orc", + "/f1=10/bucket-1/data-fd1d2255-43f2-4534-b4cc-08b29e662940-0.orc", + "/f1=20/bucket-0/data-7b3f4cc7-116b-4d2f-9c62-5dadc1f11bcb-0.orc"}; + ASSERT_OK(PrepareFakeFiles(data_files)); + + std::vector> msgs1 = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-01", + /*version=*/3); + std::vector> msgs2 = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-02", + /*version=*/3); + + // Two identifiers in a single call exercises the sort-by-identifier comparator. + std::map>> inputs; + inputs[2] = msgs2; + inputs[1] = msgs1; + ASSERT_OK_AND_ASSIGN(int32_t committed, commit_impl->FilterAndCommit(inputs, 5)); + ASSERT_EQ(2, committed); +} + +TEST_F(FileStoreCommitImplTest, CheckFilesExistenceFailsForNonImplCommitMessage) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + auto committable = std::make_shared(/*identifier=*/1); + committable->AddFileCommittable(std::make_shared()); + ASSERT_NOK_WITH_MSG(commit_impl->CheckFilesExistence({committable}), + "fail to cast commit message to impl"); +} + +TEST_F(FileStoreCommitImplTest, CheckFilesExistenceCollectsIndexFilePaths) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + const BinaryRow partition = CreateIntRow(10); + + // New index files from both the data increment and the compact increment must be + // collected for the existence check. + DataIncrement data_increment(/*new_files=*/{}, /*deleted_files=*/{}, /*changelog_files=*/{}, + /*new_index_files=*/{CreateIndexFileMeta("new-index-missing")}, + /*deleted_index_files=*/{}); + CompactIncrement compact_increment( + /*compact_before=*/{}, /*compact_after=*/{}, /*changelog_files=*/{}, + /*new_index_files=*/{CreateIndexFileMeta("compact-index-missing")}, + /*deleted_index_files=*/{}); + std::shared_ptr message = std::make_shared( + partition, /*bucket=*/0, /*total_buckets=*/2, data_increment, compact_increment); + + auto committable = std::make_shared(/*identifier=*/1); + committable->AddFileCommittable(message); + // The referenced index files were never written, so the existence check reports them missing. + ASSERT_NOK_WITH_MSG(commit_impl->CheckFilesExistence({committable}), "have been deleted"); +} + +TEST_F(FileStoreCommitImplTest, OverwriteStaticPartitionValidatesFileOwnership) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::DYNAMIC_PARTITION_OVERWRITE, "false") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + // Files whose partition matches the overwrite target are accepted. + DataIncrement matching_increment({CreateAppendDataFileMeta("static-f1-10", 1)}, {}, {}); + std::shared_ptr matching_msg = + std::make_shared(CreateIntRow(10), /*bucket=*/0, /*total_buckets=*/2, + matching_increment, CompactIncrement({}, {}, {})); + ASSERT_OK(commit_impl->Overwrite({{"f1", "10"}}, {matching_msg}, /*commit_identifier=*/1)); + + // Files belonging to a different partition than the overwrite target are rejected. + DataIncrement mismatching_increment({CreateAppendDataFileMeta("static-f1-20", 1)}, {}, {}); + std::shared_ptr mismatching_msg = + std::make_shared(CreateIntRow(20), /*bucket=*/0, /*total_buckets=*/2, + mismatching_increment, CompactIncrement({}, {}, {})); + ASSERT_NOK_WITH_MSG( + commit_impl->Overwrite({{"f1", "10"}}, {mismatching_msg}, /*commit_identifier=*/2), + "does not belong to this partition"); +} + +TEST_F(FileStoreCommitImplTest, OverwriteWithChangelogFilesLogsWarning) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + const BinaryRow partition = CreateIntRow(10); + + // Overwrite ignores changelog files but emits a warning listing them. + DataIncrement data_increment( + /*new_files=*/{CreateAppendDataFileMeta("overwrite-with-changelog", 1)}, + /*deleted_files=*/{}, + /*changelog_files=*/{CreateAppendDataFileMeta("ignored-append-changelog", 1)}, + /*new_index_files=*/{}, /*deleted_index_files=*/{}); + CompactIncrement compact_increment( + /*compact_before=*/{}, /*compact_after=*/{}, + /*changelog_files=*/{CreateAppendDataFileMeta("ignored-compact-changelog", 1)}); + std::shared_ptr message = std::make_shared( + partition, /*bucket=*/0, /*total_buckets=*/2, data_increment, compact_increment); + + ASSERT_OK(commit_impl->Overwrite({}, {message}, /*commit_identifier=*/1)); + ASSERT_OK_AND_ASSIGN(std::optional snapshot, + commit_impl->snapshot_manager_->LatestSnapshot()); + ASSERT_TRUE(snapshot.has_value()); +} + +TEST_F(FileStoreCommitImplTest, OverwriteUpgradesNonOverlappingPrimaryKeyFiles) { + auto pk_dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(pk_dir); + std::string pk_root = pk_dir->Str(); + ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(pk_root, {})); + ASSERT_OK(catalog->CreateDatabase("db", {}, false)); + + arrow::Schema pk_schema( + {arrow::field("pk", arrow::int32()), arrow::field("val", arrow::utf8())}); + ::ArrowSchema arrow_schema; + ASSERT_TRUE(arrow::ExportSchema(pk_schema, &arrow_schema).ok()); + std::map table_options = {{Options::BUCKET, "4"}}; + ASSERT_OK(catalog->CreateTable(Identifier("db", "pk_tbl"), &arrow_schema, + /*partition_keys=*/{}, /*primary_keys=*/{"pk"}, table_options, + /*ignore_if_exists=*/false)); + std::string pk_table_path = PathUtil::JoinPath(pk_root, "db.db/pk_tbl"); + + CommitContextBuilder builder(pk_table_path, "test_user"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::OVERWRITE_UPGRADE, "true") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + const BinaryRow partition = BinaryRow::EmptyRow(); + // Bucket 0: non-overlapping key ranges -> files are upgraded to a higher level. + DataIncrement bucket0_increment( + {CreateLeveledDataFileMeta("pk-b0-lo", CreateIntRow(0), CreateIntRow(10), /*level=*/0), + CreateLeveledDataFileMeta("pk-b0-hi", CreateIntRow(20), CreateIntRow(30), /*level=*/0)}, + {}, {}); + std::shared_ptr bucket0_msg = + std::make_shared(partition, /*bucket=*/0, /*total_buckets=*/4, + bucket0_increment, CompactIncrement({}, {}, {})); + // Bucket 1: overlapping key ranges -> files are kept at their original level. + DataIncrement bucket1_increment( + {CreateLeveledDataFileMeta("pk-b1-a", CreateIntRow(0), CreateIntRow(20), /*level=*/0), + CreateLeveledDataFileMeta("pk-b1-b", CreateIntRow(10), CreateIntRow(30), /*level=*/0)}, + {}, {}); + std::shared_ptr bucket1_msg = + std::make_shared(partition, /*bucket=*/1, /*total_buckets=*/4, + bucket1_increment, CompactIncrement({}, {}, {})); + + ASSERT_OK(commit_impl->Overwrite({}, {bucket0_msg, bucket1_msg}, /*commit_identifier=*/1)); + + ASSERT_OK_AND_ASSIGN(std::optional snapshot, + commit_impl->snapshot_manager_->LatestSnapshot()); + ASSERT_TRUE(snapshot.has_value()); + ASSERT_OK_AND_ASSIGN(std::vector entries, + commit_impl->GetAllFiles(snapshot.value(), {})); + ASSERT_EQ(4u, entries.size()); + + int32_t max_level = 0; + int32_t level0_count = 0; + for (const auto& entry : entries) { + max_level = std::max(max_level, entry.Level()); + if (entry.Level() == 0) { + level0_count++; + } + } + // Bucket 0 files were upgraded above level 0, bucket 1 files stayed at level 0. + ASSERT_GT(max_level, 0); + ASSERT_EQ(2, level0_count); +} + +TEST_F(FileStoreCommitImplTest, CommitWithAppendCommitCheckConflict) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .AppendCommitCheckConflict(true) + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + + std::vector> msgs = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-01", + /*version=*/3); + ASSERT_GT(msgs.size(), 0); + ASSERT_OK(commit->Commit(msgs, /*commit_identifier=*/1)); + ASSERT_OK_AND_ASSIGN( + bool exist, file_system_->Exists(PathUtil::JoinPath(table_path_, "snapshot/snapshot-1"))); + ASSERT_TRUE(exist); +} + +TEST_F(FileStoreCommitImplTest, SnapshotSequenceMaxFallsBackToManifestScan) { + // First snapshot is committed in the default (scan) mode, so it does NOT carry the + // max-sequence-number property. + { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + std::vector> msgs1 = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-01", + /*version=*/3); + ASSERT_OK(commit->Commit(msgs1, /*commit_identifier=*/1)); + } + + // Second commit uses snapshot-init mode; since the previous snapshot lacks the property, + // the max sequence number is recomputed by scanning the base manifests. + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::WRITE_SEQUENCE_NUMBER_INIT_MODE, "snapshot") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + ASSERT_OK_AND_ASSIGN(Snapshot snapshot1, commit_impl->snapshot_manager_->LoadSnapshot(1)); + if (snapshot1.Properties()) { + ASSERT_EQ(snapshot1.Properties().value().end(), + snapshot1.Properties().value().find("sequence.generation.max-sequence-number")); + } + + std::vector> msgs2 = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-02", + /*version=*/3); + ASSERT_OK(commit_impl->Commit(msgs2, /*commit_identifier=*/2)); + + ASSERT_OK_AND_ASSIGN(Snapshot snapshot2, commit_impl->snapshot_manager_->LoadSnapshot(2)); + ASSERT_TRUE(snapshot2.Properties()); + auto iter = snapshot2.Properties().value().find("sequence.generation.max-sequence-number"); + ASSERT_TRUE(iter != snapshot2.Properties().value().end()); +} + +TEST_F(FileStoreCommitImplTest, FilterAndOverwriteWithSpecifiedPartition) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + std::vector> msgs = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-01", + /*version=*/3); + // A non-empty partition spec is forwarded to the overwrite so the partition list is populated. + std::map partition_spec = {{"f1", "10"}}; + ASSERT_OK_AND_ASSIGN(int32_t actual_commit, commit_impl->FilterAndOverwrite( + partition_spec, msgs, /*commit_identifier=*/1, + /*watermark=*/10)); + ASSERT_EQ(1, actual_commit); + + ASSERT_OK_AND_ASSIGN(auto snapshot, commit_impl->snapshot_manager_->LatestSnapshot()); + ASSERT_TRUE(snapshot); + ASSERT_EQ(Snapshot::CommitKind::Overwrite(), snapshot.value().GetCommitKind()); +} + +TEST_F(FileStoreCommitImplTest, TryUpgradeReturnsInputWhenOverwriteUpgradeDisabled) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::OVERWRITE_UPGRADE, "false") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + std::vector entries = { + CreateManifestEntry("upgrade-disabled-1", FileKind::Add())}; + // overwrite-upgrade disabled: TryUpgrade returns the input unchanged. + ASSERT_OK_AND_ASSIGN(std::vector result, commit_impl->TryUpgrade(entries)); + ASSERT_EQ(entries.size(), result.size()); +} + +TEST_F(FileStoreCommitImplTest, TryUpgradeReturnsInputWhenEntryLevelAboveZero) { + auto pk_dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(pk_dir); + std::string pk_root = pk_dir->Str(); + ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(pk_root, {})); + ASSERT_OK(catalog->CreateDatabase("db", {}, false)); + + arrow::Schema pk_schema( + {arrow::field("pk", arrow::int32()), arrow::field("val", arrow::utf8())}); + ::ArrowSchema arrow_schema; + ASSERT_TRUE(arrow::ExportSchema(pk_schema, &arrow_schema).ok()); + std::map table_options = {{Options::BUCKET, "4"}}; + ASSERT_OK(catalog->CreateTable(Identifier("db", "pk_tbl"), &arrow_schema, + /*partition_keys=*/{}, /*primary_keys=*/{"pk"}, table_options, + /*ignore_if_exists=*/false)); + std::string pk_table_path = PathUtil::JoinPath(pk_root, "db.db/pk_tbl"); + + CommitContextBuilder builder(pk_table_path, "test_user"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::OVERWRITE_UPGRADE, "true") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + // A PK-table entry already at a level above 0 bypasses the upgrade and returns the input. + std::vector entries = { + CreateManifestEntry("already-upgraded", BinaryRow::EmptyRow(), FileKind::Add(), + DataFileMeta::EmptyMinKey(), DataFileMeta::EmptyMaxKey(), /*level=*/2, + /*bucket=*/0)}; + ASSERT_OK_AND_ASSIGN(std::vector result, commit_impl->TryUpgrade(entries)); + ASSERT_EQ(entries.size(), result.size()); + ASSERT_EQ(2, result[0].Level()); +} + +TEST_F(FileStoreCommitImplTest, CheckSameBucketFromSnapshotReturnsOkForEmptyDelta) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + std::vector> msgs = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-01", + /*version=*/3); + ASSERT_OK(commit_impl->Commit(msgs, /*commit_identifier=*/1)); + + ASSERT_OK_AND_ASSIGN(std::optional snapshot, + commit_impl->snapshot_manager_->LatestSnapshot()); + ASSERT_TRUE(snapshot); + // No delta entries -> no buckets to verify -> returns OK without scanning the snapshot. + ASSERT_OK(commit_impl->CheckSameBucketFromSnapshot(/*delta_entries=*/{}, snapshot)); +} + +TEST_F(FileStoreCommitImplTest, MaxSequenceNumberReturnsNulloptForEmptyManifests) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + // No manifests to scan -> no sequence number found -> returns nullopt. + ASSERT_OK_AND_ASSIGN(std::optional max_seq, commit_impl->MaxSequenceNumber({})); + ASSERT_FALSE(max_seq.has_value()); +} + +TEST_F(FileStoreCommitImplTest, RowIdCheckConflictSetsCheckSnapshotAndReturnsSelf) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = std::dynamic_pointer_cast( + std::shared_ptr(std::move(commit))); + + ASSERT_FALSE(commit_impl->conflict_detection_.HasRowIdCheckFromSnapshot()); + // RowIdCheckConflict records the snapshot to verify against and returns *this for chaining. + FileStoreCommit& returned = commit_impl->RowIdCheckConflict(/*row_id_check_from_snapshot=*/5); + ASSERT_EQ(commit_impl.get(), &returned); + ASSERT_TRUE(commit_impl->conflict_detection_.HasRowIdCheckFromSnapshot()); +} + } // namespace paimon::test