Coverage Report

Created: 2026-07-27 18:41

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/partial_update_info.cpp
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "storage/partial_update_info.h"
19
20
#include <gen_cpp/olap_file.pb.h>
21
22
#include <cstdint>
23
#include <optional>
24
25
#include "common/consts.h"
26
#include "common/logging.h"
27
#include "core/assert_cast.h"
28
#include "core/block/block.h"
29
#include "core/data_type/data_type_number.h" // IWYU pragma: keep
30
#include "core/value/bitmap_value.h"
31
#include "storage/iterator/olap_data_convertor.h"
32
#include "storage/key/row_key_encoder.h"
33
#include "storage/olap_common.h"
34
#include "storage/rowset/rowset.h"
35
#include "storage/rowset/rowset_writer_context.h"
36
#include "storage/segment/historical_row_retriever.h"
37
#include "storage/segment/vertical_segment_writer.h"
38
#include "storage/tablet/base_tablet.h"
39
#include "storage/tablet/tablet_meta.h"
40
#include "storage/tablet/tablet_schema.h"
41
#include "storage/utils.h"
42
43
namespace doris {
44
namespace {
45
46
290
ColumnBitmap* get_mutable_skip_bitmap_column(Block* block, size_t skip_bitmap_col_idx) {
47
290
    auto skip_bitmap_column =
48
290
            IColumn::mutate(std::move(block->get_by_position(skip_bitmap_col_idx).column));
49
290
    auto* skip_bitmap_column_ptr = assert_cast<ColumnBitmap*>(skip_bitmap_column.get());
50
290
    block->replace_by_position(skip_bitmap_col_idx, std::move(skip_bitmap_column));
51
290
    return skip_bitmap_column_ptr;
52
290
}
53
54
} // namespace
55
56
Status PartialUpdateInfo::init(int64_t tablet_id, int64_t txn_id, const TabletSchema& tablet_schema,
57
                               UniqueKeyUpdateModePB unique_key_update_mode,
58
                               PartialUpdateNewRowPolicyPB policy,
59
                               const std::set<std::string>& partial_update_cols,
60
                               bool is_strict_mode_, int64_t timestamp_ms_, int32_t nano_seconds_,
61
                               const std::string& timezone_,
62
                               const std::string& auto_increment_column,
63
228k
                               int32_t sequence_map_col_uid, int64_t cur_max_version) {
64
228k
    partial_update_mode = unique_key_update_mode;
65
228k
    partial_update_new_key_policy = policy;
66
228k
    partial_update_input_columns = partial_update_cols;
67
228k
    max_version_in_flush_phase = cur_max_version;
68
228k
    sequence_map_col_unqiue_id = sequence_map_col_uid;
69
228k
    timestamp_ms = timestamp_ms_;
70
228k
    nano_seconds = nano_seconds_;
71
228k
    timezone = timezone_;
72
228k
    missing_cids.clear();
73
228k
    update_cids.clear();
74
75
228k
    if (partial_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {
76
        // partial_update_cols should include all key columns
77
12.1k
        for (std::size_t i {0}; i < tablet_schema.num_key_columns(); i++) {
78
8.38k
            const auto key_col = tablet_schema.column(i);
79
8.38k
            if (!partial_update_cols.contains(key_col.name())) {
80
0
                auto msg = fmt::format(
81
0
                        "Unable to do partial update on shadow index's tablet, tablet_id={}, "
82
0
                        "txn_id={}. Missing key column {}.",
83
0
                        tablet_id, txn_id, key_col.name());
84
0
                LOG_WARNING(msg);
85
0
                return Status::Aborted<false>(msg);
86
0
            }
87
8.38k
        }
88
3.75k
    }
89
228k
    if (is_partial_update()) {
90
38.8k
        for (auto i = 0; i < tablet_schema.num_columns(); ++i) {
91
34.8k
            if (partial_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {
92
30.9k
                auto tablet_column = tablet_schema.column(i);
93
30.9k
                if (!partial_update_input_columns.contains(tablet_column.name())) {
94
17.9k
                    missing_cids.emplace_back(i);
95
17.9k
                    if (!tablet_column.has_default_value() && !tablet_column.is_nullable() &&
96
17.9k
                        tablet_schema.auto_increment_column() != tablet_column.name()) {
97
1.28k
                        can_insert_new_rows_in_partial_update = false;
98
1.28k
                    }
99
17.9k
                } else {
100
13.0k
                    update_cids.emplace_back(i);
101
13.0k
                }
102
30.9k
                if (auto_increment_column == tablet_column.name()) {
103
240
                    is_schema_contains_auto_inc_column = true;
104
240
                }
105
30.9k
            } else {
106
                // in flexible partial update, missing cids is all non sort keys' cid
107
3.90k
                if (i >= tablet_schema.num_key_columns()) {
108
3.61k
                    missing_cids.emplace_back(i);
109
3.61k
                }
110
3.90k
            }
111
34.8k
        }
112
4.00k
        _generate_default_values_for_missing_cids(tablet_schema);
113
4.00k
    }
114
228k
    is_strict_mode = is_strict_mode_;
115
228k
    is_input_columns_contains_auto_inc_column =
116
228k
            is_fixed_partial_update() &&
117
228k
            partial_update_input_columns.contains(auto_increment_column);
118
228k
    return Status::OK();
119
228k
}
120
121
506
void PartialUpdateInfo::to_pb(PartialUpdateInfoPB* partial_update_info_pb) const {
122
506
    partial_update_info_pb->set_partial_update_mode(partial_update_mode);
123
506
    partial_update_info_pb->set_partial_update_new_key_policy(partial_update_new_key_policy);
124
506
    partial_update_info_pb->set_max_version_in_flush_phase(max_version_in_flush_phase);
125
2.28k
    for (const auto& col : partial_update_input_columns) {
126
2.28k
        partial_update_info_pb->add_partial_update_input_columns(col);
127
2.28k
    }
128
2.43k
    for (auto cid : missing_cids) {
129
2.43k
        partial_update_info_pb->add_missing_cids(cid);
130
2.43k
    }
131
2.28k
    for (auto cid : update_cids) {
132
2.28k
        partial_update_info_pb->add_update_cids(cid);
133
2.28k
    }
134
506
    partial_update_info_pb->set_can_insert_new_rows_in_partial_update(
135
506
            can_insert_new_rows_in_partial_update);
136
506
    partial_update_info_pb->set_is_strict_mode(is_strict_mode);
137
506
    partial_update_info_pb->set_timestamp_ms(timestamp_ms);
138
506
    partial_update_info_pb->set_nano_seconds(nano_seconds);
139
506
    partial_update_info_pb->set_timezone(timezone);
140
506
    partial_update_info_pb->set_is_input_columns_contains_auto_inc_column(
141
506
            is_input_columns_contains_auto_inc_column);
142
506
    partial_update_info_pb->set_is_schema_contains_auto_inc_column(
143
506
            is_schema_contains_auto_inc_column);
144
2.43k
    for (const auto& value : default_values) {
145
2.43k
        partial_update_info_pb->add_default_values(value);
146
2.43k
    }
147
506
}
148
149
0
void PartialUpdateInfo::from_pb(PartialUpdateInfoPB* partial_update_info_pb) {
150
0
    if (!partial_update_info_pb->has_partial_update_mode()) {
151
        // for backward compatibility
152
0
        if (partial_update_info_pb->is_partial_update()) {
153
0
            partial_update_mode = UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS;
154
0
        } else {
155
0
            partial_update_mode = UniqueKeyUpdateModePB::UPSERT;
156
0
        }
157
0
    } else {
158
0
        partial_update_mode = partial_update_info_pb->partial_update_mode();
159
0
    }
160
0
    if (partial_update_info_pb->has_partial_update_new_key_policy()) {
161
0
        partial_update_new_key_policy = partial_update_info_pb->partial_update_new_key_policy();
162
0
    }
163
0
    max_version_in_flush_phase = partial_update_info_pb->has_max_version_in_flush_phase()
164
0
                                         ? partial_update_info_pb->max_version_in_flush_phase()
165
0
                                         : -1;
166
0
    partial_update_input_columns.clear();
167
0
    for (const auto& col : partial_update_info_pb->partial_update_input_columns()) {
168
0
        partial_update_input_columns.insert(col);
169
0
    }
170
0
    missing_cids.clear();
171
0
    for (auto cid : partial_update_info_pb->missing_cids()) {
172
0
        missing_cids.push_back(cid);
173
0
    }
174
0
    update_cids.clear();
175
0
    for (auto cid : partial_update_info_pb->update_cids()) {
176
0
        update_cids.push_back(cid);
177
0
    }
178
0
    can_insert_new_rows_in_partial_update =
179
0
            partial_update_info_pb->can_insert_new_rows_in_partial_update();
180
0
    is_strict_mode = partial_update_info_pb->is_strict_mode();
181
0
    timestamp_ms = partial_update_info_pb->timestamp_ms();
182
0
    timezone = partial_update_info_pb->timezone();
183
0
    is_input_columns_contains_auto_inc_column =
184
0
            partial_update_info_pb->is_input_columns_contains_auto_inc_column();
185
0
    is_schema_contains_auto_inc_column =
186
0
            partial_update_info_pb->is_schema_contains_auto_inc_column();
187
0
    if (partial_update_info_pb->has_nano_seconds()) {
188
0
        nano_seconds = partial_update_info_pb->nano_seconds();
189
0
    }
190
0
    default_values.clear();
191
0
    for (const auto& value : partial_update_info_pb->default_values()) {
192
0
        default_values.push_back(value);
193
0
    }
194
0
}
195
196
0
std::string PartialUpdateInfo::summary() const {
197
0
    std::string mode;
198
0
    switch (partial_update_mode) {
199
0
    case UniqueKeyUpdateModePB::UPSERT:
200
0
        mode = "upsert";
201
0
        break;
202
0
    case UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS:
203
0
        mode = "fixed partial update";
204
0
        break;
205
0
    case UniqueKeyUpdateModePB::UPDATE_FLEXIBLE_COLUMNS:
206
0
        mode = "flexible partial update";
207
0
        break;
208
0
    }
209
0
    return fmt::format(
210
0
            "mode={}, update_cids={}, missing_cids={}, is_strict_mode={}, "
211
0
            "max_version_in_flush_phase={}",
212
0
            mode, update_cids.size(), missing_cids.size(), is_strict_mode,
213
0
            max_version_in_flush_phase);
214
0
}
215
216
Status PartialUpdateInfo::handle_new_key(const TabletSchema& tablet_schema,
217
                                         const std::function<std::string()>& line,
218
555
                                         BitmapValue* skip_bitmap) {
219
555
    switch (partial_update_new_key_policy) {
220
519
    case doris::PartialUpdateNewRowPolicyPB::APPEND: {
221
519
        if (partial_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {
222
237
            if (!can_insert_new_rows_in_partial_update) {
223
10
                std::string error_column;
224
18
                for (auto cid : missing_cids) {
225
18
                    const TabletColumn& col = tablet_schema.column(cid);
226
18
                    if (!col.has_default_value() && !col.is_nullable() &&
227
18
                        !(tablet_schema.auto_increment_column() == col.name())) {
228
10
                        error_column = col.name();
229
10
                        break;
230
10
                    }
231
18
                }
232
10
                return Status::Error<ErrorCode::INVALID_SCHEMA, false>(
233
10
                        "the unmentioned column `{}` should have default value or be nullable "
234
10
                        "for newly inserted rows in non-strict mode partial update",
235
10
                        error_column);
236
10
            }
237
283
        } else if (partial_update_mode == UniqueKeyUpdateModePB::UPDATE_FLEXIBLE_COLUMNS) {
238
283
            DCHECK(skip_bitmap != nullptr);
239
283
            bool can_insert_new_row {true};
240
283
            std::string error_column;
241
4.13k
            for (auto cid : missing_cids) {
242
4.13k
                const TabletColumn& col = tablet_schema.column(cid);
243
4.13k
                if (skip_bitmap->contains(col.unique_id()) && !col.has_default_value() &&
244
4.13k
                    !col.is_nullable() && !col.is_auto_increment()) {
245
0
                    error_column = col.name();
246
0
                    can_insert_new_row = false;
247
0
                    break;
248
0
                }
249
4.13k
            }
250
283
            if (!can_insert_new_row) {
251
0
                return Status::Error<ErrorCode::INVALID_SCHEMA, false>(
252
0
                        "the unmentioned column `{}` should have default value or be "
253
0
                        "nullable for newly inserted rows in non-strict mode flexible partial "
254
0
                        "update",
255
0
                        error_column);
256
0
            }
257
283
        }
258
519
    } break;
259
509
    case doris::PartialUpdateNewRowPolicyPB::ERROR: {
260
36
        return Status::Error<ErrorCode::NEW_ROWS_IN_PARTIAL_UPDATE, false>(
261
36
                "Can't append new rows in partial update when partial_update_new_key_behavior is "
262
36
                "ERROR. Row with key=[{}] is not in table.",
263
36
                line());
264
519
    } break;
265
555
    }
266
512
    return Status::OK();
267
555
}
268
269
void PartialUpdateInfo::_generate_default_values_for_missing_cids(
270
4.01k
        const TabletSchema& tablet_schema) {
271
21.4k
    for (unsigned int cur_cid : missing_cids) {
272
21.4k
        const auto& column = tablet_schema.column(cur_cid);
273
21.4k
        if (column.has_default_value()) {
274
7.71k
            std::string default_value;
275
7.71k
            if (UNLIKELY((column.type() == FieldType::OLAP_FIELD_TYPE_DATETIMEV2 ||
276
7.71k
                          column.type() == FieldType::OLAP_FIELD_TYPE_TIMESTAMPTZ) &&
277
7.71k
                         to_lower(column.default_value()).find(to_lower("CURRENT_TIMESTAMP")) !=
278
7.71k
                                 std::string::npos)) {
279
132
                auto pos = to_lower(column.default_value()).find('(');
280
132
                if (pos == std::string::npos) {
281
64
                    DateV2Value<DateTimeV2ValueType> dtv;
282
64
                    dtv.from_unixtime(timestamp_ms / 1000, timezone);
283
64
                    default_value = dtv.to_string();
284
64
                    if (column.type() == FieldType::OLAP_FIELD_TYPE_TIMESTAMPTZ) {
285
0
                        default_value += timezone;
286
0
                    }
287
68
                } else {
288
68
                    int precision = std::stoi(column.default_value().substr(pos + 1));
289
68
                    DateV2Value<DateTimeV2ValueType> dtv;
290
68
                    dtv.from_unixtime(timestamp_ms / 1000, nano_seconds, timezone, precision);
291
68
                    default_value = dtv.to_string();
292
68
                    if (column.type() == FieldType::OLAP_FIELD_TYPE_TIMESTAMPTZ) {
293
0
                        default_value += timezone;
294
0
                    }
295
68
                }
296
7.58k
            } else if (UNLIKELY(column.type() == FieldType::OLAP_FIELD_TYPE_DATEV2 &&
297
7.58k
                                to_lower(column.default_value()).find(to_lower("CURRENT_DATE")) !=
298
7.58k
                                        std::string::npos)) {
299
82
                DateV2Value<DateV2ValueType> dv;
300
82
                dv.from_unixtime(timestamp_ms / 1000, timezone);
301
82
                default_value = dv.to_string();
302
7.50k
            } else if (UNLIKELY(column.type() == FieldType::OLAP_FIELD_TYPE_BITMAP &&
303
7.50k
                                to_lower(column.default_value()).find(to_lower("BITMAP_EMPTY")) !=
304
7.50k
                                        std::string::npos)) {
305
0
                BitmapValue v = BitmapValue {};
306
0
                default_value = v.to_string();
307
7.50k
            } else {
308
7.50k
                default_value = column.default_value();
309
7.50k
            }
310
7.71k
            default_values.emplace_back(default_value);
311
13.7k
        } else {
312
            // place an empty string here
313
13.7k
            default_values.emplace_back();
314
13.7k
        }
315
21.4k
    }
316
4.01k
    CHECK_EQ(missing_cids.size(), default_values.size());
317
4.01k
}
318
319
9
bool FixedReadPlan::empty() const {
320
9
    return plan.empty();
321
9
}
322
323
22.7k
void FixedReadPlan::prepare_to_read(const RowLocation& row_location, size_t pos) {
324
22.7k
    plan[row_location.rowset_id][row_location.segment_id].emplace_back(row_location.row_id, pos);
325
22.7k
}
326
327
// read columns by read plan
328
// read_index: ori_pos-> block_idx
329
Status FixedReadPlan::read_columns_by_plan(
330
        const TabletSchema& tablet_schema, std::vector<uint32_t> cids_to_read,
331
        const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset, Block& block,
332
        std::map<uint32_t, uint32_t>* read_index, bool force_read_old_delete_signs,
333
1.51k
        const signed char* __restrict cur_delete_signs) const {
334
1.51k
    if (force_read_old_delete_signs) {
335
        // always read delete sign column from historical data
336
1.49k
        if (block.get_position_by_name(DELETE_SIGN) == -1) {
337
960
            auto del_col_cid = tablet_schema.field_index(DELETE_SIGN);
338
960
            cids_to_read.emplace_back(del_col_cid);
339
960
            block.swap(tablet_schema.create_block_by_cids(cids_to_read));
340
960
        }
341
1.49k
    }
342
1.51k
    bool has_row_column = tablet_schema.has_row_store_for_all_columns();
343
1.51k
    std::optional<Block::ScopedMutableColumns> mutable_columns_guard;
344
1.51k
    MutableColumns* mutable_columns = nullptr;
345
1.51k
    if (!has_row_column) {
346
1.27k
        mutable_columns_guard.emplace(block);
347
1.27k
        mutable_columns = &mutable_columns_guard->mutable_columns();
348
1.27k
    }
349
1.51k
    uint32_t read_idx = 0;
350
1.51k
    for (const auto& [rowset_id, segment_row_mappings] : plan) {
351
623
        for (const auto& [segment_id, mappings] : segment_row_mappings) {
352
623
            auto rowset_iter = rsid_to_rowset.find(rowset_id);
353
623
            CHECK(rowset_iter != rsid_to_rowset.end());
354
623
            std::vector<uint32_t> rids;
355
22.7k
            for (auto [rid, pos] : mappings) {
356
22.7k
                if (cur_delete_signs && cur_delete_signs[pos]) {
357
4
                    continue;
358
4
                }
359
22.7k
                rids.emplace_back(rid);
360
22.7k
                (*read_index)[static_cast<uint32_t>(pos)] = read_idx++;
361
22.7k
            }
362
623
            if (has_row_column) {
363
178
                auto st = BaseTablet::fetch_value_through_row_column(
364
178
                        rowset_iter->second, tablet_schema, segment_id, rids, cids_to_read, block);
365
178
                if (!st.ok()) {
366
0
                    LOG(WARNING) << "failed to fetch value through row column";
367
0
                    return st;
368
0
                }
369
178
                continue;
370
178
            }
371
2.56k
            for (size_t cid = 0; cid < mutable_columns->size(); ++cid) {
372
2.12k
                TabletColumn tablet_column = tablet_schema.column(cids_to_read[cid]);
373
2.12k
                auto st = doris::BaseTablet::fetch_value_by_rowids(rowset_iter->second, segment_id,
374
2.12k
                                                                   rids, tablet_column,
375
2.12k
                                                                   (*mutable_columns)[cid]);
376
                // set read value to output block
377
2.12k
                if (!st.ok()) {
378
0
                    LOG(WARNING) << "failed to fetch value";
379
0
                    return st;
380
0
                }
381
2.12k
            }
382
445
        }
383
623
    }
384
1.51k
    return Status::OK();
385
1.51k
}
386
387
Status FixedReadPlan::fill_missing_columns(
388
        const segment_v2::HistoricalRowRetrieverContext& historical_context,
389
        const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
390
        const TabletSchema& tablet_schema, Block& full_block,
391
        const std::vector<bool>& use_default_or_null_flag, bool has_default_or_nullable,
392
1.47k
        uint32_t segment_start_pos, const Block* block) const {
393
1.47k
    auto mutable_full_columns_guard = full_block.mutate_columns_scoped();
394
1.47k
    auto& mutable_full_columns = mutable_full_columns_guard.mutable_columns();
395
    // create old value columns
396
1.47k
    DCHECK(historical_context.partial_update_info != nullptr);
397
1.47k
    DCHECK(historical_context.tablet_schema != nullptr);
398
1.47k
    const auto& partial_update_info = *historical_context.partial_update_info;
399
1.47k
    const auto& missing_cids = partial_update_info.missing_cids;
400
1.47k
    bool have_input_seq_column = false;
401
1.47k
    if (tablet_schema.has_sequence_col()) {
402
224
        const std::vector<uint32_t>& including_cids = partial_update_info.update_cids;
403
224
        have_input_seq_column =
404
224
                (std::find(including_cids.cbegin(), including_cids.cend(),
405
224
                           tablet_schema.sequence_col_idx()) != including_cids.cend());
406
224
    }
407
408
1.47k
    auto old_value_block = tablet_schema.create_block_by_cids(missing_cids);
409
1.47k
    CHECK_EQ(missing_cids.size(), old_value_block.columns());
410
411
    // segment pos to write -> rowid to read in old_value_block
412
1.47k
    std::map<uint32_t, uint32_t> read_index;
413
1.47k
    RETURN_IF_ERROR(read_columns_by_plan(tablet_schema, missing_cids, rsid_to_rowset,
414
1.47k
                                         old_value_block, &read_index, true, nullptr));
415
416
1.47k
    const auto* old_delete_signs = BaseTablet::get_delete_sign_column_data(old_value_block);
417
1.47k
    if (old_delete_signs == nullptr) {
418
0
        return Status::InternalError("old delete signs column not found, block: {}",
419
0
                                     old_value_block.dump_structure());
420
0
    }
421
    // build default value columns
422
1.47k
    auto default_value_block = old_value_block.clone_empty();
423
1.47k
    RETURN_IF_ERROR(BaseTablet::generate_default_value_block(tablet_schema, missing_cids,
424
1.47k
                                                             partial_update_info.default_values,
425
1.47k
                                                             old_value_block, default_value_block));
426
1.47k
    auto mutable_default_value_columns_guard = default_value_block.mutate_columns_scoped();
427
1.47k
    auto& mutable_default_value_columns = mutable_default_value_columns_guard.mutable_columns();
428
429
    // fill all missing value from mutable_old_columns, need to consider default value and null value
430
25.9k
    for (auto idx = 0; idx < use_default_or_null_flag.size(); idx++) {
431
24.5k
        auto segment_pos = idx + segment_start_pos;
432
24.5k
        auto pos_in_old_block = read_index[segment_pos];
433
434
116k
        for (auto i = 0; i < missing_cids.size(); ++i) {
435
            // if the column has default value, fill it with default value
436
            // otherwise, if the column is nullable, fill it with null value
437
92.0k
            const auto& tablet_column = tablet_schema.column(missing_cids[i]);
438
92.0k
            auto& missing_col = mutable_full_columns[missing_cids[i]];
439
440
92.0k
            bool should_use_default = use_default_or_null_flag[idx];
441
92.0k
            if (!should_use_default) {
442
79.1k
                bool old_row_delete_sign =
443
79.1k
                        (old_delete_signs != nullptr && old_delete_signs[pos_in_old_block] != 0);
444
79.1k
                if (old_row_delete_sign) {
445
151
                    if (!tablet_schema.has_sequence_col()) {
446
70
                        should_use_default = true;
447
81
                    } else if (have_input_seq_column || (!tablet_column.is_seqeunce_col())) {
448
                        // to keep the sequence column value not decreasing, we should read values of seq column
449
                        // from old rows even if the old row is deleted when the input don't specify the sequence column, otherwise
450
                        // it may cause the merge-on-read based compaction to produce incorrect results
451
77
                        should_use_default = true;
452
77
                    }
453
151
                }
454
79.1k
            }
455
456
92.0k
            if (should_use_default) {
457
12.9k
                if (tablet_column.has_default_value()) {
458
2.98k
                    missing_col->insert_from(*mutable_default_value_columns[i], 0);
459
10.0k
                } else if (tablet_column.is_nullable()) {
460
8.51k
                    auto* nullable_column = assert_cast<ColumnNullable*>(missing_col.get());
461
8.51k
                    nullable_column->insert_many_defaults(1);
462
8.51k
                } else if (tablet_schema.auto_increment_column() == tablet_column.name()) {
463
41
                    const auto& column = *DORIS_TRY(
464
41
                            historical_context.tablet_schema->column(tablet_column.name()));
465
41
                    DCHECK(column.type() == FieldType::OLAP_FIELD_TYPE_BIGINT);
466
41
                    auto* auto_inc_column = assert_cast<ColumnInt64*>(missing_col.get());
467
41
                    int pos = block->get_position_by_name(BeConsts::PARTIAL_UPDATE_AUTO_INC_COL);
468
41
                    if (pos == -1) {
469
0
                        return Status::InternalError("auto increment column not found in block {}",
470
0
                                                     block->dump_structure());
471
0
                    }
472
41
                    auto_inc_column->insert_from(*block->get_by_position(pos).column.get(), idx);
473
1.44k
                } else {
474
                    // If the control flow reaches this branch, the column neither has default value
475
                    // nor is nullable. It means that the row's delete sign is marked, and the value
476
                    // columns are useless and won't be read. So we can just put arbitary values in the cells
477
1.44k
                    missing_col->insert_default();
478
1.44k
                }
479
79.0k
            } else {
480
79.0k
                missing_col->insert_from(*old_value_block.get_by_position(i).column,
481
79.0k
                                         pos_in_old_block);
482
79.0k
            }
483
92.0k
        }
484
24.5k
    }
485
1.47k
    return Status::OK();
486
1.47k
}
487
488
void FlexibleReadPlan::prepare_to_read(const RowLocation& row_location, size_t pos,
489
1.79k
                                       const BitmapValue& skip_bitmap) {
490
1.79k
    if (!use_row_store) {
491
3.91k
        for (uint64_t col_uid : skip_bitmap) {
492
3.91k
            plan[row_location.rowset_id][row_location.segment_id][static_cast<uint32_t>(col_uid)]
493
3.91k
                    .emplace_back(row_location.row_id, pos);
494
3.91k
        }
495
932
    } else {
496
858
        row_store_plan[row_location.rowset_id][row_location.segment_id].emplace_back(
497
858
                row_location.row_id, pos);
498
858
    }
499
1.79k
}
500
501
Status FlexibleReadPlan::read_columns_by_plan(
502
        const TabletSchema& tablet_schema,
503
        const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset, Block& old_value_block,
504
145
        std::map<uint32_t, std::map<uint32_t, uint32_t>>* read_index) const {
505
145
    auto mutable_columns_guard = old_value_block.mutate_columns_scoped();
506
145
    auto& mutable_columns = mutable_columns_guard.mutable_columns();
507
508
    // cid -> next rid to fill in block
509
145
    std::map<uint32_t, uint32_t> next_read_idx;
510
2.16k
    for (uint32_t cid {0}; cid < tablet_schema.num_columns(); cid++) {
511
2.02k
        next_read_idx[cid] = 0;
512
2.02k
    }
513
514
182
    for (const auto& [rowset_id, segment_mappings] : plan) {
515
182
        for (const auto& [segment_id, uid_mappings] : segment_mappings) {
516
1.49k
            for (const auto& [col_uid, mappings] : uid_mappings) {
517
1.49k
                auto rowset_iter = rsid_to_rowset.find(rowset_id);
518
1.49k
                CHECK(rowset_iter != rsid_to_rowset.end());
519
1.49k
                auto cid = tablet_schema.field_index(col_uid);
520
1.49k
                DCHECK_NE(cid, -1);
521
1.49k
                DCHECK_GE(cid, tablet_schema.num_key_columns());
522
1.49k
                std::vector<uint32_t> rids;
523
3.91k
                for (auto [rid, pos] : mappings) {
524
3.91k
                    rids.emplace_back(rid);
525
3.91k
                    (*read_index)[cid][static_cast<uint32_t>(pos)] = next_read_idx[cid]++;
526
3.91k
                }
527
528
1.49k
                TabletColumn tablet_column = tablet_schema.column(cid);
529
1.49k
                auto idx = cid - tablet_schema.num_key_columns();
530
1.49k
                RETURN_IF_ERROR(doris::BaseTablet::fetch_value_by_rowids(
531
1.49k
                        rowset_iter->second, segment_id, rids, tablet_column,
532
1.49k
                        mutable_columns[idx]));
533
1.49k
            }
534
182
        }
535
182
    }
536
    // !!!ATTENTION!!!: columns in block may have different size because every row has different columns to update
537
145
    return Status::OK();
538
145
}
539
540
Status FlexibleReadPlan::read_columns_by_plan(
541
        const TabletSchema& tablet_schema, const std::vector<uint32_t>& cids_to_read,
542
        const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset, Block& old_value_block,
543
119
        std::map<uint32_t, uint32_t>* read_index) const {
544
119
    DCHECK(use_row_store);
545
119
    uint32_t read_idx = 0;
546
159
    for (const auto& [rowset_id, segment_row_mappings] : row_store_plan) {
547
159
        for (const auto& [segment_id, mappings] : segment_row_mappings) {
548
159
            auto rowset_iter = rsid_to_rowset.find(rowset_id);
549
159
            CHECK(rowset_iter != rsid_to_rowset.end());
550
159
            std::vector<uint32_t> rids;
551
859
            for (auto [rid, pos] : mappings) {
552
859
                rids.emplace_back(rid);
553
859
                (*read_index)[static_cast<uint32_t>(pos)] = read_idx++;
554
859
            }
555
159
            auto st = BaseTablet::fetch_value_through_row_column(rowset_iter->second, tablet_schema,
556
159
                                                                 segment_id, rids, cids_to_read,
557
159
                                                                 old_value_block);
558
159
            if (!st.ok()) {
559
0
                LOG(WARNING) << "failed to fetch value through row column";
560
0
                return st;
561
0
            }
562
159
        }
563
159
    }
564
119
    return Status::OK();
565
119
}
566
567
Status FlexibleReadPlan::fill_non_primary_key_columns(
568
        const segment_v2::HistoricalRowRetrieverContext& historical_context,
569
        const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
570
        const TabletSchema& tablet_schema, Block& full_block,
571
        const std::vector<bool>& use_default_or_null_flag, bool has_default_or_nullable,
572
        uint32_t segment_start_pos, uint32_t block_start_pos, const Block* block,
573
264
        std::vector<BitmapValue>* skip_bitmaps) const {
574
264
    auto mutable_full_columns_guard = full_block.mutate_columns_scoped();
575
264
    auto& mutable_full_columns = mutable_full_columns_guard.mutable_columns();
576
264
    DCHECK(historical_context.partial_update_info != nullptr);
577
578
    // missing_cids are all non sort key columns' cids
579
264
    const auto& non_sort_key_cids = historical_context.partial_update_info->missing_cids;
580
264
    auto old_value_block = tablet_schema.create_block_by_cids(non_sort_key_cids);
581
264
    CHECK_EQ(non_sort_key_cids.size(), old_value_block.columns());
582
583
264
    if (!use_row_store) {
584
145
        RETURN_IF_ERROR(fill_non_primary_key_columns_for_column_store(
585
145
                historical_context, rsid_to_rowset, tablet_schema, non_sort_key_cids,
586
145
                old_value_block, mutable_full_columns, use_default_or_null_flag,
587
145
                has_default_or_nullable, segment_start_pos, block_start_pos, block, skip_bitmaps));
588
145
    } else {
589
119
        RETURN_IF_ERROR(fill_non_primary_key_columns_for_row_store(
590
119
                historical_context, rsid_to_rowset, tablet_schema, non_sort_key_cids,
591
119
                old_value_block, mutable_full_columns, use_default_or_null_flag,
592
119
                has_default_or_nullable, segment_start_pos, block_start_pos, block, skip_bitmaps));
593
119
    }
594
264
    return Status::OK();
595
264
}
596
597
static void fill_non_primary_key_cell_for_column_store(
598
        const TabletColumn& tablet_column, uint32_t cid, MutableColumnPtr& new_col,
599
        const IColumn& default_value_col, const IColumn& old_value_col, const IColumn& cur_col,
600
        std::size_t block_pos, uint32_t segment_pos, bool skipped, bool row_has_sequence_col,
601
        bool use_default, const signed char* delete_sign_column_data,
602
        const TabletSchema& tablet_schema,
603
        std::map<uint32_t, std::map<uint32_t, uint32_t>>& read_index,
604
16.9k
        const PartialUpdateInfo* info) {
605
16.9k
    if (skipped) {
606
4.78k
        DCHECK(cid != tablet_schema.skip_bitmap_col_idx());
607
4.78k
        DCHECK(cid != tablet_schema.version_col_idx());
608
4.78k
        DCHECK(!tablet_column.is_row_store_column());
609
610
4.78k
        if (!use_default) {
611
3.91k
            if (delete_sign_column_data != nullptr) {
612
3.91k
                bool old_row_delete_sign = false;
613
3.91k
                if (auto it = read_index[tablet_schema.delete_sign_idx()].find(segment_pos);
614
3.91k
                    it != read_index[tablet_schema.delete_sign_idx()].end()) {
615
3.79k
                    old_row_delete_sign = (delete_sign_column_data[it->second] != 0);
616
3.79k
                }
617
618
3.91k
                if (old_row_delete_sign) {
619
7
                    if (!tablet_schema.has_sequence_col()) {
620
2
                        use_default = true;
621
5
                    } else if (row_has_sequence_col ||
622
5
                               (!tablet_column.is_seqeunce_col() &&
623
5
                                (tablet_column.unique_id() != info->sequence_map_col_uid()))) {
624
                        // to keep the sequence column value not decreasing, we should read values of seq column(and seq map column)
625
                        // from old rows even if the old row is deleted when the input don't specify the sequence column, otherwise
626
                        // it may cause the merge-on-read based compaction to produce incorrect results
627
3
                        use_default = true;
628
3
                    }
629
7
                }
630
3.91k
            }
631
3.91k
        }
632
4.78k
        if (!use_default && tablet_column.is_on_update_current_timestamp()) {
633
6
            use_default = true;
634
6
        }
635
4.78k
        if (use_default) {
636
873
            if (tablet_column.has_default_value()) {
637
424
                new_col->insert_from(default_value_col, 0);
638
449
            } else if (tablet_column.is_nullable()) {
639
426
                assert_cast<ColumnNullable*, TypeCheckOnRelease::DISABLE>(new_col.get())
640
426
                        ->insert_many_defaults(1);
641
426
            } else if (tablet_column.is_auto_increment()) {
642
                // In flexible partial update, the skip bitmap indicates whether a cell
643
                // is specified in the original load, so the generated auto-increment value is filled
644
                // in current block in place if needed rather than using a seperate column to
645
                // store the generated auto-increment value in fixed partial update
646
2
                new_col->insert_from(cur_col, block_pos);
647
21
            } else {
648
21
                new_col->insert_default();
649
21
            }
650
3.90k
        } else {
651
3.90k
            auto pos_in_old_block = read_index.at(cid).at(segment_pos);
652
3.90k
            new_col->insert_from(old_value_col, pos_in_old_block);
653
3.90k
        }
654
12.1k
    } else {
655
12.1k
        new_col->insert_from(cur_col, block_pos);
656
12.1k
    }
657
16.9k
}
658
659
Status FlexibleReadPlan::fill_non_primary_key_columns_for_column_store(
660
        const segment_v2::HistoricalRowRetrieverContext& historical_context,
661
        const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
662
        const TabletSchema& tablet_schema, const std::vector<uint32_t>& non_sort_key_cids,
663
        Block& old_value_block, MutableColumns& mutable_full_columns,
664
        const std::vector<bool>& use_default_or_null_flag, bool has_default_or_nullable,
665
        uint32_t segment_start_pos, uint32_t block_start_pos, const Block* block,
666
145
        std::vector<BitmapValue>* skip_bitmaps) const {
667
145
    auto* info = historical_context.partial_update_info.get();
668
145
    int32_t seq_col_unique_id = -1;
669
145
    if (tablet_schema.has_sequence_col()) {
670
22
        seq_col_unique_id = tablet_schema.column(tablet_schema.sequence_col_idx()).unique_id();
671
22
    }
672
    // cid -> segment pos to write -> rowid to read in old_value_block
673
145
    std::map<uint32_t, std::map<uint32_t, uint32_t>> read_index;
674
145
    RETURN_IF_ERROR(
675
145
            read_columns_by_plan(tablet_schema, rsid_to_rowset, old_value_block, &read_index));
676
    // !!!ATTENTION!!!: columns in old_value_block may have different size because every row has different columns to update
677
678
145
    const auto* delete_sign_column_data = BaseTablet::get_delete_sign_column_data(old_value_block);
679
    // build default value columns
680
145
    auto default_value_block = old_value_block.clone_empty();
681
145
    if (has_default_or_nullable || delete_sign_column_data != nullptr) {
682
145
        RETURN_IF_ERROR(BaseTablet::generate_default_value_block(
683
145
                tablet_schema, non_sort_key_cids, info->default_values, old_value_block,
684
145
                default_value_block));
685
145
    }
686
687
    // fill all non sort key columns from mutable_old_columns, need to consider default value and null value
688
2.00k
    for (std::size_t i {0}; i < non_sort_key_cids.size(); i++) {
689
1.86k
        auto cid = non_sort_key_cids[i];
690
1.86k
        const auto& tablet_column = tablet_schema.column(cid);
691
1.86k
        auto col_uid = tablet_column.unique_id();
692
18.8k
        for (auto idx = 0; idx < use_default_or_null_flag.size(); idx++) {
693
16.9k
            auto segment_pos = segment_start_pos + idx;
694
16.9k
            auto block_pos = block_start_pos + idx;
695
696
16.9k
            fill_non_primary_key_cell_for_column_store(
697
16.9k
                    tablet_column, cid, mutable_full_columns[cid],
698
16.9k
                    *default_value_block.get_by_position(i).column,
699
16.9k
                    *old_value_block.get_by_position(i).column, *block->get_by_position(cid).column,
700
16.9k
                    block_pos, segment_pos, skip_bitmaps->at(block_pos).contains(col_uid),
701
16.9k
                    tablet_schema.has_sequence_col()
702
16.9k
                            ? !skip_bitmaps->at(block_pos).contains(seq_col_unique_id)
703
16.9k
                            : false,
704
16.9k
                    use_default_or_null_flag[idx], delete_sign_column_data, tablet_schema,
705
16.9k
                    read_index, info);
706
16.9k
        }
707
1.86k
    }
708
145
    return Status::OK();
709
145
}
710
711
static void fill_non_primary_key_cell_for_row_store(
712
        const TabletColumn& tablet_column, uint32_t cid, MutableColumnPtr& new_col,
713
        const IColumn& default_value_col, const IColumn& old_value_col, const IColumn& cur_col,
714
        std::size_t block_pos, bool skipped, bool row_has_sequence_col, bool use_default,
715
        const signed char* delete_sign_column_data, uint32_t pos_in_old_block,
716
16.5k
        const TabletSchema& tablet_schema, const PartialUpdateInfo* info) {
717
16.5k
    if (skipped) {
718
4.16k
        DCHECK(cid != tablet_schema.skip_bitmap_col_idx());
719
4.16k
        DCHECK(cid != tablet_schema.version_col_idx());
720
4.16k
        DCHECK(!tablet_column.is_row_store_column());
721
4.16k
        if (!use_default) {
722
3.64k
            if (delete_sign_column_data != nullptr) {
723
3.64k
                bool old_row_delete_sign = (delete_sign_column_data[pos_in_old_block] != 0);
724
3.64k
                if (old_row_delete_sign) {
725
7
                    if (!tablet_schema.has_sequence_col()) {
726
2
                        use_default = true;
727
5
                    } else if (row_has_sequence_col ||
728
5
                               (!tablet_column.is_seqeunce_col() &&
729
5
                                (tablet_column.unique_id() != info->sequence_map_col_uid()))) {
730
                        // to keep the sequence column value not decreasing, we should read values of seq column(and seq map column)
731
                        // from old rows even if the old row is deleted when the input don't specify the sequence column, otherwise
732
                        // it may cause the merge-on-read based compaction to produce incorrect results
733
3
                        use_default = true;
734
3
                    }
735
7
                }
736
3.64k
            }
737
3.64k
        }
738
739
4.16k
        if (!use_default && tablet_column.is_on_update_current_timestamp()) {
740
6
            use_default = true;
741
6
        }
742
4.16k
        if (use_default) {
743
529
            if (tablet_column.has_default_value()) {
744
175
                new_col->insert_from(default_value_col, 0);
745
354
            } else if (tablet_column.is_nullable()) {
746
350
                assert_cast<ColumnNullable*, TypeCheckOnRelease::DISABLE>(new_col.get())
747
350
                        ->insert_many_defaults(1);
748
350
            } else if (tablet_column.is_auto_increment()) {
749
                // In flexible partial update, the skip bitmap indicates whether a cell
750
                // is specified in the original load, so the generated auto-increment value is filled
751
                // in current block in place if needed rather than using a seperate column to
752
                // store the generated auto-increment value in fixed partial update
753
2
                new_col->insert_from(cur_col, block_pos);
754
2
            } else {
755
2
                new_col->insert_default();
756
2
            }
757
3.63k
        } else {
758
3.63k
            new_col->insert_from(old_value_col, pos_in_old_block);
759
3.63k
        }
760
12.3k
    } else {
761
12.3k
        new_col->insert_from(cur_col, block_pos);
762
12.3k
    }
763
16.5k
}
764
765
Status FlexibleReadPlan::fill_non_primary_key_columns_for_row_store(
766
        const segment_v2::HistoricalRowRetrieverContext& historical_context,
767
        const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
768
        const TabletSchema& tablet_schema, const std::vector<uint32_t>& non_sort_key_cids,
769
        Block& old_value_block, MutableColumns& mutable_full_columns,
770
        const std::vector<bool>& use_default_or_null_flag, bool has_default_or_nullable,
771
        uint32_t segment_start_pos, uint32_t block_start_pos, const Block* block,
772
119
        std::vector<BitmapValue>* skip_bitmaps) const {
773
119
    auto* info = historical_context.partial_update_info.get();
774
119
    int32_t seq_col_unique_id = -1;
775
119
    if (tablet_schema.has_sequence_col()) {
776
3
        seq_col_unique_id = tablet_schema.column(tablet_schema.sequence_col_idx()).unique_id();
777
3
    }
778
    // segment pos to write -> rowid to read in old_value_block
779
119
    std::map<uint32_t, uint32_t> read_index;
780
119
    RETURN_IF_ERROR(read_columns_by_plan(tablet_schema, non_sort_key_cids, rsid_to_rowset,
781
119
                                         old_value_block, &read_index));
782
783
119
    const auto* delete_sign_column_data = BaseTablet::get_delete_sign_column_data(old_value_block);
784
    // build default value columns
785
119
    auto default_value_block = old_value_block.clone_empty();
786
119
    if (has_default_or_nullable || delete_sign_column_data != nullptr) {
787
119
        RETURN_IF_ERROR(BaseTablet::generate_default_value_block(
788
119
                tablet_schema, non_sort_key_cids, info->default_values, old_value_block,
789
119
                default_value_block));
790
119
    }
791
792
    // fill all non sort key columns from mutable_old_columns, need to consider default value and null value
793
1.87k
    for (std::size_t i {0}; i < non_sort_key_cids.size(); i++) {
794
1.75k
        auto cid = non_sort_key_cids[i];
795
1.75k
        const auto& tablet_column = tablet_schema.column(cid);
796
1.75k
        auto col_uid = tablet_column.unique_id();
797
18.2k
        for (auto idx = 0; idx < use_default_or_null_flag.size(); idx++) {
798
16.5k
            auto segment_pos = segment_start_pos + idx;
799
16.5k
            auto block_pos = block_start_pos + idx;
800
16.5k
            auto pos_in_old_block = read_index[segment_pos];
801
802
16.5k
            fill_non_primary_key_cell_for_row_store(
803
16.5k
                    tablet_column, cid, mutable_full_columns[cid],
804
16.5k
                    *default_value_block.get_by_position(i).column,
805
16.5k
                    *old_value_block.get_by_position(i).column, *block->get_by_position(cid).column,
806
16.5k
                    block_pos, skip_bitmaps->at(block_pos).contains(col_uid),
807
16.5k
                    tablet_schema.has_sequence_col()
808
16.5k
                            ? !skip_bitmaps->at(block_pos).contains(seq_col_unique_id)
809
16.5k
                            : false,
810
16.5k
                    use_default_or_null_flag[idx], delete_sign_column_data, pos_in_old_block,
811
16.5k
                    tablet_schema, info);
812
16.5k
        }
813
1.75k
    }
814
119
    return Status::OK();
815
119
}
816
817
BlockAggregator::BlockAggregator(segment_v2::VerticalSegmentWriter& vertical_segment_writer)
818
75.9k
        : _writer(vertical_segment_writer), _tablet_schema(*_writer._tablet_schema) {}
819
820
void BlockAggregator::merge_one_row(MutableBlock& dst_block, Block* src_block, int rid,
821
153
                                    BitmapValue& skip_bitmap) {
822
1.57k
    for (size_t cid {_tablet_schema.num_key_columns()}; cid < _tablet_schema.num_columns(); cid++) {
823
1.42k
        if (cid == _tablet_schema.skip_bitmap_col_idx()) {
824
153
            auto& cur_skip_bitmap =
825
153
                    assert_cast<ColumnBitmap*>(dst_block.mutable_columns()[cid].get())
826
153
                            ->get_data()
827
153
                            .back();
828
153
            const auto& new_row_skip_bitmap =
829
153
                    assert_cast<const ColumnBitmap*>(src_block->get_by_position(cid).column.get())
830
153
                            ->get_data()[rid];
831
153
            cur_skip_bitmap &= new_row_skip_bitmap;
832
153
            continue;
833
153
        }
834
1.27k
        if (!skip_bitmap.contains(_tablet_schema.column(cid).unique_id())) {
835
460
            dst_block.mutable_columns()[cid]->pop_back(1);
836
460
            dst_block.mutable_columns()[cid]->insert_from(*src_block->get_by_position(cid).column,
837
460
                                                          rid);
838
460
        }
839
1.27k
    }
840
153
    VLOG_DEBUG << fmt::format("merge a row, after merge, output_block.rows()={}, state: {}",
841
0
                              dst_block.rows(), _state.to_string());
842
153
}
843
844
252
void BlockAggregator::append_one_row(MutableBlock& dst_block, Block* src_block, int rid) {
845
252
    dst_block.add_row(src_block, rid);
846
252
    _state.rows++;
847
252
    VLOG_DEBUG << fmt::format("append a new row, after append, output_block.rows()={}, state: {}",
848
0
                              dst_block.rows(), _state.to_string());
849
252
}
850
851
123
void BlockAggregator::remove_last_n_rows(MutableBlock& dst_block, int n) {
852
123
    if (n > 0) {
853
968
        for (size_t cid {0}; cid < _tablet_schema.num_columns(); cid++) {
854
880
            DCHECK_GE(dst_block.mutable_columns()[cid]->size(), n);
855
880
            dst_block.mutable_columns()[cid]->pop_back(n);
856
880
        }
857
88
    }
858
123
}
859
860
void BlockAggregator::append_or_merge_row(MutableBlock& dst_block, Block* src_block, int rid,
861
405
                                          BitmapValue& skip_bitmap, bool have_delete_sign) {
862
405
    if (have_delete_sign) {
863
        // remove all the previous batched rows
864
123
        remove_last_n_rows(dst_block, _state.rows);
865
123
        _state.rows = 0;
866
123
        _state.has_row_with_delete_sign = true;
867
868
123
        append_one_row(dst_block, src_block, rid);
869
282
    } else {
870
282
        if (_state.should_merge()) {
871
153
            merge_one_row(dst_block, src_block, rid, skip_bitmap);
872
153
        } else {
873
129
            append_one_row(dst_block, src_block, rid);
874
129
        }
875
282
    }
876
405
};
877
878
Status BlockAggregator::aggregate_rows(
879
        MutableBlock& output_block, Block* block, int start, int end, std::string key,
880
        std::vector<BitmapValue>* skip_bitmaps, const signed char* delete_signs,
881
        IOlapColumnDataAccessor* seq_column, const std::vector<RowsetSharedPtr>& specified_rowsets,
882
133
        std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches) {
883
133
    VLOG_DEBUG << fmt::format("merge rows in range=[{}-{})", start, end);
884
133
    if (end - start == 1) {
885
34
        output_block.add_row(block, start);
886
34
        VLOG_DEBUG << fmt::format("append a row directly, rid={}", start);
887
34
        return Status::OK();
888
34
    }
889
890
99
    auto seq_col_unique_id = _tablet_schema.column(_tablet_schema.sequence_col_idx()).unique_id();
891
99
    auto delete_sign_col_unique_id =
892
99
            _tablet_schema.column(_tablet_schema.delete_sign_idx()).unique_id();
893
894
99
    _state.reset();
895
896
99
    RowLocation loc;
897
99
    RowsetSharedPtr rowset;
898
99
    std::string previous_encoded_seq_value {};
899
99
    Status st = _writer._tablet->lookup_row_key(
900
99
            key, &_tablet_schema, false, specified_rowsets, &loc, _writer._mow_context->max_version,
901
99
            segment_caches, &rowset, true, &previous_encoded_seq_value);
902
99
    int pos = start;
903
99
    bool is_expected_st = (st.is<ErrorCode::KEY_NOT_FOUND>() || st.ok());
904
99
    DCHECK(is_expected_st || st.is<ErrorCode::MEM_LIMIT_EXCEEDED>())
905
0
            << "[BlockAggregator::aggregate_rows] unexpected error status while lookup_row_key:"
906
0
            << st;
907
99
    if (!is_expected_st) {
908
0
        return st;
909
0
    }
910
911
99
    std::string cur_seq_val;
912
99
    if (st.ok()) {
913
97
        for (pos = start; pos < end; pos++) {
914
97
            auto& skip_bitmap = skip_bitmaps->at(pos);
915
97
            bool row_has_sequence_col = (!skip_bitmap.contains(seq_col_unique_id));
916
            // Discard all the rows whose seq value is smaller than previous_encoded_seq_value.
917
97
            if (row_has_sequence_col) {
918
56
                std::string seq_val {};
919
56
                _writer._key_encoder.append_seq_suffix(&seq_val, seq_column, pos);
920
56
                if (Slice {seq_val}.compare(Slice {previous_encoded_seq_value}) < 0) {
921
19
                    continue;
922
19
                }
923
37
                cur_seq_val = std::move(seq_val);
924
37
                break;
925
56
            }
926
41
            cur_seq_val = std::move(previous_encoded_seq_value);
927
41
            break;
928
97
        }
929
78
    } else {
930
21
        pos = start;
931
21
        auto& skip_bitmap = skip_bitmaps->at(pos);
932
21
        bool row_has_sequence_col = (!skip_bitmap.contains(seq_col_unique_id));
933
21
        if (row_has_sequence_col) {
934
15
            std::string seq_val {};
935
            // for rows that don't specify seqeunce col, seq_val will be encoded to minial value
936
15
            _writer._key_encoder.append_seq_suffix(&seq_val, seq_column, pos);
937
15
            cur_seq_val = std::move(seq_val);
938
15
        } else {
939
6
            cur_seq_val.clear();
940
6
            RETURN_IF_ERROR(_writer._generate_encoded_default_seq_value(
941
6
                    _tablet_schema, *_writer._opts.rowset_ctx->partial_update_info, &cur_seq_val));
942
6
        }
943
21
    }
944
945
562
    for (int rid {pos}; rid < end; rid++) {
946
463
        auto& skip_bitmap = skip_bitmaps->at(rid);
947
463
        bool row_has_sequence_col = (!skip_bitmap.contains(seq_col_unique_id));
948
463
        bool have_delete_sign =
949
463
                (!skip_bitmap.contains(delete_sign_col_unique_id) && delete_signs[rid] != 0);
950
463
        if (!row_has_sequence_col) {
951
193
            append_or_merge_row(output_block, block, rid, skip_bitmap, have_delete_sign);
952
270
        } else {
953
270
            std::string seq_val {};
954
270
            _writer._key_encoder.append_seq_suffix(&seq_val, seq_column, rid);
955
270
            if (Slice {seq_val}.compare(Slice {cur_seq_val}) >= 0) {
956
212
                append_or_merge_row(output_block, block, rid, skip_bitmap, have_delete_sign);
957
212
                cur_seq_val = std::move(seq_val);
958
212
            } else {
959
58
                VLOG_DEBUG << fmt::format(
960
0
                        "skip rid={} becasue its seq value is lower than the previous", rid);
961
58
            }
962
270
        }
963
463
    }
964
99
    return Status::OK();
965
99
};
966
967
Status BlockAggregator::aggregate_for_sequence_column(
968
        Block* block, int num_rows, const std::vector<IOlapColumnDataAccessor*>& key_columns,
969
        IOlapColumnDataAccessor* seq_column, const std::vector<RowsetSharedPtr>& specified_rowsets,
970
25
        std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches) {
971
25
    DCHECK_EQ(block->columns(), _tablet_schema.num_columns());
972
    // the process logic here is the same as MemTable::_aggregate_for_flexible_partial_update_without_seq_col()
973
    // after this function, there will be at most 2 rows for a specified key
974
25
    std::vector<BitmapValue>* skip_bitmaps =
975
25
            &get_mutable_skip_bitmap_column(block, _tablet_schema.skip_bitmap_col_idx())
976
25
                     ->get_data();
977
25
    const auto* delete_signs = BaseTablet::get_delete_sign_column_data(*block, num_rows);
978
979
25
    auto filtered_block = _tablet_schema.create_block();
980
25
    MutableBlock output_block = MutableBlock::build_mutable_block(std::move(filtered_block));
981
982
25
    int same_key_rows {0};
983
25
    std::string previous_key {};
984
541
    for (int block_pos {0}; block_pos < num_rows; block_pos++) {
985
516
        std::string key = _writer._key_encoder.full_encode(key_columns, block_pos);
986
516
        if (block_pos > 0 && previous_key == key) {
987
383
            same_key_rows++;
988
383
        } else {
989
133
            if (same_key_rows > 0) {
990
108
                RETURN_IF_ERROR(aggregate_rows(output_block, block, block_pos - same_key_rows,
991
108
                                               block_pos, std::move(previous_key), skip_bitmaps,
992
108
                                               delete_signs, seq_column, specified_rowsets,
993
108
                                               segment_caches));
994
108
            }
995
133
            same_key_rows = 1;
996
133
        }
997
516
        previous_key = std::move(key);
998
516
    }
999
25
    if (same_key_rows > 0) {
1000
25
        RETURN_IF_ERROR(aggregate_rows(output_block, block, num_rows - same_key_rows, num_rows,
1001
25
                                       std::move(previous_key), skip_bitmaps, delete_signs,
1002
25
                                       seq_column, specified_rowsets, segment_caches));
1003
25
    }
1004
1005
25
    block->swap(output_block.to_block());
1006
25
    return Status::OK();
1007
25
}
1008
1009
Status BlockAggregator::fill_sequence_column(Block* block, size_t num_rows,
1010
                                             const FixedReadPlan& read_plan,
1011
4
                                             std::vector<BitmapValue>& skip_bitmaps) {
1012
4
    DCHECK(_tablet_schema.has_sequence_col());
1013
4
    std::vector<uint32_t> cids {static_cast<uint32_t>(_tablet_schema.sequence_col_idx())};
1014
4
    auto seq_col_unique_id = _tablet_schema.column(_tablet_schema.sequence_col_idx()).unique_id();
1015
1016
4
    auto seq_col_block = _tablet_schema.create_block_by_cids(cids);
1017
4
    auto tmp_block = _tablet_schema.create_block_by_cids(cids);
1018
4
    std::map<uint32_t, uint32_t> read_index;
1019
4
    RETURN_IF_ERROR(read_plan.read_columns_by_plan(_tablet_schema, cids, _writer._rsid_to_rowset,
1020
4
                                                   seq_col_block, &read_index, false));
1021
1022
4
    auto new_seq_col_ptr = tmp_block.get_by_position(0).column->assert_mutable();
1023
4
    const auto& old_seq_col_ptr = *seq_col_block.get_by_position(0).column;
1024
4
    const auto& cur_seq_col_ptr = *block->get_by_position(_tablet_schema.sequence_col_idx()).column;
1025
80
    for (uint32_t block_pos {0}; block_pos < num_rows; block_pos++) {
1026
76
        if (read_index.contains(block_pos)) {
1027
14
            new_seq_col_ptr->insert_from(old_seq_col_ptr, read_index[block_pos]);
1028
14
            skip_bitmaps[block_pos].remove(seq_col_unique_id);
1029
62
        } else {
1030
62
            new_seq_col_ptr->insert_from(cur_seq_col_ptr, block_pos);
1031
62
        }
1032
76
    }
1033
4
    block->replace_by_position(_tablet_schema.sequence_col_idx(), std::move(new_seq_col_ptr));
1034
4
    return Status::OK();
1035
4
}
1036
1037
Status BlockAggregator::aggregate_for_insert_after_delete(
1038
        Block* block, size_t num_rows, const std::vector<IOlapColumnDataAccessor*>& key_columns,
1039
        const std::vector<RowsetSharedPtr>& specified_rowsets,
1040
265
        std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches) {
1041
265
    DCHECK_EQ(block->columns(), _tablet_schema.num_columns());
1042
    // there will be at most 2 rows for a specified key in block when control flow reaches here
1043
    // after this function, there will not be duplicate rows in block
1044
1045
265
    std::vector<BitmapValue>* skip_bitmaps =
1046
265
            &get_mutable_skip_bitmap_column(block, _tablet_schema.skip_bitmap_col_idx())
1047
265
                     ->get_data();
1048
265
    const auto* delete_signs = BaseTablet::get_delete_sign_column_data(*block, num_rows);
1049
1050
265
    auto filter_column = ColumnUInt8::create(num_rows, 1);
1051
265
    auto* __restrict filter_map = filter_column->get_data().data();
1052
265
    std::string previous_key {};
1053
265
    bool previous_has_delete_sign {false};
1054
265
    int duplicate_rows {0};
1055
265
    int32_t delete_sign_col_unique_id =
1056
265
            _tablet_schema.column(_tablet_schema.delete_sign_idx()).unique_id();
1057
265
    auto seq_col_unique_id =
1058
265
            (_tablet_schema.sequence_col_idx() != -1)
1059
265
                    ? _tablet_schema.column(_tablet_schema.sequence_col_idx()).unique_id()
1060
265
                    : -1;
1061
265
    FixedReadPlan read_plan;
1062
2.45k
    for (size_t block_pos {0}; block_pos < num_rows; block_pos++) {
1063
2.19k
        size_t delta_pos = block_pos;
1064
2.19k
        auto& skip_bitmap = skip_bitmaps->at(block_pos);
1065
2.19k
        std::string key = _writer._key_encoder.full_encode(key_columns, delta_pos);
1066
2.19k
        bool have_delete_sign =
1067
2.19k
                (!skip_bitmap.contains(delete_sign_col_unique_id) && delete_signs[block_pos] != 0);
1068
2.19k
        if (delta_pos > 0 && previous_key == key) {
1069
            // !!ATTENTION!!: We can only remove the row with delete sign if there is a insert with the same key after this row.
1070
            // If there is only a row with delete sign, we should keep it and can't remove it from block, because
1071
            // compaction will not use the delete bitmap when reading data. So there may still be rows with delete sign
1072
            // in later process
1073
43
            DCHECK(previous_has_delete_sign);
1074
43
            DCHECK(!have_delete_sign);
1075
43
            ++duplicate_rows;
1076
43
            RowLocation loc;
1077
43
            RowsetSharedPtr rowset;
1078
43
            Status st = _writer._tablet->lookup_row_key(
1079
43
                    key, &_tablet_schema, false, specified_rowsets, &loc,
1080
43
                    _writer._mow_context->max_version, segment_caches, &rowset, true);
1081
43
            bool is_expected_st = (st.is<ErrorCode::KEY_NOT_FOUND>() || st.ok());
1082
43
            DCHECK(is_expected_st || st.is<ErrorCode::MEM_LIMIT_EXCEEDED>())
1083
0
                    << "[BlockAggregator::aggregate_for_insert_after_delete] unexpected error "
1084
0
                       "status while lookup_row_key:"
1085
0
                    << st;
1086
43
            if (!is_expected_st) {
1087
0
                return st;
1088
0
            }
1089
1090
43
            Slice previous_seq_slice {};
1091
43
            if (st.ok()) {
1092
38
                if (_tablet_schema.has_sequence_col()) {
1093
                    // if the insert row doesn't specify the sequence column, we need to
1094
                    // read the historical's sequence column value so that we don't need
1095
                    // to handle seqeunce column in append_block_with_flexible_content()
1096
                    // for this row
1097
34
                    bool row_has_sequence_col = (!skip_bitmap.contains(seq_col_unique_id));
1098
34
                    if (!row_has_sequence_col) {
1099
14
                        read_plan.prepare_to_read(loc, block_pos);
1100
14
                        _writer._rsid_to_rowset.emplace(rowset->rowset_id(), rowset);
1101
14
                    }
1102
34
                }
1103
                // delete the existing row
1104
38
                _writer._mow_context->delete_bitmap->add(
1105
38
                        {loc.rowset_id, loc.segment_id, DeleteBitmap::TEMP_VERSION_COMMON},
1106
38
                        loc.row_id);
1107
38
            }
1108
            // and remove the row with delete sign from the current block
1109
43
            filter_map[block_pos - 1] = 0;
1110
43
        }
1111
2.19k
        previous_has_delete_sign = have_delete_sign;
1112
2.19k
        previous_key = std::move(key);
1113
2.19k
    }
1114
265
    if (duplicate_rows > 0) {
1115
9
        if (!read_plan.empty()) {
1116
            // fill sequence column value for some rows
1117
4
            RETURN_IF_ERROR(fill_sequence_column(block, num_rows, read_plan, *skip_bitmaps));
1118
4
        }
1119
9
        RETURN_IF_ERROR(filter_block(block, num_rows, std::move(filter_column), duplicate_rows,
1120
9
                                     "__filter_insert_after_delete_col__"));
1121
9
    }
1122
265
    return Status::OK();
1123
265
}
1124
1125
Status BlockAggregator::filter_block(Block* block, size_t num_rows, MutableColumnPtr filter_column,
1126
9
                                     int duplicate_rows, std::string col_name) {
1127
9
    auto num_cols = block->columns();
1128
9
    block->insert({std::move(filter_column), std::make_shared<DataTypeUInt8>(), col_name});
1129
9
    RETURN_IF_ERROR(Block::filter_block(block, num_cols, num_cols));
1130
9
    DCHECK_EQ(num_cols, block->columns());
1131
9
    size_t merged_rows = num_rows - block->rows();
1132
9
    if (duplicate_rows != merged_rows) {
1133
0
        auto msg = fmt::format(
1134
0
                "filter_block_for_flexible_partial_update {}: duplicate_rows != merged_rows, "
1135
0
                "duplicate_keys={}, merged_rows={}, num_rows={}, mutable_block->rows()={}",
1136
0
                col_name, duplicate_rows, merged_rows, num_rows, block->rows());
1137
0
        DCHECK(false) << msg;
1138
0
        return Status::InternalError<false>(msg);
1139
0
    }
1140
9
    return Status::OK();
1141
9
}
1142
1143
Status BlockAggregator::convert_pk_columns(Block* block, size_t row_pos, size_t num_rows,
1144
545
                                           std::vector<IOlapColumnDataAccessor*>& key_columns) {
1145
545
    key_columns.clear();
1146
1.14k
    for (uint32_t cid {0}; cid < _tablet_schema.num_key_columns(); cid++) {
1147
603
        RETURN_IF_ERROR(_writer._olap_data_convertor->set_source_content_with_specifid_column(
1148
603
                block->get_by_position(cid), row_pos, num_rows, cid));
1149
603
        auto [status, column] = _writer._olap_data_convertor->convert_column_data(cid);
1150
603
        if (!status.ok()) {
1151
0
            return status;
1152
0
        }
1153
603
        key_columns.push_back(column);
1154
603
    }
1155
545
    return Status::OK();
1156
545
}
1157
1158
Status BlockAggregator::convert_seq_column(Block* block, size_t row_pos, size_t num_rows,
1159
545
                                           IOlapColumnDataAccessor*& seq_column) {
1160
545
    seq_column = nullptr;
1161
545
    if (_tablet_schema.has_sequence_col()) {
1162
65
        auto seq_col_idx = _tablet_schema.sequence_col_idx();
1163
65
        RETURN_IF_ERROR(_writer._olap_data_convertor->set_source_content_with_specifid_column(
1164
65
                block->get_by_position(seq_col_idx), row_pos, num_rows, seq_col_idx));
1165
65
        auto [status, column] = _writer._olap_data_convertor->convert_column_data(seq_col_idx);
1166
65
        if (!status.ok()) {
1167
0
            return status;
1168
0
        }
1169
65
        seq_column = column;
1170
65
    }
1171
545
    return Status::OK();
1172
545
};
1173
1174
Status BlockAggregator::aggregate_for_flexible_partial_update(
1175
        Block* block, size_t num_rows, const std::vector<RowsetSharedPtr>& specified_rowsets,
1176
265
        std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches) {
1177
265
    std::vector<IOlapColumnDataAccessor*> key_columns {};
1178
265
    IOlapColumnDataAccessor* seq_column {nullptr};
1179
1180
265
    RETURN_IF_ERROR(convert_pk_columns(block, 0, num_rows, key_columns));
1181
265
    RETURN_IF_ERROR(convert_seq_column(block, 0, num_rows, seq_column));
1182
1183
    // 1. merge duplicate rows when table has sequence column
1184
    // When there are multiple rows with the same keys in memtable, some of them specify specify the sequence column,
1185
    // some of them don't. We can't do the de-duplication in memtable because we don't know the historical data. We must
1186
    // de-duplicate them here.
1187
265
    if (_tablet_schema.has_sequence_col()) {
1188
25
        RETURN_IF_ERROR(aggregate_for_sequence_column(block, static_cast<int>(num_rows),
1189
25
                                                      key_columns, seq_column, specified_rowsets,
1190
25
                                                      segment_caches));
1191
25
    }
1192
1193
    // 2. merge duplicate rows and handle insert after delete
1194
265
    if (block->rows() != num_rows) {
1195
15
        num_rows = block->rows();
1196
        // data in block has changed, should re-encode key columns, sequence column
1197
15
        _writer._olap_data_convertor->clear_source_content();
1198
15
        RETURN_IF_ERROR(convert_pk_columns(block, 0, num_rows, key_columns));
1199
15
        RETURN_IF_ERROR(convert_seq_column(block, 0, num_rows, seq_column));
1200
15
    }
1201
265
    RETURN_IF_ERROR(aggregate_for_insert_after_delete(block, num_rows, key_columns,
1202
265
                                                      specified_rowsets, segment_caches));
1203
265
    return Status::OK();
1204
265
}
1205
1206
} // namespace doris