Coverage Report

Created: 2024-11-21 20:39

/root/doris/be/src/olap/tablet_reader.cpp
Line
Count
Source (jump to first uncovered line)
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/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 <shared_mutex>
31
32
#include "common/compiler_util.h" // IWYU pragma: keep
33
#include "common/config.h"
34
#include "common/exception.h"
35
#include "common/logging.h"
36
#include "common/status.h"
37
#include "exprs/bitmapfilter_predicate.h"
38
#include "exprs/bloom_filter_func.h"
39
#include "exprs/create_predicate_function.h"
40
#include "exprs/hybrid_set.h"
41
#include "olap/column_predicate.h"
42
#include "olap/itoken_extractor.h"
43
#include "olap/like_column_predicate.h"
44
#include "olap/olap_common.h"
45
#include "olap/olap_define.h"
46
#include "olap/predicate_creator.h"
47
#include "olap/row_cursor.h"
48
#include "olap/rowset/segment_v2/bloom_filter.h"
49
#include "olap/schema.h"
50
#include "olap/tablet.h"
51
#include "olap/tablet_meta.h"
52
#include "olap/tablet_schema.h"
53
#include "runtime/query_context.h"
54
#include "runtime/runtime_predicate.h"
55
#include "runtime/runtime_state.h"
56
#include "vec/common/arena.h"
57
#include "vec/core/block.h"
58
59
namespace doris {
60
using namespace ErrorCode;
61
62
163
void TabletReader::ReaderParams::check_validation() const {
63
163
    if (UNLIKELY(version.first == -1 && is_segcompaction == false)) {
64
0
        LOG(FATAL) << "version is not set. tablet=" << tablet->tablet_id();
65
0
    }
66
163
}
67
68
0
std::string TabletReader::ReaderParams::to_string() const {
69
0
    std::stringstream ss;
70
0
    ss << "tablet=" << tablet->tablet_id() << " reader_type=" << int(reader_type)
71
0
       << " aggregation=" << aggregation << " version=" << version
72
0
       << " start_key_include=" << start_key_include << " end_key_include=" << end_key_include;
73
74
0
    for (const auto& key : start_key) {
75
0
        ss << " keys=" << key;
76
0
    }
77
78
0
    for (const auto& key : end_key) {
79
0
        ss << " end_keys=" << key;
80
0
    }
81
82
0
    for (auto& condition : conditions) {
83
0
        ss << " conditions=" << apache::thrift::ThriftDebugString(condition);
84
0
    }
85
86
0
    return ss.str();
87
0
}
88
89
0
std::string TabletReader::KeysParam::to_string() const {
90
0
    std::stringstream ss;
91
0
    ss << "start_key_include=" << start_key_include << " end_key_include=" << end_key_include;
92
93
0
    for (const auto& start_key : start_keys) {
94
0
        ss << " keys=" << start_key.to_string();
95
0
    }
96
0
    for (const auto& end_key : end_keys) {
97
0
        ss << " end_keys=" << end_key.to_string();
98
0
    }
99
100
0
    return ss.str();
101
0
}
102
103
141
void TabletReader::ReadSource::fill_delete_predicates() {
104
141
    DCHECK_EQ(delete_predicates.size(), 0);
105
423
    for (auto&& split : rs_splits) {
106
423
        auto& rs_meta = split.rs_reader->rowset()->rowset_meta();
107
423
        if (rs_meta->has_delete_predicate()) {
108
77
            delete_predicates.push_back(rs_meta);
109
77
        }
110
423
    }
111
141
}
112
113
163
TabletReader::~TabletReader() {
114
163
    for (auto* pred : _col_predicates) {
115
0
        delete pred;
116
0
    }
117
163
    for (auto* pred : _value_col_predicates) {
118
0
        delete pred;
119
0
    }
120
163
}
121
122
163
Status TabletReader::init(const ReaderParams& read_params) {
123
163
    _predicate_arena = std::make_unique<vectorized::Arena>();
124
125
163
    Status res = _init_params(read_params);
126
163
    if (!res.ok()) {
127
0
        LOG(WARNING) << "fail to init reader when init params. res:" << res
128
0
                     << ", tablet_id:" << read_params.tablet->tablet_id()
129
0
                     << ", schema_hash:" << read_params.tablet->schema_hash()
130
0
                     << ", reader type:" << int(read_params.reader_type)
131
0
                     << ", version:" << read_params.version;
132
0
    }
133
163
    return res;
134
163
}
135
136
// When only one rowset has data, and this rowset is nonoverlapping, we can read directly without aggregation
137
bool TabletReader::_optimize_for_single_rowset(
138
0
        const std::vector<RowsetReaderSharedPtr>& rs_readers) {
139
0
    bool has_delete_rowset = false;
140
0
    bool has_overlapping = false;
141
0
    int nonoverlapping_count = 0;
142
0
    for (const auto& rs_reader : rs_readers) {
143
0
        if (rs_reader->rowset()->rowset_meta()->delete_flag()) {
144
0
            has_delete_rowset = true;
145
0
            break;
146
0
        }
147
0
        if (rs_reader->rowset()->rowset_meta()->num_rows() > 0) {
148
0
            if (rs_reader->rowset()->rowset_meta()->is_segments_overlapping()) {
149
                // when there are overlapping segments, can not do directly read
150
0
                has_overlapping = true;
151
0
                break;
152
0
            } else if (++nonoverlapping_count > 1) {
153
0
                break;
154
0
            }
155
0
        }
156
0
    }
157
158
0
    return !has_overlapping && nonoverlapping_count == 1 && !has_delete_rowset;
159
0
}
160
161
141
Status TabletReader::_capture_rs_readers(const ReaderParams& read_params) {
162
141
    if (read_params.rs_splits.empty()) {
163
0
        return Status::InternalError("fail to acquire data sources. tablet={}",
164
0
                                     _tablet->tablet_id());
165
0
    }
166
167
141
    bool eof = false;
168
141
    bool is_lower_key_included = _keys_param.start_key_include;
169
141
    bool is_upper_key_included = _keys_param.end_key_include;
170
171
141
    for (int i = 0; i < _keys_param.start_keys.size(); ++i) {
172
        // lower bound
173
0
        RowCursor& start_key = _keys_param.start_keys[i];
174
0
        RowCursor& end_key = _keys_param.end_keys[i];
175
176
0
        if (!is_lower_key_included) {
177
0
            if (compare_row_key(start_key, end_key) >= 0) {
178
0
                VLOG_NOTICE << "return EOF when lower key not include"
179
0
                            << ", start_key=" << start_key.to_string()
180
0
                            << ", end_key=" << end_key.to_string();
181
0
                eof = true;
182
0
                break;
183
0
            }
184
0
        } else {
185
0
            if (compare_row_key(start_key, end_key) > 0) {
186
0
                VLOG_NOTICE << "return EOF when lower key include="
187
0
                            << ", start_key=" << start_key.to_string()
188
0
                            << ", end_key=" << end_key.to_string();
189
0
                eof = true;
190
0
                break;
191
0
            }
192
0
        }
193
194
0
        _is_lower_keys_included.push_back(is_lower_key_included);
195
0
        _is_upper_keys_included.push_back(is_upper_key_included);
196
0
    }
197
198
141
    if (eof) {
199
0
        return Status::OK();
200
0
    }
201
202
141
    bool need_ordered_result = true;
203
141
    if (read_params.reader_type == ReaderType::READER_QUERY) {
204
0
        if (_tablet_schema->keys_type() == DUP_KEYS) {
205
            // duplicated keys are allowed, no need to merge sort keys in rowset
206
0
            need_ordered_result = false;
207
0
        }
208
0
        if (_tablet_schema->keys_type() == UNIQUE_KEYS &&
209
0
            _tablet->enable_unique_key_merge_on_write()) {
210
            // unique keys with merge on write, no need to merge sort keys in rowset
211
0
            need_ordered_result = false;
212
0
        }
213
0
        if (_aggregation) {
214
            // compute engine will aggregate rows with the same key,
215
            // it's ok for rowset to return unordered result
216
0
            need_ordered_result = false;
217
0
        }
218
219
0
        if (_direct_mode) {
220
            // direct mode indicates that the storage layer does not need to merge,
221
            // it's ok for rowset to return unordered result
222
0
            need_ordered_result = false;
223
0
        }
224
225
0
        if (read_params.read_orderby_key) {
226
0
            need_ordered_result = true;
227
0
        }
228
0
    }
229
230
141
    _reader_context.reader_type = read_params.reader_type;
231
141
    _reader_context.version = read_params.version;
232
141
    _reader_context.tablet_schema = _tablet_schema;
233
141
    _reader_context.need_ordered_result = need_ordered_result;
234
141
    _reader_context.topn_filter_source_node_ids = read_params.topn_filter_source_node_ids;
235
141
    _reader_context.topn_filter_target_node_id = read_params.topn_filter_target_node_id;
236
141
    _reader_context.read_orderby_key_reverse = read_params.read_orderby_key_reverse;
237
141
    _reader_context.read_orderby_key_limit = read_params.read_orderby_key_limit;
238
141
    _reader_context.filter_block_conjuncts = read_params.filter_block_conjuncts;
239
141
    _reader_context.return_columns = &_return_columns;
240
141
    _reader_context.read_orderby_key_columns =
241
141
            !_orderby_key_columns.empty() ? &_orderby_key_columns : nullptr;
242
141
    _reader_context.predicates = &_col_predicates;
243
141
    _reader_context.value_predicates = &_value_col_predicates;
244
141
    _reader_context.lower_bound_keys = &_keys_param.start_keys;
245
141
    _reader_context.is_lower_keys_included = &_is_lower_keys_included;
246
141
    _reader_context.upper_bound_keys = &_keys_param.end_keys;
247
141
    _reader_context.is_upper_keys_included = &_is_upper_keys_included;
248
141
    _reader_context.delete_handler = &_delete_handler;
249
141
    _reader_context.stats = &_stats;
250
141
    _reader_context.use_page_cache = read_params.use_page_cache;
251
141
    _reader_context.sequence_id_idx = _sequence_col_idx;
252
141
    _reader_context.is_unique = tablet()->keys_type() == UNIQUE_KEYS;
253
141
    _reader_context.merged_rows = &_merged_rows;
254
141
    _reader_context.delete_bitmap = read_params.delete_bitmap;
255
141
    _reader_context.enable_unique_key_merge_on_write = tablet()->enable_unique_key_merge_on_write();
256
141
    _reader_context.record_rowids = read_params.record_rowids;
257
141
    _reader_context.rowid_conversion = read_params.rowid_conversion;
258
141
    _reader_context.is_key_column_group = read_params.is_key_column_group;
259
141
    _reader_context.remaining_conjunct_roots = read_params.remaining_conjunct_roots;
260
141
    _reader_context.common_expr_ctxs_push_down = read_params.common_expr_ctxs_push_down;
261
141
    _reader_context.output_columns = &read_params.output_columns;
262
141
    _reader_context.push_down_agg_type_opt = read_params.push_down_agg_type_opt;
263
141
    _reader_context.ttl_seconds = _tablet->ttl_seconds();
264
265
141
    return Status::OK();
266
141
}
267
268
0
TabletColumn TabletReader::materialize_column(const TabletColumn& orig) {
269
0
    if (!orig.is_variant_type()) {
270
0
        return orig;
271
0
    }
272
0
    TabletColumn column_with_cast_type = orig;
273
0
    auto cast_type = _reader_context.target_cast_type_for_variants.at(orig.name());
274
0
    FieldType filed_type = TabletColumn::get_field_type_by_type(cast_type.type);
275
0
    if (filed_type == FieldType::OLAP_FIELD_TYPE_UNKNOWN) {
276
0
        throw doris::Exception(ErrorCode::INTERNAL_ERROR, "Invalid type for variant column: {}",
277
0
                               cast_type.type);
278
0
    }
279
0
    column_with_cast_type.set_type(filed_type);
280
0
    return column_with_cast_type;
281
0
}
282
283
163
Status TabletReader::_init_params(const ReaderParams& read_params) {
284
163
    read_params.check_validation();
285
286
163
    _direct_mode = read_params.direct_mode;
287
163
    _aggregation = read_params.aggregation;
288
163
    _reader_type = read_params.reader_type;
289
163
    _tablet = read_params.tablet;
290
163
    _tablet_schema = read_params.tablet_schema;
291
163
    _reader_context.runtime_state = read_params.runtime_state;
292
163
    _reader_context.target_cast_type_for_variants = read_params.target_cast_type_for_variants;
293
294
163
    RETURN_IF_ERROR(_init_conditions_param(read_params));
295
296
163
    Status res = _init_delete_condition(read_params);
297
163
    if (!res.ok()) {
298
0
        LOG(WARNING) << "fail to init delete param. res = " << res;
299
0
        return res;
300
0
    }
301
302
163
    res = _init_return_columns(read_params);
303
163
    if (!res.ok()) {
304
0
        LOG(WARNING) << "fail to init return columns. res = " << res;
305
0
        return res;
306
0
    }
307
308
163
    res = _init_keys_param(read_params);
309
163
    if (!res.ok()) {
310
0
        LOG(WARNING) << "fail to init keys param. res=" << res;
311
0
        return res;
312
0
    }
313
163
    res = _init_orderby_keys_param(read_params);
314
163
    if (!res.ok()) {
315
0
        LOG(WARNING) << "fail to init orderby keys param. res=" << res;
316
0
        return res;
317
0
    }
318
163
    if (_tablet_schema->has_sequence_col()) {
319
8
        auto sequence_col_idx = _tablet_schema->sequence_col_idx();
320
8
        DCHECK_NE(sequence_col_idx, -1);
321
16
        for (auto col : _return_columns) {
322
            // query has sequence col
323
16
            if (col == sequence_col_idx) {
324
4
                _sequence_col_idx = sequence_col_idx;
325
4
                break;
326
4
            }
327
16
        }
328
8
    }
329
330
163
    return res;
331
163
}
332
333
163
Status TabletReader::_init_return_columns(const ReaderParams& read_params) {
334
163
    if (read_params.reader_type == ReaderType::READER_QUERY) {
335
22
        _return_columns = read_params.return_columns;
336
22
        _tablet_columns_convert_to_null_set = read_params.tablet_columns_convert_to_null_set;
337
37
        for (auto id : read_params.return_columns) {
338
37
            if (_tablet_schema->column(id).is_key()) {
339
22
                _key_cids.push_back(id);
340
22
            } else {
341
15
                _value_cids.push_back(id);
342
15
            }
343
37
        }
344
141
    } else if (read_params.return_columns.empty()) {
345
0
        for (size_t i = 0; i < _tablet_schema->num_columns(); ++i) {
346
0
            _return_columns.push_back(i);
347
0
            if (_tablet_schema->column(i).is_key()) {
348
0
                _key_cids.push_back(i);
349
0
            } else {
350
0
                _value_cids.push_back(i);
351
0
            }
352
0
        }
353
0
        VLOG_NOTICE << "return column is empty, using full column as default.";
354
141
    } else if ((read_params.reader_type == ReaderType::READER_CUMULATIVE_COMPACTION ||
355
141
                read_params.reader_type == ReaderType::READER_SEGMENT_COMPACTION ||
356
141
                read_params.reader_type == ReaderType::READER_BASE_COMPACTION ||
357
141
                read_params.reader_type == ReaderType::READER_FULL_COMPACTION ||
358
141
                read_params.reader_type == ReaderType::READER_COLD_DATA_COMPACTION ||
359
141
                read_params.reader_type == ReaderType::READER_ALTER_TABLE) &&
360
141
               !read_params.return_columns.empty()) {
361
141
        _return_columns = read_params.return_columns;
362
265
        for (auto id : read_params.return_columns) {
363
265
            if (_tablet_schema->column(id).is_key()) {
364
92
                _key_cids.push_back(id);
365
173
            } else {
366
173
                _value_cids.push_back(id);
367
173
            }
368
265
        }
369
141
    } else if (read_params.reader_type == ReaderType::READER_CHECKSUM) {
370
0
        _return_columns = read_params.return_columns;
371
0
        for (auto id : read_params.return_columns) {
372
0
            if (_tablet_schema->column(id).is_key()) {
373
0
                _key_cids.push_back(id);
374
0
            } else {
375
0
                _value_cids.push_back(id);
376
0
            }
377
0
        }
378
0
    } else {
379
0
        return Status::Error<INVALID_ARGUMENT>(
380
0
                "fail to init return columns. reader_type={}, return_columns_size={}",
381
0
                int(read_params.reader_type), read_params.return_columns.size());
382
0
    }
383
384
163
    std::sort(_key_cids.begin(), _key_cids.end(), std::greater<>());
385
386
163
    return Status::OK();
387
163
}
388
389
163
Status TabletReader::_init_keys_param(const ReaderParams& read_params) {
390
163
    if (read_params.start_key.empty()) {
391
163
        return Status::OK();
392
163
    }
393
394
0
    _keys_param.start_key_include = read_params.start_key_include;
395
0
    _keys_param.end_key_include = read_params.end_key_include;
396
397
0
    size_t start_key_size = read_params.start_key.size();
398
    //_keys_param.start_keys.resize(start_key_size);
399
0
    std::vector<RowCursor>(start_key_size).swap(_keys_param.start_keys);
400
401
0
    size_t scan_key_size = read_params.start_key.front().size();
402
0
    if (scan_key_size > _tablet_schema->num_columns()) {
403
0
        return Status::Error<INVALID_ARGUMENT>(
404
0
                "Input param are invalid. Column count is bigger than num_columns of schema. "
405
0
                "column_count={}, schema.num_columns={}",
406
0
                scan_key_size, _tablet_schema->num_columns());
407
0
    }
408
409
0
    std::vector<uint32_t> columns(scan_key_size);
410
0
    std::iota(columns.begin(), columns.end(), 0);
411
412
0
    std::shared_ptr<Schema> schema = std::make_shared<Schema>(_tablet_schema->columns(), columns);
413
414
0
    for (size_t i = 0; i < start_key_size; ++i) {
415
0
        if (read_params.start_key[i].size() != scan_key_size) {
416
0
            return Status::Error<INVALID_ARGUMENT>(
417
0
                    "The start_key.at({}).size={}, not equals the scan_key_size={}", i,
418
0
                    read_params.start_key[i].size(), scan_key_size);
419
0
        }
420
421
0
        Status res = _keys_param.start_keys[i].init_scan_key(
422
0
                _tablet_schema, read_params.start_key[i].values(), schema);
423
0
        if (!res.ok()) {
424
0
            LOG(WARNING) << "fail to init row cursor. res = " << res;
425
0
            return res;
426
0
        }
427
0
        res = _keys_param.start_keys[i].from_tuple(read_params.start_key[i]);
428
0
        if (!res.ok()) {
429
0
            LOG(WARNING) << "fail to init row cursor from Keys. res=" << res << "key_index=" << i;
430
0
            return res;
431
0
        }
432
0
    }
433
434
0
    size_t end_key_size = read_params.end_key.size();
435
    //_keys_param.end_keys.resize(end_key_size);
436
0
    std::vector<RowCursor>(end_key_size).swap(_keys_param.end_keys);
437
0
    for (size_t i = 0; i < end_key_size; ++i) {
438
0
        if (read_params.end_key[i].size() != scan_key_size) {
439
0
            return Status::Error<INVALID_ARGUMENT>(
440
0
                    "The end_key.at({}).size={}, not equals the scan_key_size={}", i,
441
0
                    read_params.end_key[i].size(), scan_key_size);
442
0
        }
443
444
0
        Status res = _keys_param.end_keys[i].init_scan_key(_tablet_schema,
445
0
                                                           read_params.end_key[i].values(), schema);
446
0
        if (!res.ok()) {
447
0
            LOG(WARNING) << "fail to init row cursor. res = " << res;
448
0
            return res;
449
0
        }
450
451
0
        res = _keys_param.end_keys[i].from_tuple(read_params.end_key[i]);
452
0
        if (!res.ok()) {
453
0
            LOG(WARNING) << "fail to init row cursor from Keys. res=" << res << " key_index=" << i;
454
0
            return res;
455
0
        }
456
0
    }
457
458
    //TODO:check the valid of start_key and end_key.(eg. start_key <= end_key)
459
460
0
    return Status::OK();
461
0
}
462
463
163
Status TabletReader::_init_orderby_keys_param(const ReaderParams& read_params) {
464
    // UNIQUE_KEYS will compare all keys as before
465
163
    if (_tablet_schema->keys_type() == DUP_KEYS || (_tablet_schema->keys_type() == UNIQUE_KEYS &&
466
116
                                                    _tablet->enable_unique_key_merge_on_write())) {
467
        // find index in vector _return_columns
468
        //   for the read_orderby_key_num_prefix_columns orderby keys
469
103
        for (uint32_t i = 0; i < read_params.read_orderby_key_num_prefix_columns; i++) {
470
0
            for (uint32_t idx = 0; idx < _return_columns.size(); idx++) {
471
0
                if (_return_columns[idx] == i) {
472
0
                    _orderby_key_columns.push_back(idx);
473
0
                    break;
474
0
                }
475
0
            }
476
0
        }
477
103
        if (read_params.read_orderby_key_num_prefix_columns != _orderby_key_columns.size()) {
478
0
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
479
0
                    "read_orderby_key_num_prefix_columns != _orderby_key_columns.size, "
480
0
                    "read_params.read_orderby_key_num_prefix_columns={}, "
481
0
                    "_orderby_key_columns.size()={}",
482
0
                    read_params.read_orderby_key_num_prefix_columns, _orderby_key_columns.size());
483
0
        }
484
103
    }
485
486
163
    return Status::OK();
487
163
}
488
489
163
Status TabletReader::_init_conditions_param(const ReaderParams& read_params) {
490
163
    std::vector<ColumnPredicate*> predicates;
491
163
    for (const auto& condition : read_params.conditions) {
492
0
        TCondition tmp_cond = condition;
493
0
        RETURN_IF_ERROR(_tablet_schema->have_column(tmp_cond.column_name));
494
        // The "column" parameter might represent a column resulting from the decomposition of a variant column.
495
        // Instead of using a "unique_id" for identification, we are utilizing a "path" to denote this column.
496
0
        const auto& column = *DORIS_TRY(_tablet_schema->column(tmp_cond.column_name));
497
0
        const auto& mcolumn = materialize_column(column);
498
0
        uint32_t index = _tablet_schema->field_index(tmp_cond.column_name);
499
0
        ColumnPredicate* predicate =
500
0
                parse_to_predicate(mcolumn, index, tmp_cond, _predicate_arena.get());
501
        // record condition value into predicate_params in order to pushdown segment_iterator,
502
        // _gen_predicate_result_sign will build predicate result unique sign with condition value
503
0
        auto predicate_params = predicate->predicate_params();
504
0
        predicate_params->values = condition.condition_values;
505
0
        predicate_params->marked_by_runtime_filter = condition.marked_by_runtime_filter;
506
0
        predicates.push_back(predicate);
507
0
    }
508
509
    // Only key column bloom filter will push down to storage engine
510
163
    for (const auto& filter : read_params.bloom_filters) {
511
0
        ColumnPredicate* predicate = _parse_to_predicate(filter);
512
0
        predicate->predicate_params()->marked_by_runtime_filter = true;
513
0
        predicates.emplace_back(predicate);
514
0
    }
515
516
163
    for (const auto& filter : read_params.bitmap_filters) {
517
0
        ColumnPredicate* predicate = _parse_to_predicate(filter);
518
0
        predicate->predicate_params()->marked_by_runtime_filter = true;
519
0
        predicates.emplace_back(predicate);
520
0
    }
521
522
163
    for (const auto& filter : read_params.in_filters) {
523
0
        ColumnPredicate* predicate = _parse_to_predicate(filter);
524
0
        predicate->predicate_params()->marked_by_runtime_filter = true;
525
0
        predicates.emplace_back(predicate);
526
0
    }
527
528
    // Function filter push down to storage engine
529
163
    auto is_like_predicate = [](ColumnPredicate* _pred) {
530
0
        return dynamic_cast<LikeColumnPredicate<TYPE_CHAR>*>(_pred) != nullptr ||
531
0
               dynamic_cast<LikeColumnPredicate<TYPE_STRING>*>(_pred) != nullptr;
532
0
    };
533
534
163
    for (const auto& filter : read_params.function_filters) {
535
0
        predicates.emplace_back(_parse_to_predicate(filter));
536
0
        auto* pred = predicates.back();
537
538
0
        const auto& col = _tablet_schema->column(pred->column_id());
539
0
        const auto* tablet_index = _tablet_schema->get_ngram_bf_index(col.unique_id());
540
0
        if (is_like_predicate(pred) && tablet_index && config::enable_query_like_bloom_filter) {
541
0
            std::unique_ptr<segment_v2::BloomFilter> ng_bf;
542
0
            std::string pattern = pred->get_search_str();
543
0
            auto gram_bf_size = tablet_index->get_gram_bf_size();
544
0
            auto gram_size = tablet_index->get_gram_size();
545
546
0
            RETURN_IF_ERROR(segment_v2::BloomFilter::create(segment_v2::NGRAM_BLOOM_FILTER, &ng_bf,
547
0
                                                            gram_bf_size));
548
0
            NgramTokenExtractor _token_extractor(gram_size);
549
550
0
            if (_token_extractor.string_like_to_bloom_filter(pattern.data(), pattern.length(),
551
0
                                                             *ng_bf)) {
552
0
                pred->set_page_ng_bf(std::move(ng_bf));
553
0
            }
554
0
        }
555
0
    }
556
557
163
    for (auto* predicate : predicates) {
558
0
        auto column = _tablet_schema->column(predicate->column_id());
559
0
        if (column.aggregation() != FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE) {
560
0
            _value_col_predicates.push_back(predicate);
561
0
        } else {
562
0
            _col_predicates.push_back(predicate);
563
0
        }
564
0
    }
565
566
163
    for (int id : read_params.topn_filter_source_node_ids) {
567
0
        auto& runtime_predicate =
568
0
                read_params.runtime_state->get_query_ctx()->get_runtime_predicate(id);
569
0
        RETURN_IF_ERROR(runtime_predicate.set_tablet_schema(read_params.topn_filter_target_node_id,
570
0
                                                            _tablet_schema));
571
0
    }
572
163
    return Status::OK();
573
163
}
574
575
ColumnPredicate* TabletReader::_parse_to_predicate(
576
0
        const std::pair<std::string, std::shared_ptr<BloomFilterFuncBase>>& bloom_filter) {
577
0
    int32_t index = _tablet_schema->field_index(bloom_filter.first);
578
0
    if (index < 0) {
579
0
        return nullptr;
580
0
    }
581
0
    const TabletColumn& column = materialize_column(_tablet_schema->column(index));
582
0
    return create_column_predicate(index, bloom_filter.second, column.type(),
583
0
                                   _reader_context.runtime_state->be_exec_version(), &column);
584
0
}
585
586
ColumnPredicate* TabletReader::_parse_to_predicate(
587
0
        const std::pair<std::string, std::shared_ptr<HybridSetBase>>& in_filter) {
588
0
    int32_t index = _tablet_schema->field_index(in_filter.first);
589
0
    if (index < 0) {
590
0
        return nullptr;
591
0
    }
592
0
    const TabletColumn& column = materialize_column(_tablet_schema->column(index));
593
0
    return create_column_predicate(index, in_filter.second, column.type(),
594
0
                                   _reader_context.runtime_state->be_exec_version(), &column);
595
0
}
596
597
ColumnPredicate* TabletReader::_parse_to_predicate(
598
0
        const std::pair<std::string, std::shared_ptr<BitmapFilterFuncBase>>& bitmap_filter) {
599
0
    int32_t index = _tablet_schema->field_index(bitmap_filter.first);
600
0
    if (index < 0) {
601
0
        return nullptr;
602
0
    }
603
0
    const TabletColumn& column = materialize_column(_tablet_schema->column(index));
604
0
    return create_column_predicate(index, bitmap_filter.second, column.type(),
605
0
                                   _reader_context.runtime_state->be_exec_version(), &column);
606
0
}
607
608
0
ColumnPredicate* TabletReader::_parse_to_predicate(const FunctionFilter& function_filter) {
609
0
    int32_t index = _tablet_schema->field_index(function_filter._col_name);
610
0
    if (index < 0) {
611
0
        return nullptr;
612
0
    }
613
0
    const TabletColumn& column = materialize_column(_tablet_schema->column(index));
614
0
    return create_column_predicate(index, std::make_shared<FunctionFilter>(function_filter),
615
0
                                   column.type(), _reader_context.runtime_state->be_exec_version(),
616
0
                                   &column);
617
0
}
618
619
163
Status TabletReader::_init_delete_condition(const ReaderParams& read_params) {
620
    // If it's cumu and not allow do delete when cumu
621
163
    if (read_params.reader_type == ReaderType::READER_SEGMENT_COMPACTION ||
622
163
        (read_params.reader_type == ReaderType::READER_CUMULATIVE_COMPACTION &&
623
163
         !config::enable_delete_when_cumu_compaction)) {
624
1
        return Status::OK();
625
1
    }
626
162
    bool cumu_delete = read_params.reader_type == ReaderType::READER_CUMULATIVE_COMPACTION &&
627
162
                       config::enable_delete_when_cumu_compaction;
628
    // Delete sign could not be applied when delete on cumu compaction is enabled, bucause it is meant for delete with predicates.
629
    // If delete design is applied on cumu compaction, it will lose effect when doing base compaction.
630
    // `_delete_sign_available` indicates the condition where we could apply delete signs to data.
631
162
    _delete_sign_available = (read_params.reader_type == ReaderType::READER_BASE_COMPACTION ||
632
162
                              read_params.reader_type == ReaderType::READER_COLD_DATA_COMPACTION ||
633
162
                              read_params.reader_type == ReaderType::READER_CHECKSUM);
634
635
    // `_filter_delete` indicates the condition where we should execlude deleted tuples when reading data.
636
    // However, queries will not use this condition but generate special where predicates to filter data.
637
    // (Though a lille bit confused, it is how the current logic working...)
638
162
    _filter_delete = _delete_sign_available || cumu_delete;
639
162
    return _delete_handler.init(_tablet_schema, read_params.delete_predicates,
640
162
                                read_params.version.second);
641
163
}
642
643
Status TabletReader::init_reader_params_and_create_block(
644
        TabletSharedPtr tablet, ReaderType reader_type,
645
        const std::vector<RowsetSharedPtr>& input_rowsets,
646
0
        TabletReader::ReaderParams* reader_params, vectorized::Block* block) {
647
0
    reader_params->tablet = tablet;
648
0
    reader_params->reader_type = reader_type;
649
0
    reader_params->version =
650
0
            Version(input_rowsets.front()->start_version(), input_rowsets.back()->end_version());
651
652
0
    ReadSource read_source;
653
0
    for (const auto& rowset : input_rowsets) {
654
0
        RowsetReaderSharedPtr rs_reader;
655
0
        RETURN_IF_ERROR(rowset->create_reader(&rs_reader));
656
0
        read_source.rs_splits.emplace_back(std::move(rs_reader));
657
0
    }
658
0
    read_source.fill_delete_predicates();
659
0
    reader_params->set_read_source(std::move(read_source));
660
661
0
    std::vector<RowsetMetaSharedPtr> rowset_metas(input_rowsets.size());
662
0
    std::transform(input_rowsets.begin(), input_rowsets.end(), rowset_metas.begin(),
663
0
                   [](const RowsetSharedPtr& rowset) { return rowset->rowset_meta(); });
664
0
    TabletSchemaSPtr read_tablet_schema =
665
0
            tablet->tablet_schema_with_merged_max_schema_version(rowset_metas);
666
0
    TabletSchemaSPtr merge_tablet_schema = std::make_shared<TabletSchema>();
667
0
    merge_tablet_schema->copy_from(*read_tablet_schema);
668
669
    // Merge the columns in delete predicate that not in latest schema in to current tablet schema
670
0
    for (auto& del_pred : reader_params->delete_predicates) {
671
0
        merge_tablet_schema->merge_dropped_columns(*del_pred->tablet_schema());
672
0
    }
673
0
    reader_params->tablet_schema = merge_tablet_schema;
674
0
    if (tablet->enable_unique_key_merge_on_write()) {
675
0
        reader_params->delete_bitmap = &tablet->tablet_meta()->delete_bitmap();
676
0
    }
677
678
0
    reader_params->return_columns.resize(read_tablet_schema->num_columns());
679
0
    std::iota(reader_params->return_columns.begin(), reader_params->return_columns.end(), 0);
680
0
    reader_params->origin_return_columns = &reader_params->return_columns;
681
682
0
    *block = read_tablet_schema->create_block();
683
684
0
    return Status::OK();
685
0
}
686
687
} // namespace doris