fix

Author: lucasfang

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

@@ -289,6 +289,8 @@ set(PAIMON_CORE_SRCS
     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

src/paimon/core/manifest/manifest_committable.h

@@ -40,15 +40,13 @@ class ManifestCommittable {
         : ManifestCommittable(identifier, std::nullopt) {}
 
     ManifestCommittable(int64_t identifier, std::optional<int64_t> watermark)
-        : ManifestCommittable(identifier, watermark, {}, {}, {}) {}
+        : ManifestCommittable(identifier, watermark, {}, {}) {}
 
     ManifestCommittable(int64_t identifier, std::optional<int64_t> watermark,
-                        const std::map<int32_t, int64_t>& log_offsets,
                         const std::map<std::string, std::string>& properties,
                         const std::vector<std::shared_ptr<CommitMessage>>& commit_messages)
         : identifier_(identifier),
           watermark_(watermark),
-          log_offsets_(log_offsets),
           properties_(properties),
           commit_messages_(commit_messages) {}
 
@@ -60,10 +58,6 @@ class ManifestCommittable {
         return watermark_;
     }
 
-    const std::map<int32_t, int64_t>& LogOffsets() const {
-        return log_offsets_;
-    }
-
     const std::map<std::string, std::string>& Properties() const {
         return properties_;
     }
@@ -86,29 +80,22 @@ class ManifestCommittable {
         std::string watermark_str =
             watermark_ == std::nullopt ? "null" : std::to_string(watermark_.value());
 
-        std::vector<std::string> log_offsets_str;
-        log_offsets_str.reserve(log_offsets_.size());
-        for (const auto& [key, value] : log_offsets_) {
-            log_offsets_str.emplace_back(fmt::format("{}: {}", key, value));
-        }
-
         std::vector<std::string> properties_str;
         properties_str.reserve(properties_.size());
         for (const auto& [key, value] : properties_) {
             properties_str.emplace_back(fmt::format("{}: {}", key, value));
         }
 
         return fmt::format(
-            "ManifestCommittable {{identifier = {}, watermark = {}, logOffsets = {}, "
+            "ManifestCommittable {{identifier = {}, watermark = {}, "
             "commitMessages = {}, properties = {}}}",
-            identifier_, watermark_str, fmt::join(log_offsets_str, ", "),
-            fmt::join(commit_messages_str, ", "), fmt::join(properties_str, ", "));
+            identifier_, watermark_str, fmt::join(commit_messages_str, ", "),
+            fmt::join(properties_str, ", "));
     }
 
  private:
     int64_t identifier_;
     std::optional<int64_t> watermark_;
-    std::map<int32_t, int64_t> log_offsets_;
     std::map<std::string, std::string> properties_;
     std::vector<std::shared_ptr<CommitMessage>> commit_messages_;
 };

src/paimon/core/manifest/manifest_committable_test.cpp

@@ -29,28 +29,6 @@ namespace paimon::test {
 
 class ManifestCommittableTest : public testing::Test {
  private:
-    bool IsEqualMap(const std::map<int32_t, int64_t>& actual_map,
-                    const std::map<int32_t, int64_t>& expected_map) {
-        if (expected_map.size() != actual_map.size()) {
-            return false;
-        }
-        for (const auto& kv : expected_map) {
-            const auto& key = kv.first;
-            const auto& value = kv.second;
-            auto iter = actual_map.find(key);
-            if (iter != actual_map.end()) {
-                if (iter->second == value) {
-                    continue;
-                } else {
-                    return false;
-                }
-            } else {
-                return false;
-            }
-        }
-        return true;
-    }
-
     std::vector<std::shared_ptr<CommitMessage>> GetCommitMessages(const std::string& path,
                                                                   int32_t version) const {
         auto file_system = std::make_shared<LocalFileSystem>();
@@ -99,26 +77,23 @@ TEST_F(ManifestCommittableTest, TestSimple) {
         ASSERT_EQ(committable.Watermark().value(), 456);
     }
     {
-        std::map<int32_t, int64_t> log_offsets = {{123, 444}, {234, 555}};
         std::map<std::string, std::string> properties = {};
         std::vector<std::shared_ptr<CommitMessage>> msgs =
             GetCommitMessages(paimon::test::GetDataDir() +
                                   "/orc/append_09.db/append_09/commit_messages/commit_messages-01",
                               /*version=*/3);
-        ManifestCommittable committable(/*identifier=*/123, /*watermark=*/456, log_offsets,
-                                        properties, msgs);
-        ASSERT_TRUE(IsEqualMap(committable.LogOffsets(), log_offsets));
+        ManifestCommittable committable(/*identifier=*/123, /*watermark=*/456, properties, msgs);
+        ASSERT_EQ(committable.Properties(), properties);
         ASSERT_TRUE(IsEqualMsgs(msgs, committable.FileCommittables()));
     }
     {
-        std::map<int32_t, int64_t> log_offsets = {};
         std::map<std::string, std::string> properties = {{"key1", "value1"}, {"key2", "value2"}};
         std::vector<std::shared_ptr<CommitMessage>> msgs =
             GetCommitMessages(paimon::test::GetDataDir() +
                                   "/orc/append_09.db/append_09/commit_messages/commit_messages-01",
                               /*version=*/3);
         ManifestCommittable committable(/*identifier=*/123, /*watermark=*/456,
-                                        /*log_offsets=*/{}, properties, msgs);
+                                        properties, msgs);
         ASSERT_EQ(committable.Properties(), properties);
         ASSERT_TRUE(IsEqualMsgs(msgs, committable.FileCommittables()));
     }

src/paimon/core/operation/abstract_file_store_write.cpp

@@ -281,7 +281,8 @@ int32_t AbstractFileStoreWrite::GetDefaultBucketNum() const {
 
 Result<std::shared_ptr<RestoreFiles>> AbstractFileStoreWrite::ScanExistingFileMetas(
     const BinaryRow& partition, int32_t bucket) const {
-    PAIMON_ASSIGN_OR_RAISE(auto part_values,
+    std::vector<std::pair<std::string, std::string>> part_values;
+    PAIMON_ASSIGN_OR_RAISE(part_values,
                            file_store_path_factory_->GeneratePartitionVector(partition));
     std::map<std::string, std::string> part_values_map;
     for (const auto& [key, value] : part_values) {

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2024-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/commit_changes_provider.h"
+
+#include <utility>
+
+namespace paimon {
+
+StaticCommitChangesProvider::StaticCommitChangesProvider(
+    const std::vector<ManifestEntry>& delta_files,
+    const std::vector<ManifestEntry>& changelog_files,
+    const std::vector<IndexManifestEntry>& index_entries)
+    : delta_files_(delta_files),
+      changelog_files_(changelog_files),
+      index_entries_(index_entries) {}
+
+Status StaticCommitChangesProvider::Provide(const std::optional<Snapshot>&,
+                                            std::vector<ManifestEntry>* delta_files,
+                                            std::vector<ManifestEntry>* changelog_files,
+                                            std::vector<IndexManifestEntry>* index_entries) const {
+    *delta_files = delta_files_;
+    *changelog_files = changelog_files_;
+    *index_entries = index_entries_;
+    return Status::OK();
+}
+
+}  // namespace paimon

src/paimon/core/operation/commit/commit_changes_provider.h

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2024-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 <functional>
+#include <optional>
+#include <vector>
+
+#include "paimon/core/manifest/index_manifest_entry.h"
+#include "paimon/core/manifest/manifest_entry.h"
+#include "paimon/core/snapshot.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+class CommitChangesProvider {
+ public:
+    virtual ~CommitChangesProvider() = default;
+
+    virtual Status Provide(const std::optional<Snapshot>& latest_snapshot,
+                           std::vector<ManifestEntry>* delta_files,
+                           std::vector<ManifestEntry>* changelog_files,
+                           std::vector<IndexManifestEntry>* index_entries) const = 0;
+};
+
+class StaticCommitChangesProvider final : public CommitChangesProvider {
+ public:
+    StaticCommitChangesProvider(const std::vector<ManifestEntry>& delta_files,
+                                const std::vector<ManifestEntry>& changelog_files,
+                                const std::vector<IndexManifestEntry>& index_entries);
+
+    Status Provide(const std::optional<Snapshot>& latest_snapshot,
+                   std::vector<ManifestEntry>* delta_files,
+                   std::vector<ManifestEntry>* changelog_files,
+                   std::vector<IndexManifestEntry>* index_entries) const override;
+
+ private:
+    const std::vector<ManifestEntry>& delta_files_;
+    const std::vector<ManifestEntry>& changelog_files_;
+    const std::vector<IndexManifestEntry>& index_entries_;
+};
+
+}  // namespace paimon