Coverage Report

Created: 2025-07-23 20:08

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/root/doris/be/src/olap/base_tablet.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 "olap/base_tablet.h"
19
20
#include <bthread/mutex.h>
21
#include <fmt/format.h>
22
#include <rapidjson/prettywriter.h>
23
24
#include <algorithm>
25
#include <cstdint>
26
#include <iterator>
27
#include <random>
28
#include <shared_mutex>
29
30
#include "cloud/cloud_tablet.h"
31
#include "cloud/config.h"
32
#include "common/cast_set.h"
33
#include "common/logging.h"
34
#include "common/status.h"
35
#include "olap/calc_delete_bitmap_executor.h"
36
#include "olap/cumulative_compaction_time_series_policy.h"
37
#include "olap/delete_bitmap_calculator.h"
38
#include "olap/iterators.h"
39
#include "olap/memtable.h"
40
#include "olap/partial_update_info.h"
41
#include "olap/primary_key_index.h"
42
#include "olap/rowid_conversion.h"
43
#include "olap/rowset/beta_rowset.h"
44
#include "olap/rowset/rowset.h"
45
#include "olap/rowset/rowset_fwd.h"
46
#include "olap/rowset/rowset_reader.h"
47
#include "olap/tablet_fwd.h"
48
#include "olap/txn_manager.h"
49
#include "service/point_query_executor.h"
50
#include "util/bvar_helper.h"
51
#include "util/crc32c.h"
52
#include "util/debug_points.h"
53
#include "util/doris_metrics.h"
54
#include "util/key_util.h"
55
#include "vec/common/assert_cast.h"
56
#include "vec/common/schema_util.h"
57
#include "vec/data_types/data_type_factory.hpp"
58
#include "vec/jsonb/serialize.h"
59
60
namespace doris {
61
#include "common/compile_check_begin.h"
62
63
using namespace ErrorCode;
64
65
namespace {
66
67
bvar::LatencyRecorder g_tablet_commit_phase_update_delete_bitmap_latency(
68
        "doris_pk", "commit_phase_update_delete_bitmap");
69
bvar::LatencyRecorder g_tablet_lookup_rowkey_latency("doris_pk", "tablet_lookup_rowkey");
70
bvar::Adder<uint64_t> g_tablet_pk_not_found("doris_pk", "lookup_not_found");
71
bvar::PerSecond<bvar::Adder<uint64_t>> g_tablet_pk_not_found_per_second(
72
        "doris_pk", "lookup_not_found_per_second", &g_tablet_pk_not_found, 60);
73
bvar::LatencyRecorder g_tablet_update_delete_bitmap_latency("doris_pk", "update_delete_bitmap");
74
75
static bvar::Adder<size_t> g_total_tablet_num("doris_total_tablet_num");
76
77
Status _get_segment_column_iterator(const BetaRowsetSharedPtr& rowset, uint32_t segid,
78
                                    const TabletColumn& target_column,
79
                                    SegmentCacheHandle* segment_cache_handle,
80
                                    std::unique_ptr<segment_v2::ColumnIterator>* column_iterator,
81
0
                                    OlapReaderStatistics* stats) {
82
0
    RETURN_IF_ERROR(SegmentLoader::instance()->load_segments(rowset, segment_cache_handle, true));
83
    // find segment
84
0
    auto it = std::find_if(
85
0
            segment_cache_handle->get_segments().begin(),
86
0
            segment_cache_handle->get_segments().end(),
87
0
            [&segid](const segment_v2::SegmentSharedPtr& seg) { return seg->id() == segid; });
88
0
    if (it == segment_cache_handle->get_segments().end()) {
89
0
        return Status::NotFound(fmt::format("rowset {} 's segemnt not found, seg_id {}",
90
0
                                            rowset->rowset_id().to_string(), segid));
91
0
    }
92
0
    segment_v2::SegmentSharedPtr segment = *it;
93
0
    StorageReadOptions opts;
94
0
    opts.stats = stats;
95
0
    RETURN_IF_ERROR(segment->new_column_iterator(target_column, column_iterator, &opts));
96
0
    segment_v2::ColumnIteratorOptions opt {
97
0
            .use_page_cache = !config::disable_storage_page_cache,
98
0
            .file_reader = segment->file_reader().get(),
99
0
            .stats = stats,
100
0
            .io_ctx = io::IOContext {.reader_type = ReaderType::READER_QUERY,
101
0
                                     .file_cache_stats = &stats->file_cache_stats},
102
0
    };
103
0
    RETURN_IF_ERROR((*column_iterator)->init(opt));
104
0
    return Status::OK();
105
0
}
106
107
} // namespace
108
109
extern MetricPrototype METRIC_query_scan_bytes;
110
extern MetricPrototype METRIC_query_scan_rows;
111
extern MetricPrototype METRIC_query_scan_count;
112
DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(flush_bytes, MetricUnit::BYTES);
113
DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(flush_finish_count, MetricUnit::OPERATIONS);
114
115
560
BaseTablet::BaseTablet(TabletMetaSharedPtr tablet_meta) : _tablet_meta(std::move(tablet_meta)) {
116
560
    _metric_entity = DorisMetrics::instance()->metric_registry()->register_entity(
117
560
            fmt::format("Tablet.{}", tablet_id()), {{"tablet_id", std::to_string(tablet_id())}},
118
560
            MetricEntityType::kTablet);
119
560
    INT_COUNTER_METRIC_REGISTER(_metric_entity, query_scan_bytes);
120
560
    INT_COUNTER_METRIC_REGISTER(_metric_entity, query_scan_rows);
121
560
    INT_COUNTER_METRIC_REGISTER(_metric_entity, query_scan_count);
122
560
    INT_COUNTER_METRIC_REGISTER(_metric_entity, flush_bytes);
123
560
    INT_COUNTER_METRIC_REGISTER(_metric_entity, flush_finish_count);
124
125
    // construct _timestamped_versioned_tracker from rs and stale rs meta
126
560
    _timestamped_version_tracker.construct_versioned_tracker(_tablet_meta->all_rs_metas(),
127
560
                                                             _tablet_meta->all_stale_rs_metas());
128
129
    // if !_tablet_meta->all_rs_metas()[0]->tablet_schema(),
130
    // that mean the tablet_meta is still no upgrade to doris 1.2 versions.
131
    // Before doris 1.2 version, rowset metas don't have tablet schema.
132
    // And when upgrade to doris 1.2 version,
133
    // all rowset metas will be set the tablet schmea from tablet meta.
134
560
    if (_tablet_meta->all_rs_metas().empty() || !_tablet_meta->all_rs_metas()[0]->tablet_schema()) {
135
523
        _max_version_schema = _tablet_meta->tablet_schema();
136
523
    } else {
137
37
        _max_version_schema =
138
37
                tablet_schema_with_merged_max_schema_version(_tablet_meta->all_rs_metas());
139
37
    }
140
560
    DCHECK(_max_version_schema);
141
560
    g_total_tablet_num << 1;
142
560
}
143
144
560
BaseTablet::~BaseTablet() {
145
560
    DorisMetrics::instance()->metric_registry()->deregister_entity(_metric_entity);
146
560
    g_total_tablet_num << -1;
147
560
}
148
149
TabletSchemaSPtr BaseTablet::tablet_schema_with_merged_max_schema_version(
150
79
        const std::vector<RowsetMetaSharedPtr>& rowset_metas) {
151
79
    RowsetMetaSharedPtr max_schema_version_rs = *std::max_element(
152
79
            rowset_metas.begin(), rowset_metas.end(),
153
248
            [](const RowsetMetaSharedPtr& a, const RowsetMetaSharedPtr& b) {
154
248
                return !a->tablet_schema()
155
248
                               ? true
156
248
                               : (!b->tablet_schema()
157
248
                                          ? false
158
248
                                          : a->tablet_schema()->schema_version() <
159
248
                                                    b->tablet_schema()->schema_version());
160
248
            });
161
79
    TabletSchemaSPtr target_schema = max_schema_version_rs->tablet_schema();
162
79
    if (target_schema->num_variant_columns() > 0) {
163
        // For variant columns tablet schema need to be the merged wide tablet schema
164
0
        std::vector<TabletSchemaSPtr> schemas;
165
0
        std::transform(rowset_metas.begin(), rowset_metas.end(), std::back_inserter(schemas),
166
0
                       [](const RowsetMetaSharedPtr& rs_meta) { return rs_meta->tablet_schema(); });
167
0
        static_cast<void>(
168
0
                vectorized::schema_util::get_least_common_schema(schemas, nullptr, target_schema));
169
0
        VLOG_DEBUG << "dump schema: " << target_schema->dump_full_schema();
170
0
    }
171
79
    return target_schema;
172
79
}
173
174
255
Status BaseTablet::set_tablet_state(TabletState state) {
175
255
    if (_tablet_meta->tablet_state() == TABLET_SHUTDOWN && state != TABLET_SHUTDOWN) {
176
0
        return Status::Error<META_INVALID_ARGUMENT>(
177
0
                "could not change tablet state from shutdown to {}", state);
178
0
    }
179
255
    _tablet_meta->set_tablet_state(state);
180
255
    return Status::OK();
181
255
}
182
183
0
void BaseTablet::update_max_version_schema(const TabletSchemaSPtr& tablet_schema) {
184
0
    std::lock_guard wrlock(_meta_lock);
185
    // Double Check for concurrent update
186
0
    if (!_max_version_schema ||
187
0
        tablet_schema->schema_version() > _max_version_schema->schema_version()) {
188
0
        _max_version_schema = tablet_schema;
189
0
    }
190
0
}
191
192
0
Status BaseTablet::update_by_least_common_schema(const TabletSchemaSPtr& update_schema) {
193
0
    std::lock_guard wrlock(_meta_lock);
194
0
    CHECK(_max_version_schema->schema_version() >= update_schema->schema_version());
195
0
    TabletSchemaSPtr final_schema;
196
0
    bool check_column_size = true;
197
0
    RETURN_IF_ERROR(vectorized::schema_util::get_least_common_schema(
198
0
            {_max_version_schema, update_schema}, _max_version_schema, final_schema,
199
0
            check_column_size));
200
0
    _max_version_schema = final_schema;
201
0
    VLOG_DEBUG << "dump updated tablet schema: " << final_schema->dump_structure();
202
0
    return Status::OK();
203
0
}
204
205
233
uint32_t BaseTablet::get_real_compaction_score() const {
206
233
    std::shared_lock l(_meta_lock);
207
233
    const auto& rs_metas = _tablet_meta->all_rs_metas();
208
233
    return std::accumulate(rs_metas.begin(), rs_metas.end(), 0,
209
10.7k
                           [](uint32_t score, const RowsetMetaSharedPtr& rs_meta) {
210
10.7k
                               return score + rs_meta->get_compaction_score();
211
10.7k
                           });
212
233
}
213
214
Status BaseTablet::capture_rs_readers_unlocked(const Versions& version_path,
215
1
                                               std::vector<RowSetSplits>* rs_splits) const {
216
1
    DCHECK(rs_splits != nullptr && rs_splits->empty());
217
1
    for (auto version : version_path) {
218
1
        auto it = _rs_version_map.find(version);
219
1
        if (it == _rs_version_map.end()) {
220
0
            VLOG_NOTICE << "fail to find Rowset in rs_version for version. tablet=" << tablet_id()
221
0
                        << ", version='" << version.first << "-" << version.second;
222
223
0
            it = _stale_rs_version_map.find(version);
224
0
            if (it == _stale_rs_version_map.end()) {
225
0
                return Status::Error<CAPTURE_ROWSET_READER_ERROR>(
226
0
                        "fail to find Rowset in stale_rs_version for version. tablet={}, "
227
0
                        "version={}-{}",
228
0
                        tablet_id(), version.first, version.second);
229
0
            }
230
0
        }
231
1
        RowsetReaderSharedPtr rs_reader;
232
1
        auto res = it->second->create_reader(&rs_reader);
233
1
        if (!res.ok()) {
234
0
            return Status::Error<CAPTURE_ROWSET_READER_ERROR>(
235
0
                    "failed to create reader for rowset:{}", it->second->rowset_id().to_string());
236
0
        }
237
1
        rs_splits->emplace_back(std::move(rs_reader));
238
1
    }
239
1
    return Status::OK();
240
1
}
241
242
// snapshot manager may call this api to check if version exists, so that
243
// the version maybe not exist
244
RowsetSharedPtr BaseTablet::get_rowset_by_version(const Version& version,
245
1
                                                  bool find_in_stale) const {
246
1
    auto iter = _rs_version_map.find(version);
247
1
    if (iter == _rs_version_map.end()) {
248
0
        if (find_in_stale) {
249
0
            return get_stale_rowset_by_version(version);
250
0
        }
251
0
        return nullptr;
252
0
    }
253
1
    return iter->second;
254
1
}
255
256
0
RowsetSharedPtr BaseTablet::get_stale_rowset_by_version(const Version& version) const {
257
0
    auto iter = _stale_rs_version_map.find(version);
258
0
    if (iter == _stale_rs_version_map.end()) {
259
0
        VLOG_NOTICE << "no rowset for version:" << version << ", tablet: " << tablet_id();
260
0
        return nullptr;
261
0
    }
262
0
    return iter->second;
263
0
}
264
265
// Already under _meta_lock
266
42
RowsetSharedPtr BaseTablet::get_rowset_with_max_version() const {
267
42
    Version max_version = _tablet_meta->max_version();
268
42
    if (max_version.first == -1) {
269
0
        return nullptr;
270
0
    }
271
272
42
    auto iter = _rs_version_map.find(max_version);
273
42
    if (iter == _rs_version_map.end()) {
274
0
        DCHECK(false) << "invalid version:" << max_version;
275
0
        return nullptr;
276
0
    }
277
42
    return iter->second;
278
42
}
279
280
0
Status BaseTablet::get_all_rs_id(int64_t max_version, RowsetIdUnorderedSet* rowset_ids) const {
281
0
    std::shared_lock rlock(_meta_lock);
282
0
    return get_all_rs_id_unlocked(max_version, rowset_ids);
283
0
}
284
285
Status BaseTablet::get_all_rs_id_unlocked(int64_t max_version,
286
9
                                          RowsetIdUnorderedSet* rowset_ids) const {
287
    //  Ensure that the obtained versions of rowsets are continuous
288
9
    Version spec_version(0, max_version);
289
9
    Versions version_path;
290
9
    auto st = _timestamped_version_tracker.capture_consistent_versions(spec_version, &version_path);
291
9
    if (!st.ok()) [[unlikely]] {
292
0
        return st;
293
0
    }
294
295
11
    for (auto& ver : version_path) {
296
11
        if (ver.second == 1) {
297
            // [0-1] rowset is empty for each tablet, skip it
298
9
            continue;
299
9
        }
300
2
        auto it = _rs_version_map.find(ver);
301
2
        if (it == _rs_version_map.end()) {
302
0
            return Status::Error<CAPTURE_ROWSET_ERROR, false>(
303
0
                    "fail to find Rowset for version. tablet={}, version={}", tablet_id(),
304
0
                    ver.to_string());
305
0
        }
306
2
        rowset_ids->emplace(it->second->rowset_id());
307
2
    }
308
9
    return Status::OK();
309
9
}
310
311
0
Versions BaseTablet::get_missed_versions(int64_t spec_version) const {
312
0
    DCHECK(spec_version > 0) << "invalid spec_version: " << spec_version;
313
314
0
    Versions existing_versions;
315
0
    {
316
0
        std::shared_lock rdlock(_meta_lock);
317
0
        for (const auto& rs : _tablet_meta->all_rs_metas()) {
318
0
            existing_versions.emplace_back(rs->version());
319
0
        }
320
0
    }
321
0
    return calc_missed_versions(spec_version, std::move(existing_versions));
322
0
}
323
324
2
Versions BaseTablet::get_missed_versions_unlocked(int64_t spec_version) const {
325
2
    DCHECK(spec_version > 0) << "invalid spec_version: " << spec_version;
326
327
2
    Versions existing_versions;
328
8
    for (const auto& rs : _tablet_meta->all_rs_metas()) {
329
8
        existing_versions.emplace_back(rs->version());
330
8
    }
331
2
    return calc_missed_versions(spec_version, std::move(existing_versions));
332
2
}
333
334
1
void BaseTablet::_print_missed_versions(const Versions& missed_versions) const {
335
1
    std::stringstream ss;
336
1
    ss << tablet_id() << " has " << missed_versions.size() << " missed version:";
337
    // print at most 10 version
338
3
    for (int i = 0; i < 10 && i < missed_versions.size(); ++i) {
339
2
        ss << missed_versions[i] << ",";
340
2
    }
341
1
    LOG(WARNING) << ss.str();
342
1
}
343
344
1
bool BaseTablet::_reconstruct_version_tracker_if_necessary() {
345
1
    double orphan_vertex_ratio = _timestamped_version_tracker.get_orphan_vertex_ratio();
346
1
    if (orphan_vertex_ratio >= config::tablet_version_graph_orphan_vertex_ratio) {
347
1
        _timestamped_version_tracker.construct_versioned_tracker(
348
1
                _tablet_meta->all_rs_metas(), _tablet_meta->all_stale_rs_metas());
349
1
        return true;
350
1
    }
351
0
    return false;
352
1
}
353
354
// should use this method to get a copy of current tablet meta
355
// there are some rowset meta in local meta store and in in-memory tablet meta
356
// but not in tablet meta in local meta store
357
0
void BaseTablet::generate_tablet_meta_copy(TabletMeta& new_tablet_meta) const {
358
0
    TabletMetaPB tablet_meta_pb;
359
0
    {
360
0
        std::shared_lock rdlock(_meta_lock);
361
0
        _tablet_meta->to_meta_pb(&tablet_meta_pb);
362
0
    }
363
0
    generate_tablet_meta_copy_unlocked(new_tablet_meta);
364
0
}
365
366
// this is a unlocked version of generate_tablet_meta_copy()
367
// some method already hold the _meta_lock before calling this,
368
// such as EngineCloneTask::_finish_clone -> tablet->revise_tablet_meta
369
4
void BaseTablet::generate_tablet_meta_copy_unlocked(TabletMeta& new_tablet_meta) const {
370
4
    TabletMetaPB tablet_meta_pb;
371
4
    _tablet_meta->to_meta_pb(&tablet_meta_pb);
372
4
    new_tablet_meta.init_from_pb(tablet_meta_pb);
373
4
}
374
375
Status BaseTablet::calc_delete_bitmap_between_segments(
376
        RowsetId rowset_id, const std::vector<segment_v2::SegmentSharedPtr>& segments,
377
0
        DeleteBitmapPtr delete_bitmap) {
378
0
    size_t const num_segments = segments.size();
379
0
    if (num_segments < 2) {
380
0
        return Status::OK();
381
0
    }
382
383
0
    OlapStopWatch watch;
384
0
    size_t seq_col_length = 0;
385
0
    if (_tablet_meta->tablet_schema()->has_sequence_col()) {
386
0
        auto seq_col_idx = _tablet_meta->tablet_schema()->sequence_col_idx();
387
0
        seq_col_length = _tablet_meta->tablet_schema()->column(seq_col_idx).length() + 1;
388
0
    }
389
0
    size_t rowid_length = 0;
390
0
    if (!_tablet_meta->tablet_schema()->cluster_key_uids().empty()) {
391
0
        rowid_length = PrimaryKeyIndexReader::ROW_ID_LENGTH;
392
0
    }
393
394
0
    MergeIndexDeleteBitmapCalculator calculator;
395
0
    RETURN_IF_ERROR(calculator.init(rowset_id, segments, seq_col_length, rowid_length));
396
397
0
    RETURN_IF_ERROR(calculator.calculate_all(delete_bitmap));
398
399
0
    delete_bitmap->add(
400
0
            {rowset_id, DeleteBitmap::INVALID_SEGMENT_ID, DeleteBitmap::TEMP_VERSION_COMMON},
401
0
            DeleteBitmap::ROWSET_SENTINEL_MARK);
402
0
    LOG(INFO) << fmt::format(
403
0
            "construct delete bitmap between segments, "
404
0
            "tablet: {}, rowset: {}, number of segments: {}, bitmap count: {}, bitmap cardinality: "
405
0
            "{}, cost {} (us)",
406
0
            tablet_id(), rowset_id.to_string(), num_segments,
407
0
            delete_bitmap->get_delete_bitmap_count(), delete_bitmap->cardinality(),
408
0
            watch.get_elapse_time_us());
409
0
    return Status::OK();
410
0
}
411
412
std::vector<RowsetSharedPtr> BaseTablet::get_rowset_by_ids(
413
72
        const RowsetIdUnorderedSet* specified_rowset_ids) {
414
72
    std::vector<RowsetSharedPtr> rowsets;
415
72
    for (auto& rs : _rs_version_map) {
416
15
        if (!specified_rowset_ids ||
417
15
            specified_rowset_ids->find(rs.second->rowset_id()) != specified_rowset_ids->end()) {
418
1
            rowsets.push_back(rs.second);
419
1
        }
420
15
    }
421
422
72
    std::sort(rowsets.begin(), rowsets.end(), [](RowsetSharedPtr& lhs, RowsetSharedPtr& rhs) {
423
0
        return lhs->end_version() > rhs->end_version();
424
0
    });
425
72
    return rowsets;
426
72
}
427
428
Status BaseTablet::lookup_row_data(const Slice& encoded_key, const RowLocation& row_location,
429
                                   RowsetSharedPtr input_rowset, OlapReaderStatistics& stats,
430
0
                                   std::string& values, bool write_to_cache) {
431
0
    MonotonicStopWatch watch;
432
0
    size_t row_size = 1;
433
0
    watch.start();
434
0
    Defer _defer([&]() {
435
0
        LOG_EVERY_N(INFO, 500) << "get a single_row, cost(us):" << watch.elapsed_time() / 1000
436
0
                               << ", row_size:" << row_size;
437
0
    });
438
439
0
    BetaRowsetSharedPtr rowset = std::static_pointer_cast<BetaRowset>(input_rowset);
440
0
    CHECK(rowset);
441
0
    const TabletSchemaSPtr tablet_schema = rowset->tablet_schema();
442
0
    SegmentCacheHandle segment_cache_handle;
443
0
    std::unique_ptr<segment_v2::ColumnIterator> column_iterator;
444
0
    const auto& column = *DORIS_TRY(tablet_schema->column(BeConsts::ROW_STORE_COL));
445
0
    RETURN_IF_ERROR(_get_segment_column_iterator(rowset, row_location.segment_id, column,
446
0
                                                 &segment_cache_handle, &column_iterator, &stats));
447
    // get and parse tuple row
448
0
    vectorized::MutableColumnPtr column_ptr = vectorized::ColumnString::create();
449
0
    std::vector<segment_v2::rowid_t> rowids {static_cast<segment_v2::rowid_t>(row_location.row_id)};
450
0
    RETURN_IF_ERROR(column_iterator->read_by_rowids(rowids.data(), 1, column_ptr));
451
0
    assert(column_ptr->size() == 1);
452
0
    auto* string_column = static_cast<vectorized::ColumnString*>(column_ptr.get());
453
0
    StringRef value = string_column->get_data_at(0);
454
0
    values = value.to_string();
455
0
    if (write_to_cache) {
456
0
        RowCache::instance()->insert({tablet_id(), encoded_key}, Slice {value.data, value.size});
457
0
    }
458
0
    return Status::OK();
459
0
}
460
461
Status BaseTablet::lookup_row_key(const Slice& encoded_key, TabletSchema* latest_schema,
462
                                  bool with_seq_col,
463
                                  const std::vector<RowsetSharedPtr>& specified_rowsets,
464
                                  RowLocation* row_location, uint32_t version,
465
                                  std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches,
466
                                  RowsetSharedPtr* rowset, bool with_rowid,
467
                                  std::string* encoded_seq_value, OlapReaderStatistics* stats,
468
2
                                  DeleteBitmapPtr delete_bitmap) {
469
2
    SCOPED_BVAR_LATENCY(g_tablet_lookup_rowkey_latency);
470
2
    size_t seq_col_length = 0;
471
    // use the latest tablet schema to decide if the tablet has sequence column currently
472
2
    const TabletSchema* schema =
473
2
            (latest_schema == nullptr ? _tablet_meta->tablet_schema().get() : latest_schema);
474
2
    if (schema->has_sequence_col() && with_seq_col) {
475
2
        seq_col_length = schema->column(schema->sequence_col_idx()).length() + 1;
476
2
    }
477
2
    size_t rowid_length = 0;
478
2
    if (with_rowid && !schema->cluster_key_uids().empty()) {
479
0
        rowid_length = PrimaryKeyIndexReader::ROW_ID_LENGTH;
480
0
    }
481
2
    Slice key_without_seq =
482
2
            Slice(encoded_key.get_data(), encoded_key.get_size() - seq_col_length - rowid_length);
483
2
    RowLocation loc;
484
485
2
    auto tablet_delete_bitmap =
486
2
            delete_bitmap == nullptr ? _tablet_meta->delete_bitmap_ptr() : delete_bitmap;
487
2
    for (size_t i = 0; i < specified_rowsets.size(); i++) {
488
2
        const auto& rs = specified_rowsets[i];
489
2
        std::vector<KeyBoundsPB> segments_key_bounds;
490
2
        rs->rowset_meta()->get_segments_key_bounds(&segments_key_bounds);
491
2
        int num_segments = cast_set<int>(rs->num_segments());
492
2
        DCHECK_EQ(segments_key_bounds.size(), num_segments);
493
2
        std::vector<uint32_t> picked_segments;
494
4
        for (int j = num_segments - 1; j >= 0; j--) {
495
2
            if (key_is_not_in_segment(key_without_seq, segments_key_bounds[j],
496
2
                                      rs->rowset_meta()->is_segments_key_bounds_truncated())) {
497
0
                continue;
498
0
            }
499
2
            picked_segments.emplace_back(j);
500
2
        }
501
2
        if (picked_segments.empty()) {
502
0
            continue;
503
0
        }
504
505
2
        if (UNLIKELY(segment_caches[i] == nullptr)) {
506
1
            segment_caches[i] = std::make_unique<SegmentCacheHandle>();
507
1
            RETURN_IF_ERROR(SegmentLoader::instance()->load_segments(
508
1
                    std::static_pointer_cast<BetaRowset>(rs), segment_caches[i].get(), true, true));
509
1
        }
510
2
        auto& segments = segment_caches[i]->get_segments();
511
2
        DCHECK_EQ(segments.size(), num_segments);
512
513
2
        for (auto id : picked_segments) {
514
2
            Status s = segments[id]->lookup_row_key(encoded_key, schema, with_seq_col, with_rowid,
515
2
                                                    &loc, stats, encoded_seq_value);
516
2
            if (s.is<KEY_NOT_FOUND>()) {
517
0
                continue;
518
0
            }
519
2
            if (!s.ok() && !s.is<KEY_ALREADY_EXISTS>()) {
520
0
                return s;
521
0
            }
522
2
            if (s.ok() && tablet_delete_bitmap->contains_agg_without_cache(
523
1
                                  {loc.rowset_id, loc.segment_id, version}, loc.row_id)) {
524
                // if has sequence col, we continue to compare the sequence_id of
525
                // all rowsets, util we find an existing key.
526
0
                if (schema->has_sequence_col()) {
527
0
                    continue;
528
0
                }
529
                // The key is deleted, we don't need to search for it any more.
530
0
                break;
531
0
            }
532
            // `st` is either OK or KEY_ALREADY_EXISTS now.
533
            // for partial update, even if the key is already exists, we still need to
534
            // read it's original values to keep all columns align.
535
2
            *row_location = loc;
536
2
            if (rowset) {
537
                // return it's rowset
538
2
                *rowset = rs;
539
2
            }
540
            // find it and return
541
2
            return s;
542
2
        }
543
2
    }
544
0
    g_tablet_pk_not_found << 1;
545
0
    return Status::Error<ErrorCode::KEY_NOT_FOUND>("can't find key in all rowsets");
546
2
}
547
548
// if user pass a token, then all calculation works will submit to a threadpool,
549
// user can get all delete bitmaps from that token.
550
// if `token` is nullptr, the calculation will run in local, and user can get the result
551
// delete bitmap from `delete_bitmap` directly.
552
Status BaseTablet::calc_delete_bitmap(
553
        const BaseTabletSPtr& tablet, RowsetSharedPtr rowset,
554
        const std::vector<segment_v2::SegmentSharedPtr>& segments,
555
        const std::vector<RowsetSharedPtr>& specified_rowsets, DeleteBitmapPtr delete_bitmap,
556
        int64_t end_version, CalcDeleteBitmapToken* token, RowsetWriter* rowset_writer,
557
        DeleteBitmapPtr tablet_delete_bitmap,
558
69
        std::function<void(segment_v2::SegmentSharedPtr, Status)> callback) {
559
69
    if (specified_rowsets.empty() || segments.empty()) {
560
68
        return Status::OK();
561
68
    }
562
563
1
    OlapStopWatch watch;
564
1
    for (const auto& segment : segments) {
565
1
        const auto& seg = segment;
566
1
        if (token != nullptr) {
567
1
            RETURN_IF_ERROR(token->submit(tablet, rowset, seg, specified_rowsets, end_version,
568
1
                                          delete_bitmap, rowset_writer, tablet_delete_bitmap,
569
1
                                          callback));
570
1
        } else {
571
0
            RETURN_IF_ERROR(tablet->calc_segment_delete_bitmap(
572
0
                    rowset, segment, specified_rowsets, delete_bitmap, end_version, rowset_writer,
573
0
                    tablet_delete_bitmap));
574
0
        }
575
1
    }
576
577
1
    return Status::OK();
578
1
}
579
580
Status BaseTablet::calc_segment_delete_bitmap(RowsetSharedPtr rowset,
581
                                              const segment_v2::SegmentSharedPtr& seg,
582
                                              const std::vector<RowsetSharedPtr>& specified_rowsets,
583
                                              DeleteBitmapPtr delete_bitmap, int64_t end_version,
584
                                              RowsetWriter* rowset_writer,
585
1
                                              DeleteBitmapPtr tablet_delete_bitmap) {
586
1
    OlapStopWatch watch;
587
1
    auto rowset_id = rowset->rowset_id();
588
1
    Version dummy_version(end_version + 1, end_version + 1);
589
1
    auto rowset_schema = rowset->tablet_schema();
590
591
1
    PartialUpdateInfo* partial_update_info {nullptr};
592
1
    bool is_partial_update = rowset_writer && rowset_writer->is_partial_update();
593
    // `have_input_seq_column` is for fixed partial update only. For flexible partial update, we should use
594
    // the skip bitmap to determine wheather a row has specified the sequence column
595
1
    bool have_input_seq_column = false;
596
    // `rids_be_overwritten` is for flexible partial update only, it records row ids that is overwritten by
597
    // another row with higher seqeucne value
598
1
    std::set<uint32_t> rids_be_overwritten;
599
1
    if (is_partial_update) {
600
0
        partial_update_info = rowset_writer->get_partial_update_info().get();
601
0
        if (partial_update_info->is_fixed_partial_update() && rowset_schema->has_sequence_col()) {
602
0
            std::vector<uint32_t> including_cids =
603
0
                    rowset_writer->get_partial_update_info()->update_cids;
604
0
            have_input_seq_column =
605
0
                    rowset_schema->has_sequence_col() &&
606
0
                    (std::find(including_cids.cbegin(), including_cids.cend(),
607
0
                               rowset_schema->sequence_col_idx()) != including_cids.cend());
608
0
        }
609
0
    }
610
611
1
    DBUG_EXECUTE_IF("BaseTablet::calc_segment_delete_bitmap.sleep", {
612
1
        auto target_tablet_id = dp->param<int64_t>("tablet_id", -1);
613
1
        auto sleep = dp->param<int64_t>("sleep", 10);
614
1
        if (target_tablet_id == tablet_id()) {
615
1
            std::this_thread::sleep_for(std::chrono::seconds(sleep));
616
1
        }
617
1
    });
618
619
1
    if (rowset_schema->num_variant_columns() > 0) {
620
        // During partial updates, the extracted columns of a variant should not be included in the rowset schema.
621
        // This is because the partial update for a variant needs to ignore the extracted columns.
622
        // Otherwise, the schema types in different rowsets might be inconsistent. When performing a partial update,
623
        // the complete variant is constructed by reading all the sub-columns of the variant.
624
0
        rowset_schema = rowset_schema->copy_without_variant_extracted_columns();
625
0
    }
626
    // use for partial update
627
1
    FixedReadPlan read_plan_ori;
628
1
    FixedReadPlan read_plan_update;
629
1
    int64_t conflict_rows = 0;
630
1
    int64_t new_generated_rows = 0;
631
632
1
    std::map<RowsetId, RowsetSharedPtr> rsid_to_rowset;
633
1
    rsid_to_rowset[rowset_id] = rowset;
634
1
    vectorized::Block block = rowset_schema->create_block();
635
1
    vectorized::Block ordered_block = block.clone_empty();
636
1
    uint32_t pos = 0;
637
638
1
    RETURN_IF_ERROR(seg->load_pk_index_and_bf(nullptr)); // We need index blocks to iterate
639
1
    const auto* pk_idx = seg->get_primary_key_index();
640
1
    int total = pk_idx->num_rows();
641
1
    uint32_t row_id = 0;
642
1
    int32_t remaining = total;
643
1
    bool exact_match = false;
644
1
    std::string last_key;
645
1
    int batch_size = 1024;
646
    // The data for each segment may be lookup multiple times. Creating a SegmentCacheHandle
647
    // will update the lru cache, and there will be obvious lock competition in multithreading
648
    // scenarios, so using a segment_caches to cache SegmentCacheHandle.
649
1
    std::vector<std::unique_ptr<SegmentCacheHandle>> segment_caches(specified_rowsets.size());
650
2
    while (remaining > 0) {
651
1
        std::unique_ptr<segment_v2::IndexedColumnIterator> iter;
652
1
        RETURN_IF_ERROR(pk_idx->new_iterator(&iter, nullptr));
653
654
1
        size_t num_to_read = std::min(batch_size, remaining);
655
1
        auto index_type = vectorized::DataTypeFactory::instance().create_data_type(
656
1
                pk_idx->type_info()->type(), 1, 0);
657
1
        auto index_column = index_type->create_column();
658
1
        Slice last_key_slice(last_key);
659
1
        RETURN_IF_ERROR(iter->seek_at_or_after(&last_key_slice, &exact_match));
660
1
        auto current_ordinal = iter->get_current_ordinal();
661
1
        DCHECK(total == remaining + current_ordinal)
662
0
                << "total: " << total << ", remaining: " << remaining
663
0
                << ", current_ordinal: " << current_ordinal;
664
665
1
        size_t num_read = num_to_read;
666
1
        RETURN_IF_ERROR(iter->next_batch(&num_read, index_column));
667
1
        DCHECK(num_to_read == num_read)
668
0
                << "num_to_read: " << num_to_read << ", num_read: " << num_read;
669
1
        last_key = index_column->get_data_at(num_read - 1).to_string();
670
671
        // exclude last_key, last_key will be read in next batch.
672
1
        if (num_read == batch_size && num_read != remaining) {
673
0
            num_read -= 1;
674
0
        }
675
3
        for (size_t i = 0; i < num_read; i++, row_id++) {
676
2
            Slice key = Slice(index_column->get_data_at(i).data, index_column->get_data_at(i).size);
677
2
            RowLocation loc;
678
            // calculate row id
679
2
            if (!_tablet_meta->tablet_schema()->cluster_key_uids().empty()) {
680
0
                size_t seq_col_length = 0;
681
0
                if (_tablet_meta->tablet_schema()->has_sequence_col()) {
682
0
                    seq_col_length =
683
0
                            _tablet_meta->tablet_schema()
684
0
                                    ->column(_tablet_meta->tablet_schema()->sequence_col_idx())
685
0
                                    .length() +
686
0
                            1;
687
0
                }
688
0
                size_t rowid_length = PrimaryKeyIndexReader::ROW_ID_LENGTH;
689
0
                Slice key_without_seq =
690
0
                        Slice(key.get_data(), key.get_size() - seq_col_length - rowid_length);
691
0
                Slice rowid_slice =
692
0
                        Slice(key.get_data() + key_without_seq.get_size() + seq_col_length + 1,
693
0
                              rowid_length - 1);
694
0
                const auto* type_info =
695
0
                        get_scalar_type_info<FieldType::OLAP_FIELD_TYPE_UNSIGNED_INT>();
696
0
                const auto* rowid_coder = get_key_coder(type_info->type());
697
0
                RETURN_IF_ERROR(rowid_coder->decode_ascending(&rowid_slice, rowid_length,
698
0
                                                              (uint8_t*)&row_id));
699
0
            }
700
            // same row in segments should be filtered
701
2
            if (delete_bitmap->contains({rowset_id, seg->id(), DeleteBitmap::TEMP_VERSION_COMMON},
702
2
                                        row_id)) {
703
0
                continue;
704
0
            }
705
706
2
            DBUG_EXECUTE_IF("BaseTablet::calc_segment_delete_bitmap.inject_err", {
707
2
                auto p = dp->param("percent", 0.01);
708
2
                std::mt19937 gen {std::random_device {}()};
709
2
                std::bernoulli_distribution inject_fault {p};
710
2
                if (inject_fault(gen)) {
711
2
                    return Status::InternalError(
712
2
                            "injection error in calc_segment_delete_bitmap, "
713
2
                            "tablet_id={}, rowset_id={}",
714
2
                            tablet_id(), rowset_id.to_string());
715
2
                }
716
2
            });
717
718
2
            RowsetSharedPtr rowset_find;
719
2
            Status st = Status::OK();
720
2
            if (tablet_delete_bitmap == nullptr) {
721
2
                st = lookup_row_key(key, rowset_schema.get(), true, specified_rowsets, &loc,
722
2
                                    cast_set<uint32_t>(dummy_version.first - 1), segment_caches,
723
2
                                    &rowset_find);
724
2
            } else {
725
0
                st = lookup_row_key(key, rowset_schema.get(), true, specified_rowsets, &loc,
726
0
                                    cast_set<uint32_t>(dummy_version.first - 1), segment_caches,
727
0
                                    &rowset_find, true, nullptr, nullptr, tablet_delete_bitmap);
728
0
            }
729
2
            bool expected_st = st.ok() || st.is<KEY_NOT_FOUND>() || st.is<KEY_ALREADY_EXISTS>();
730
            // It's a defensive DCHECK, we need to exclude some common errors to avoid core-dump
731
            // while stress test
732
2
            DCHECK(expected_st || st.is<MEM_LIMIT_EXCEEDED>())
733
0
                    << "unexpected error status while lookup_row_key:" << st;
734
2
            if (!expected_st) {
735
0
                return st;
736
0
            }
737
2
            if (st.is<KEY_NOT_FOUND>()) {
738
0
                continue;
739
0
            }
740
741
2
            ++conflict_rows;
742
2
            if (st.is<KEY_ALREADY_EXISTS>() &&
743
2
                (!is_partial_update ||
744
1
                 (partial_update_info->is_fixed_partial_update() && have_input_seq_column))) {
745
                // `st.is<KEY_ALREADY_EXISTS>()` means that there exists a row with the same key and larger value
746
                // in seqeunce column.
747
                // - If the current load is not a partial update, we just delete current row.
748
                // - Otherwise, it means that we are doing the alignment process in publish phase due to conflicts
749
                // during concurrent partial updates. And there exists another load which introduces a row with
750
                // the same keys and larger sequence column value published successfully after the commit phase
751
                // of the current load.
752
                //     - If the columns we update include sequence column, we should delete the current row becase the
753
                //       partial update on the current row has been `overwritten` by the previous one with larger sequence
754
                //       column value.
755
                //     - Otherwise, we should combine the values of the missing columns in the previous row and the values
756
                //       of the including columns in the current row into a new row.
757
1
                delete_bitmap->add({rowset_id, seg->id(), DeleteBitmap::TEMP_VERSION_COMMON},
758
1
                                   row_id);
759
1
                continue;
760
                // NOTE: for partial update which doesn't specify the sequence column, we can't use the sequence column value filled in flush phase
761
                // as its final value. Otherwise it may cause inconsistency between replicas.
762
1
            }
763
1
            if (is_partial_update && rowset_writer != nullptr) {
764
                // In publish version, record rows to be deleted for concurrent update
765
                // For example, if version 5 and 6 update a row, but version 6 only see
766
                // version 4 when write, and when publish version, version 5's value will
767
                // be marked as deleted and it's update is losed.
768
                // So here we should read version 5's columns and build a new row, which is
769
                // consists of version 6's update columns and version 5's origin columns
770
                // here we build 2 read plan for ori values and update values
771
772
                // - for fixed partial update, we should read update columns from current load's rowset
773
                // and read missing columns from previous rowsets to create the final block
774
                // - for flexible partial update, we should read all columns from current load's rowset
775
                // and read non sort key columns from previous rowsets to create the final block
776
                // So we only need to record rows to read for both mode partial update
777
0
                read_plan_ori.prepare_to_read(loc, pos);
778
0
                read_plan_update.prepare_to_read(RowLocation {rowset_id, seg->id(), row_id}, pos);
779
780
                // For flexible partial update, we should use skip bitmap to determine wheather
781
                // a row has specified the sequence column. But skip bitmap should be read from the segment.
782
                // So we record these row ids and process and filter them in `generate_new_block_for_flexible_partial_update()`
783
0
                if (st.is<KEY_ALREADY_EXISTS>() &&
784
0
                    partial_update_info->is_flexible_partial_update()) {
785
0
                    rids_be_overwritten.insert(pos);
786
0
                }
787
788
0
                rsid_to_rowset[rowset_find->rowset_id()] = rowset_find;
789
0
                ++pos;
790
791
                // delete bitmap will be calculate when memtable flush and
792
                // publish. The two stages may see different versions.
793
                // When there is sequence column, the currently imported data
794
                // of rowset may be marked for deletion at memtablet flush or
795
                // publish because the seq column is smaller than the previous
796
                // rowset.
797
                // just set 0 as a unified temporary version number, and update to
798
                // the real version number later.
799
0
                delete_bitmap->add(
800
0
                        {loc.rowset_id, loc.segment_id, DeleteBitmap::TEMP_VERSION_COMMON},
801
0
                        loc.row_id);
802
0
                delete_bitmap->add({rowset_id, seg->id(), DeleteBitmap::TEMP_VERSION_COMMON},
803
0
                                   row_id);
804
0
                ++new_generated_rows;
805
0
                continue;
806
0
            }
807
            // when st = ok
808
1
            delete_bitmap->add({loc.rowset_id, loc.segment_id, DeleteBitmap::TEMP_VERSION_COMMON},
809
1
                               loc.row_id);
810
1
        }
811
1
        remaining -= num_read;
812
1
    }
813
    // DCHECK_EQ(total, row_id) << "segment total rows: " << total << " row_id:" << row_id;
814
815
1
    if (config::enable_merge_on_write_correctness_check) {
816
1
        RowsetIdUnorderedSet rowsetids;
817
1
        for (const auto& specified_rowset : specified_rowsets) {
818
1
            rowsetids.emplace(specified_rowset->rowset_id());
819
1
            VLOG_NOTICE << "[tabletID:" << tablet_id() << "]"
820
0
                        << "[add_sentinel_mark_to_delete_bitmap][end_version:" << end_version << "]"
821
0
                        << "add:" << specified_rowset->rowset_id();
822
1
        }
823
1
        add_sentinel_mark_to_delete_bitmap(delete_bitmap.get(), rowsetids);
824
1
    }
825
826
1
    if (pos > 0) {
827
0
        DCHECK(partial_update_info);
828
0
        if (partial_update_info->is_fixed_partial_update()) {
829
0
            RETURN_IF_ERROR(generate_new_block_for_partial_update(
830
0
                    rowset_schema, partial_update_info, read_plan_ori, read_plan_update,
831
0
                    rsid_to_rowset, &block));
832
0
        } else {
833
0
            RETURN_IF_ERROR(generate_new_block_for_flexible_partial_update(
834
0
                    rowset_schema, partial_update_info, rids_be_overwritten, read_plan_ori,
835
0
                    read_plan_update, rsid_to_rowset, &block));
836
0
        }
837
0
        RETURN_IF_ERROR(sort_block(block, ordered_block));
838
0
        RETURN_IF_ERROR(rowset_writer->flush_single_block(&ordered_block));
839
0
        auto cost_us = watch.get_elapse_time_us();
840
0
        if (config::enable_mow_verbose_log || cost_us > 10 * 1000) {
841
0
            LOG(INFO) << "calc segment delete bitmap for "
842
0
                      << partial_update_info->partial_update_mode_str()
843
0
                      << ", tablet: " << tablet_id() << " rowset: " << rowset_id
844
0
                      << " seg_id: " << seg->id() << " dummy_version: " << end_version + 1
845
0
                      << " rows: " << seg->num_rows() << " conflict rows: " << conflict_rows
846
0
                      << " new generated rows: " << new_generated_rows
847
0
                      << " bitmap num: " << delete_bitmap->get_delete_bitmap_count()
848
0
                      << " bitmap cardinality: " << delete_bitmap->cardinality()
849
0
                      << " cost: " << cost_us << "(us)";
850
0
        }
851
0
        return Status::OK();
852
0
    }
853
1
    auto cost_us = watch.get_elapse_time_us();
854
1
    if (config::enable_mow_verbose_log || cost_us > 10 * 1000) {
855
0
        LOG(INFO) << "calc segment delete bitmap, tablet: " << tablet_id()
856
0
                  << " rowset: " << rowset_id << " seg_id: " << seg->id()
857
0
                  << " dummy_version: " << end_version + 1 << " rows: " << seg->num_rows()
858
0
                  << " conflict rows: " << conflict_rows
859
0
                  << " bitmap num: " << delete_bitmap->get_delete_bitmap_count()
860
0
                  << " bitmap cardinality: " << delete_bitmap->cardinality() << " cost: " << cost_us
861
0
                  << "(us)";
862
0
    }
863
1
    return Status::OK();
864
1
}
865
866
0
Status BaseTablet::sort_block(vectorized::Block& in_block, vectorized::Block& output_block) {
867
0
    vectorized::MutableBlock mutable_input_block =
868
0
            vectorized::MutableBlock::build_mutable_block(&in_block);
869
0
    vectorized::MutableBlock mutable_output_block =
870
0
            vectorized::MutableBlock::build_mutable_block(&output_block);
871
872
0
    std::shared_ptr<RowInBlockComparator> vec_row_comparator =
873
0
            std::make_shared<RowInBlockComparator>(_tablet_meta->tablet_schema());
874
0
    vec_row_comparator->set_block(&mutable_input_block);
875
876
0
    std::vector<std::unique_ptr<RowInBlock>> row_in_blocks;
877
0
    DCHECK(in_block.rows() <= std::numeric_limits<int>::max());
878
0
    row_in_blocks.reserve(in_block.rows());
879
0
    for (size_t i = 0; i < in_block.rows(); ++i) {
880
0
        row_in_blocks.emplace_back(std::make_unique<RowInBlock>(i));
881
0
    }
882
0
    std::sort(row_in_blocks.begin(), row_in_blocks.end(),
883
0
              [&](const std::unique_ptr<RowInBlock>& l,
884
0
                  const std::unique_ptr<RowInBlock>& r) -> bool {
885
0
                  auto value = (*vec_row_comparator)(l.get(), r.get());
886
0
                  DCHECK(value != 0) << "value equel when sort block, l_pos: " << l->_row_pos
887
0
                                     << " r_pos: " << r->_row_pos;
888
0
                  return value < 0;
889
0
              });
890
0
    std::vector<uint32_t> row_pos_vec;
891
0
    row_pos_vec.reserve(in_block.rows());
892
0
    for (auto& block : row_in_blocks) {
893
0
        row_pos_vec.emplace_back(block->_row_pos);
894
0
    }
895
0
    return mutable_output_block.add_rows(&in_block, row_pos_vec.data(),
896
0
                                         row_pos_vec.data() + in_block.rows());
897
0
}
898
899
// fetch value by row column
900
Status BaseTablet::fetch_value_through_row_column(RowsetSharedPtr input_rowset,
901
                                                  const TabletSchema& tablet_schema, uint32_t segid,
902
                                                  const std::vector<uint32_t>& rowids,
903
                                                  const std::vector<uint32_t>& cids,
904
0
                                                  vectorized::Block& block) {
905
0
    MonotonicStopWatch watch;
906
0
    watch.start();
907
0
    Defer _defer([&]() {
908
0
        LOG_EVERY_N(INFO, 500) << "fetch_value_by_rowids, cost(us):" << watch.elapsed_time() / 1000
909
0
                               << ", row_batch_size:" << rowids.size();
910
0
    });
911
912
0
    BetaRowsetSharedPtr rowset = std::static_pointer_cast<BetaRowset>(input_rowset);
913
0
    CHECK(rowset);
914
0
    CHECK(tablet_schema.has_row_store_for_all_columns());
915
0
    SegmentCacheHandle segment_cache_handle;
916
0
    std::unique_ptr<segment_v2::ColumnIterator> column_iterator;
917
0
    OlapReaderStatistics stats;
918
0
    const auto& column = *DORIS_TRY(tablet_schema.column(BeConsts::ROW_STORE_COL));
919
0
    RETURN_IF_ERROR(_get_segment_column_iterator(rowset, segid, column, &segment_cache_handle,
920
0
                                                 &column_iterator, &stats));
921
    // get and parse tuple row
922
0
    vectorized::MutableColumnPtr column_ptr = vectorized::ColumnString::create();
923
0
    RETURN_IF_ERROR(column_iterator->read_by_rowids(rowids.data(), rowids.size(), column_ptr));
924
0
    assert(column_ptr->size() == rowids.size());
925
0
    auto* string_column = static_cast<vectorized::ColumnString*>(column_ptr.get());
926
0
    vectorized::DataTypeSerDeSPtrs serdes;
927
0
    serdes.resize(cids.size());
928
0
    std::unordered_map<uint32_t, uint32_t> col_uid_to_idx;
929
0
    std::vector<std::string> default_values;
930
0
    default_values.resize(cids.size());
931
0
    for (int i = 0; i < cids.size(); ++i) {
932
0
        const TabletColumn& tablet_column = tablet_schema.column(cids[i]);
933
0
        vectorized::DataTypePtr type =
934
0
                vectorized::DataTypeFactory::instance().create_data_type(tablet_column);
935
0
        col_uid_to_idx[tablet_column.unique_id()] = i;
936
0
        default_values[i] = tablet_column.default_value();
937
0
        serdes[i] = type->get_serde();
938
0
    }
939
0
    RETURN_IF_ERROR(vectorized::JsonbSerializeUtil::jsonb_to_block(
940
0
            serdes, *string_column, col_uid_to_idx, block, default_values, {}));
941
0
    return Status::OK();
942
0
}
943
944
Status BaseTablet::fetch_value_by_rowids(RowsetSharedPtr input_rowset, uint32_t segid,
945
                                         const std::vector<uint32_t>& rowids,
946
                                         const TabletColumn& tablet_column,
947
0
                                         vectorized::MutableColumnPtr& dst) {
948
0
    MonotonicStopWatch watch;
949
0
    watch.start();
950
0
    Defer _defer([&]() {
951
0
        LOG_EVERY_N(INFO, 500) << "fetch_value_by_rowids, cost(us):" << watch.elapsed_time() / 1000
952
0
                               << ", row_batch_size:" << rowids.size();
953
0
    });
954
955
    // read row data
956
0
    BetaRowsetSharedPtr rowset = std::static_pointer_cast<BetaRowset>(input_rowset);
957
0
    CHECK(rowset);
958
0
    SegmentCacheHandle segment_cache_handle;
959
0
    std::unique_ptr<segment_v2::ColumnIterator> column_iterator;
960
0
    OlapReaderStatistics stats;
961
0
    RETURN_IF_ERROR(_get_segment_column_iterator(rowset, segid, tablet_column,
962
0
                                                 &segment_cache_handle, &column_iterator, &stats));
963
0
    RETURN_IF_ERROR(column_iterator->read_by_rowids(rowids.data(), rowids.size(), dst));
964
0
    return Status::OK();
965
0
}
966
967
const signed char* BaseTablet::get_delete_sign_column_data(const vectorized::Block& block,
968
0
                                                           size_t rows_at_least) {
969
0
    if (const vectorized::ColumnWithTypeAndName* delete_sign_column =
970
0
                block.try_get_by_name(DELETE_SIGN);
971
0
        delete_sign_column != nullptr) {
972
0
        const auto& delete_sign_col =
973
0
                reinterpret_cast<const vectorized::ColumnInt8&>(*(delete_sign_column->column));
974
0
        if (delete_sign_col.size() >= rows_at_least) {
975
0
            return delete_sign_col.get_data().data();
976
0
        }
977
0
    }
978
0
    return nullptr;
979
0
};
980
981
Status BaseTablet::generate_default_value_block(const TabletSchema& schema,
982
                                                const std::vector<uint32_t>& cids,
983
                                                const std::vector<std::string>& default_values,
984
                                                const vectorized::Block& ref_block,
985
0
                                                vectorized::Block& default_value_block) {
986
0
    auto mutable_default_value_columns = default_value_block.mutate_columns();
987
0
    for (auto i = 0; i < cids.size(); ++i) {
988
0
        const auto& column = schema.column(cids[i]);
989
0
        if (column.has_default_value()) {
990
0
            const auto& default_value = default_values[i];
991
0
            vectorized::ReadBuffer rb(const_cast<char*>(default_value.c_str()),
992
0
                                      default_value.size());
993
0
            RETURN_IF_ERROR(ref_block.get_by_position(i).type->from_string(
994
0
                    rb, mutable_default_value_columns[i].get()));
995
0
        }
996
0
    }
997
0
    default_value_block.set_columns(std::move(mutable_default_value_columns));
998
0
    return Status::OK();
999
0
}
1000
1001
Status BaseTablet::generate_new_block_for_partial_update(
1002
        TabletSchemaSPtr rowset_schema, const PartialUpdateInfo* partial_update_info,
1003
        const FixedReadPlan& read_plan_ori, const FixedReadPlan& read_plan_update,
1004
        const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
1005
0
        vectorized::Block* output_block) {
1006
    // do partial update related works
1007
    // 1. read columns by read plan
1008
    // 2. generate new block
1009
    // 3. write a new segment and modify rowset meta
1010
    // 4. mark current keys deleted
1011
0
    CHECK(output_block);
1012
0
    auto full_mutable_columns = output_block->mutate_columns();
1013
0
    const auto& missing_cids = partial_update_info->missing_cids;
1014
0
    const auto& update_cids = partial_update_info->update_cids;
1015
0
    auto old_block = rowset_schema->create_block_by_cids(missing_cids);
1016
0
    auto update_block = rowset_schema->create_block_by_cids(update_cids);
1017
1018
0
    bool have_input_seq_column = false;
1019
0
    if (rowset_schema->has_sequence_col()) {
1020
0
        have_input_seq_column =
1021
0
                (std::find(update_cids.cbegin(), update_cids.cend(),
1022
0
                           rowset_schema->sequence_col_idx()) != update_cids.cend());
1023
0
    }
1024
1025
    // rowid in the final block(start from 0, increase continuously) -> rowid to read in update_block
1026
0
    std::map<uint32_t, uint32_t> read_index_update;
1027
1028
    // read current rowset first, if a row in the current rowset has delete sign mark
1029
    // we don't need to read values from old block
1030
0
    RETURN_IF_ERROR(read_plan_update.read_columns_by_plan(
1031
0
            *rowset_schema, update_cids, rsid_to_rowset, update_block, &read_index_update, false));
1032
0
    size_t update_rows = read_index_update.size();
1033
0
    for (auto i = 0; i < update_cids.size(); ++i) {
1034
0
        for (auto idx = 0; idx < update_rows; ++idx) {
1035
0
            full_mutable_columns[update_cids[i]]->insert_from(
1036
0
                    *update_block.get_by_position(i).column, read_index_update[idx]);
1037
0
        }
1038
0
    }
1039
1040
    // if there is sequence column in the table, we need to read the sequence column,
1041
    // otherwise it may cause the merge-on-read based compaction policy to produce incorrect results
1042
0
    const auto* __restrict new_block_delete_signs =
1043
0
            rowset_schema->has_sequence_col()
1044
0
                    ? nullptr
1045
0
                    : get_delete_sign_column_data(update_block, update_rows);
1046
1047
    // rowid in the final block(start from 0, increase, may not continuous becasue we skip to read some rows) -> rowid to read in old_block
1048
0
    std::map<uint32_t, uint32_t> read_index_old;
1049
0
    RETURN_IF_ERROR(read_plan_ori.read_columns_by_plan(*rowset_schema, missing_cids, rsid_to_rowset,
1050
0
                                                       old_block, &read_index_old, true,
1051
0
                                                       new_block_delete_signs));
1052
0
    size_t old_rows = read_index_old.size();
1053
0
    const auto* __restrict old_block_delete_signs =
1054
0
            get_delete_sign_column_data(old_block, old_rows);
1055
0
    DCHECK(old_block_delete_signs != nullptr);
1056
    // build default value block
1057
0
    auto default_value_block = old_block.clone_empty();
1058
0
    RETURN_IF_ERROR(BaseTablet::generate_default_value_block(*rowset_schema, missing_cids,
1059
0
                                                             partial_update_info->default_values,
1060
0
                                                             old_block, default_value_block));
1061
1062
0
    CHECK(update_rows >= old_rows);
1063
1064
    // build full block
1065
0
    for (auto i = 0; i < missing_cids.size(); ++i) {
1066
0
        const auto& rs_column = rowset_schema->column(missing_cids[i]);
1067
0
        auto& mutable_column = full_mutable_columns[missing_cids[i]];
1068
0
        for (auto idx = 0; idx < update_rows; ++idx) {
1069
            // There are two cases we don't need to read values from old data:
1070
            //     1. if the conflicting new row's delete sign is marked, which means the value columns
1071
            //     of the row will not be read. So we don't need to read the missing values from the previous rows.
1072
            //     2. if the conflicting old row's delete sign is marked, which means that the key is not exist now,
1073
            //     we should not read old values from the deleted data, and should use default value instead.
1074
            //     NOTE: since now we are in the publishing phase, all data is commited
1075
            //         before, even the `strict_mode` is true (which requires partial update
1076
            //         load job can't insert new keys), this "new" key MUST be written into
1077
            //         the new generated segment file.
1078
0
            bool use_default = false;
1079
0
            bool new_row_delete_sign =
1080
0
                    (new_block_delete_signs != nullptr && new_block_delete_signs[idx]);
1081
0
            bool old_row_delete_sign = (old_block_delete_signs != nullptr &&
1082
0
                                        old_block_delete_signs[read_index_old[idx]] != 0);
1083
0
            if (old_row_delete_sign) {
1084
0
                if (!rowset_schema->has_sequence_col()) {
1085
0
                    use_default = true;
1086
0
                } else if (have_input_seq_column || !rs_column.is_seqeunce_col()) {
1087
                    // to keep the sequence column value not decreasing, we should read values of seq column
1088
                    // from old rows even if the old row is deleted when the input don't specify the sequence column, otherwise
1089
                    // it may cause the merge-on-read based compaction to produce incorrect results
1090
0
                    use_default = true;
1091
0
                }
1092
0
            }
1093
1094
0
            if (new_row_delete_sign) {
1095
0
                mutable_column->insert_default();
1096
0
            } else if (use_default) {
1097
0
                if (rs_column.has_default_value()) {
1098
0
                    mutable_column->insert_from(*default_value_block.get_by_position(i).column, 0);
1099
0
                } else if (rs_column.is_nullable()) {
1100
0
                    assert_cast<vectorized::ColumnNullable*, TypeCheckOnRelease::DISABLE>(
1101
0
                            mutable_column.get())
1102
0
                            ->insert_default();
1103
0
                } else {
1104
0
                    mutable_column->insert(rs_column.get_vec_type()->get_default());
1105
0
                }
1106
0
            } else {
1107
0
                mutable_column->insert_from(*old_block.get_by_position(i).column,
1108
0
                                            read_index_old[idx]);
1109
0
            }
1110
0
        }
1111
0
    }
1112
0
    output_block->set_columns(std::move(full_mutable_columns));
1113
0
    VLOG_DEBUG << "full block when publish: " << output_block->dump_data();
1114
0
    return Status::OK();
1115
0
}
1116
1117
Status BaseTablet::generate_new_block_for_flexible_partial_update(
1118
        TabletSchemaSPtr rowset_schema, const PartialUpdateInfo* partial_update_info,
1119
        std::set<uint32_t>& rids_be_overwritten, const FixedReadPlan& read_plan_ori,
1120
        const FixedReadPlan& read_plan_update,
1121
        const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
1122
0
        vectorized::Block* output_block) {
1123
0
    CHECK(output_block);
1124
1125
0
    int32_t seq_col_unique_id = -1;
1126
0
    if (rowset_schema->has_sequence_col()) {
1127
0
        seq_col_unique_id = rowset_schema->column(rowset_schema->sequence_col_idx()).unique_id();
1128
0
    }
1129
0
    const auto& non_sort_key_cids = partial_update_info->missing_cids;
1130
0
    std::vector<uint32_t> all_cids(rowset_schema->num_columns());
1131
0
    std::iota(all_cids.begin(), all_cids.end(), 0);
1132
0
    auto old_block = rowset_schema->create_block_by_cids(non_sort_key_cids);
1133
0
    auto update_block = rowset_schema->create_block_by_cids(all_cids);
1134
1135
    // rowid in the final block(start from 0, increase continuously) -> rowid to read in update_block
1136
0
    std::map<uint32_t, uint32_t> read_index_update;
1137
1138
    // 1. read the current rowset first, if a row in the current rowset has delete sign mark
1139
    // we don't need to read values from old block for that row
1140
0
    RETURN_IF_ERROR(read_plan_update.read_columns_by_plan(*rowset_schema, all_cids, rsid_to_rowset,
1141
0
                                                          update_block, &read_index_update, true));
1142
0
    size_t update_rows = read_index_update.size();
1143
1144
    // TODO(bobhan1): add the delete sign optimazation here
1145
    // // if there is sequence column in the table, we need to read the sequence column,
1146
    // // otherwise it may cause the merge-on-read based compaction policy to produce incorrect results
1147
    // const auto* __restrict new_block_delete_signs =
1148
    //         rowset_schema->has_sequence_col()
1149
    //                 ? nullptr
1150
    //                 : get_delete_sign_column_data(update_block, update_rows);
1151
1152
    // 2. read previous rowsets
1153
    // rowid in the final block(start from 0, increase, may not continuous becasue we skip to read some rows) -> rowid to read in old_block
1154
0
    std::map<uint32_t, uint32_t> read_index_old;
1155
0
    RETURN_IF_ERROR(read_plan_ori.read_columns_by_plan(
1156
0
            *rowset_schema, non_sort_key_cids, rsid_to_rowset, old_block, &read_index_old, true));
1157
0
    size_t old_rows = read_index_old.size();
1158
0
    DCHECK(update_rows == old_rows);
1159
0
    const auto* __restrict old_block_delete_signs =
1160
0
            get_delete_sign_column_data(old_block, old_rows);
1161
0
    DCHECK(old_block_delete_signs != nullptr);
1162
1163
    // 3. build default value block
1164
0
    auto default_value_block = old_block.clone_empty();
1165
0
    RETURN_IF_ERROR(BaseTablet::generate_default_value_block(*rowset_schema, non_sort_key_cids,
1166
0
                                                             partial_update_info->default_values,
1167
0
                                                             old_block, default_value_block));
1168
1169
    // 4. build the final block
1170
0
    auto full_mutable_columns = output_block->mutate_columns();
1171
0
    DCHECK(rowset_schema->has_skip_bitmap_col());
1172
0
    auto skip_bitmap_col_idx = rowset_schema->skip_bitmap_col_idx();
1173
0
    const std::vector<BitmapValue>* skip_bitmaps =
1174
0
            &(assert_cast<const vectorized::ColumnBitmap*, TypeCheckOnRelease::DISABLE>(
1175
0
                      update_block.get_by_position(skip_bitmap_col_idx).column->get_ptr().get())
1176
0
                      ->get_data());
1177
1178
0
    if (rowset_schema->has_sequence_col() && !rids_be_overwritten.empty()) {
1179
        // If the row specifies the sequence column, we should delete the current row becase the
1180
        // flexible partial update on the current row has been `overwritten` by the previous one with larger sequence
1181
        // column value.
1182
0
        for (auto it = rids_be_overwritten.begin(); it != rids_be_overwritten.end();) {
1183
0
            auto rid = *it;
1184
0
            if (!skip_bitmaps->at(rid).contains(seq_col_unique_id)) {
1185
0
                ++it;
1186
0
            } else {
1187
0
                it = rids_be_overwritten.erase(it);
1188
0
            }
1189
0
        }
1190
0
    }
1191
1192
0
    auto fill_one_cell = [&read_index_old, &read_index_update, &rowset_schema, partial_update_info](
1193
0
                                 const TabletColumn& tablet_column, std::size_t idx,
1194
0
                                 vectorized::MutableColumnPtr& new_col,
1195
0
                                 const vectorized::IColumn& default_value_col,
1196
0
                                 const vectorized::IColumn& old_value_col,
1197
0
                                 const vectorized::IColumn& cur_col, bool skipped,
1198
0
                                 bool row_has_sequence_col,
1199
0
                                 const signed char* delete_sign_column_data) {
1200
0
        if (skipped) {
1201
0
            bool use_default = false;
1202
0
            bool old_row_delete_sign =
1203
0
                    (delete_sign_column_data != nullptr &&
1204
0
                     delete_sign_column_data[read_index_old[cast_set<uint32_t>(idx)]] != 0);
1205
0
            if (old_row_delete_sign) {
1206
0
                if (!rowset_schema->has_sequence_col()) {
1207
0
                    use_default = true;
1208
0
                } else if (row_has_sequence_col ||
1209
0
                           (!tablet_column.is_seqeunce_col() &&
1210
0
                            (tablet_column.unique_id() !=
1211
0
                             partial_update_info->sequence_map_col_uid()))) {
1212
                    // to keep the sequence column value not decreasing, we should read values of seq column(and seq map column)
1213
                    // from old rows even if the old row is deleted when the input don't specify the sequence column, otherwise
1214
                    // it may cause the merge-on-read based compaction to produce incorrect results
1215
0
                    use_default = true;
1216
0
                }
1217
0
            }
1218
0
            if (use_default) {
1219
0
                if (tablet_column.has_default_value()) {
1220
0
                    new_col->insert_from(default_value_col, 0);
1221
0
                } else if (tablet_column.is_nullable()) {
1222
0
                    assert_cast<vectorized::ColumnNullable*, TypeCheckOnRelease::DISABLE>(
1223
0
                            new_col.get())
1224
0
                            ->insert_many_defaults(1);
1225
0
                } else if (tablet_column.is_auto_increment()) {
1226
                    // For auto-increment column, its default value(generated value) is filled in current block in flush phase
1227
                    // when the load doesn't specify the auto-increment column
1228
                    //     - if the previous conflicting row is deleted, we should use the value in current block as its final value
1229
                    //     - if the previous conflicting row is an insert, we should use the value in old block as its final value to
1230
                    //       keep consistency between replicas
1231
0
                    new_col->insert_from(cur_col, read_index_update[cast_set<uint32_t>(idx)]);
1232
0
                } else {
1233
0
                    new_col->insert(tablet_column.get_vec_type()->get_default());
1234
0
                }
1235
0
            } else {
1236
0
                new_col->insert_from(old_value_col, read_index_old[cast_set<uint32_t>(idx)]);
1237
0
            }
1238
0
        } else {
1239
0
            new_col->insert_from(cur_col, read_index_update[cast_set<uint32_t>(idx)]);
1240
0
        }
1241
0
    };
1242
1243
0
    for (std::size_t cid {0}; cid < rowset_schema->num_columns(); cid++) {
1244
0
        vectorized::MutableColumnPtr& new_col = full_mutable_columns[cid];
1245
0
        const vectorized::IColumn& cur_col = *update_block.get_by_position(cid).column;
1246
0
        const auto& rs_column = rowset_schema->column(cid);
1247
0
        auto col_uid = rs_column.unique_id();
1248
0
        for (auto idx = 0; idx < update_rows; ++idx) {
1249
0
            if (cid < rowset_schema->num_key_columns()) {
1250
0
                new_col->insert_from(cur_col, read_index_update[idx]);
1251
0
            } else {
1252
0
                const vectorized::IColumn& default_value_col =
1253
0
                        *default_value_block.get_by_position(cid - rowset_schema->num_key_columns())
1254
0
                                 .column;
1255
0
                const vectorized::IColumn& old_value_col =
1256
0
                        *old_block.get_by_position(cid - rowset_schema->num_key_columns()).column;
1257
0
                if (rids_be_overwritten.contains(idx)) {
1258
0
                    new_col->insert_from(old_value_col, read_index_old[idx]);
1259
0
                } else {
1260
0
                    fill_one_cell(rs_column, idx, new_col, default_value_col, old_value_col,
1261
0
                                  cur_col, skip_bitmaps->at(idx).contains(col_uid),
1262
0
                                  rowset_schema->has_sequence_col()
1263
0
                                          ? !skip_bitmaps->at(idx).contains(seq_col_unique_id)
1264
0
                                          : false,
1265
0
                                  old_block_delete_signs);
1266
0
                }
1267
0
            }
1268
0
        }
1269
0
        DCHECK_EQ(full_mutable_columns[cid]->size(), update_rows);
1270
0
    }
1271
1272
0
    output_block->set_columns(std::move(full_mutable_columns));
1273
0
    VLOG_DEBUG << "full block when publish: " << output_block->dump_data();
1274
0
    return Status::OK();
1275
0
}
1276
1277
Status BaseTablet::commit_phase_update_delete_bitmap(
1278
        const BaseTabletSPtr& tablet, const RowsetSharedPtr& rowset,
1279
        RowsetIdUnorderedSet& pre_rowset_ids, DeleteBitmapPtr delete_bitmap,
1280
        const std::vector<segment_v2::SegmentSharedPtr>& segments, int64_t txn_id,
1281
3
        CalcDeleteBitmapToken* token, RowsetWriter* rowset_writer) {
1282
3
    DBUG_EXECUTE_IF("BaseTablet::commit_phase_update_delete_bitmap.enable_spin_wait", {
1283
3
        auto tok = dp->param<std::string>("token", "invalid_token");
1284
3
        while (DebugPoints::instance()->is_enable(
1285
3
                "BaseTablet::commit_phase_update_delete_bitmap.block")) {
1286
3
            auto block_dp = DebugPoints::instance()->get_debug_point(
1287
3
                    "BaseTablet::commit_phase_update_delete_bitmap.block");
1288
3
            if (block_dp) {
1289
3
                auto pass_token = block_dp->param<std::string>("pass_token", "");
1290
3
                if (pass_token == tok) {
1291
3
                    break;
1292
3
                }
1293
3
            }
1294
3
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
1295
3
        }
1296
3
    });
1297
3
    SCOPED_BVAR_LATENCY(g_tablet_commit_phase_update_delete_bitmap_latency);
1298
3
    RowsetIdUnorderedSet cur_rowset_ids;
1299
3
    RowsetIdUnorderedSet rowset_ids_to_add;
1300
3
    RowsetIdUnorderedSet rowset_ids_to_del;
1301
3
    int64_t cur_version;
1302
1303
3
    std::vector<RowsetSharedPtr> specified_rowsets;
1304
3
    {
1305
        // to prevent seeing intermediate state of a tablet
1306
3
        std::unique_lock<bthread::Mutex> sync_lock;
1307
3
        if (config::is_cloud_mode()) {
1308
0
            sync_lock = std::unique_lock<bthread::Mutex>(
1309
0
                    std::static_pointer_cast<CloudTablet>(tablet)->get_sync_meta_lock());
1310
0
        }
1311
3
        std::shared_lock meta_rlock(tablet->_meta_lock);
1312
3
        if (tablet->tablet_state() == TABLET_NOTREADY) {
1313
            // tablet is under alter process. The delete bitmap will be calculated after conversion.
1314
0
            LOG(INFO) << "tablet is under alter process, delete bitmap will be calculated later, "
1315
0
                         "tablet_id: "
1316
0
                      << tablet->tablet_id() << " txn_id: " << txn_id;
1317
0
            return Status::OK();
1318
0
        }
1319
3
        cur_version = tablet->max_version_unlocked();
1320
3
        RETURN_IF_ERROR(tablet->get_all_rs_id_unlocked(cur_version, &cur_rowset_ids));
1321
3
        _rowset_ids_difference(cur_rowset_ids, pre_rowset_ids, &rowset_ids_to_add,
1322
3
                               &rowset_ids_to_del);
1323
3
        specified_rowsets = tablet->get_rowset_by_ids(&rowset_ids_to_add);
1324
3
    }
1325
0
    for (const auto& to_del : rowset_ids_to_del) {
1326
0
        delete_bitmap->remove({to_del, 0, 0}, {to_del, UINT32_MAX, INT64_MAX});
1327
0
    }
1328
1329
3
    RETURN_IF_ERROR(calc_delete_bitmap(tablet, rowset, segments, specified_rowsets, delete_bitmap,
1330
3
                                       cur_version, token, rowset_writer));
1331
3
    size_t total_rows = std::accumulate(
1332
3
            segments.begin(), segments.end(), 0,
1333
3
            [](size_t sum, const segment_v2::SegmentSharedPtr& s) { return sum += s->num_rows(); });
1334
3
    LOG(INFO) << "[Before Commit] construct delete bitmap tablet: " << tablet->tablet_id()
1335
3
              << ", rowset_ids to add: " << rowset_ids_to_add.size()
1336
3
              << ", rowset_ids to del: " << rowset_ids_to_del.size()
1337
3
              << ", cur max_version: " << cur_version << ", transaction_id: " << txn_id
1338
3
              << ", total rows: " << total_rows;
1339
3
    pre_rowset_ids = cur_rowset_ids;
1340
3
    return Status::OK();
1341
3
}
1342
1343
void BaseTablet::add_sentinel_mark_to_delete_bitmap(DeleteBitmap* delete_bitmap,
1344
5
                                                    const RowsetIdUnorderedSet& rowsetids) {
1345
5
    for (const auto& rowsetid : rowsetids) {
1346
5
        delete_bitmap->add(
1347
5
                {rowsetid, DeleteBitmap::INVALID_SEGMENT_ID, DeleteBitmap::TEMP_VERSION_COMMON},
1348
5
                DeleteBitmap::ROWSET_SENTINEL_MARK);
1349
5
    }
1350
5
}
1351
1352
void BaseTablet::_rowset_ids_difference(const RowsetIdUnorderedSet& cur,
1353
                                        const RowsetIdUnorderedSet& pre,
1354
                                        RowsetIdUnorderedSet* to_add,
1355
6
                                        RowsetIdUnorderedSet* to_del) {
1356
6
    for (const auto& id : cur) {
1357
2
        if (pre.find(id) == pre.end()) {
1358
1
            to_add->insert(id);
1359
1
        }
1360
2
    }
1361
6
    for (const auto& id : pre) {
1362
1
        if (cur.find(id) == cur.end()) {
1363
0
            to_del->insert(id);
1364
0
        }
1365
1
    }
1366
6
}
1367
1368
Status BaseTablet::_capture_consistent_rowsets_unlocked(
1369
4
        const std::vector<Version>& version_path, std::vector<RowsetSharedPtr>* rowsets) const {
1370
4
    DCHECK(rowsets != nullptr);
1371
4
    rowsets->reserve(version_path.size());
1372
8
    for (const auto& version : version_path) {
1373
8
        bool is_find = false;
1374
8
        do {
1375
8
            auto it = _rs_version_map.find(version);
1376
8
            if (it != _rs_version_map.end()) {
1377
8
                is_find = true;
1378
8
                rowsets->push_back(it->second);
1379
8
                break;
1380
8
            }
1381
1382
0
            auto it_expired = _stale_rs_version_map.find(version);
1383
0
            if (it_expired != _stale_rs_version_map.end()) {
1384
0
                is_find = true;
1385
0
                rowsets->push_back(it_expired->second);
1386
0
                break;
1387
0
            }
1388
0
        } while (false);
1389
1390
8
        if (!is_find) {
1391
0
            return Status::Error<CAPTURE_ROWSET_ERROR>(
1392
0
                    "fail to find Rowset for version. tablet={}, version={}", tablet_id(),
1393
0
                    version.to_string());
1394
0
        }
1395
8
    }
1396
4
    return Status::OK();
1397
4
}
1398
1399
Status BaseTablet::check_delete_bitmap_correctness(DeleteBitmapPtr delete_bitmap,
1400
                                                   int64_t max_version, int64_t txn_id,
1401
                                                   const RowsetIdUnorderedSet& rowset_ids,
1402
6
                                                   std::vector<RowsetSharedPtr>* rowsets) {
1403
6
    RowsetIdUnorderedSet missing_ids;
1404
6
    for (const auto& rowsetid : rowset_ids) {
1405
2
        if (!delete_bitmap->delete_bitmap.contains({rowsetid, DeleteBitmap::INVALID_SEGMENT_ID,
1406
2
                                                    DeleteBitmap::TEMP_VERSION_COMMON})) {
1407
0
            missing_ids.insert(rowsetid);
1408
0
        }
1409
2
    }
1410
1411
6
    if (!missing_ids.empty()) {
1412
0
        LOG(WARNING) << "[txn_id:" << txn_id << "][tablet_id:" << tablet_id()
1413
0
                     << "][max_version: " << max_version
1414
0
                     << "] check delete bitmap correctness failed!";
1415
0
        rapidjson::Document root;
1416
0
        root.SetObject();
1417
0
        rapidjson::Document required_rowsets_arr;
1418
0
        required_rowsets_arr.SetArray();
1419
0
        rapidjson::Document missing_rowsets_arr;
1420
0
        missing_rowsets_arr.SetArray();
1421
1422
0
        if (rowsets != nullptr) {
1423
0
            for (const auto& rowset : *rowsets) {
1424
0
                rapidjson::Value value;
1425
0
                std::string version_str = rowset->get_rowset_info_str();
1426
0
                value.SetString(version_str.c_str(),
1427
0
                                cast_set<rapidjson::SizeType>(version_str.length()),
1428
0
                                required_rowsets_arr.GetAllocator());
1429
0
                required_rowsets_arr.PushBack(value, required_rowsets_arr.GetAllocator());
1430
0
            }
1431
0
        } else {
1432
0
            std::vector<RowsetSharedPtr> tablet_rowsets;
1433
0
            {
1434
0
                std::shared_lock meta_rlock(_meta_lock);
1435
0
                tablet_rowsets = get_rowset_by_ids(&rowset_ids);
1436
0
            }
1437
0
            for (const auto& rowset : tablet_rowsets) {
1438
0
                rapidjson::Value value;
1439
0
                std::string version_str = rowset->get_rowset_info_str();
1440
0
                value.SetString(version_str.c_str(),
1441
0
                                cast_set<rapidjson::SizeType>(version_str.length()),
1442
0
                                required_rowsets_arr.GetAllocator());
1443
0
                required_rowsets_arr.PushBack(value, required_rowsets_arr.GetAllocator());
1444
0
            }
1445
0
        }
1446
0
        for (const auto& missing_rowset_id : missing_ids) {
1447
0
            rapidjson::Value miss_value;
1448
0
            std::string rowset_id_str = missing_rowset_id.to_string();
1449
0
            miss_value.SetString(rowset_id_str.c_str(),
1450
0
                                 cast_set<rapidjson::SizeType>(rowset_id_str.length()),
1451
0
                                 missing_rowsets_arr.GetAllocator());
1452
0
            missing_rowsets_arr.PushBack(miss_value, missing_rowsets_arr.GetAllocator());
1453
0
        }
1454
1455
0
        root.AddMember("required_rowsets", required_rowsets_arr, root.GetAllocator());
1456
0
        root.AddMember("missing_rowsets", missing_rowsets_arr, root.GetAllocator());
1457
0
        rapidjson::StringBuffer strbuf;
1458
0
        rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(strbuf);
1459
0
        root.Accept(writer);
1460
0
        std::string rowset_status_string = std::string(strbuf.GetString());
1461
0
        LOG_EVERY_SECOND(WARNING) << rowset_status_string;
1462
        // let it crash if correctness check failed in Debug mode
1463
0
        DCHECK(false) << "delete bitmap correctness check failed in publish phase!";
1464
0
        return Status::InternalError("check delete bitmap failed!");
1465
0
    }
1466
6
    return Status::OK();
1467
6
}
1468
1469
Status BaseTablet::update_delete_bitmap(const BaseTabletSPtr& self, TabletTxnInfo* txn_info,
1470
                                        int64_t txn_id, int64_t txn_expiration,
1471
3
                                        DeleteBitmapPtr tablet_delete_bitmap) {
1472
3
    SCOPED_BVAR_LATENCY(g_tablet_update_delete_bitmap_latency);
1473
3
    RowsetIdUnorderedSet cur_rowset_ids;
1474
3
    RowsetIdUnorderedSet rowset_ids_to_add;
1475
3
    RowsetIdUnorderedSet rowset_ids_to_del;
1476
3
    RowsetSharedPtr rowset = txn_info->rowset;
1477
3
    int64_t cur_version = rowset->start_version();
1478
1479
3
    std::unique_ptr<RowsetWriter> transient_rs_writer;
1480
3
    DeleteBitmapPtr delete_bitmap = txn_info->delete_bitmap;
1481
3
    bool is_partial_update =
1482
3
            txn_info->partial_update_info && txn_info->partial_update_info->is_partial_update();
1483
3
    if (is_partial_update) {
1484
0
        transient_rs_writer = DORIS_TRY(self->create_transient_rowset_writer(
1485
0
                *rowset, txn_info->partial_update_info, txn_expiration));
1486
0
        DBUG_EXECUTE_IF("BaseTablet::update_delete_bitmap.after.create_transient_rs_writer",
1487
0
                        DBUG_BLOCK);
1488
        // Partial update might generate new segments when there is conflicts while publish, and mark
1489
        // the same key in original segments as delete.
1490
        // When the new segment flush fails or the rowset build fails, the deletion marker for the
1491
        // duplicate key of the original segment should not remain in `txn_info->delete_bitmap`,
1492
        // so we need to make a copy of `txn_info->delete_bitmap` and make changes on it.
1493
0
        delete_bitmap = std::make_shared<DeleteBitmap>(*(txn_info->delete_bitmap));
1494
0
    }
1495
1496
3
    OlapStopWatch watch;
1497
3
    std::vector<segment_v2::SegmentSharedPtr> segments;
1498
3
    RETURN_IF_ERROR(std::dynamic_pointer_cast<BetaRowset>(rowset)->load_segments(&segments));
1499
3
    auto t1 = watch.get_elapse_time_us();
1500
1501
3
    int64_t next_visible_version = txn_info->is_txn_load ? txn_info->next_visible_version
1502
3
                                                         : txn_info->rowset->start_version();
1503
3
    {
1504
3
        std::shared_lock meta_rlock(self->_meta_lock);
1505
        // tablet is under alter process. The delete bitmap will be calculated after conversion.
1506
3
        if (self->tablet_state() == TABLET_NOTREADY) {
1507
0
            LOG(INFO) << "tablet is under alter process, update delete bitmap later, tablet_id="
1508
0
                      << self->tablet_id();
1509
0
            return Status::OK();
1510
0
        }
1511
3
        RETURN_IF_ERROR(self->get_all_rs_id_unlocked(next_visible_version - 1, &cur_rowset_ids));
1512
3
    }
1513
3
    auto t2 = watch.get_elapse_time_us();
1514
1515
3
    _rowset_ids_difference(cur_rowset_ids, txn_info->rowset_ids, &rowset_ids_to_add,
1516
3
                           &rowset_ids_to_del);
1517
3
    for (const auto& to_del : rowset_ids_to_del) {
1518
0
        delete_bitmap->remove({to_del, 0, 0}, {to_del, UINT32_MAX, INT64_MAX});
1519
0
    }
1520
1521
3
    std::vector<RowsetSharedPtr> specified_rowsets;
1522
3
    {
1523
3
        std::shared_lock meta_rlock(self->_meta_lock);
1524
3
        specified_rowsets = self->get_rowset_by_ids(&rowset_ids_to_add);
1525
3
    }
1526
3
    if (txn_info->is_txn_load) {
1527
0
        for (auto invisible_rowset : txn_info->invisible_rowsets) {
1528
0
            specified_rowsets.emplace_back(invisible_rowset);
1529
0
        }
1530
0
        std::sort(specified_rowsets.begin(), specified_rowsets.end(),
1531
0
                  [](RowsetSharedPtr& lhs, RowsetSharedPtr& rhs) {
1532
0
                      return lhs->end_version() > rhs->end_version();
1533
0
                  });
1534
0
    }
1535
3
    auto t3 = watch.get_elapse_time_us();
1536
1537
    // If a rowset is produced by compaction before the commit phase of the partial update load
1538
    // and is not included in txn_info->rowset_ids, we can skip the alignment process of that rowset
1539
    // because data remains the same before and after compaction. But we still need to calculate the
1540
    // the delete bitmap for that rowset.
1541
3
    std::vector<RowsetSharedPtr> rowsets_skip_alignment;
1542
3
    if (is_partial_update) {
1543
0
        int64_t max_version_in_flush_phase =
1544
0
                txn_info->partial_update_info->max_version_in_flush_phase;
1545
0
        DCHECK(max_version_in_flush_phase != -1);
1546
0
        std::vector<RowsetSharedPtr> remained_rowsets;
1547
0
        for (const auto& specified_rowset : specified_rowsets) {
1548
0
            if (specified_rowset->end_version() <= max_version_in_flush_phase &&
1549
0
                specified_rowset->produced_by_compaction()) {
1550
0
                rowsets_skip_alignment.emplace_back(specified_rowset);
1551
0
            } else {
1552
0
                remained_rowsets.emplace_back(specified_rowset);
1553
0
            }
1554
0
        }
1555
0
        if (!rowsets_skip_alignment.empty()) {
1556
0
            specified_rowsets = std::move(remained_rowsets);
1557
0
        }
1558
0
    }
1559
1560
3
    DBUG_EXECUTE_IF("BaseTablet::update_delete_bitmap.enable_spin_wait", {
1561
3
        auto token = dp->param<std::string>("token", "invalid_token");
1562
3
        while (DebugPoints::instance()->is_enable("BaseTablet::update_delete_bitmap.block")) {
1563
3
            auto block_dp = DebugPoints::instance()->get_debug_point(
1564
3
                    "BaseTablet::update_delete_bitmap.block");
1565
3
            if (block_dp) {
1566
3
                auto wait_token = block_dp->param<std::string>("wait_token", "");
1567
3
                if (wait_token != token) {
1568
3
                    break;
1569
3
                }
1570
3
            }
1571
3
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
1572
3
        }
1573
3
    });
1574
1575
3
    if (!rowsets_skip_alignment.empty()) {
1576
0
        auto token = self->calc_delete_bitmap_executor()->create_token();
1577
        // set rowset_writer to nullptr to skip the alignment process
1578
0
        RETURN_IF_ERROR(calc_delete_bitmap(self, rowset, segments, rowsets_skip_alignment,
1579
0
                                           delete_bitmap, cur_version - 1, token.get(), nullptr,
1580
0
                                           tablet_delete_bitmap));
1581
0
        RETURN_IF_ERROR(token->wait());
1582
0
    }
1583
1584
    // When there is only one segment, it will be calculated in the current thread.
1585
    // Otherwise, it will be submitted to the thread pool for calculation.
1586
3
    if (segments.size() <= 1) {
1587
3
        RETURN_IF_ERROR(calc_delete_bitmap(self, rowset, segments, specified_rowsets, delete_bitmap,
1588
3
                                           cur_version - 1, nullptr, transient_rs_writer.get(),
1589
3
                                           tablet_delete_bitmap));
1590
1591
3
    } else {
1592
0
        auto token = self->calc_delete_bitmap_executor()->create_token();
1593
0
        RETURN_IF_ERROR(calc_delete_bitmap(self, rowset, segments, specified_rowsets, delete_bitmap,
1594
0
                                           cur_version - 1, token.get(), transient_rs_writer.get(),
1595
0
                                           tablet_delete_bitmap));
1596
0
        RETURN_IF_ERROR(token->wait());
1597
0
    }
1598
1599
3
    std::stringstream ss;
1600
3
    ss << "cost(us): (load segments: " << t1 << ", get all rsid: " << t2 - t1
1601
3
       << ", get rowsets: " << t3 - t2
1602
3
       << ", calc delete bitmap: " << watch.get_elapse_time_us() - t3 << ")";
1603
1604
3
    if (config::enable_merge_on_write_correctness_check && rowset->num_rows() != 0) {
1605
        // only do correctness check if the rowset has at least one row written
1606
        // check if all the rowset has ROWSET_SENTINEL_MARK
1607
3
        auto st = self->check_delete_bitmap_correctness(delete_bitmap, cur_version - 1, -1,
1608
3
                                                        cur_rowset_ids, &specified_rowsets);
1609
3
        if (!st.ok()) {
1610
0
            LOG(WARNING) << fmt::format("delete bitmap correctness check failed in publish phase!");
1611
0
        }
1612
3
    }
1613
1614
3
    if (transient_rs_writer) {
1615
0
        auto t4 = watch.get_elapse_time_us();
1616
0
        DBUG_EXECUTE_IF("Tablet.update_delete_bitmap.partial_update_write_rowset_fail", {
1617
0
            if (rand() % 100 < (100 * dp->param("percent", 0.5))) {
1618
0
                LOG_WARNING("Tablet.update_delete_bitmap.partial_update_write_rowset random failed")
1619
0
                        .tag("txn_id", txn_id);
1620
0
                return Status::InternalError(
1621
0
                        "debug update_delete_bitmap partial update write rowset random failed");
1622
0
            }
1623
0
        });
1624
        // build rowset writer and merge transient rowset
1625
0
        RETURN_IF_ERROR(transient_rs_writer->flush());
1626
0
        RowsetSharedPtr transient_rowset;
1627
0
        RETURN_IF_ERROR(transient_rs_writer->build(transient_rowset));
1628
0
        auto old_segments = rowset->num_segments();
1629
0
        rowset->merge_rowset_meta(*transient_rowset->rowset_meta());
1630
0
        auto new_segments = rowset->num_segments();
1631
0
        ss << ", " << txn_info->partial_update_info->partial_update_mode_str()
1632
0
           << " flush rowset (old segment num: " << old_segments
1633
0
           << ", new segment num: " << new_segments << ")"
1634
0
           << ", cost:" << watch.get_elapse_time_us() - t4 << "(us)";
1635
1636
        // update the shared_ptr to new bitmap, which is consistent with current rowset.
1637
0
        txn_info->delete_bitmap = delete_bitmap;
1638
        // erase segment cache cause we will add a segment to rowset
1639
0
        SegmentLoader::instance()->erase_segments(rowset->rowset_id(), rowset->num_segments());
1640
0
    }
1641
1642
3
    size_t total_rows = std::accumulate(
1643
3
            segments.begin(), segments.end(), 0,
1644
3
            [](size_t sum, const segment_v2::SegmentSharedPtr& s) { return sum += s->num_rows(); });
1645
3
    auto t5 = watch.get_elapse_time_us();
1646
3
    int64_t lock_id = txn_info->is_txn_load ? txn_info->lock_id : -1;
1647
3
    RETURN_IF_ERROR(self->save_delete_bitmap(txn_info, txn_id, delete_bitmap,
1648
3
                                             transient_rs_writer.get(), cur_rowset_ids, lock_id,
1649
3
                                             next_visible_version));
1650
1651
    // defensive check, check that the delete bitmap cache we wrote is correct
1652
3
    RETURN_IF_ERROR(self->check_delete_bitmap_cache(txn_id, delete_bitmap.get()));
1653
1654
3
    LOG(INFO) << "[Publish] construct delete bitmap tablet: " << self->tablet_id()
1655
3
              << ", rowset_ids to add: "
1656
3
              << (specified_rowsets.size() + rowsets_skip_alignment.size())
1657
3
              << ", rowset_ids to del: " << rowset_ids_to_del.size()
1658
3
              << ", cur version: " << cur_version << ", transaction_id: " << txn_id << ","
1659
3
              << ss.str() << " , total rows: " << total_rows
1660
3
              << ", update delete_bitmap cost: " << watch.get_elapse_time_us() - t5 << "(us)";
1661
3
    return Status::OK();
1662
3
}
1663
1664
void BaseTablet::calc_compaction_output_rowset_delete_bitmap(
1665
        const std::vector<RowsetSharedPtr>& input_rowsets, const RowIdConversion& rowid_conversion,
1666
        uint64_t start_version, uint64_t end_version, std::set<RowLocation>* missed_rows,
1667
        std::map<RowsetSharedPtr, std::list<std::pair<RowLocation, RowLocation>>>* location_map,
1668
0
        const DeleteBitmap& input_delete_bitmap, DeleteBitmap* output_rowset_delete_bitmap) {
1669
0
    RowLocation src;
1670
0
    RowLocation dst;
1671
0
    for (auto& rowset : input_rowsets) {
1672
0
        src.rowset_id = rowset->rowset_id();
1673
0
        for (uint32_t seg_id = 0; seg_id < rowset->num_segments(); ++seg_id) {
1674
0
            src.segment_id = seg_id;
1675
0
            DeleteBitmap subset_map(tablet_id());
1676
0
            input_delete_bitmap.subset({rowset->rowset_id(), seg_id, start_version},
1677
0
                                       {rowset->rowset_id(), seg_id, end_version}, &subset_map);
1678
            // traverse all versions and convert rowid
1679
0
            for (auto iter = subset_map.delete_bitmap.begin();
1680
0
                 iter != subset_map.delete_bitmap.end(); ++iter) {
1681
0
                auto cur_version = std::get<2>(iter->first);
1682
0
                for (auto index = iter->second.begin(); index != iter->second.end(); ++index) {
1683
0
                    src.row_id = *index;
1684
0
                    if (rowid_conversion.get(src, &dst) != 0) {
1685
0
                        VLOG_CRITICAL << "Can't find rowid, may be deleted by the delete_handler, "
1686
0
                                      << " src loaction: |" << src.rowset_id << "|"
1687
0
                                      << src.segment_id << "|" << src.row_id
1688
0
                                      << " version: " << cur_version;
1689
0
                        if (missed_rows) {
1690
0
                            missed_rows->insert(src);
1691
0
                        }
1692
0
                        continue;
1693
0
                    }
1694
0
                    VLOG_DEBUG << "calc_compaction_output_rowset_delete_bitmap dst location: |"
1695
0
                               << dst.rowset_id << "|" << dst.segment_id << "|" << dst.row_id
1696
0
                               << " src location: |" << src.rowset_id << "|" << src.segment_id
1697
0
                               << "|" << src.row_id << " start version: " << start_version
1698
0
                               << "end version" << end_version;
1699
0
                    if (location_map) {
1700
0
                        (*location_map)[rowset].emplace_back(src, dst);
1701
0
                    }
1702
0
                    output_rowset_delete_bitmap->add({dst.rowset_id, dst.segment_id, cur_version},
1703
0
                                                     dst.row_id);
1704
0
                }
1705
0
            }
1706
0
        }
1707
0
    }
1708
0
}
1709
1710
Status BaseTablet::check_rowid_conversion(
1711
        RowsetSharedPtr dst_rowset,
1712
        const std::map<RowsetSharedPtr, std::list<std::pair<RowLocation, RowLocation>>>&
1713
0
                location_map) {
1714
0
    if (location_map.empty()) {
1715
0
        VLOG_DEBUG << "check_rowid_conversion, location_map is empty";
1716
0
        return Status::OK();
1717
0
    }
1718
0
    std::vector<segment_v2::SegmentSharedPtr> dst_segments;
1719
1720
0
    RETURN_IF_ERROR(
1721
0
            std::dynamic_pointer_cast<BetaRowset>(dst_rowset)->load_segments(&dst_segments));
1722
0
    std::unordered_map<RowsetId, std::vector<segment_v2::SegmentSharedPtr>> input_rowsets_segment;
1723
1724
0
    VLOG_DEBUG << "check_rowid_conversion, dst_segments size: " << dst_segments.size();
1725
0
    for (auto [src_rowset, locations] : location_map) {
1726
0
        std::vector<segment_v2::SegmentSharedPtr>& segments =
1727
0
                input_rowsets_segment[src_rowset->rowset_id()];
1728
0
        if (segments.empty()) {
1729
0
            RETURN_IF_ERROR(
1730
0
                    std::dynamic_pointer_cast<BetaRowset>(src_rowset)->load_segments(&segments));
1731
0
        }
1732
0
        for (auto& [src, dst] : locations) {
1733
0
            std::string src_key;
1734
0
            std::string dst_key;
1735
0
            Status s = segments[src.segment_id]->read_key_by_rowid(src.row_id, &src_key);
1736
0
            if (UNLIKELY(s.is<NOT_IMPLEMENTED_ERROR>())) {
1737
0
                LOG(INFO) << "primary key index of old version does not "
1738
0
                             "support reading key by rowid";
1739
0
                break;
1740
0
            }
1741
0
            if (UNLIKELY(!s)) {
1742
0
                LOG(WARNING) << "failed to get src key: |" << src.rowset_id << "|" << src.segment_id
1743
0
                             << "|" << src.row_id << " status: " << s;
1744
0
                DCHECK(false);
1745
0
                return s;
1746
0
            }
1747
1748
0
            s = dst_segments[dst.segment_id]->read_key_by_rowid(dst.row_id, &dst_key);
1749
0
            if (UNLIKELY(!s)) {
1750
0
                LOG(WARNING) << "failed to get dst key: |" << dst.rowset_id << "|" << dst.segment_id
1751
0
                             << "|" << dst.row_id << " status: " << s;
1752
0
                DCHECK(false);
1753
0
                return s;
1754
0
            }
1755
1756
0
            VLOG_DEBUG << "check_rowid_conversion, src: |" << src.rowset_id << "|" << src.segment_id
1757
0
                       << "|" << src.row_id << "|" << src_key << " dst: |" << dst.rowset_id << "|"
1758
0
                       << dst.segment_id << "|" << dst.row_id << "|" << dst_key;
1759
0
            if (UNLIKELY(src_key.compare(dst_key) != 0)) {
1760
0
                LOG(WARNING) << "failed to check key, src key: |" << src.rowset_id << "|"
1761
0
                             << src.segment_id << "|" << src.row_id << "|" << src_key
1762
0
                             << " dst key: |" << dst.rowset_id << "|" << dst.segment_id << "|"
1763
0
                             << dst.row_id << "|" << dst_key;
1764
0
                DCHECK(false);
1765
0
                return Status::InternalError("failed to check rowid conversion");
1766
0
            }
1767
0
        }
1768
0
    }
1769
0
    return Status::OK();
1770
0
}
1771
1772
// The caller should hold _rowset_update_lock and _meta_lock lock.
1773
Status BaseTablet::update_delete_bitmap_without_lock(
1774
        const BaseTabletSPtr& self, const RowsetSharedPtr& rowset,
1775
0
        const std::vector<RowsetSharedPtr>* specified_base_rowsets) {
1776
0
    DBUG_EXECUTE_IF("BaseTablet.update_delete_bitmap_without_lock.random_failed", {
1777
0
        auto rnd = rand() % 100;
1778
0
        auto percent = dp->param("percent", 0.1);
1779
0
        if (rnd < (100 * percent)) {
1780
0
            LOG(WARNING) << "BaseTablet.update_delete_bitmap_without_lock.random_failed";
1781
0
            return Status::InternalError(
1782
0
                    "debug tablet update delete bitmap without lock random failed");
1783
0
        } else {
1784
0
            LOG(INFO) << "BaseTablet.update_delete_bitmap_without_lock.random_failed not "
1785
0
                         "triggered"
1786
0
                      << ", rnd:" << rnd << ", percent: " << percent;
1787
0
        }
1788
0
    });
1789
0
    int64_t cur_version = rowset->start_version();
1790
0
    std::vector<segment_v2::SegmentSharedPtr> segments;
1791
0
    RETURN_IF_ERROR(std::dynamic_pointer_cast<BetaRowset>(rowset)->load_segments(&segments));
1792
1793
    // If this rowset does not have a segment, there is no need for an update.
1794
0
    if (segments.empty()) {
1795
0
        LOG(INFO) << "[Schema Change or Clone] skip to construct delete bitmap tablet: "
1796
0
                  << self->tablet_id() << " cur max_version: " << cur_version;
1797
0
        return Status::OK();
1798
0
    }
1799
1800
    // calculate delete bitmap between segments if necessary.
1801
0
    DeleteBitmapPtr delete_bitmap = std::make_shared<DeleteBitmap>(self->tablet_id());
1802
0
    RETURN_IF_ERROR(self->calc_delete_bitmap_between_segments(rowset->rowset_id(), segments,
1803
0
                                                              delete_bitmap));
1804
1805
    // get all base rowsets to calculate on
1806
0
    std::vector<RowsetSharedPtr> specified_rowsets;
1807
0
    RowsetIdUnorderedSet cur_rowset_ids;
1808
0
    if (specified_base_rowsets == nullptr) {
1809
0
        RETURN_IF_ERROR(self->get_all_rs_id_unlocked(cur_version - 1, &cur_rowset_ids));
1810
0
        specified_rowsets = self->get_rowset_by_ids(&cur_rowset_ids);
1811
0
    } else {
1812
0
        specified_rowsets = *specified_base_rowsets;
1813
0
    }
1814
1815
0
    OlapStopWatch watch;
1816
0
    auto token = self->calc_delete_bitmap_executor()->create_token();
1817
0
    RETURN_IF_ERROR(calc_delete_bitmap(self, rowset, segments, specified_rowsets, delete_bitmap,
1818
0
                                       cur_version - 1, token.get()));
1819
0
    RETURN_IF_ERROR(token->wait());
1820
0
    size_t total_rows = std::accumulate(
1821
0
            segments.begin(), segments.end(), 0,
1822
0
            [](size_t sum, const segment_v2::SegmentSharedPtr& s) { return sum += s->num_rows(); });
1823
0
    LOG(INFO) << "[Schema Change or Clone] construct delete bitmap tablet: " << self->tablet_id()
1824
0
              << ", rowset_ids: " << cur_rowset_ids.size() << ", cur max_version: " << cur_version
1825
0
              << ", transaction_id: " << -1 << ", cost: " << watch.get_elapse_time_us()
1826
0
              << "(us), total rows: " << total_rows;
1827
0
    if (config::enable_merge_on_write_correctness_check) {
1828
        // check if all the rowset has ROWSET_SENTINEL_MARK
1829
0
        auto st = self->check_delete_bitmap_correctness(delete_bitmap, cur_version - 1, -1,
1830
0
                                                        cur_rowset_ids, &specified_rowsets);
1831
0
        if (!st.ok()) {
1832
0
            LOG(WARNING) << fmt::format("delete bitmap correctness check failed in publish phase!");
1833
0
        }
1834
0
        delete_bitmap->remove_sentinel_marks();
1835
0
    }
1836
0
    for (auto& iter : delete_bitmap->delete_bitmap) {
1837
0
        self->_tablet_meta->delete_bitmap().merge(
1838
0
                {std::get<0>(iter.first), std::get<1>(iter.first), cur_version}, iter.second);
1839
0
    }
1840
1841
0
    return Status::OK();
1842
0
}
1843
1844
void BaseTablet::agg_delete_bitmap_for_stale_rowsets(
1845
0
        Version version, DeleteBitmapKeyRanges& remove_delete_bitmap_key_ranges) {
1846
0
    if (!config::enable_agg_and_remove_pre_rowsets_delete_bitmap) {
1847
0
        return;
1848
0
    }
1849
0
    if (!(keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write())) {
1850
0
        return;
1851
0
    }
1852
0
    int64_t start_version = version.first;
1853
0
    int64_t end_version = version.second;
1854
0
    if (start_version == end_version) {
1855
0
        return;
1856
0
    }
1857
0
    DCHECK(start_version < end_version)
1858
0
            << ". start_version: " << start_version << ", end_version: " << end_version;
1859
    // get pre rowsets
1860
0
    std::vector<RowsetSharedPtr> pre_rowsets {};
1861
0
    {
1862
0
        std::shared_lock rdlock(_meta_lock);
1863
0
        for (const auto& it2 : _rs_version_map) {
1864
0
            if (it2.first.second < start_version) {
1865
0
                pre_rowsets.emplace_back(it2.second);
1866
0
            }
1867
0
        }
1868
0
    }
1869
0
    std::sort(pre_rowsets.begin(), pre_rowsets.end(), Rowset::comparator);
1870
    // do agg for pre rowsets
1871
0
    DeleteBitmapPtr new_delete_bitmap = std::make_shared<DeleteBitmap>(tablet_id());
1872
0
    for (auto& rowset : pre_rowsets) {
1873
0
        for (uint32_t seg_id = 0; seg_id < rowset->num_segments(); ++seg_id) {
1874
0
            auto d = tablet_meta()->delete_bitmap().get_agg_without_cache(
1875
0
                    {rowset->rowset_id(), seg_id, end_version}, start_version);
1876
0
            if (d->isEmpty()) {
1877
0
                continue;
1878
0
            }
1879
0
            VLOG_DEBUG << "agg delete bitmap for tablet_id=" << tablet_id()
1880
0
                       << ", rowset_id=" << rowset->rowset_id() << ", seg_id=" << seg_id
1881
0
                       << ", rowset_version=" << rowset->version().to_string()
1882
0
                       << ". compaction start_version=" << start_version
1883
0
                       << ", end_version=" << end_version << ", delete_bitmap=" << d->cardinality();
1884
0
            DeleteBitmap::BitmapKey start_key {rowset->rowset_id(), seg_id, start_version};
1885
0
            DeleteBitmap::BitmapKey end_key {rowset->rowset_id(), seg_id, end_version};
1886
0
            new_delete_bitmap->set(end_key, *d);
1887
0
            remove_delete_bitmap_key_ranges.emplace_back(start_key, end_key);
1888
0
        }
1889
0
    }
1890
0
    DBUG_EXECUTE_IF("BaseTablet.agg_delete_bitmap_for_stale_rowsets.merge_delete_bitmap.block",
1891
0
                    DBUG_BLOCK);
1892
0
    tablet_meta()->delete_bitmap().merge(*new_delete_bitmap);
1893
0
}
1894
1895
void BaseTablet::check_agg_delete_bitmap_for_stale_rowsets(int64_t& useless_rowset_count,
1896
0
                                                           int64_t& useless_rowset_version_count) {
1897
0
    std::set<RowsetId> rowset_ids;
1898
0
    std::set<int64_t> end_versions;
1899
0
    traverse_rowsets(
1900
0
            [&rowset_ids, &end_versions](const RowsetSharedPtr& rs) {
1901
0
                rowset_ids.emplace(rs->rowset_id());
1902
0
                end_versions.emplace(rs->end_version());
1903
0
            },
1904
0
            true);
1905
1906
0
    std::set<RowsetId> useless_rowsets;
1907
0
    std::map<RowsetId, std::vector<int64_t>> useless_rowset_versions;
1908
0
    {
1909
0
        _tablet_meta->delete_bitmap().traverse_rowset_and_version(
1910
                // 0: rowset and rowset with version exists
1911
                // -1: rowset does not exist
1912
                // -2: rowset exist, rowset with version does not exist
1913
0
                [&](const RowsetId& rowset_id, int64_t version) {
1914
0
                    if (rowset_ids.find(rowset_id) == rowset_ids.end()) {
1915
0
                        useless_rowsets.emplace(rowset_id);
1916
0
                        return -1;
1917
0
                    }
1918
0
                    if (end_versions.find(version) == end_versions.end()) {
1919
0
                        if (useless_rowset_versions.find(rowset_id) ==
1920
0
                            useless_rowset_versions.end()) {
1921
0
                            useless_rowset_versions[rowset_id] = {};
1922
0
                        }
1923
0
                        useless_rowset_versions[rowset_id].emplace_back(version);
1924
0
                        return -2;
1925
0
                    }
1926
0
                    return 0;
1927
0
                });
1928
0
    }
1929
0
    useless_rowset_count = useless_rowsets.size();
1930
0
    useless_rowset_version_count = useless_rowset_versions.size();
1931
0
    if (!useless_rowsets.empty() || !useless_rowset_versions.empty()) {
1932
0
        std::stringstream ss;
1933
0
        if (!useless_rowsets.empty()) {
1934
0
            ss << "useless rowsets: {";
1935
0
            for (auto it = useless_rowsets.begin(); it != useless_rowsets.end(); ++it) {
1936
0
                if (it != useless_rowsets.begin()) {
1937
0
                    ss << ", ";
1938
0
                }
1939
0
                ss << it->to_string();
1940
0
            }
1941
0
            ss << "}. ";
1942
0
        }
1943
0
        if (!useless_rowset_versions.empty()) {
1944
0
            ss << "useless rowset versions: {";
1945
0
            for (auto iter = useless_rowset_versions.begin(); iter != useless_rowset_versions.end();
1946
0
                 ++iter) {
1947
0
                if (iter != useless_rowset_versions.begin()) {
1948
0
                    ss << ", ";
1949
0
                }
1950
0
                ss << iter->first.to_string() << ": [";
1951
                // some versions are continuous, such as [8, 9, 10, 11, 13, 17, 18]
1952
                // print as [8-11, 13, 17-18]
1953
0
                int64_t last_start_version = -1;
1954
0
                int64_t last_end_version = -1;
1955
0
                for (int64_t version : iter->second) {
1956
0
                    if (last_start_version == -1) {
1957
0
                        last_start_version = version;
1958
0
                        last_end_version = version;
1959
0
                        continue;
1960
0
                    }
1961
0
                    if (last_end_version + 1 == version) {
1962
0
                        last_end_version = version;
1963
0
                    } else {
1964
0
                        if (last_start_version == last_end_version) {
1965
0
                            ss << last_start_version << ", ";
1966
0
                        } else {
1967
0
                            ss << last_start_version << "-" << last_end_version << ", ";
1968
0
                        }
1969
0
                        last_start_version = version;
1970
0
                        last_end_version = version;
1971
0
                    }
1972
0
                }
1973
0
                if (last_start_version == last_end_version) {
1974
0
                    ss << last_start_version;
1975
0
                } else {
1976
0
                    ss << last_start_version << "-" << last_end_version;
1977
0
                }
1978
1979
0
                ss << "]";
1980
0
            }
1981
0
            ss << "}.";
1982
0
        }
1983
0
        LOG(WARNING) << "failed check_agg_delete_bitmap_for_stale_rowsets for tablet_id="
1984
0
                     << tablet_id() << ". " << ss.str();
1985
0
    } else {
1986
0
        LOG(INFO) << "succeed check_agg_delete_bitmap_for_stale_rowsets for tablet_id="
1987
0
                  << tablet_id();
1988
0
    }
1989
0
}
1990
1991
0
RowsetSharedPtr BaseTablet::get_rowset(const RowsetId& rowset_id) {
1992
0
    std::shared_lock rdlock(_meta_lock);
1993
0
    for (auto& version_rowset : _rs_version_map) {
1994
0
        if (version_rowset.second->rowset_id() == rowset_id) {
1995
0
            return version_rowset.second;
1996
0
        }
1997
0
    }
1998
0
    for (auto& stale_version_rowset : _stale_rs_version_map) {
1999
0
        if (stale_version_rowset.second->rowset_id() == rowset_id) {
2000
0
            return stale_version_rowset.second;
2001
0
        }
2002
0
    }
2003
0
    return nullptr;
2004
0
}
2005
2006
1
std::vector<RowsetSharedPtr> BaseTablet::get_snapshot_rowset(bool include_stale_rowset) const {
2007
1
    std::shared_lock rdlock(_meta_lock);
2008
1
    std::vector<RowsetSharedPtr> rowsets;
2009
1
    std::transform(_rs_version_map.cbegin(), _rs_version_map.cend(), std::back_inserter(rowsets),
2010
28
                   [](auto& kv) { return kv.second; });
2011
1
    if (include_stale_rowset) {
2012
0
        std::transform(_stale_rs_version_map.cbegin(), _stale_rs_version_map.cend(),
2013
0
                       std::back_inserter(rowsets), [](auto& kv) { return kv.second; });
2014
0
    }
2015
1
    return rowsets;
2016
1
}
2017
2018
void BaseTablet::calc_consecutive_empty_rowsets(
2019
        std::vector<RowsetSharedPtr>* empty_rowsets,
2020
4
        const std::vector<RowsetSharedPtr>& candidate_rowsets, int64_t limit) {
2021
4
    int len = cast_set<int>(candidate_rowsets.size());
2022
12
    for (int i = 0; i < len - 1; ++i) {
2023
9
        auto rowset = candidate_rowsets[i];
2024
9
        auto next_rowset = candidate_rowsets[i + 1];
2025
2026
        // identify two consecutive rowsets that are empty
2027
9
        if (rowset->num_segments() == 0 && next_rowset->num_segments() == 0 &&
2028
9
            !rowset->rowset_meta()->has_delete_predicate() &&
2029
9
            !next_rowset->rowset_meta()->has_delete_predicate() &&
2030
9
            rowset->end_version() == next_rowset->start_version() - 1) {
2031
1
            empty_rowsets->emplace_back(rowset);
2032
1
            empty_rowsets->emplace_back(next_rowset);
2033
1
            rowset = next_rowset;
2034
1
            int next_index = i + 2;
2035
2036
            // keep searching for consecutive empty rowsets
2037
6
            while (next_index < len && candidate_rowsets[next_index]->num_segments() == 0 &&
2038
6
                   !candidate_rowsets[next_index]->rowset_meta()->has_delete_predicate() &&
2039
6
                   rowset->end_version() == candidate_rowsets[next_index]->start_version() - 1) {
2040
5
                empty_rowsets->emplace_back(candidate_rowsets[next_index]);
2041
5
                rowset = candidate_rowsets[next_index++];
2042
5
            }
2043
            // if the number of consecutive empty rowset reach the limit,
2044
            // and there are still rowsets following them
2045
1
            if (empty_rowsets->size() >= limit && next_index < len) {
2046
1
                return;
2047
1
            } else {
2048
                // current rowset is not empty, start searching from that rowset in the next
2049
0
                i = next_index - 1;
2050
0
                empty_rowsets->clear();
2051
0
            }
2052
1
        }
2053
9
    }
2054
4
}
2055
2056
Status BaseTablet::calc_file_crc(uint32_t* crc_value, int64_t start_version, int64_t end_version,
2057
0
                                 uint32_t* rowset_count, int64_t* file_count) {
2058
0
    Version v(start_version, end_version);
2059
0
    std::vector<RowsetSharedPtr> rowsets;
2060
0
    traverse_rowsets([&rowsets, &v](const auto& rs) {
2061
        // get all rowsets
2062
0
        if (v.contains(rs->version())) {
2063
0
            rowsets.emplace_back(rs);
2064
0
        }
2065
0
    });
2066
0
    std::sort(rowsets.begin(), rowsets.end(), Rowset::comparator);
2067
0
    *rowset_count = cast_set<uint32_t>(rowsets.size());
2068
2069
0
    *crc_value = 0;
2070
0
    *file_count = 0;
2071
0
    for (const auto& rs : rowsets) {
2072
0
        uint32_t rs_crc_value = 0;
2073
0
        int64_t rs_file_count = 0;
2074
0
        auto rowset = std::static_pointer_cast<BetaRowset>(rs);
2075
0
        auto st = rowset->calc_file_crc(&rs_crc_value, &rs_file_count);
2076
0
        if (!st.ok()) {
2077
0
            return st;
2078
0
        }
2079
        // crc_value is calculated based on the crc_value of each rowset.
2080
0
        *crc_value = crc32c::Extend(*crc_value, reinterpret_cast<const char*>(&rs_crc_value),
2081
0
                                    sizeof(rs_crc_value));
2082
0
        *file_count += rs_file_count;
2083
0
    }
2084
0
    return Status::OK();
2085
0
}
2086
2087
0
Status BaseTablet::show_nested_index_file(std::string* json_meta) {
2088
0
    Version v(0, max_version_unlocked());
2089
0
    std::vector<RowsetSharedPtr> rowsets;
2090
0
    traverse_rowsets([&rowsets, &v](const auto& rs) {
2091
        // get all rowsets
2092
0
        if (v.contains(rs->version())) {
2093
0
            rowsets.emplace_back(rs);
2094
0
        }
2095
0
    });
2096
0
    std::sort(rowsets.begin(), rowsets.end(), Rowset::comparator);
2097
2098
0
    rapidjson::Document doc;
2099
0
    doc.SetObject();
2100
0
    rapidjson::Document::AllocatorType& allocator = doc.GetAllocator();
2101
0
    rapidjson::Value tabletIdValue(tablet_id());
2102
0
    doc.AddMember("tablet_id", tabletIdValue, allocator);
2103
2104
0
    rapidjson::Value rowsets_value(rapidjson::kArrayType);
2105
2106
0
    for (const auto& rs : rowsets) {
2107
0
        rapidjson::Value rowset_value(rapidjson::kObjectType);
2108
2109
0
        auto rowset = std::static_pointer_cast<BetaRowset>(rs);
2110
0
        RETURN_IF_ERROR(rowset->show_nested_index_file(&rowset_value, allocator));
2111
0
        rowsets_value.PushBack(rowset_value, allocator);
2112
0
    }
2113
0
    doc.AddMember("rowsets", rowsets_value, allocator);
2114
2115
0
    rapidjson::StringBuffer buffer;
2116
0
    rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(buffer);
2117
0
    doc.Accept(writer);
2118
0
    *json_meta = std::string(buffer.GetString());
2119
2120
0
    return Status::OK();
2121
0
}
2122
2123
void BaseTablet::get_base_rowset_delete_bitmap_count(
2124
        uint64_t* max_base_rowset_delete_bitmap_score,
2125
0
        int64_t* max_base_rowset_delete_bitmap_score_tablet_id) {
2126
0
    std::vector<RowsetSharedPtr> rowsets_;
2127
0
    std::string base_rowset_id_str;
2128
0
    {
2129
0
        std::shared_lock rowset_ldlock(this->get_header_lock());
2130
0
        for (const auto& it : _rs_version_map) {
2131
0
            rowsets_.emplace_back(it.second);
2132
0
        }
2133
0
    }
2134
0
    std::sort(rowsets_.begin(), rowsets_.end(), Rowset::comparator);
2135
0
    if (!rowsets_.empty()) {
2136
0
        bool base_found = false;
2137
0
        for (auto& rowset : rowsets_) {
2138
0
            if (rowset->start_version() > 2) {
2139
0
                break;
2140
0
            }
2141
0
            base_found = true;
2142
0
            uint64_t base_rowset_delete_bitmap_count =
2143
0
                    this->tablet_meta()->delete_bitmap().get_count_with_range(
2144
0
                            {rowset->rowset_id(), 0, 0},
2145
0
                            {rowset->rowset_id(), UINT32_MAX, UINT64_MAX});
2146
0
            if (base_rowset_delete_bitmap_count > *max_base_rowset_delete_bitmap_score) {
2147
0
                *max_base_rowset_delete_bitmap_score = base_rowset_delete_bitmap_count;
2148
0
                *max_base_rowset_delete_bitmap_score_tablet_id = this->tablet_id();
2149
0
            }
2150
0
        }
2151
0
        if (!base_found) {
2152
0
            LOG(WARNING) << "can not found base rowset for tablet " << tablet_id();
2153
0
        }
2154
0
    }
2155
0
}
2156
2157
32
int32_t BaseTablet::max_version_config() {
2158
32
    int32_t max_version = tablet_meta()->compaction_policy() == CUMULATIVE_TIME_SERIES_POLICY
2159
32
                                  ? std::max(config::time_series_max_tablet_version_num,
2160
0
                                             config::max_tablet_version_num)
2161
32
                                  : config::max_tablet_version_num;
2162
32
    return max_version;
2163
32
}
2164
2165
} // namespace doris