[Fix](partial update) abort partial update on shadow index's tablet when the including columns miss key columns on new schema

be/src/cloud/cloud_rowset_builder.cpp

@@ -51,8 +51,8 @@ Status CloudRowsetBuilder::init() {
             duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();
 
     // build tablet schema in request level
-    _build_current_tablet_schema(_req.index_id, _req.table_schema_param.get(),
-                                 *_tablet->tablet_schema());
+    RETURN_IF_ERROR(_build_current_tablet_schema(_req.index_id, _req.table_schema_param.get(),
+                                                 *_tablet->tablet_schema()));
 
     RowsetWriterContext context;
     context.txn_id = _req.txn_id;

be/src/olap/delta_writer_v2.cpp

@@ -102,8 +102,8 @@ Status DeltaWriterV2::init() {
     if (_streams.size() == 0 || _streams[0]->tablet_schema(_req.index_id) == nullptr) {
         return Status::InternalError("failed to find tablet schema for {}", _req.index_id);
     }
-    _build_current_tablet_schema(_req.index_id, _req.table_schema_param.get(),
-                                 *_streams[0]->tablet_schema(_req.index_id));
+    RETURN_IF_ERROR(_build_current_tablet_schema(_req.index_id, _req.table_schema_param.get(),
+                                                 *_streams[0]->tablet_schema(_req.index_id)));
     RowsetWriterContext context;
     context.txn_id = _req.txn_id;
     context.load_id = _req.load_id;
@@ -210,9 +210,9 @@ Status DeltaWriterV2::cancel_with_status(const Status& st) {
     return Status::OK();
 }
 
-void DeltaWriterV2::_build_current_tablet_schema(int64_t index_id,
-                                                 const OlapTableSchemaParam* table_schema_param,
-                                                 const TabletSchema& ori_tablet_schema) {
+Status DeltaWriterV2::_build_current_tablet_schema(int64_t index_id,
+                                                   const OlapTableSchemaParam* table_schema_param,
+                                                   const TabletSchema& ori_tablet_schema) {
     _tablet_schema->copy_from(ori_tablet_schema);
     // find the right index id
     int i = 0;
@@ -236,12 +236,14 @@ void DeltaWriterV2::_build_current_tablet_schema(int64_t index_id,
     }
     // set partial update columns info
     _partial_update_info = std::make_shared<PartialUpdateInfo>();
-    _partial_update_info->init(*_tablet_schema, table_schema_param->unique_key_update_mode(),
-                               table_schema_param->partial_update_input_columns(),
-                               table_schema_param->is_strict_mode(),
-                               table_schema_param->timestamp_ms(),
-                               table_schema_param->nano_seconds(), table_schema_param->timezone(),
-                               table_schema_param->auto_increment_coulumn());
+    RETURN_IF_ERROR(_partial_update_info->init(
+            _req.tablet_id, _req.txn_id, *_tablet_schema,
+            table_schema_param->unique_key_update_mode(),
+            table_schema_param->partial_update_input_columns(),
+            table_schema_param->is_strict_mode(), table_schema_param->timestamp_ms(),
+            table_schema_param->nano_seconds(), table_schema_param->timezone(),
+            table_schema_param->auto_increment_coulumn()));
+    return Status::OK();
 }
 
 } // namespace doris

be/src/olap/delta_writer_v2.h

@@ -85,9 +85,9 @@ class DeltaWriterV2 {
     Status cancel_with_status(const Status& st);
 
 private:
-    void _build_current_tablet_schema(int64_t index_id,
-                                      const OlapTableSchemaParam* table_schema_param,
-                                      const TabletSchema& ori_tablet_schema);
+    Status _build_current_tablet_schema(int64_t index_id,
+                                        const OlapTableSchemaParam* table_schema_param,
+                                        const TabletSchema& ori_tablet_schema);
 
     void _update_profile(RuntimeProfile* profile);
 

be/src/olap/partial_update_info.cpp

@@ -20,6 +20,7 @@
 #include <gen_cpp/olap_file.pb.h>
 
 #include "common/consts.h"
+#include "common/logging.h"
 #include "olap/base_tablet.h"
 #include "olap/olap_common.h"
 #include "olap/rowset/rowset.h"
@@ -32,12 +33,13 @@
 
 namespace doris {
 
-void PartialUpdateInfo::init(const TabletSchema& tablet_schema,
-                             UniqueKeyUpdateModePB unique_key_update_mode,
-                             const std::set<string>& partial_update_cols, bool is_strict_mode,
-                             int64_t timestamp_ms, int32_t nano_seconds,
-                             const std::string& timezone, const std::string& auto_increment_column,
-                             int32_t sequence_map_col_uid, int64_t cur_max_version) {
+Status PartialUpdateInfo::init(int64_t tablet_id, int64_t txn_id, const TabletSchema& tablet_schema,
+                               UniqueKeyUpdateModePB unique_key_update_mode,
+                               const std::set<string>& partial_update_cols, bool is_strict_mode,
+                               int64_t timestamp_ms, int32_t nano_seconds,
+                               const std::string& timezone,
+                               const std::string& auto_increment_column,
+                               int32_t sequence_map_col_uid, int64_t cur_max_version) {
     partial_update_mode = unique_key_update_mode;
     partial_update_input_columns = partial_update_cols;
     max_version_in_flush_phase = cur_max_version;
@@ -48,6 +50,21 @@ void PartialUpdateInfo::init(const TabletSchema& tablet_schema,
     missing_cids.clear();
     update_cids.clear();
 
+    if (partial_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {
+        // partial_update_cols should include all key columns
+        for (std::size_t i {0}; i < tablet_schema.num_key_columns(); i++) {
+            const auto key_col = tablet_schema.column(i);
+            if (!partial_update_cols.contains(key_col.name())) {
+                auto msg = fmt::format(
+                        "Unable to do partial update on shadow index's tablet, tablet_id={}, "
+                        "txn_id={}. Missing key column {}.",
+                        tablet_id, txn_id, key_col.name());
+                LOG_WARNING(msg);
+                return Status::Aborted<false>(msg);
+            }
+        }
+    }
+
     for (auto i = 0; i < tablet_schema.num_columns(); ++i) {
         if (partial_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {
             auto tablet_column = tablet_schema.column(i);
@@ -75,6 +92,7 @@ void PartialUpdateInfo::init(const TabletSchema& tablet_schema,
             is_fixed_partial_update() &&
             partial_update_input_columns.contains(auto_increment_column);
     _generate_default_values_for_missing_cids(tablet_schema);
+    return Status::OK();
 }
 
 void PartialUpdateInfo::to_pb(PartialUpdateInfoPB* partial_update_info_pb) const {

be/src/olap/partial_update_info.h

@@ -42,11 +42,12 @@ struct RowsetId;
 class BitmapValue;
 
 struct PartialUpdateInfo {
-    void init(const TabletSchema& tablet_schema, UniqueKeyUpdateModePB unique_key_update_mode,
-              const std::set<std::string>& partial_update_cols, bool is_strict_mode,
-              int64_t timestamp_ms, int32_t nano_seconds, const std::string& timezone,
-              const std::string& auto_increment_column, int32_t sequence_map_col_uid = -1,
-              int64_t cur_max_version = -1);
+    Status init(int64_t tablet_id, int64_t txn_id, const TabletSchema& tablet_schema,
+                UniqueKeyUpdateModePB unique_key_update_mode,
+                const std::set<std::string>& partial_update_cols, bool is_strict_mode,
+                int64_t timestamp_ms, int32_t nano_seconds, const std::string& timezone,
+                const std::string& auto_increment_column, int32_t sequence_map_col_uid = -1,
+                int64_t cur_max_version = -1);
     void to_pb(PartialUpdateInfoPB* partial_update_info) const;
     void from_pb(PartialUpdateInfoPB* partial_update_info);
     Status handle_non_strict_mode_not_found_error(const TabletSchema& tablet_schema,

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

@@ -60,6 +60,7 @@
 #include "service/point_query_executor.h"
 #include "util/coding.h"
 #include "util/crc32c.h"
+#include "util/debug_points.h"
 #include "util/faststring.h"
 #include "util/key_util.h"
 #include "vec/columns/column_nullable.h"
@@ -472,6 +473,8 @@ Status VerticalSegmentWriter::_partial_update_preconditions_check(size_t row_pos
 // 3. set columns to data convertor and then write all columns
 Status VerticalSegmentWriter::_append_block_with_partial_content(RowsInBlock& data,
                                                                  vectorized::Block& full_block) {
+    DBUG_EXECUTE_IF("_append_block_with_partial_content.block", DBUG_BLOCK);
+
     RETURN_IF_ERROR(_partial_update_preconditions_check(data.row_pos, false));
     // create full block and fill with input columns
     full_block = _tablet_schema->create_block();
@@ -1340,6 +1343,11 @@ void VerticalSegmentWriter::_encode_rowid(const uint32_t rowid, string* encoded_
 std::string VerticalSegmentWriter::_full_encode_keys(
         const std::vector<vectorized::IOlapColumnDataAccessor*>& key_columns, size_t pos) {
     assert(_key_index_size.size() == _num_sort_key_columns);
+    if (!(key_columns.size() == _num_sort_key_columns &&
+          _key_coders.size() == _num_sort_key_columns)) {
+        LOG_INFO("key_columns.size()={}, _key_coders.size()={}, _num_sort_key_columns={}, ",
+                 key_columns.size(), _key_coders.size(), _num_sort_key_columns);
+    }
     assert(key_columns.size() == _num_sort_key_columns &&
            _key_coders.size() == _num_sort_key_columns);
     return _full_encode_keys(_key_coders, key_columns, pos);

be/src/olap/rowset_builder.cpp

@@ -221,8 +221,8 @@ Status RowsetBuilder::init() {
         };
     })
     // build tablet schema in request level
-    _build_current_tablet_schema(_req.index_id, _req.table_schema_param.get(),
-                                 *_tablet->tablet_schema());
+    RETURN_IF_ERROR(_build_current_tablet_schema(_req.index_id, _req.table_schema_param.get(),
+                                                 *_tablet->tablet_schema()));
     RowsetWriterContext context;
     context.txn_id = _req.txn_id;
     context.load_id = _req.load_id;
@@ -396,9 +396,9 @@ Status BaseRowsetBuilder::cancel() {
     return Status::OK();
 }
 
-void BaseRowsetBuilder::_build_current_tablet_schema(int64_t index_id,
-                                                     const OlapTableSchemaParam* table_schema_param,
-                                                     const TabletSchema& ori_tablet_schema) {
+Status BaseRowsetBuilder::_build_current_tablet_schema(
+        int64_t index_id, const OlapTableSchemaParam* table_schema_param,
+        const TabletSchema& ori_tablet_schema) {
     // find the right index id
     int i = 0;
     auto indexes = table_schema_param->indexes();
@@ -438,13 +438,15 @@ void BaseRowsetBuilder::_build_current_tablet_schema(int64_t index_id,
     }
     // set partial update columns info
     _partial_update_info = std::make_shared<PartialUpdateInfo>();
-    _partial_update_info->init(
-            *_tablet_schema, table_schema_param->unique_key_update_mode(),
+    RETURN_IF_ERROR(_partial_update_info->init(
+            tablet()->tablet_id(), _req.txn_id, *_tablet_schema,
+            table_schema_param->unique_key_update_mode(),
             table_schema_param->partial_update_input_columns(),
             table_schema_param->is_strict_mode(), table_schema_param->timestamp_ms(),
             table_schema_param->nano_seconds(), table_schema_param->timezone(),
             table_schema_param->auto_increment_coulumn(),
-            table_schema_param->sequence_map_col_uid(), _max_version_in_flush_phase);
+            table_schema_param->sequence_map_col_uid(), _max_version_in_flush_phase));
+    return Status::OK();
 }
 
 } // namespace doris

be/src/olap/rowset_builder.h

@@ -84,9 +84,9 @@ class BaseRowsetBuilder {
     Status init_mow_context(std::shared_ptr<MowContext>& mow_context);
 
 protected:
-    void _build_current_tablet_schema(int64_t index_id,
-                                      const OlapTableSchemaParam* table_schema_param,
-                                      const TabletSchema& ori_tablet_schema);
+    Status _build_current_tablet_schema(int64_t index_id,
+                                        const OlapTableSchemaParam* table_schema_param,
+                                        const TabletSchema& ori_tablet_schema);
 
     virtual void _init_profile(RuntimeProfile* profile);
 

be/src/olap/schema_change.cpp

@@ -805,6 +805,7 @@ Status SchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& request) {
 
     Status res = _do_process_alter_tablet(request);
     LOG(INFO) << "finished alter tablet process, res=" << res;
+    DBUG_EXECUTE_IF("SchemaChangeJob::process_alter_tablet.leave.sleep", { sleep(5); });
     return res;
 }
 

fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java

@@ -612,6 +612,7 @@ protected void runRunningJob() throws AlterCancelException {
         Env.getCurrentEnv().getGroupCommitManager().blockTable(tableId);
         Env.getCurrentEnv().getGroupCommitManager().waitWalFinished(tableId);
         Env.getCurrentEnv().getGroupCommitManager().unblockTable(tableId);
+
         /*
          * all tasks are finished. check the integrity.
          * we just check whether all new replicas are healthy.

regression-test/data/fault_injection_p0/partial_update/test_add_key_partial_update.out

@@ -0,0 +1,17 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+1	1	1
+2	2	2
+3	3	3
+4	4	4
+5	5	5
+6	6	6
+
+-- !sql --
+1	\N	1	1	2	0
+2	\N	2	2	2	0
+3	\N	3	3	2	0
+4	\N	4	4	5	0
+5	\N	5	5	5	0
+6	\N	6	6	5	0
+

regression-test/suites/fault_injection_p0/partial_update/test_add_key_partial_update.groovy

@@ -0,0 +1,74 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+import org.junit.Assert
+import java.util.concurrent.TimeUnit
+import org.awaitility.Awaitility
+
+suite("test_add_key_partial_update", "nonConcurrent") {
+
+    def table1 = "test_add_key_partial_update"
+    sql "DROP TABLE IF EXISTS ${table1} FORCE;"
+    sql """ CREATE TABLE IF NOT EXISTS ${table1} (
+            `k1` int NOT NULL,
+            `c1` int,
+            `c2` int,
+        )UNIQUE KEY(k1)
+        DISTRIBUTED BY HASH(k1) BUCKETS 1
+        PROPERTIES (
+            "enable_mow_light_delete" = "false",
+            "disable_auto_compaction" = "true",
+            "replication_num" = "1"); """
+
+    sql "insert into ${table1} values(1,1,1),(2,2,2),(3,3,3);"
+    sql "insert into ${table1} values(4,4,4),(5,5,5),(6,6,6);"
+    sql "insert into ${table1} values(4,4,4),(5,5,5),(6,6,6);"
+    sql "insert into ${table1} values(4,4,4),(5,5,5),(6,6,6);"
+    sql "sync;"
+    order_qt_sql "select * from ${table1};"
+
+    try {
+        GetDebugPoint().clearDebugPointsForAllFEs()
+        GetDebugPoint().clearDebugPointsForAllBEs()
+
+        // block the schema change process before it change the shadow index to base index
+        GetDebugPoint().enableDebugPointForAllBEs("SchemaChangeJob::process_alter_tablet.leave.sleep")
+
+        sql "alter table ${table1} ADD COLUMN k2 int key;"
+
+        Thread.sleep(1000)
+        test {
+            sql "delete from ${table1} where k1<=3;"
+            exception "Unable to do partial update on shadow index's tablet"
+        }
+
+        waitForSchemaChangeDone {
+            sql """ SHOW ALTER TABLE COLUMN WHERE TableName='${table1}' ORDER BY createtime DESC LIMIT 1 """
+            time 1000
+        }
+
+        sql "set skip_delete_sign=true;"
+        sql "sync;"
+        qt_sql "select k1,k2,c1,c2,__DORIS_VERSION_COL__,__DORIS_DELETE_SIGN__ from ${table1} order by k1,k2,__DORIS_VERSION_COL__;"
+    } catch(Exception e) {
+        logger.info(e.getMessage())
+        throw e
+    } finally {
+        GetDebugPoint().clearDebugPointsForAllFEs()
+        GetDebugPoint().clearDebugPointsForAllBEs()
+    }
+}