Coverage Report

Created: 2026-07-29 18:03

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/tablet/tablet_reader.cpp
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "storage/tablet/tablet_reader.h"
19
20
#include <gen_cpp/olap_file.pb.h>
21
#include <gen_cpp/segment_v2.pb.h>
22
#include <thrift/protocol/TDebugProtocol.h>
23
24
#include <algorithm>
25
#include <functional>
26
#include <iterator>
27
#include <memory>
28
#include <numeric>
29
#include <ostream>
30
#include <set>
31
#include <shared_mutex>
32
#include <unordered_map>
33
34
#include "common/compiler_util.h" // IWYU pragma: keep
35
#include "common/config.h"
36
#include "common/exception.h"
37
#include "common/logging.h"
38
#include "common/status.h"
39
#include "core/arena.h"
40
#include "core/block/block.h"
41
#include "exec/common/variant_util.h"
42
#include "exprs/bloom_filter_func.h"
43
#include "exprs/create_predicate_function.h"
44
#include "exprs/hybrid_set.h"
45
#include "runtime/query_context.h"
46
#include "runtime/runtime_predicate.h"
47
#include "runtime/runtime_state.h"
48
#include "storage/delete/delete_handler.h"
49
#include "storage/index/bloom_filter/bloom_filter.h"
50
#include "storage/itoken_extractor.h"
51
#include "storage/olap_common.h"
52
#include "storage/olap_define.h"
53
#include "storage/predicate/block_column_predicate.h"
54
#include "storage/predicate/column_predicate.h"
55
#include "storage/predicate/like_column_predicate.h"
56
#include "storage/predicate/predicate_creator.h"
57
#include "storage/row_cursor.h"
58
#include "storage/schema.h"
59
#include "storage/tablet/tablet.h"
60
#include "storage/tablet/tablet_meta.h"
61
#include "storage/tablet/tablet_schema.h"
62
63
namespace doris {
64
using namespace ErrorCode;
65
66
1.33M
void TabletReader::ReaderParams::check_validation() const {
67
1.33M
    if (UNLIKELY(version.first == -1 && is_segcompaction == false)) {
68
0
        throw Exception(Status::FatalError("version is not set. tablet={}", tablet->tablet_id()));
69
0
    }
70
1.33M
}
71
72
1.33M
Status TabletReader::init(const ReaderParams& read_params) {
73
1.33M
    Status res = _init_params(read_params);
74
1.33M
    if (!res.ok()) {
75
0
        LOG(WARNING) << "fail to init reader when init params. res:" << res
76
0
                     << ", tablet_id:" << read_params.tablet->tablet_id()
77
0
                     << ", schema_hash:" << read_params.tablet->schema_hash()
78
0
                     << ", reader type:" << int(read_params.reader_type)
79
0
                     << ", version:" << read_params.version;
80
0
    }
81
1.33M
    return res;
82
1.33M
}
83
84
void TabletReader::remove_delete_columns_from_access_paths(
85
        const DeleteHandler& delete_handler, const TabletSchema& tablet_schema,
86
80
        std::map<int32_t, TColumnAccessPaths>& all_access_paths) {
87
80
    auto delete_predicates = AndBlockColumnPredicate::create_shared();
88
80
    std::unordered_map<int32_t, std::vector<std::shared_ptr<const ColumnPredicate>>>
89
80
            del_predicates_for_zone_map;
90
80
    delete_handler.get_delete_conditions_after_version(0, delete_predicates.get(),
91
80
                                                       &del_predicates_for_zone_map);
92
80
    std::set<ColumnId> delete_column_ids;
93
80
    delete_predicates->get_all_column_ids(delete_column_ids);
94
128
    for (auto cid : delete_column_ids) {
95
128
        all_access_paths.erase(tablet_schema.column(cid).unique_id());
96
128
    }
97
80
}
98
99
1.33M
Status TabletReader::_capture_rs_readers(const ReaderParams& read_params) {
100
1.33M
    SCOPED_RAW_TIMER(&_stats.tablet_reader_capture_rs_readers_timer_ns);
101
1.33M
    if (read_params.rs_splits.empty()) {
102
0
        return Status::InternalError("fail to acquire data sources. tablet={}",
103
0
                                     _tablet->tablet_id());
104
0
    }
105
106
1.33M
    bool eof = false;
107
1.33M
    bool is_lower_key_included = _keys_param.start_key_include;
108
1.33M
    bool is_upper_key_included = _keys_param.end_key_include;
109
110
3.00M
    for (int i = 0; i < _keys_param.start_keys.size(); ++i) {
111
        // lower bound
112
1.67M
        RowCursor& start_key = _keys_param.start_keys[i];
113
1.67M
        RowCursor& end_key = _keys_param.end_keys[i];
114
115
1.67M
        if (!is_lower_key_included) {
116
1.58k
            if (compare_row_key(start_key, end_key) >= 0) {
117
0
                VLOG_NOTICE << "return EOF when lower key not include"
118
0
                            << ", start_key=" << start_key.to_string()
119
0
                            << ", end_key=" << end_key.to_string();
120
0
                eof = true;
121
0
                break;
122
0
            }
123
1.67M
        } else {
124
1.67M
            if (compare_row_key(start_key, end_key) > 0) {
125
0
                VLOG_NOTICE << "return EOF when lower key include="
126
0
                            << ", start_key=" << start_key.to_string()
127
0
                            << ", end_key=" << end_key.to_string();
128
0
                eof = true;
129
0
                break;
130
0
            }
131
1.67M
        }
132
133
1.67M
        _is_lower_keys_included.push_back(is_lower_key_included);
134
1.67M
        _is_upper_keys_included.push_back(is_upper_key_included);
135
1.67M
    }
136
137
1.33M
    if (eof) {
138
0
        return Status::EndOfFile("reach end of scan range. tablet={}", _tablet->tablet_id());
139
0
    }
140
141
1.33M
    bool need_ordered_result = true;
142
1.33M
    if (read_params.reader_type == ReaderType::READER_QUERY ||
143
1.33M
        read_params.reader_type == ReaderType::READER_BINLOG) {
144
1.30M
        if (_tablet_schema->keys_type() == DUP_KEYS) {
145
            // duplicated keys are allowed, no need to merge sort keys in rowset
146
400k
            need_ordered_result = false;
147
400k
        }
148
1.30M
        if (_tablet_schema->keys_type() == UNIQUE_KEYS &&
149
1.30M
            _tablet->enable_unique_key_merge_on_write()) {
150
            // unique keys with merge on write, no need to merge sort keys in rowset
151
863k
            need_ordered_result = false;
152
863k
        }
153
1.30M
        if (_aggregation) {
154
            // compute engine will aggregate rows with the same key,
155
            // it's ok for rowset to return unordered result
156
1.26M
            need_ordered_result = false;
157
1.26M
        }
158
159
1.30M
        if (_direct_mode) {
160
            // direct mode indicates that the storage layer does not need to merge,
161
            // it's ok for rowset to return unordered result
162
1.29M
            need_ordered_result = false;
163
1.29M
        }
164
165
1.30M
        if (read_params.read_orderby_key) {
166
741
            need_ordered_result = true;
167
741
        }
168
1.30M
    }
169
170
1.33M
    _reader_context.reader_type = read_params.reader_type;
171
1.33M
    _reader_context.version = read_params.version;
172
1.33M
    _reader_context.tablet_schema = _tablet_schema;
173
1.33M
    _reader_context.need_ordered_result = need_ordered_result;
174
1.33M
    _reader_context.topn_filter_source_node_ids = read_params.topn_filter_source_node_ids;
175
1.33M
    _reader_context.topn_filter_target_node_id = read_params.topn_filter_target_node_id;
176
1.33M
    _reader_context.read_orderby_key_reverse = read_params.read_orderby_key_reverse;
177
1.33M
    _reader_context.use_insert_order_when_same =
178
1.33M
            read_params.use_insert_order_when_same ||
179
1.33M
            read_params.reader_type == ReaderType::READER_BINLOG ||
180
1.33M
            read_params.reader_type == ReaderType::READER_BINLOG_COMPACTION;
181
1.33M
    _reader_context.force_key_ordered_read = read_params.force_key_ordered_read;
182
1.33M
    _reader_context.read_orderby_key_limit = read_params.read_orderby_key_limit;
183
1.33M
    _reader_context.return_columns = &_return_columns;
184
1.33M
    _reader_context.tso_predicate_column_id = read_params.tso_predicate_column_id;
185
1.33M
    _reader_context.start_tso = read_params.start_tso;
186
1.33M
    _reader_context.end_tso = read_params.end_tso;
187
1.33M
    _reader_context.read_orderby_key_columns =
188
1.33M
            !_orderby_key_columns.empty() ? &_orderby_key_columns : nullptr;
189
1.33M
    _reader_context.predicates = &_col_predicates;
190
1.33M
    _reader_context.value_predicates = &_value_col_predicates;
191
1.33M
    _reader_context.lower_bound_keys = &_keys_param.start_keys;
192
1.33M
    _reader_context.is_lower_keys_included = &_is_lower_keys_included;
193
1.33M
    _reader_context.upper_bound_keys = &_keys_param.end_keys;
194
1.33M
    _reader_context.is_upper_keys_included = &_is_upper_keys_included;
195
1.33M
    _reader_context.delete_handler = &_delete_handler;
196
1.33M
    _reader_context.stats = &_stats;
197
1.33M
    _reader_context.use_page_cache = read_params.use_page_cache;
198
1.33M
    _reader_context.sequence_id_idx = _sequence_col_idx;
199
1.33M
    _reader_context.is_unique = tablet()->keys_type() == UNIQUE_KEYS;
200
1.33M
    _reader_context.merged_rows = &_merged_rows;
201
1.33M
    _reader_context.delete_bitmap = read_params.delete_bitmap;
202
1.33M
    _reader_context.enable_unique_key_merge_on_write = tablet()->enable_unique_key_merge_on_write();
203
1.33M
    _reader_context.enable_mor_value_predicate_pushdown =
204
1.33M
            read_params.enable_mor_value_predicate_pushdown;
205
1.33M
    _reader_context.record_rowids = read_params.record_rowids;
206
1.33M
    _reader_context.rowid_conversion = read_params.rowid_conversion;
207
1.33M
    _reader_context.is_key_column_group = read_params.is_key_column_group;
208
1.33M
    _reader_context.common_expr_ctxs_push_down = read_params.common_expr_ctxs_push_down;
209
1.33M
    _reader_context.output_columns = &read_params.output_columns;
210
1.33M
    _reader_context.extra_columns = read_params.extra_columns;
211
1.33M
    _reader_context.push_down_agg_type_opt = read_params.push_down_agg_type_opt;
212
1.33M
    _reader_context.ttl_seconds = _tablet->ttl_seconds();
213
1.33M
    _reader_context.score_runtime = read_params.score_runtime;
214
1.33M
    _reader_context.collection_statistics = read_params.collection_statistics;
215
216
1.33M
    _reader_context.virtual_column_exprs = read_params.virtual_column_exprs;
217
1.33M
    _reader_context.ann_topn_runtime = read_params.ann_topn_runtime;
218
219
1.33M
    _reader_context.condition_cache_digest = read_params.condition_cache_digest;
220
1.33M
    _reader_context.all_access_paths = read_params.all_access_paths;
221
1.33M
    _reader_context.predicate_access_paths = read_params.predicate_access_paths;
222
223
    // Force a full read of delete-condition columns: the FE can't see storage deletes and may
224
    // mark them meta-only (OFFSET/NULL), whose content-less read makes the delete predicate
225
    // match nothing and leak deleted rows.
226
1.33M
    if (!_delete_handler.empty() && !_reader_context.all_access_paths.empty()) {
227
84
        remove_delete_columns_from_access_paths(_delete_handler, *_tablet_schema,
228
84
                                                _reader_context.all_access_paths);
229
84
    }
230
231
    // Propagate general read limit for DUP_KEYS and UNIQUE_KEYS with MOW
232
1.33M
    _reader_context.general_read_limit = read_params.general_read_limit;
233
234
1.33M
    return Status::OK();
235
1.33M
}
236
237
268
TabletColumn TabletReader::materialize_column(const TabletColumn& orig) {
238
276
    if (!orig.is_variant_type()) {
239
276
        return orig;
240
276
    }
241
18.4E
    TabletColumn column_with_cast_type = orig;
242
18.4E
    auto cast_type = _reader_context.target_cast_type_for_variants.at(orig.name());
243
18.4E
    return variant_util::get_column_by_type(cast_type, orig.name(),
244
18.4E
                                            {
245
18.4E
                                                    .unique_id = orig.unique_id(),
246
18.4E
                                                    .parent_unique_id = orig.parent_unique_id(),
247
18.4E
                                                    .path_info = *orig.path_info_ptr(),
248
18.4E
                                            });
249
268
}
250
251
1.33M
Status TabletReader::_init_params(const ReaderParams& read_params) {
252
1.33M
    read_params.check_validation();
253
254
1.33M
    _direct_mode = read_params.direct_mode;
255
1.33M
    _aggregation = read_params.aggregation;
256
1.33M
    _reader_type = read_params.reader_type;
257
1.33M
    _tablet = read_params.tablet;
258
1.33M
    _tablet_schema = read_params.tablet_schema;
259
1.33M
    _reader_context.runtime_state = read_params.runtime_state;
260
1.33M
    _reader_context.target_cast_type_for_variants = read_params.target_cast_type_for_variants;
261
262
1.33M
    RETURN_IF_ERROR(_init_conditions_param(read_params));
263
264
1.33M
    Status res = _init_delete_condition(read_params);
265
1.33M
    if (!res.ok()) {
266
0
        LOG(WARNING) << "fail to init delete param. res = " << res;
267
0
        return res;
268
0
    }
269
270
1.33M
    res = _init_return_columns(read_params);
271
1.33M
    if (!res.ok()) {
272
0
        LOG(WARNING) << "fail to init return columns. res = " << res;
273
0
        return res;
274
0
    }
275
276
1.33M
    res = _init_keys_param(read_params);
277
1.33M
    if (!res.ok()) {
278
0
        LOG(WARNING) << "fail to init keys param. res=" << res;
279
0
        return res;
280
0
    }
281
1.33M
    res = _init_orderby_keys_param(read_params);
282
1.33M
    if (!res.ok()) {
283
0
        LOG(WARNING) << "fail to init orderby keys param. res=" << res;
284
0
        return res;
285
0
    }
286
1.33M
    if (_tablet_schema->has_sequence_col()) {
287
6.60k
        auto sequence_col_idx = _tablet_schema->sequence_col_idx();
288
6.60k
        DCHECK_NE(sequence_col_idx, -1);
289
23.7k
        for (auto col : _return_columns) {
290
            // query has sequence col
291
23.7k
            if (col == sequence_col_idx) {
292
409
                _sequence_col_idx = sequence_col_idx;
293
409
                break;
294
409
            }
295
23.7k
        }
296
6.60k
    }
297
298
1.33M
    return res;
299
1.33M
}
300
301
1.33M
Status TabletReader::_init_return_columns(const ReaderParams& read_params) {
302
1.33M
    SCOPED_RAW_TIMER(&_stats.tablet_reader_init_return_columns_timer_ns);
303
1.33M
    if (read_params.reader_type == ReaderType::READER_QUERY ||
304
1.33M
        read_params.reader_type == ReaderType::READER_BINLOG) {
305
1.30M
        _return_columns = read_params.return_columns;
306
1.30M
        _tablet_columns_convert_to_null_set = read_params.tablet_columns_convert_to_null_set;
307
13.9M
        for (auto id : read_params.return_columns) {
308
13.9M
            if (_tablet_schema->column(id).is_key()) {
309
6.14M
                _key_cids.push_back(id);
310
7.81M
            } else {
311
7.81M
                _value_cids.push_back(id);
312
7.81M
            }
313
13.9M
        }
314
1.30M
    } else if (read_params.return_columns.empty()) {
315
0
        for (uint32_t i = 0; i < _tablet_schema->num_columns(); ++i) {
316
0
            _return_columns.push_back(i);
317
0
            if (_tablet_schema->column(i).is_key()) {
318
0
                _key_cids.push_back(i);
319
0
            } else {
320
0
                _value_cids.push_back(i);
321
0
            }
322
0
        }
323
0
        VLOG_NOTICE << "return column is empty, using full column as default.";
324
22.6k
    } else if ((read_params.reader_type == ReaderType::READER_CUMULATIVE_COMPACTION ||
325
22.6k
                read_params.reader_type == ReaderType::READER_SEGMENT_COMPACTION ||
326
22.6k
                read_params.reader_type == ReaderType::READER_BASE_COMPACTION ||
327
22.6k
                read_params.reader_type == ReaderType::READER_FULL_COMPACTION ||
328
22.6k
                read_params.reader_type == ReaderType::READER_BINLOG_COMPACTION ||
329
22.6k
                read_params.reader_type == ReaderType::READER_COLD_DATA_COMPACTION ||
330
22.6k
                read_params.reader_type == ReaderType::READER_ALTER_TABLE) &&
331
29.5k
               !read_params.return_columns.empty()) {
332
29.5k
        _return_columns = read_params.return_columns;
333
112k
        for (auto id : read_params.return_columns) {
334
112k
            if (_tablet_schema->column(id).is_key()) {
335
39.6k
                _key_cids.push_back(id);
336
72.8k
            } else {
337
72.8k
                _value_cids.push_back(id);
338
72.8k
            }
339
112k
        }
340
18.4E
    } else if (read_params.reader_type == ReaderType::READER_CHECKSUM) {
341
0
        _return_columns = read_params.return_columns;
342
0
        for (auto id : read_params.return_columns) {
343
0
            if (_tablet_schema->column(id).is_key()) {
344
0
                _key_cids.push_back(id);
345
0
            } else {
346
0
                _value_cids.push_back(id);
347
0
            }
348
0
        }
349
18.4E
    } else {
350
18.4E
        return Status::Error<INVALID_ARGUMENT>(
351
18.4E
                "fail to init return columns. reader_type={}, return_columns_size={}",
352
18.4E
                int(read_params.reader_type), read_params.return_columns.size());
353
18.4E
    }
354
355
1.33M
    std::sort(_key_cids.begin(), _key_cids.end(), std::greater<>());
356
357
1.33M
    return Status::OK();
358
1.33M
}
359
360
1.33M
Status TabletReader::_init_keys_param(const ReaderParams& read_params) {
361
1.33M
    SCOPED_RAW_TIMER(&_stats.tablet_reader_init_keys_param_timer_ns);
362
1.33M
    if (read_params.start_key.empty()) {
363
254k
        return Status::OK();
364
254k
    }
365
366
1.07M
    _keys_param.start_key_include = read_params.start_key_include;
367
1.07M
    _keys_param.end_key_include = read_params.end_key_include;
368
369
1.07M
    size_t start_key_size = read_params.start_key.size();
370
    //_keys_param.start_keys.resize(start_key_size);
371
1.07M
    std::vector<RowCursor>(start_key_size).swap(_keys_param.start_keys);
372
373
1.07M
    size_t scan_key_size = read_params.start_key.front().size();
374
1.07M
    if (scan_key_size > _tablet_schema->num_columns()) {
375
0
        return Status::Error<INVALID_ARGUMENT>(
376
0
                "Input param are invalid. Column count is bigger than num_columns of schema. "
377
0
                "column_count={}, schema.num_columns={}",
378
0
                scan_key_size, _tablet_schema->num_columns());
379
0
    }
380
381
2.75M
    for (size_t i = 0; i < start_key_size; ++i) {
382
1.67M
        if (read_params.start_key[i].size() != scan_key_size) {
383
0
            return Status::Error<INVALID_ARGUMENT>(
384
0
                    "The start_key.at({}).size={}, not equals the scan_key_size={}", i,
385
0
                    read_params.start_key[i].size(), scan_key_size);
386
0
        }
387
388
1.67M
        Status res = _keys_param.start_keys[i].init(_tablet_schema, read_params.start_key[i]);
389
1.67M
        if (!res.ok()) {
390
0
            LOG(WARNING) << "fail to init row cursor. res = " << res;
391
0
            return res;
392
0
        }
393
1.67M
    }
394
395
1.07M
    size_t end_key_size = read_params.end_key.size();
396
    //_keys_param.end_keys.resize(end_key_size);
397
1.07M
    std::vector<RowCursor>(end_key_size).swap(_keys_param.end_keys);
398
2.75M
    for (size_t i = 0; i < end_key_size; ++i) {
399
1.67M
        if (read_params.end_key[i].size() != scan_key_size) {
400
0
            return Status::Error<INVALID_ARGUMENT>(
401
0
                    "The end_key.at({}).size={}, not equals the scan_key_size={}", i,
402
0
                    read_params.end_key[i].size(), scan_key_size);
403
0
        }
404
405
1.67M
        Status res = _keys_param.end_keys[i].init(_tablet_schema, read_params.end_key[i]);
406
1.67M
        if (!res.ok()) {
407
0
            LOG(WARNING) << "fail to init row cursor. res = " << res;
408
0
            return res;
409
0
        }
410
1.67M
    }
411
412
    //TODO:check the valid of start_key and end_key.(eg. start_key <= end_key)
413
414
1.07M
    return Status::OK();
415
1.07M
}
416
417
1.33M
Status TabletReader::_init_orderby_keys_param(const ReaderParams& read_params) {
418
1.33M
    SCOPED_RAW_TIMER(&_stats.tablet_reader_init_orderby_keys_param_timer_ns);
419
    // UNIQUE_KEYS will compare all keys as before
420
1.33M
    if (_tablet_schema->keys_type() == DUP_KEYS || (_tablet_schema->keys_type() == UNIQUE_KEYS &&
421
1.29M
                                                    _tablet->enable_unique_key_merge_on_write())) {
422
1.29M
        if (!_tablet_schema->cluster_key_uids().empty()) {
423
3.81k
            if (read_params.read_orderby_key_num_prefix_columns >
424
3.81k
                _tablet_schema->cluster_key_uids().size()) {
425
0
                return Status::Error<ErrorCode::INTERNAL_ERROR>(
426
0
                        "read_orderby_key_num_prefix_columns={} > cluster_keys.size()={}",
427
0
                        read_params.read_orderby_key_num_prefix_columns,
428
0
                        _tablet_schema->cluster_key_uids().size());
429
0
            }
430
3.81k
            for (uint32_t i = 0; i < read_params.read_orderby_key_num_prefix_columns; i++) {
431
0
                auto cid = _tablet_schema->cluster_key_uids()[i];
432
0
                auto index = _tablet_schema->field_index(cid);
433
0
                if (index < 0) {
434
0
                    return Status::Error<ErrorCode::INTERNAL_ERROR>(
435
0
                            "could not find cluster key column with unique_id=" +
436
0
                            std::to_string(cid) +
437
0
                            " in tablet schema, tablet_id=" + std::to_string(_tablet->tablet_id()));
438
0
                }
439
0
                for (uint32_t idx = 0; idx < _return_columns.size(); idx++) {
440
0
                    if (_return_columns[idx] == index) {
441
0
                        _orderby_key_columns.push_back(idx);
442
0
                        break;
443
0
                    }
444
0
                }
445
0
            }
446
1.28M
        } else {
447
            // find index in vector _return_columns
448
            //   for the read_orderby_key_num_prefix_columns orderby keys
449
1.28M
            for (uint32_t i = 0; i < read_params.read_orderby_key_num_prefix_columns; i++) {
450
1.34k
                for (uint32_t idx = 0; idx < _return_columns.size(); idx++) {
451
1.34k
                    if (_return_columns[idx] == i) {
452
1.01k
                        _orderby_key_columns.push_back(idx);
453
1.01k
                        break;
454
1.01k
                    }
455
1.34k
                }
456
1.01k
            }
457
1.28M
        }
458
1.29M
        if (read_params.read_orderby_key_num_prefix_columns != _orderby_key_columns.size()) {
459
0
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
460
0
                    "read_orderby_key_num_prefix_columns != _orderby_key_columns.size, "
461
0
                    "read_params.read_orderby_key_num_prefix_columns={}, "
462
0
                    "_orderby_key_columns.size()={}",
463
0
                    read_params.read_orderby_key_num_prefix_columns, _orderby_key_columns.size());
464
0
        }
465
1.29M
    }
466
467
1.33M
    return Status::OK();
468
1.33M
}
469
470
1.33M
Status TabletReader::_init_conditions_param(const ReaderParams& read_params) {
471
1.33M
    SCOPED_RAW_TIMER(&_stats.tablet_reader_init_conditions_param_timer_ns);
472
1.33M
    std::vector<std::shared_ptr<ColumnPredicate>> predicates;
473
1.33M
    std::copy(read_params.predicates.cbegin(), read_params.predicates.cend(),
474
1.33M
              std::inserter(predicates, predicates.begin()));
475
    // Function filter push down to storage engine
476
1.33M
    auto is_like_predicate = [](std::shared_ptr<ColumnPredicate> _pred) {
477
285
        return dynamic_cast<LikeColumnPredicate*>(_pred.get()) != nullptr;
478
285
    };
479
480
1.33M
    for (const auto& filter : read_params.function_filters) {
481
277
        predicates.emplace_back(_parse_to_predicate(filter));
482
277
        auto pred = predicates.back();
483
484
277
        const auto& col = _tablet_schema->column(pred->column_id());
485
277
        const auto* tablet_index = _tablet_schema->get_ngram_bf_index(col.unique_id());
486
284
        if (is_like_predicate(pred) && tablet_index && config::enable_query_like_bloom_filter) {
487
13
            std::unique_ptr<segment_v2::BloomFilter> ng_bf;
488
13
            std::string pattern = pred->get_search_str();
489
13
            auto gram_bf_size = tablet_index->get_gram_bf_size();
490
13
            auto gram_size = tablet_index->get_gram_size();
491
492
13
            RETURN_IF_ERROR(segment_v2::BloomFilter::create(segment_v2::NGRAM_BLOOM_FILTER, &ng_bf,
493
13
                                                            gram_bf_size));
494
13
            NgramTokenExtractor _token_extractor(gram_size);
495
496
13
            if (_token_extractor.string_like_to_bloom_filter(pattern.data(), pattern.length(),
497
13
                                                             *ng_bf)) {
498
13
                pred->set_page_ng_bf(std::move(ng_bf));
499
13
            }
500
13
        }
501
277
    }
502
503
1.33M
    int32_t delete_sign_idx = _tablet_schema->delete_sign_idx();
504
1.33M
    for (auto predicate : predicates) {
505
1.03M
        auto column = _tablet_schema->column(predicate->column_id());
506
1.03M
        if (column.aggregation() != FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE) {
507
            // When MOR value predicate pushdown is enabled, drop __DORIS_DELETE_SIGN__
508
            // from storage-layer predicates entirely. Delete sign must only be evaluated
509
            // post-merge via VExpr to prevent deleted rows from reappearing.
510
2.98k
            if (read_params.enable_mor_value_predicate_pushdown && delete_sign_idx >= 0 &&
511
2.98k
                predicate->column_id() == static_cast<uint32_t>(delete_sign_idx)) {
512
25
                continue;
513
25
            }
514
2.95k
            _value_col_predicates.push_back(predicate);
515
1.03M
        } else {
516
1.03M
            _col_predicates.push_back(predicate);
517
1.03M
        }
518
1.03M
    }
519
520
1.33M
    return Status::OK();
521
1.33M
}
522
523
std::shared_ptr<ColumnPredicate> TabletReader::_parse_to_predicate(
524
267
        const FunctionFilter& function_filter) {
525
267
    int32_t index = _tablet_schema->field_index(function_filter._col_name);
526
267
    if (index < 0) {
527
0
        throw Exception(Status::InternalError("Column {} not found in tablet schema",
528
0
                                              function_filter._col_name));
529
0
        return nullptr;
530
0
    }
531
267
    const TabletColumn& column = materialize_column(_tablet_schema->column(index));
532
267
    return create_column_predicate(index, std::make_shared<FunctionFilter>(function_filter),
533
267
                                   column.type(), &column);
534
267
}
535
536
1.33M
Status TabletReader::_init_delete_condition(const ReaderParams& read_params) {
537
1.33M
    SCOPED_RAW_TIMER(&_stats.tablet_reader_init_delete_condition_param_timer_ns);
538
    // If it's cumu and not allow do delete when cumu
539
1.33M
    if (read_params.reader_type == ReaderType::READER_SEGMENT_COMPACTION ||
540
1.33M
        (read_params.reader_type == ReaderType::READER_CUMULATIVE_COMPACTION &&
541
1.33M
         !config::enable_delete_when_cumu_compaction)) {
542
27.4k
        return Status::OK();
543
27.4k
    }
544
1.30M
    bool cumu_delete = read_params.reader_type == ReaderType::READER_CUMULATIVE_COMPACTION &&
545
1.30M
                       config::enable_delete_when_cumu_compaction;
546
    // Delete sign could not be applied when delete on cumu compaction is enabled, bucause it is meant for delete with predicates.
547
    // If delete design is applied on cumu compaction, it will lose effect when doing base compaction.
548
    // `_delete_sign_available` indicates the condition where we could apply delete signs to data.
549
1.30M
    _delete_sign_available = (((read_params.reader_type == ReaderType::READER_BASE_COMPACTION ||
550
1.30M
                                read_params.reader_type == ReaderType::READER_FULL_COMPACTION) &&
551
1.30M
                               config::enable_prune_delete_sign_when_base_compaction) ||
552
1.30M
                              read_params.reader_type == ReaderType::READER_COLD_DATA_COMPACTION ||
553
1.30M
                              read_params.reader_type == ReaderType::READER_CHECKSUM);
554
555
    // `_filter_delete` indicates the condition where we should execlude deleted tuples when reading data.
556
    // However, queries will not use this condition but generate special where predicates to filter data.
557
    // (Though a lille bit confused, it is how the current logic working...)
558
1.30M
    _filter_delete = _delete_sign_available || cumu_delete;
559
1.30M
    return _delete_handler.init(_tablet_schema, read_params.delete_predicates,
560
1.30M
                                read_params.version.second);
561
1.33M
}
562
563
Status TabletReader::init_reader_params_and_create_block(
564
        TabletSharedPtr tablet, ReaderType reader_type,
565
        const std::vector<RowsetSharedPtr>& input_rowsets,
566
0
        TabletReader::ReaderParams* reader_params, Block* block) {
567
0
    reader_params->tablet = tablet;
568
0
    reader_params->reader_type = reader_type;
569
0
    reader_params->version =
570
0
            Version(input_rowsets.front()->start_version(), input_rowsets.back()->end_version());
571
572
0
    TabletReadSource read_source;
573
0
    for (const auto& rowset : input_rowsets) {
574
0
        RowsetReaderSharedPtr rs_reader;
575
0
        RETURN_IF_ERROR(rowset->create_reader(&rs_reader));
576
0
        read_source.rs_splits.emplace_back(std::move(rs_reader));
577
0
    }
578
0
    read_source.fill_delete_predicates();
579
0
    reader_params->set_read_source(std::move(read_source));
580
581
0
    std::vector<RowsetMetaSharedPtr> rowset_metas(input_rowsets.size());
582
0
    std::transform(input_rowsets.begin(), input_rowsets.end(), rowset_metas.begin(),
583
0
                   [](const RowsetSharedPtr& rowset) { return rowset->rowset_meta(); });
584
0
    TabletSchemaSPtr read_tablet_schema =
585
0
            tablet->tablet_schema_with_merged_max_schema_version(rowset_metas);
586
0
    TabletSchemaSPtr merge_tablet_schema = std::make_shared<TabletSchema>();
587
0
    merge_tablet_schema->copy_from(*read_tablet_schema);
588
589
    // Merge the columns in delete predicate that not in latest schema in to current tablet schema
590
0
    for (auto& del_pred : reader_params->delete_predicates) {
591
0
        merge_tablet_schema->merge_dropped_columns(*del_pred->tablet_schema());
592
0
    }
593
0
    reader_params->tablet_schema = merge_tablet_schema;
594
595
0
    reader_params->return_columns.resize(read_tablet_schema->num_columns());
596
0
    std::iota(reader_params->return_columns.begin(), reader_params->return_columns.end(), 0);
597
0
    reader_params->origin_return_columns = &reader_params->return_columns;
598
599
0
    *block = read_tablet_schema->create_block();
600
601
0
    return Status::OK();
602
0
}
603
604
} // namespace doris