[branch-2.0](cherry-pick) partial update should not read old fileds from rows with delete sign (#36210)

be/src/olap/partial_update_info.h

@@ -18,6 +18,7 @@
 #pragma once
 
 #include "olap/tablet_schema.h"
+#include "util/string_util.h"
 
 namespace doris {
 
@@ -27,6 +28,7 @@ struct PartialUpdateInfo {
               int64_t timestamp_ms, const std::string& timezone) {
         is_partial_update = partial_update;
         partial_update_input_columns = partial_update_cols;
+
         this->timestamp_ms = timestamp_ms;
         this->timezone = timezone;
         missing_cids.clear();
@@ -43,8 +45,37 @@ struct PartialUpdateInfo {
             }
         }
         this->is_strict_mode = is_strict_mode;
+        _generate_default_values_for_missing_cids(tablet_schema);
+    }
+
+private:
+    void _generate_default_values_for_missing_cids(const TabletSchema& tablet_schema) {
+        for (auto i = 0; i < missing_cids.size(); ++i) {
+            auto cur_cid = missing_cids[i];
+            const auto& column = tablet_schema.column(cur_cid);
+            if (column.has_default_value()) {
+                std::string default_value;
+                if (UNLIKELY(tablet_schema.column(cur_cid).type() ==
+                                     FieldType::OLAP_FIELD_TYPE_DATETIMEV2 &&
+                             to_lower(tablet_schema.column(cur_cid).default_value())
+                                             .find(to_lower("CURRENT_TIMESTAMP")) !=
+                                     std::string::npos)) {
+                    vectorized::DateV2Value<vectorized::DateTimeV2ValueType> dtv;
+                    dtv.from_unixtime(timestamp_ms / 1000, timezone);
+                    default_value = dtv.debug_string();
+                } else {
+                    default_value = tablet_schema.column(cur_cid).default_value();
+                }
+                default_values.emplace_back(default_value);
+            } else {
+                // place an empty string here
+                default_values.emplace_back();
+            }
+        }
+        CHECK_EQ(missing_cids.size(), default_values.size());
     }
 
+public:
     bool is_partial_update {false};
     std::set<std::string> partial_update_input_columns;
     std::vector<uint32_t> missing_cids;
@@ -55,5 +86,8 @@ struct PartialUpdateInfo {
     bool is_strict_mode {false};
     int64_t timestamp_ms {0};
     std::string timezone;
+
+    // default values for missing cids
+    std::vector<std::string> default_values;
 };
 } // namespace doris

be/src/olap/rowset/segment_v2/segment_writer.cpp

@@ -625,7 +625,7 @@ Status SegmentWriter::fill_missing_columns(vectorized::MutableColumns& mutable_f
     const vectorized::Int8* delete_sign_column_data = nullptr;
     if (const vectorized::ColumnWithTypeAndName* delete_sign_column =
                 old_value_block.try_get_by_name(DELETE_SIGN);
-        delete_sign_column != nullptr && _tablet_schema->has_sequence_col()) {
+        delete_sign_column != nullptr) {
         auto& delete_sign_col =
                 reinterpret_cast<const vectorized::ColumnInt8&>(*(delete_sign_column->column));
         delete_sign_column_data = delete_sign_col.get_data().data();
@@ -635,19 +635,8 @@ Status SegmentWriter::fill_missing_columns(vectorized::MutableColumns& mutable_f
         for (auto i = 0; i < cids_missing.size(); ++i) {
             const auto& column = _tablet_schema->column(cids_missing[i]);
             if (column.has_default_value()) {
-                std::string default_value;
-                if (UNLIKELY(_tablet_schema->column(cids_missing[i]).type() ==
-                                     FieldType::OLAP_FIELD_TYPE_DATETIMEV2 &&
-                             to_lower(_tablet_schema->column(cids_missing[i]).default_value())
-                                             .find(to_lower("CURRENT_TIMESTAMP")) !=
-                                     std::string::npos)) {
-                    vectorized::DateV2Value<vectorized::DateTimeV2ValueType> dtv;
-                    dtv.from_unixtime(_opts.rowset_ctx->partial_update_info->timestamp_ms / 1000,
-                                      _opts.rowset_ctx->partial_update_info->timezone);
-                    default_value = dtv.debug_string();
-                } else {
-                    default_value = _tablet_schema->column(cids_missing[i]).default_value();
-                }
+                const auto& default_value =
+                        _opts.rowset_ctx->partial_update_info->default_values[i];
                 vectorized::ReadBuffer rb(const_cast<char*>(default_value.c_str()),
                                           default_value.size());
                 old_value_block.get_by_position(i).type->from_string(

be/src/olap/tablet.cpp

@@ -3292,8 +3292,8 @@ Status Tablet::calc_segment_delete_bitmap(RowsetSharedPtr rowset,
         auto partial_update_info = rowset_writer->get_partial_update_info();
         DCHECK(partial_update_info);
         RETURN_IF_ERROR(generate_new_block_for_partial_update(
-                rowset_schema, partial_update_info->missing_cids, partial_update_info->update_cids,
-                read_plan_ori, read_plan_update, rsid_to_rowset, &block));
+                rowset_schema, partial_update_info.get(), read_plan_ori, read_plan_update,
+                rsid_to_rowset, &block));
         sort_block(block, ordered_block);
         int64_t size;
         RETURN_IF_ERROR(rowset_writer->flush_single_memtable(&ordered_block, &size));
@@ -3359,9 +3359,8 @@ std::vector<RowsetSharedPtr> Tablet::get_rowset_by_ids(
 }
 
 Status Tablet::generate_new_block_for_partial_update(
-        TabletSchemaSPtr rowset_schema, const std::vector<uint32>& missing_cids,
-        const std::vector<uint32>& update_cids, const PartialUpdateReadPlan& read_plan_ori,
-        const PartialUpdateReadPlan& read_plan_update,
+        TabletSchemaSPtr rowset_schema, const PartialUpdateInfo* partial_update_info,
+        const PartialUpdateReadPlan& read_plan_ori, const PartialUpdateReadPlan& read_plan_update,
         const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
         vectorized::Block* output_block) {
     // do partial update related works
@@ -3371,6 +3370,8 @@ Status Tablet::generate_new_block_for_partial_update(
     // 4. mark current keys deleted
     CHECK(output_block);
     auto full_mutable_columns = output_block->mutate_columns();
+    const auto& missing_cids = partial_update_info->missing_cids;
+    const auto& update_cids = partial_update_info->update_cids;
     auto old_block = rowset_schema->create_block_by_cids(missing_cids);
     auto update_block = rowset_schema->create_block_by_cids(update_cids);
 
@@ -3382,10 +3383,57 @@ Status Tablet::generate_new_block_for_partial_update(
     RETURN_IF_ERROR(read_columns_by_plan(rowset_schema, update_cids, read_plan_update,
                                          rsid_to_rowset, update_block, &read_index_update));
 
+    const vectorized::Int8* delete_sign_column_data = nullptr;
+    if (const vectorized::ColumnWithTypeAndName* delete_sign_column =
+                old_block.try_get_by_name(DELETE_SIGN);
+        delete_sign_column != nullptr) {
+        auto& delete_sign_col =
+                reinterpret_cast<const vectorized::ColumnInt8&>(*(delete_sign_column->column));
+        delete_sign_column_data = delete_sign_col.get_data().data();
+    }
+
+    // build default value block
+    auto default_value_block = old_block.clone_empty();
+    auto mutable_default_value_columns = default_value_block.mutate_columns();
+    if (delete_sign_column_data != nullptr) {
+        for (auto i = 0; i < missing_cids.size(); ++i) {
+            const auto& column = rowset_schema->column(missing_cids[i]);
+            if (column.has_default_value()) {
+                const auto& default_value = partial_update_info->default_values[i];
+                vectorized::ReadBuffer rb(const_cast<char*>(default_value.c_str()),
+                                          default_value.size());
+                RETURN_IF_ERROR(old_block.get_by_position(i).type->from_string(
+                        rb, mutable_default_value_columns[i].get()));
+            }
+        }
+    }
+
     // build full block
     CHECK(read_index_old.size() == read_index_update.size());
+
     for (auto i = 0; i < missing_cids.size(); ++i) {
+        const auto& rs_column = rowset_schema->column(missing_cids[i]);
         for (auto idx = 0; idx < read_index_old.size(); ++idx) {
+            // if the conflict update is a delete sign, which means that the key is
+            // not exist now, we should not read old values from the deleted data,
+            // and should use default value instead.
+            // NOTE: since now we are in the publishing phase, all data is commited
+            // before, even the `strict_mode` is true (which requires partial update
+            // load job can't insert new keys), this "new" key MUST be written into
+            // the new generated segment file.
+            if (delete_sign_column_data != nullptr &&
+                delete_sign_column_data[read_index_old[idx]] != 0) {
+                auto& mutable_column = full_mutable_columns[missing_cids[i]];
+                if (rs_column.has_default_value()) {
+                    mutable_column->insert_from(*mutable_default_value_columns[i].get(), 0);
+                } else if (rs_column.is_nullable()) {
+                    assert_cast<vectorized::ColumnNullable*>(mutable_column.get())
+                            ->insert_null_elements(1);
+                } else {
+                    mutable_column->insert_default();
+                }
+                continue;
+            }
             full_mutable_columns[missing_cids[i]]->insert_from(
                     *old_block.get_columns_with_type_and_name()[i].column.get(),
                     read_index_old[idx]);

be/src/olap/tablet.h

@@ -491,8 +491,8 @@ class Tablet : public BaseTablet {
     void prepare_to_read(const RowLocation& row_location, size_t pos,
                          PartialUpdateReadPlan* read_plan);
     Status generate_new_block_for_partial_update(
-            TabletSchemaSPtr rowset_schema, const std::vector<uint32>& missing_cids,
-            const std::vector<uint32>& update_cids, const PartialUpdateReadPlan& read_plan_ori,
+            TabletSchemaSPtr rowset_schema, const PartialUpdateInfo* partial_update_info,
+            const PartialUpdateReadPlan& read_plan_ori,
             const PartialUpdateReadPlan& read_plan_update,
             const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
             vectorized::Block* output_block);

regression-test/data/unique_with_mow_p0/partial_update/partial_update_parallel_with_delete_sign.csv

@@ -0,0 +1,5 @@
+1,10,1
+2,20,0
+3,30,1
+4,40,0
+5,50,1

regression-test/data/unique_with_mow_p0/partial_update/test_partial_update_delete_sign.out

@@ -10,7 +10,7 @@
 2	2	2	2	2
 4	4	4	4	4
 
--- !with_delete_sign --
+-- !1 --
 1	\N	\N	\N	\N	1
 1	1	1	1	1	0
 2	2	2	2	2	0
@@ -21,12 +21,51 @@
 5	5	5	5	5	0
 6	\N	\N	\N	\N	1
 
+-- !2 --
+1	\N	\N	\N	\N	1
+2	2	2	2	2	0
+3	\N	\N	\N	\N	1
+4	4	4	4	4	0
+5	\N	\N	\N	\N	1
+6	\N	\N	\N	\N	1
+
+-- !sql --
+1	1	1	1	1
+2	2	2	2	2
+3	3	3	3	3
+4	4	4	4	4
+5	5	5	5	5
+
+-- !after_delete --
+2	2	2	2	2
+4	4	4	4	4
+
+-- !1 --
+1	1	1	1	1	0
+1	1	1	1	1	1
+2	2	2	2	2	0
+3	3	3	3	3	0
+3	3	3	3	3	1
+4	4	4	4	4	0
+5	5	5	5	5	0
+5	5	5	5	5	1
+6	\N	\N	\N	\N	1
+
+-- !2 --
+1	1	1	1	1	1
+2	2	2	2	2	0
+3	3	3	3	3	1
+4	4	4	4	4	0
+5	5	5	5	5	1
+6	\N	\N	\N	\N	1
+
 -- !1 --
 1	1	1
 
 -- !2 --
 
 -- !3 --
+1	2	\N
 
 -- !1 --
 1	1	1	1
@@ -47,7 +86,7 @@
 2	2	2	2	2
 4	4	4	4	4
 
--- !with_delete_sign --
+-- !1 --
 1	\N	\N	\N	\N	1
 1	1	1	1	1	0
 2	2	2	2	2	0
@@ -58,12 +97,51 @@
 5	5	5	5	5	0
 6	\N	\N	\N	\N	1
 
+-- !2 --
+1	\N	\N	\N	\N	1
+2	2	2	2	2	0
+3	\N	\N	\N	\N	1
+4	4	4	4	4	0
+5	\N	\N	\N	\N	1
+6	\N	\N	\N	\N	1
+
+-- !sql --
+1	1	1	1	1
+2	2	2	2	2
+3	3	3	3	3
+4	4	4	4	4
+5	5	5	5	5
+
+-- !after_delete --
+2	2	2	2	2
+4	4	4	4	4
+
+-- !1 --
+1	1	1	1	1	0
+1	1	1	1	1	1
+2	2	2	2	2	0
+3	3	3	3	3	0
+3	3	3	3	3	1
+4	4	4	4	4	0
+5	5	5	5	5	0
+5	5	5	5	5	1
+6	\N	\N	\N	\N	1
+
+-- !2 --
+1	1	1	1	1	1
+2	2	2	2	2	0
+3	3	3	3	3	1
+4	4	4	4	4	0
+5	5	5	5	5	1
+6	\N	\N	\N	\N	1
+
 -- !1 --
 1	1	1
 
 -- !2 --
 
 -- !3 --
+1	2	\N
 
 -- !1 --
 1	1	1	1

regression-test/data/unique_with_mow_p0/partial_update/test_partial_update_delete_sign_with_conflict.out

@@ -0,0 +1,19 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+1	1	1	1	1
+2	2	2	2	2
+3	3	3	3	3
+4	4	4	4	4
+5	5	5	5	5
+
+-- !sql --
+2	20	\N	\N	foo
+4	40	\N	\N	foo
+
+-- !sql --
+1	100	10	\N	foo
+2	20	20	\N	foo
+3	100	30	\N	foo
+4	40	40	\N	foo
+5	100	50	\N	foo
+