remove copy constructor for KeyValue

Author: lszskye

src/paimon/common/data/generic_row.h

@@ -103,8 +103,8 @@ class GenericRow : public InternalRow {
         return DataDefine::IsVariantNull(fields_[pos]);
     }
 
-    void AddDataHolder(const std::shared_ptr<InternalRow>& holder) {
-        holders_.push_back(holder);
+    void AddDataHolder(std::unique_ptr<InternalRow>&& holder) {
+        holders_.push_back(std::move(holder));
     }
 
     bool GetBoolean(int32_t pos) const override {
@@ -219,7 +219,7 @@ class GenericRow : public InternalRow {
     std::vector<VariantType> fields_;
     /// As GenericRow only holds string view for string data to avoid deep copy, original data must
     /// be held in holders_
-    std::vector<std::shared_ptr<InternalRow>> holders_;
+    std::vector<std::unique_ptr<InternalRow>> holders_;
     /// The kind of change that a row describes in a changelog.
     const RowKind* kind_;
 };

src/paimon/core/key_value.h

@@ -38,29 +38,13 @@ struct KeyValue {
     KeyValue() = default;
 
     KeyValue(const RowKind* _value_kind, int64_t _sequence_number, int32_t _level,
-             std::shared_ptr<InternalRow> _key, std::shared_ptr<InternalRow> _value)
+             std::shared_ptr<InternalRow> _key, std::unique_ptr<InternalRow> _value)
         : value_kind(_value_kind),
           sequence_number(_sequence_number),
           level(_level),
           key(std::move(_key)),
           value(std::move(_value)) {}
 
-    KeyValue(const KeyValue& other) noexcept {
-        *this = other;
-    }
-
-    KeyValue& operator=(const KeyValue& other) noexcept {
-        if (&other == this) {
-            return *this;
-        }
-        value_kind = other.value_kind;
-        sequence_number = other.sequence_number;
-        level = other.level;
-        key = other.key;
-        value = other.value;
-        return *this;
-    }
-
     KeyValue(KeyValue&& other) noexcept {
         *this = std::move(other);
     }
@@ -83,7 +67,7 @@ struct KeyValue {
     // determined after read from file
     int32_t level = -1;
     std::shared_ptr<InternalRow> key;
-    std::shared_ptr<InternalRow> value;
+    std::unique_ptr<InternalRow> value;
 };
 
 struct KeyValueBatch {

src/paimon/core/mergetree/compact/lookup_changelog_merge_function_wrapper.h

@@ -70,24 +70,25 @@ class LookupChangelogMergeFunctionWrapper : public MergeFunctionWrapper<KeyValue
 
     Result<std::optional<KeyValue>> GetResult() override {
         // 1. Find the latest high level record and compute containLevel0
-        std::optional<KeyValue> high_level = merge_function_->PickHighLevel();
+        std::optional<int32_t> high_level_idx = merge_function_->PickHighLevelIdx();
 
         // 2. Lookup if latest high level record is absent
-        if (high_level == std::nullopt) {
+        if (high_level_idx == std::nullopt) {
+            std::optional<KeyValue> lookup_high_level;
             PAIMON_ASSIGN_OR_RAISE(std::optional<T> lookup_result,
                                    lookup_(merge_function_->GetKey()));
             if (lookup_result) {
                 std::string file_name;
                 int64_t row_position = -1;
                 if constexpr (std::is_same_v<T, PositionedKeyValue>) {
-                    high_level = lookup_result->key_value;
+                    lookup_high_level = std::move(lookup_result->key_value);
                     file_name = lookup_result->file_name;
                     row_position = lookup_result->row_position;
                 } else if constexpr (std::is_same_v<T, FilePosition>) {
                     file_name = lookup_result->file_name;
                     row_position = lookup_result->row_position;
                 } else if constexpr (std::is_same_v<T, KeyValue>) {
-                    high_level = lookup_result;
+                    lookup_high_level = std::move(lookup_result);
                 } else {
                     return Status::Invalid(
                         "deletion vector mode must have PositionedKeyValue or FilePosition "
@@ -98,8 +99,8 @@ class LookupChangelogMergeFunctionWrapper : public MergeFunctionWrapper<KeyValue
                         deletion_vectors_maintainer_->NotifyNewDeletion(file_name, row_position));
                 }
             }
-            if (high_level) {
-                merge_function_->InsertInto(std::move(high_level), comparator_);
+            if (lookup_high_level) {
+                merge_function_->InsertInto(std::move(lookup_high_level), comparator_);
             }
         }
 

src/paimon/core/mergetree/compact/lookup_merge_function.h

@@ -58,21 +58,13 @@ class LookupMergeFunction : public MergeFunction {
         return current_key_;
     }
 
-    std::optional<KeyValue> PickHighLevel() const {
-        int32_t idx = PickHighLevelIdx();
-        if (idx == -1) {
-            return std::optional<KeyValue>();
-        }
-        return candidates_[idx];
-    }
-
     Result<std::optional<KeyValue>> GetResult() override {
         merge_function_->Reset();
-        int32_t high_level_idx = PickHighLevelIdx();
+        std::optional<int32_t> high_level_idx = PickHighLevelIdx();
         for (int32_t i = 0; i < static_cast<int32_t>(candidates_.size()); ++i) {
             // records that has not been stored on the disk yet, such as the data in the write
             // buffer being at level -1
-            if (candidates_[i].level <= 0 || i == high_level_idx) {
+            if (candidates_[i].level <= 0 || i == high_level_idx.value()) {
                 PAIMON_RETURN_NOT_OK(merge_function_->Add(std::move(candidates_[i])));
             }
         }
@@ -88,9 +80,8 @@ class LookupMergeFunction : public MergeFunction {
         std::sort(candidates_.begin(), candidates_.end(), cmp_function);
     }
 
- private:
-    int32_t PickHighLevelIdx() const {
-        int32_t high_level_idx = -1;
+    std::optional<int32_t> PickHighLevelIdx() const {
+        std::optional<int32_t> high_level_idx;
         for (int32_t i = 0; i < static_cast<int32_t>(candidates_.size()); i++) {
             const auto& kv = candidates_[i];
             // records that has not been stored on the disk yet, such as the data in the write
@@ -100,7 +91,7 @@ class LookupMergeFunction : public MergeFunction {
             }
             // For high-level comparison logic (not involving Level 0), only the value of the
             // minimum Level should be selected
-            if (high_level_idx == -1 || kv.level < candidates_[high_level_idx].level) {
+            if (!high_level_idx || kv.level < candidates_[high_level_idx.value()].level) {
                 high_level_idx = i;
             }
         }

src/paimon/core/mergetree/compact/lookup_merge_function_test.cpp

@@ -127,7 +127,7 @@ TEST(LookupMergeFunctionTest, TestPickHighLevel) {
     auto pool = GetDefaultPool();
 
     merge_func->Reset();
-    ASSERT_FALSE(merge_func->PickHighLevel());
+    ASSERT_FALSE(merge_func->PickHighLevelIdx());
     ASSERT_FALSE(merge_func->ContainLevel0());
 
     merge_func->Reset();
@@ -145,12 +145,8 @@ TEST(LookupMergeFunctionTest, TestPickHighLevel) {
     ASSERT_OK(merge_func->Add(std::move(kv2)));
     ASSERT_OK(merge_func->Add(std::move(kv3)));
     ASSERT_TRUE(merge_func->ContainLevel0());
-    ASSERT_EQ(merge_func->GetKey()->GetInt(0), 10);
-    auto result_kv = std::move(merge_func->PickHighLevel().value());
-    KeyValue expected(RowKind::Insert(), /*sequence_number=*/2, /*level=*/1, /*key=*/
-                      BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
-                      /*value=*/BinaryRowGenerator::GenerateRowPtr({10, 200}, pool.get()));
-    KeyValueChecker::CheckResult(expected, result_kv, /*key_arity=*/1, /*value_arity=*/2);
+    ASSERT_EQ(merge_func->GetKey()->GetInt(0), 10);    
+    ASSERT_EQ(merge_func->PickHighLevelIdx(), 1);
 }
 
 TEST(LookupMergeFunctionTest, TestInsertInto) {