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 in 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;
 
@@ -135,6 +135,17 @@ class PAIMON_EXPORT FileStoreCommit {
     virtual Status DropPartition(const std::vector<std::map<std::string, std::string>>& partitions,
                                  int64_t commit_identifier) = 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
+    /// conflicts against snapshots after that id. Passing std::nullopt disables this behavior.
+    ///
+    /// @param row_id_check_from_snapshot Snapshot id to start row-id conflict checks from, or
+    ///     std::nullopt to disable.
+    /// @return Current commit object for chaining.
+    virtual FileStoreCommit& RowIdCheckConflict(
+        std::optional<int64_t> row_id_check_from_snapshot) = 0;
+
     /// Retrieve metrics related to commit operations.
     ///
     /// @return A shared pointer to a `Metrics` object containing commit metrics.

src/paimon/CMakeLists.txt

@@ -287,6 +287,17 @@ 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/metrics/commit_metrics.cpp
+    core/operation/commit/conflict_detection.cpp
+    core/operation/commit/commit_scanner.cpp
+    core/operation/commit/commit_changes_provider.cpp
+    core/operation/commit/compacted_changelog_path_resolver.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
@@ -318,6 +329,7 @@ set(PAIMON_CORE_SRCS
     core/stats/simple_stats.cpp
     core/stats/simple_stats_evolution.cpp
     core/table/table.cpp
+    core/table/bucket_mode.cpp
     core/table/sink/commit_message.cpp
     core/table/sink/commit_message_impl.cpp
     core/table/sink/commit_message_serializer.cpp
@@ -693,13 +705,24 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/spill_reader_writer_test.cpp
                     core/migrate/file_meta_utils_test.cpp
                     core/operation/metrics/compaction_metrics_test.cpp
+                    core/operation/metrics/commit_stats_test.cpp
                     core/operation/data_evolution_file_store_scan_test.cpp
                     core/operation/data_evolution_split_read_test.cpp
+                    core/operation/commit/compacted_changelog_path_resolver_test.cpp
+                    core/operation/commit/commit_changes_provider_test.cpp
+                    core/operation/commit/commit_scanner_test.cpp
+                    core/operation/commit/conflict_detection_test.cpp
+                    core/operation/commit/manifest_entry_changes_test.cpp
+                    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/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
@@ -728,6 +751,7 @@ if(PAIMON_BUILD_TESTS)
                     core/stats/simple_stats_collector_test.cpp
                     core/stats/simple_stats_test.cpp
                     core/table/table_test.cpp
+                    core/table/bucket_mode_test.cpp
                     core/table/sink/commit_message_test.cpp
                     core/table/sink/commit_message_impl_test.cpp
                     core/table/source/fallback_data_split_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/common/table/special_fields.h

@@ -22,6 +22,7 @@
 
 #include "arrow/type_fwd.h"
 #include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/string_utils.h"
 #include "paimon/utils/special_field_ids.h"
 
 namespace paimon {
@@ -63,6 +64,15 @@ struct SpecialFields {
         return data_field;
     }
 
+    static bool IsSystemField(const std::string& field_name) {
+        if (StringUtils::StartsWith(field_name, KEY_FIELD_PREFIX)) {
+            return true;
+        }
+        return field_name == SequenceNumber().Name() || field_name == ValueKind().Name() ||
+               field_name == RowKind().Name() || field_name == RowId().Name() ||
+               field_name == IndexScore().Name();
+    }
+
     static bool IsSpecialFieldName(const std::string& field_name) {
         if (field_name == SequenceNumber().Name() || field_name == ValueKind().Name() ||
             field_name == RowKind().Name() || field_name == RowId().Name() ||

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=*/{}));