feat(commit): align functionality with java paimon commit

include/paimon/commit_context.h

@@ -37,7 +37,7 @@ class PAIMON_EXPORT CommitContext {
  public:
     CommitContext(const std::string& root_path, const std::string& commit_user,
                   bool ignore_empty_commit, bool use_rest_catalog_commit,
-                  const std::shared_ptr<MemoryPool>& memory_pool,
+                  bool append_commit_check_conflict, const std::shared_ptr<MemoryPool>& memory_pool,
                   const std::shared_ptr<Executor>& executor,
                   const std::shared_ptr<FileSystem>& specific_file_system,
                   const std::map<std::string, std::string>& options);
@@ -59,6 +59,10 @@ class PAIMON_EXPORT CommitContext {
         return use_rest_catalog_commit_;
     }
 
+    bool AppendCommitCheckConflict() const {
+        return append_commit_check_conflict_;
+    }
+
     std::shared_ptr<MemoryPool> GetMemoryPool() const {
         return memory_pool_;
     }
@@ -80,6 +84,7 @@ class PAIMON_EXPORT CommitContext {
     std::string commit_user_;
     bool ignore_empty_commit_;
     bool use_rest_catalog_commit_;
+    bool append_commit_check_conflict_;
     std::shared_ptr<MemoryPool> memory_pool_;
     std::shared_ptr<Executor> executor_;
     std::shared_ptr<FileSystem> specific_file_system_;
@@ -125,6 +130,11 @@ class PAIMON_EXPORT CommitContextBuilder {
     /// @return Reference to this builder for method chaining.
     CommitContextBuilder& UseRESTCatalogCommit(bool use_rest_catalog_commit);
 
+    /// Sets whether append commits should perform conflict checking (default is false).
+    /// @param append_commit_check_conflict True to enable append conflict checks.
+    /// @return Reference to this builder for method chaining.
+    CommitContextBuilder& AppendCommitCheckConflict(bool append_commit_check_conflict);
+
     /// Sets the memory pool to be used for memory allocation during commit operations.
     /// @param memory_pool Shared pointer to the memory pool instance.
     /// @return Reference to this builder for method chaining.

include/paimon/defs.h

@@ -230,6 +230,16 @@ struct PAIMON_EXPORT Options {
     /// "commit.max-retries" - Maximum number of retries when commit failed. Default value is 10.
     static const char COMMIT_MAX_RETRIES[];
 
+    /// "commit.min-retry-wait" - Min retry wait time when commit failed. Default value is 10ms.
+    static const char COMMIT_MIN_RETRY_WAIT[];
+
+    /// "commit.max-retry-wait" - Max retry wait time when commit failed. Default value is 10s.
+    static const char COMMIT_MAX_RETRY_WAIT[];
+
+    /// "commit.discard-duplicate-files" - Whether to discard duplicate files on APPEND commit.
+    /// Default value is "false".
+    static const char COMMIT_DISCARD_DUPLICATE_FILES[];
+
     /// "compaction.max-size-amplification-percent" - The size amplification is defined as the
     /// amount (in percentage) of additional storage needed to store a single byte of data in the
     /// merge tree for changelog mode table. Default value is 200.
@@ -265,6 +275,15 @@ struct PAIMON_EXPORT Options {
     /// level 0 files in candidates. Default value is false.
     static const char COMPACTION_FORCE_UP_LEVEL_0[];
 
+    /// "overwrite-upgrade" - Whether to try upgrading the data files after overwriting a
+    /// primary key table. Default value is true.
+    static const char OVERWRITE_UPGRADE[];
+
+    /// "dynamic-partition-overwrite" - Whether only overwrite dynamic partition when
+    /// overwriting a partitioned table with dynamic partition columns. Works only when
+    /// the table has partition keys. Default value is true.
+    static const char DYNAMIC_PARTITION_OVERWRITE[];
+
     /// "lookup-compact.max-interval" - The max interval for a gentle mode lookup compaction to be
     /// triggered. For every interval, a forced lookup compaction will be performed to flush L0
     /// files to higher level. This option is only valid when lookup-compact mode is gentle. No
@@ -444,6 +463,12 @@ struct PAIMON_EXPORT Options {
     /// "write-only" - If set to "true", compactions and snapshot expiration will be skipped. This
     /// option is used along with dedicated compact jobs. Default value is "false".
     static const char WRITE_ONLY[];
+    /// "bucket-append-ordered" - Whether append writes in fixed bucket mode are ordered. This
+    /// option is used by commit conflict checks. Default value is "false".
+    static const char BUCKET_APPEND_ORDERED[];
+    /// "write.sequence-number-init-mode" - Specify how to initialize the next sequence number for
+    /// primary key table writers. Values can be: "scan", "snapshot". Default value is "scan".
+    static const char WRITE_SEQUENCE_NUMBER_INIT_MODE[];
     /// "compaction.min.file-num" - For file set [f_0,...,f_N], the minimum file number to trigger a
     /// compaction for append-only table. Default value is 5.
     static const char COMPACTION_MIN_FILE_NUM[];

include/paimon/file_store_commit.h

@@ -83,7 +83,7 @@ class PAIMON_EXPORT FileStoreCommit {
 
     /// Overwrite from manifest committable and partition.
     ///
-    /// @param partitions A single partition maps each partition key to a partition value. Depending
+    /// @param partition A single partition maps each partition key to a partition value. Depending
     ///     on the user-defined statement, the partition might not include all partition keys. Also
     ///     note that this partition does not necessarily equal to the partitions of the newly added
     ///     key-values. This is just the partition to be cleaned up.
@@ -92,22 +92,22 @@ class PAIMON_EXPORT FileStoreCommit {
     /// @param watermark An optional event-time watermark used to indicate the progress of data
     ///     processing. Default is std::nullopt.
     /// @return Result of the operation.
-    virtual Status Overwrite(const std::vector<std::map<std::string, std::string>>& partitions,
+    virtual Status Overwrite(const std::map<std::string, std::string>& partition,
                              const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
                              int64_t commit_identifier,
                              std::optional<int64_t> watermark = std::nullopt) = 0;
 
     /// This is a temporary interface for internal use. It will be removed in a future version.
     /// Please do not rely on it for long-term use.
     ///
-    /// @param partitions Description of the partitions.
+    /// @param partition Description of the partition.
     /// @param commit_messages Description of the commit messages.
     /// @param commit_identifier Unique identifier.
     /// @param watermark An optional event-time watermark used to indicate the progress of data
     ///     processing. Default is std::nullopt.
     /// @return Result of the operation.
     virtual Result<int32_t> FilterAndOverwrite(
-        const std::vector<std::map<std::string, std::string>>& partitions,
+        const std::map<std::string, std::string>& partition,
         const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
         int64_t commit_identifier, std::optional<int64_t> watermark = std::nullopt) = 0;
 

src/paimon/CMakeLists.txt

@@ -287,6 +287,15 @@ set(PAIMON_CORE_SRCS
     core/operation/abstract_split_read.cpp
     core/operation/append_only_file_store_scan.cpp
     core/operation/append_only_file_store_write.cpp
+    core/operation/commit/conflict_detection.cpp
+    core/operation/commit/commit_scanner.cpp
+    core/operation/commit/commit_changes_provider.cpp
+    core/operation/commit/overwrite_changes_provider.cpp
+    core/operation/commit/row_id_column_conflict_checker.cpp
+    core/operation/commit/manifest_entry_changes.cpp
+    core/operation/commit/row_tracking_commit_utils.cpp
+    core/operation/commit/sequence_snapshot_properties.cpp
+    core/operation/commit/retry_waiter.cpp
     core/operation/commit_context.cpp
     core/operation/expire_snapshots.cpp
     core/operation/file_store_commit.cpp
@@ -695,11 +704,16 @@ if(PAIMON_BUILD_TESTS)
                     core/operation/metrics/compaction_metrics_test.cpp
                     core/operation/data_evolution_file_store_scan_test.cpp
                     core/operation/data_evolution_split_read_test.cpp
+                    core/operation/commit/conflict_detection_test.cpp
+                    core/operation/commit/manifest_entry_changes_test.cpp
+                    core/operation/commit/row_tracking_commit_utils_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
                     core/operation/abstract_split_read_test.cpp
                     core/operation/append_only_file_store_write_test.cpp
                     core/operation/commit_metrics_test.cpp
+                    core/operation/commit_context_test.cpp
                     core/operation/expire_snapshots_test.cpp
                     core/operation/file_store_commit_impl_test.cpp
                     core/operation/file_store_commit_test.cpp

src/paimon/common/data/binary_row.h

@@ -156,6 +156,19 @@ struct hash<std::tuple<paimon::BinaryRow, int32_t, std::string>> {
     }
 };
 
+/// for std::unordered_map<std::tuple<paimon::BinaryRow, int32_t, int32_t>, ...>
+template <>
+struct hash<std::tuple<paimon::BinaryRow, int32_t, int32_t>> {
+    size_t operator()(
+        const std::tuple<paimon::BinaryRow, int32_t, int32_t>& partition_bucket_level) const {
+        const auto& [partition, bucket, level] = partition_bucket_level;
+        size_t hash = paimon::MurmurHashUtils::HashUnsafeBytes(
+            reinterpret_cast<const void*>(&bucket), 0, sizeof(bucket), partition.HashCode());
+        return paimon::MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const void*>(&level), 0,
+                                                        sizeof(level), hash);
+    }
+};
+
 template <>
 struct hash<paimon::BinaryRow> {
     size_t operator()(const paimon::BinaryRow& row) const {

src/paimon/common/defs.cpp

@@ -63,6 +63,9 @@ const char Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES[] = "snapshot.clean-empty-d
 const char Options::COMMIT_FORCE_COMPACT[] = "commit.force-compact";
 const char Options::COMMIT_TIMEOUT[] = "commit.timeout";
 const char Options::COMMIT_MAX_RETRIES[] = "commit.max-retries";
+const char Options::COMMIT_MIN_RETRY_WAIT[] = "commit.min-retry-wait";
+const char Options::COMMIT_MAX_RETRY_WAIT[] = "commit.max-retry-wait";
+const char Options::COMMIT_DISCARD_DUPLICATE_FILES[] = "commit.discard-duplicate-files";
 const char Options::SEQUENCE_FIELD[] = "sequence.field";
 const char Options::SEQUENCE_FIELD_SORT_ORDER[] = "sequence.field.sort-order";
 const char Options::MERGE_ENGINE[] = "merge-engine";
@@ -111,6 +114,8 @@ const char Options::SCAN_TIMESTAMP_MILLIS[] = "scan.timestamp-millis";
 const char Options::SCAN_TIMESTAMP[] = "scan.timestamp";
 const char Options::SCAN_TAG_NAME[] = "scan.tag-name";
 const char Options::WRITE_ONLY[] = "write-only";
+const char Options::BUCKET_APPEND_ORDERED[] = "bucket-append-ordered";
+const char Options::WRITE_SEQUENCE_NUMBER_INIT_MODE[] = "write.sequence-number-init-mode";
 const char Options::COMPACTION_MIN_FILE_NUM[] = "compaction.min.file-num";
 const char Options::COMPACTION_FORCE_REWRITE_ALL_FILES[] = "compaction.force-rewrite-all-files";
 const char Options::COMPACTION_OPTIMIZATION_INTERVAL[] = "compaction.optimization-interval";
@@ -135,6 +140,8 @@ const char Options::NUM_SORTED_RUNS_COMPACTION_TRIGGER[] = "num-sorted-run.compa
 const char Options::NUM_SORTED_RUNS_STOP_TRIGGER[] = "num-sorted-run.stop-trigger";
 const char Options::NUM_LEVELS[] = "num-levels";
 const char Options::COMPACTION_FORCE_UP_LEVEL_0[] = "compaction.force-up-level-0";
+const char Options::OVERWRITE_UPGRADE[] = "overwrite-upgrade";
+const char Options::DYNAMIC_PARTITION_OVERWRITE[] = "dynamic-partition-overwrite";
 const char Options::LOOKUP_WAIT[] = "lookup-wait";
 const char Options::LOOKUP_COMPACT[] = "lookup-compact";
 const char Options::LOOKUP_COMPACT_MAX_INTERVAL[] = "lookup-compact.max-interval";

src/paimon/core/catalog/commit_table_request_test.cpp

@@ -37,7 +37,7 @@ TEST(CommitTableRequestTest, TestSimple) {
         /*changelog_manifest_list_size=*/std::nullopt, /*index_manifest=*/std::nullopt,
         /*commit_user=*/"commit_user_1", /*commit_identifier=*/9223372036854775807,
         /*commit_kind=*/Snapshot::CommitKind::Append(), /*time_millis=*/1758097357597,
-        /*log_offsets=*/std::map<int32_t, int64_t>(), /*total_record_count=*/5,
+        /*total_record_count=*/5,
         /*delta_record_count=*/5, /*changelog_record_count=*/0, /*watermark=*/std::nullopt,
         /*statistics=*/std::nullopt, /*properties=*/std::nullopt, /*next_row_id=*/0);
     std::vector<PartitionStatistics> partition_statistics = {
@@ -61,7 +61,6 @@ TEST(CommitTableRequestTest, TestSimple) {
         "commitIdentifier": 9223372036854775807,
         "commitKind": "APPEND",
         "timeMillis": 1758097357597,
-        "logOffsets": {},
         "totalRecordCount": 5,
         "deltaRecordCount": 5,
         "changelogRecordCount": 0,

src/paimon/core/catalog/renaming_snapshot_commit_test.cpp

@@ -46,7 +46,7 @@ TEST(RenamingSnapshotCommitTest, TestSimple) {
         /*changelog_manifest_list_size=*/std::nullopt, /*index_manifest=*/std::nullopt,
         /*commit_user=*/"commit_user_1", /*commit_identifier=*/9223372036854775807,
         /*commit_kind=*/Snapshot::CommitKind::Append(), /*time_millis=*/1758097357597,
-        /*log_offsets=*/std::map<int32_t, int64_t>(), /*total_record_count=*/5,
+        /*total_record_count=*/5,
         /*delta_record_count=*/5, /*changelog_record_count=*/0, /*watermark=*/std::nullopt,
         /*statistics=*/std::nullopt, /*properties=*/std::nullopt, /*next_row_id=*/0);
     ASSERT_OK_AND_ASSIGN(bool success, commit->Commit(snapshot, /*statistics=*/{}));

src/paimon/core/core_options.cpp

@@ -368,6 +368,8 @@ struct CoreOptions::Impl {
     int64_t manifest_full_compaction_file_size = 16 * 1024 * 1024;
     int64_t write_buffer_size = 256 * 1024 * 1024;
     int64_t commit_timeout = std::numeric_limits<int64_t>::max();
+    int64_t commit_min_retry_wait = 10;
+    int64_t commit_max_retry_wait = 10 * 1000;
 
     std::shared_ptr<FileFormat> file_format;
     std::shared_ptr<FileSystem> file_system;
@@ -427,6 +429,9 @@ struct CoreOptions::Impl {
     bool ignore_delete = false;
     bool write_buffer_spillable = true;
     bool write_only = false;
+    bool bucket_append_ordered = false;
+    CoreOptions::SequenceNumberInitMode write_sequence_number_init_mode =
+        CoreOptions::SequenceNumberInitMode::SCAN;
     bool deletion_vectors_enabled = false;
     bool deletion_vectors_bitmap64 = false;
     bool force_lookup = false;
@@ -445,6 +450,9 @@ struct CoreOptions::Impl {
     bool global_index_enabled = true;
     std::optional<int32_t> global_index_thread_num;
     bool commit_force_compact = false;
+    bool commit_discard_duplicate_files = false;
+    bool dynamic_partition_overwrite = true;
+    bool overwrite_upgrade = true;
     bool compaction_force_rewrite_all_files = false;
     bool compaction_force_up_level_0 = false;
     std::optional<std::string> global_index_external_path;
@@ -519,6 +527,9 @@ struct CoreOptions::Impl {
                                                     specified_file_system, &file_system));
         // Parse write-only - if true, compactions and snapshot expiration will be skipped
         PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::WRITE_ONLY, &write_only));
+        // Parse bucket-append-ordered - append writes in fixed-bucket mode are ordered
+        PAIMON_RETURN_NOT_OK(
+            parser.Parse<bool>(Options::BUCKET_APPEND_ORDERED, &bucket_append_ordered));
         // Parse partition.legacy-name - use legacy ToString for partition names, default true
         PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::PARTITION_GENERATE_LEGACY_NAME,
                                                 &legacy_partition_name_enabled));
@@ -637,6 +648,22 @@ struct CoreOptions::Impl {
         PAIMON_RETURN_NOT_OK(parser.ParseTimeDuration(Options::COMMIT_TIMEOUT, &commit_timeout));
         // Parse commit.max-retries - maximum retries when commit failed, default 10
         PAIMON_RETURN_NOT_OK(parser.Parse(Options::COMMIT_MAX_RETRIES, &commit_max_retries));
+        // Parse commit.min-retry-wait - minimum retry wait when commit failed, default 10ms
+        PAIMON_RETURN_NOT_OK(
+            parser.ParseTimeDuration(Options::COMMIT_MIN_RETRY_WAIT, &commit_min_retry_wait));
+        // Parse commit.max-retry-wait - maximum retry wait when commit failed, default 10s
+        PAIMON_RETURN_NOT_OK(
+            parser.ParseTimeDuration(Options::COMMIT_MAX_RETRY_WAIT, &commit_max_retry_wait));
+        // Parse commit.discard-duplicate-files - whether to discard duplicate files on append
+        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::COMMIT_DISCARD_DUPLICATE_FILES,
+                                                &commit_discard_duplicate_files));
+        // Parse dynamic-partition-overwrite - whether overwrite only dynamic partitions
+        // for partitioned table overwrite.
+        PAIMON_RETURN_NOT_OK(
+            parser.Parse<bool>(Options::DYNAMIC_PARTITION_OVERWRITE, &dynamic_partition_overwrite));
+        // Parse overwrite-upgrade - whether to try upgrading data files after overwrite on
+        // primary key table
+        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::OVERWRITE_UPGRADE, &overwrite_upgrade));
         return Status::OK();
     }
 
@@ -647,6 +674,19 @@ 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
+        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));
+        write_sequence_init_mode_str = StringUtils::ToLowerCase(write_sequence_init_mode_str);
+        if (write_sequence_init_mode_str == "scan") {
+            write_sequence_number_init_mode = CoreOptions::SequenceNumberInitMode::SCAN;
+        } else if (write_sequence_init_mode_str == "snapshot") {
+            write_sequence_number_init_mode = CoreOptions::SequenceNumberInitMode::SNAPSHOT;
+        } else {
+            return Status::Invalid(fmt::format("invalid write sequence number init mode: {}",
+                                               write_sequence_init_mode_str));
+        }
         // Parse sort-engine - sort engine for primary key table, default "loser-tree"
         PAIMON_RETURN_NOT_OK(parser.ParseSortEngine(&sort_engine));
         // Parse merge-engine - merge engine for primary key table, default "deduplicate"
@@ -1037,6 +1077,26 @@ int32_t CoreOptions::GetCommitMaxRetries() const {
     return impl_->commit_max_retries;
 }
 
+int64_t CoreOptions::GetCommitMinRetryWait() const {
+    return impl_->commit_min_retry_wait;
+}
+
+int64_t CoreOptions::GetCommitMaxRetryWait() const {
+    return impl_->commit_max_retry_wait;
+}
+
+bool CoreOptions::CommitDiscardDuplicateFiles() const {
+    return impl_->commit_discard_duplicate_files;
+}
+
+bool CoreOptions::DynamicPartitionOverwrite() const {
+    return impl_->dynamic_partition_overwrite;
+}
+
+bool CoreOptions::OverwriteUpgrade() const {
+    return impl_->overwrite_upgrade;
+}
+
 int32_t CoreOptions::GetCompactionMinFileNum() const {
     return impl_->compaction_min_file_num;
 }
@@ -1133,6 +1193,14 @@ bool CoreOptions::WriteOnly() const {
     return impl_->write_only;
 }
 
+bool CoreOptions::BucketAppendOrdered() const {
+    return impl_->bucket_append_ordered;
+}
+
+CoreOptions::SequenceNumberInitMode CoreOptions::WriteSequenceNumberInitMode() const {
+    return impl_->write_sequence_number_init_mode;
+}
+
 std::optional<std::string> CoreOptions::GetFieldsDefaultFunc() const {
     return impl_->field_default_func;
 }