Merge branch 'main' into codex/build-lumina-from-release-package

Author: lucasfang

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/commit_scanner_test.cpp

@@ -112,8 +112,8 @@ TEST_F(CommitScannerTest, TestReadAllEntriesFromChangedPartitionsEmptyFastExit)
     ASSERT_OK_AND_ASSIGN(std::vector<ManifestEntry> 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