fix: Align PartialUpdateMergeFunction with Java: add initRow/meetInsert logic (alibaba#233)

lxy-9602 · web-flow · commit e9ccec0a8085 · 2026-04-17T18:16:30.000+08:00
diff --git a/src/paimon/core/mergetree/compact/partial_update_merge_function.cpp b/src/paimon/core/mergetree/compact/partial_update_merge_function.cpp
@@ -265,6 +265,18 @@ PartialUpdateMergeFunction::CreateFieldAggregators(
     return aggregators;
 }
 
+void PartialUpdateMergeFunction::Reset() {
+    current_key_.reset();
+    meet_insert_ = false;
+    not_null_column_filled_ = false;
+    row_ = std::make_unique<GenericRow>(getters_.size());
+    last_seq_num_ = 0;
+    for (auto& [_, agg] : field_aggregators_) {
+        assert(agg);
+        agg->Reset();
+    }
+}
+
 Status PartialUpdateMergeFunction::Add(KeyValue&& moved_kv) {
     // refresh key object to avoid reference overwritten
     KeyValue kv = std::move(moved_kv);
@@ -275,33 +287,50 @@ Status PartialUpdateMergeFunction::Add(KeyValue&& moved_kv) {
         // In 0.7- versions, the delete records might be written into data file even when
         // ignore-delete configured, so ignoreDelete still needs to be checked
         if (ignore_delete_) {
+            if (!not_null_column_filled_) {
+                InitRowAndHoldData(std::move(kv.value));
+                not_null_column_filled_ = true;
+            }
             return Status::OK();
         }
+
+        last_seq_num_ = kv.sequence_number;
+
         if (field_sequence_enabled_) {
+            // RetractWithSequenceGroup handles InitRow and AddDataHolder internally
             PAIMON_RETURN_NOT_OK(RetractWithSequenceGroup(std::move(kv)));
             return Status::OK();
         }
         if (remove_record_on_delete_) {
             if (kv.value_kind == RowKind::Delete()) {
                 current_delete_row_ = true;
                 row_ = std::make_unique<GenericRow>(getters_.size());
+                InitRowAndHoldData(std::move(kv.value));
+            } else if (!not_null_column_filled_) {
+                InitRowAndHoldData(std::move(kv.value));
+                not_null_column_filled_ = true;
             }
             return Status::OK();
         }
 
         return Status::Invalid(
             "By default, Partial update can not accept delete records, you can choose one of "
-            "the following solutions:1. Configure ignore-delete to ignore delete records; 2. "
-            "Configure partial-update.remove-record-on-delete to remove the whole row when "
-            "receiving delete records; 3. Configure sequence-group to retract partial "
-            "columns.");
+            "the following solutions:\n"
+            "1. Configure 'ignore-delete' to ignore delete records.\n"
+            "2. Configure 'partial-update.remove-record-on-delete' to remove the whole row "
+            "when receiving delete records.\n"
+            "3. Configure 'sequence-group's to retract partial columns. Also configure "
+            "'partial-update.remove-record-on-sequence-group' to remove the whole row when "
+            "receiving deleted records of specified sequence group.");
     }
     last_seq_num_ = kv.sequence_number;
     if (field_comparators_.empty()) {
         UpdateNonNullFields(std::move(kv));
     } else {
         UpdateWithSequenceGroup(std::move(kv));
     }
+    meet_insert_ = true;
+    not_null_column_filled_ = true;
     return Status::OK();
 }
 
@@ -356,6 +385,15 @@ void PartialUpdateMergeFunction::UpdateWithSequenceGroup(KeyValue&& kv) {
 }
 
 Status PartialUpdateMergeFunction::RetractWithSequenceGroup(KeyValue&& kv) {
+    // Initialize row with all field values if this is the first record.
+    // InitRow only reads field values (string_view etc.) from kv.value without taking ownership.
+    // kv.value remains alive throughout this method, so the views are safe until AddDataHolder
+    // at the end transfers ownership to row_.
+    if (!not_null_column_filled_) {
+        InitRow(*(kv.value));
+        not_null_column_filled_ = true;
+    }
+
     std::set<int32_t> updated_sequence_fields;
     for (size_t i = 0; i < getters_.size(); ++i) {
         auto cmp_iter = field_comparators_.find(i);
@@ -382,6 +420,7 @@ Status PartialUpdateMergeFunction::RetractWithSequenceGroup(KeyValue&& kv) {
                                     sequence_group_partial_delete_.end()) {
                                 current_delete_row_ = true;
                                 row_ = std::make_unique<GenericRow>(getters_.size());
+                                InitRowAndHoldData(std::move(kv.value));
                                 return Status::OK();
                             } else {
                                 row_->SetField(field_idx, getters_[field_idx](*(kv.value)));
@@ -413,6 +452,18 @@ Status PartialUpdateMergeFunction::RetractWithSequenceGroup(KeyValue&& kv) {
     return Status::OK();
 }
 
+void PartialUpdateMergeFunction::InitRow(const InternalRow& value) {
+    for (size_t i = 0; i < getters_.size(); ++i) {
+        VariantType field = getters_[i](value);
+        row_->SetField(i, field);
+    }
+}
+
+void PartialUpdateMergeFunction::InitRowAndHoldData(std::unique_ptr<InternalRow>&& value) {
+    InitRow(*value);
+    row_->AddDataHolder(std::move(value));
+}
+
 bool PartialUpdateMergeFunction::IsEmptySequenceGroup(
     const KeyValue& kv, const std::shared_ptr<FieldsComparator>& comparator) const {
     for (const auto& field_idx : comparator->CompareFields()) {
diff --git a/src/paimon/core/mergetree/compact/partial_update_merge_function.h b/src/paimon/core/mergetree/compact/partial_update_merge_function.h
@@ -67,21 +67,15 @@ class PartialUpdateMergeFunction : public MergeFunction {
         const std::map<std::string, std::vector<std::string>>& value_field_to_seq_group_field,
         const std::set<std::string>& seq_group_key_set);
 
-    void Reset() override {
-        current_key_.reset();
-        row_ = std::make_unique<GenericRow>(getters_.size());
-        for (auto& [_, agg] : field_aggregators_) {
-            assert(agg);
-            agg->Reset();
-        }
-    }
+    void Reset() override;
 
     Status Add(KeyValue&& kv) override;
 
     Result<std::optional<KeyValue>> GetResult() override {
-        return std::optional<KeyValue>(
-            KeyValue(current_delete_row_ ? RowKind::Delete() : RowKind::Insert(), last_seq_num_,
-                     KeyValue::UNKNOWN_LEVEL, std::move(current_key_), std::move(row_)));
+        const RowKind* row_kind =
+            (current_delete_row_ || !meet_insert_) ? RowKind::Delete() : RowKind::Insert();
+        return std::optional<KeyValue>(KeyValue(row_kind, last_seq_num_, KeyValue::UNKNOWN_LEVEL,
+                                                std::move(current_key_), std::move(row_)));
     };
 
  private:
@@ -105,6 +99,13 @@ class PartialUpdateMergeFunction : public MergeFunction {
     bool IsEmptySequenceGroup(const KeyValue& kv,
                               const std::shared_ptr<FieldsComparator>& comparator) const;
 
+    /// Initialize row_ with all field values from the given InternalRow.
+    /// This only reads field values without taking ownership of the source row.
+    void InitRow(const InternalRow& value);
+
+    /// Initialize row_ with all field values and transfer data ownership to row_.
+    void InitRowAndHoldData(std::unique_ptr<InternalRow>&& value);
+
     void UpdateNonNullFields(KeyValue&& kv);
 
     void UpdateWithSequenceGroup(KeyValue&& kv);
@@ -124,5 +125,13 @@ class PartialUpdateMergeFunction : public MergeFunction {
     std::unique_ptr<GenericRow> row_;
     int64_t last_seq_num_ = KeyValue::UNKNOWN_SEQUENCE;
     bool current_delete_row_ = false;
+
+    /// If the first value is retract, and no insert record is received, the row kind should be
+    /// RowKind::Delete. (Partial update sequence group may not correctly set current_delete_row_
+    /// if no RowKind::Insert value is received)
+    bool meet_insert_ = false;
+
+    /// Whether the row has been initialized with non-null column values from the first record.
+    bool not_null_column_filled_ = false;
 };
 }  // namespace paimon
diff --git a/src/paimon/core/mergetree/compact/partial_update_merge_function_test.cpp b/src/paimon/core/mergetree/compact/partial_update_merge_function_test.cpp
@@ -177,13 +177,11 @@ TEST_F(PartialUpdateMergeFunctionTest, TestSequenceGroupPartialDelete) {
     Add(mfunc, RowKind::Delete(), {1, 1, 1, 3, 1, 1, NullType()});
     CheckResult(mfunc, {1, NullType(), NullType(), 3, 3, 3, 3});
     Add(mfunc, RowKind::Delete(), {1, 1, 1, 3, 1, 1, 4});
-    CheckResult(mfunc, {NullType(), NullType(), NullType(), NullType(), NullType(), NullType(),
-                        NullType()});
+    CheckResult(mfunc, {1, 1, 1, 3, 1, 1, 4});
     Add(mfunc, {1, 4, 4, 4, 5, 5, 5});
     CheckResult(mfunc, {1, 4, 4, 4, 5, 5, 5});
     Add(mfunc, RowKind::Delete(), {1, 1, 1, 6, 1, 1, 6});
-    CheckResult(mfunc, {NullType(), NullType(), NullType(), NullType(), NullType(), NullType(),
-                        NullType()});
+    CheckResult(mfunc, {1, 1, 1, 6, 1, 1, 6});
 }
 
 TEST_F(PartialUpdateMergeFunctionTest, TestMultiSequenceFields) {
@@ -794,4 +792,84 @@ TEST_F(PartialUpdateMergeFunctionTest, TestCreateFieldAggregatorsWithoutDefaultA
     // test no agg: f2
     ASSERT_TRUE(aggs.find(3) == aggs.end());
 }
+
+TEST_F(PartialUpdateMergeFunctionTest, TestSequenceGroupPartialDeleteWithProjection) {
+    std::map<std::string, std::string> options_map = {
+        {"fields.f3.sequence-group", "f1,f2"},
+        {"fields.f6.sequence-group", "f4,f5"},
+        {"partial-update.remove-record-on-sequence-group", "f3,f6"}};
+
+    auto table_fields = CreateDataFields(/*value_arity=*/7);
+    // Reordered fields: f3, f6, f0, f1, f2, f4, f5
+    std::vector<DataField> value_fields = {table_fields[3], table_fields[6], table_fields[0],
+                                           table_fields[1], table_fields[2], table_fields[4],
+                                           table_fields[5]};
+    std::vector<DataField> expected_completed_value_fields = value_fields;
+
+    auto mfunc = CreateMergeFunctionWithProjection(table_fields, value_fields, options_map,
+                                                   expected_completed_value_fields);
+    mfunc->Reset();
+    // Reordered: f3=11, f6=22, f0=100, f1=200, f2=1, f4=12, f5=21
+    Add(mfunc, {11, 22, 100, 200, 1, 12, 21});
+    Add(mfunc, RowKind::Delete(), {11, 22, 100, 200, 1, 12, 21});
+    CheckResult(mfunc, {11, 22, 100, 200, 1, 12, 21});
+}
+
+TEST_F(PartialUpdateMergeFunctionTest, TestAdjustProjectionNonProject) {
+    std::map<std::string, std::string> options_map = {{"fields.f4.sequence-group", "f1,f3"},
+                                                      {"fields.f5.sequence-group", "f7"}};
+    auto table_fields = CreateDataFields(/*value_arity=*/8);
+    // Non-projection: use all fields
+    std::vector<DataField> value_fields = table_fields;
+    std::vector<DataField> expected_completed_value_fields = table_fields;
+
+    auto mfunc = CreateMergeFunctionWithProjection(table_fields, value_fields, options_map,
+                                                   expected_completed_value_fields);
+    mfunc->Reset();
+    Add(mfunc, {1, 1, 1, 1, 1, 1, 1, 1});
+    Add(mfunc, {4, 2, 4, 2, 2, 0, NullType(), 3});
+    CheckResult(mfunc, {4, 2, 4, 2, 2, 1, 1, 1});
+}
+
+TEST_F(PartialUpdateMergeFunctionTest, TestDeleteReproduceCorrectSequenceNumber) {
+    std::map<std::string, std::string> options_map = {
+        {"partial-update.remove-record-on-delete", "true"}};
+    auto mfunc = CreateMergeFunction(/*value_arity=*/5, options_map);
+    mfunc->Reset();
+
+    Add(mfunc, {1, 1, 1, 1, 1});
+    Add(mfunc, RowKind::Delete(), {1, 1, 1, 1, 1});
+
+    ASSERT_OK_AND_ASSIGN(auto result, mfunc->GetResult());
+    ASSERT_TRUE(result.has_value());
+    ASSERT_EQ(result->sequence_number, 1);
+}
+
+TEST_F(PartialUpdateMergeFunctionTest, TestInitRowWithNullableFieldOnDelete) {
+    std::map<std::string, std::string> options_map = {
+        {"partial-update.remove-record-on-delete", "true"}};
+    // f0 and f1 are not nullable, f2 and f3 are nullable
+    std::vector<DataField> data_fields = {
+        DataField(0, arrow::field("f0", arrow::int32(), /*nullable=*/false)),
+        DataField(1, arrow::field("f1", arrow::int32(), /*nullable=*/false)),
+        DataField(2, arrow::field("f2", arrow::int32(), /*nullable=*/true)),
+        DataField(3, arrow::field("f3", arrow::int32(), /*nullable=*/true))};
+    auto value_schema = DataField::ConvertDataFieldsToArrowSchema(data_fields);
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap(options_map));
+    std::map<std::string, std::vector<std::string>> value_field_to_seq_group_field;
+    std::set<std::string> seq_group_key_set;
+    ASSERT_OK(PartialUpdateMergeFunction::ParseSequenceGroupFields(
+        options, &value_field_to_seq_group_field, &seq_group_key_set));
+    ASSERT_OK_AND_ASSIGN(auto mfunc, PartialUpdateMergeFunction::Create(
+                                         value_schema, /*primary_keys=*/{"f0"}, options,
+                                         value_field_to_seq_group_field, seq_group_key_set));
+    mfunc->Reset();
+
+    // insert some data first
+    Add(mfunc, {1, 3, 5, 7});
+    // send a DELETE with nullable field as null, triggers initRow
+    Add(mfunc, RowKind::Delete(), {1, 2, 2, NullType()});
+    // after delete with removeRecordOnDelete, row is re-initialized via initRow
+    CheckResult(mfunc, {1, 2, 2, NullType()});
+}
 }  // namespace paimon::test
