feat(commit): support TruncateTable/Abort/RollbackToAsLatest and add ut

include/paimon/file_store_commit.h

@@ -135,6 +135,30 @@ class PAIMON_EXPORT FileStoreCommit {
     virtual Status DropPartition(const std::vector<std::map<std::string, std::string>>& 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<std::shared_ptr<CommitMessage>>& 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<bool>; 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<bool> 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

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

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 <string>
+
+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

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));

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<LongCounter> AppendDataFileWriterFactory::ResolveSeqNumCounter() const {
+    return options_.DataEvolutionEnabled() ? std::make_shared<LongCounter>(0) : seq_num_counter_;
+}
+
 Result<std::unique_ptr<SingleFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>>
 AppendDataFileWriterFactory::CreateWriter() const {
-    std::shared_ptr<LongCounter> seq_num_counter =
-        options_.DataEvolutionEnabled() ? std::make_shared<LongCounter>(0) : seq_num_counter_;
+    std::shared_ptr<LongCounter> seq_num_counter = ResolveSeqNumCounter();
     PAIMON_ASSIGN_OR_RAISE(WriterResources resources,
                            CreateWriterResources(*options_.GetFileFormat(), write_schema_,
                                                  /*create_stats_extractor=*/true));

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<LongCounter> ResolveSeqNumCounter() const;
+
     std::shared_ptr<arrow::Schema> write_schema_;
     std::optional<std::vector<std::string>> write_cols_;
     std::shared_ptr<LongCounter> seq_num_counter_;

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<LongCounter> seq_num_counter =
-        options_.DataEvolutionEnabled() ? std::make_shared<LongCounter>(0) : seq_num_counter_;
+    std::shared_ptr<LongCounter> seq_num_counter = ResolveSeqNumCounter();
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<MapSharedShreddingBatchConverter> converter,
                            MapSharedShreddingBatchConverter::Create(
                                write_schema_, shredding_context_, options_, pool_));

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

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 =

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

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 {

src/paimon/core/operation/commit/row_tracking_commit_utils.cpp

@@ -23,17 +23,14 @@
 #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"
 
 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<DataFileMeta>(*entry.File());
     return ManifestEntry(entry.Kind(), entry.Partition(), entry.Bucket(), entry.TotalBuckets(),
@@ -124,7 +121,7 @@ Result<int64_t> 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 {} "