Coverage Report

Created: 2026-07-15 15:08

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/segment/segment_iterator.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/segment/segment_iterator.h"
19
20
#include <gen_cpp/Exprs_types.h>
21
#include <gen_cpp/Opcodes_types.h>
22
#include <gen_cpp/Types_types.h>
23
#include <gen_cpp/olap_file.pb.h>
24
#include <glog/logging.h>
25
26
#include <algorithm>
27
#include <boost/iterator/iterator_facade.hpp>
28
#include <cassert>
29
#include <cstdint>
30
#include <memory>
31
#include <numeric>
32
#include <optional>
33
#include <set>
34
#include <unordered_map>
35
#include <utility>
36
#include <vector>
37
38
#include "cloud/config.h"
39
#include "common/compiler_util.h" // IWYU pragma: keep
40
#include "common/config.h"
41
#include "common/consts.h"
42
#include "common/exception.h"
43
#include "common/logging.h"
44
#include "common/metrics/doris_metrics.h"
45
#include "common/object_pool.h"
46
#include "common/status.h"
47
#include "core/assert_cast.h"
48
#include "core/block/column_with_type_and_name.h"
49
#include "core/column/column.h"
50
#include "core/column/column_const.h"
51
#include "core/column/column_nothing.h"
52
#include "core/column/column_nullable.h"
53
#include "core/column/column_string.h"
54
#include "core/column/column_variant.h"
55
#include "core/column/column_vector.h"
56
#include "core/data_type/data_type.h"
57
#include "core/data_type/data_type_factory.hpp"
58
#include "core/data_type/data_type_number.h"
59
#include "core/data_type/define_primitive_type.h"
60
#include "core/field.h"
61
#include "core/string_ref.h"
62
#include "core/typeid_cast.h"
63
#include "core/types.h"
64
#include "exprs/expr_zonemap_filter.h"
65
#include "exprs/function/array/function_array_index.h"
66
#include "exprs/runtime_filter_expr.h"
67
#include "exprs/vexpr.h"
68
#include "exprs/vexpr_context.h"
69
#include "exprs/virtual_slot_ref.h"
70
#include "exprs/vliteral.h"
71
#include "exprs/vslot_ref.h"
72
#include "io/cache/cached_remote_file_reader.h"
73
#include "io/fs/file_reader.h"
74
#include "io/io_common.h"
75
#include "runtime/query_context.h"
76
#include "runtime/runtime_predicate.h"
77
#include "runtime/runtime_state.h"
78
#include "runtime/thread_context.h"
79
#include "storage/binlog.h"
80
#include "storage/compaction/collection_similarity.h"
81
#include "storage/id_manager.h"
82
#include "storage/index/ann/ann_index.h"
83
#include "storage/index/ann/ann_index_iterator.h"
84
#include "storage/index/ann/ann_index_reader.h"
85
#include "storage/index/ann/ann_topn_runtime.h"
86
#include "storage/index/index_file_reader.h"
87
#include "storage/index/index_iterator.h"
88
#include "storage/index/index_query_context.h"
89
#include "storage/index/index_reader_helper.h"
90
#include "storage/index/indexed_column_reader.h"
91
#include "storage/index/inverted/inverted_index_reader.h"
92
#include "storage/index/ordinal_page_index.h"
93
#include "storage/index/primary_key_index.h"
94
#include "storage/index/short_key_index.h"
95
#include "storage/index/zone_map/zone_map_index.h"
96
#include "storage/index/zone_map/zonemap_eval_context.h"
97
#include "storage/iterators.h"
98
#include "storage/olap_common.h"
99
#include "storage/predicate/bloom_filter_predicate.h"
100
#include "storage/predicate/column_predicate.h"
101
#include "storage/predicate/like_column_predicate.h"
102
#include "storage/schema.h"
103
#include "storage/segment/column_reader.h"
104
#include "storage/segment/column_reader_cache.h"
105
#include "storage/segment/condition_cache.h"
106
#include "storage/segment/row_ranges.h"
107
#include "storage/segment/segment.h"
108
#include "storage/segment/segment_prefetcher.h"
109
#include "storage/segment/variant/variant_column_reader.h"
110
#include "storage/segment/virtual_column_iterator.h"
111
#include "storage/tablet/tablet_schema.h"
112
#include "storage/types.h"
113
#include "storage/utils.h"
114
#include "util/concurrency_stats.h"
115
#include "util/defer_op.h"
116
#include "util/simd/bits.h"
117
118
namespace doris {
119
using namespace ErrorCode;
120
namespace segment_v2 {
121
122
class ScopedColumnIteratorReadPhase {
123
public:
124
    ScopedColumnIteratorReadPhase(ColumnIterator* column_iter, ColumnIterator::ReadPhase mode)
125
28.0k
            : _column_iter(column_iter) {
126
28.0k
        DORIS_CHECK(_column_iter != nullptr);
127
28.0k
        _column_iter->set_read_phase(mode);
128
28.0k
    }
129
130
    ScopedColumnIteratorReadPhase(const ScopedColumnIteratorReadPhase&) = delete;
131
    ScopedColumnIteratorReadPhase& operator=(const ScopedColumnIteratorReadPhase&) = delete;
132
133
28.0k
    ~ScopedColumnIteratorReadPhase() {
134
        // ReadPhase is a per-read phase knob. SegmentIterator only needs a
135
        // temporary PREDICATE/LAZY mode while reading one column in one phase; it
136
        // must be restored before the next column or later normal reads reuse the
137
        // same ColumnIterator. Keep the restoration in one scoped helper instead
138
        // of open-coding the same Defer block at every call site.
139
28.0k
        _column_iter->set_read_phase(ColumnIterator::ReadPhase::NORMAL);
140
28.0k
    }
141
142
private:
143
    ColumnIterator* _column_iter = nullptr;
144
};
145
146
3.04k
SegmentIterator::~SegmentIterator() = default;
147
148
3.03k
void SegmentIterator::_init_row_bitmap_by_condition_cache() {
149
    // Only dispose need column predicate and expr cal in condition cache
150
3.03k
    if (!_col_predicates.empty() || !_common_expr_ctxs_push_down.empty()) {
151
84
        if (_opts.condition_cache_digest) {
152
0
            auto* condition_cache = ConditionCache::instance();
153
0
            ConditionCache::CacheKey cache_key(_opts.rowset_id, _segment->id(),
154
0
                                               _opts.condition_cache_digest);
155
156
            // Increment search count when digest != 0
157
0
            DorisMetrics::instance()->condition_cache_search_count->increment(1);
158
159
0
            ConditionCacheHandle handle;
160
0
            _find_condition_cache = condition_cache->lookup(cache_key, &handle);
161
162
            // Increment hit count if cache lookup is successful
163
0
            if (_find_condition_cache) {
164
0
                DorisMetrics::instance()->condition_cache_hit_count->increment(1);
165
0
                if (_opts.runtime_state) {
166
0
                    VLOG_DEBUG << "Condition cache hit, query id: "
167
0
                               << print_id(_opts.runtime_state->query_id())
168
0
                               << ", segment id: " << _segment->id()
169
0
                               << ", cache digest: " << _opts.condition_cache_digest
170
0
                               << ", rowset id: " << _opts.rowset_id.to_string();
171
0
                }
172
0
            }
173
174
0
            auto num_rows = _segment->num_rows();
175
0
            if (_find_condition_cache) {
176
0
                const auto& filter_result = *(handle.get_filter_result());
177
0
                int64_t filtered_blocks = 0;
178
0
                for (int i = 0; i < filter_result.size(); i++) {
179
0
                    if (!filter_result[i]) {
180
0
                        _row_bitmap.removeRange(
181
0
                                i * CONDITION_CACHE_OFFSET,
182
0
                                i * CONDITION_CACHE_OFFSET + CONDITION_CACHE_OFFSET);
183
0
                        filtered_blocks++;
184
0
                    }
185
0
                }
186
                // Record condition_cache hit segment number
187
0
                _opts.stats->condition_cache_hit_seg_nums++;
188
                // Record rows filtered by condition cache hit
189
0
                _opts.stats->condition_cache_filtered_rows +=
190
0
                        filtered_blocks * SegmentIterator::CONDITION_CACHE_OFFSET;
191
0
            } else {
192
0
                _condition_cache = std::make_shared<std::vector<bool>>(
193
0
                        num_rows / CONDITION_CACHE_OFFSET + 1, false);
194
0
            }
195
0
        }
196
2.94k
    } else {
197
2.94k
        _opts.condition_cache_digest = 0;
198
2.94k
    }
199
3.03k
}
200
201
// A fast range iterator for roaring bitmap. Output ranges use closed-open form, like [from, to).
202
// Example:
203
//   input bitmap:  [0 1 4 5 6 7 10 15 16 17 18 19]
204
//   output ranges: [0,2), [4,8), [10,11), [15,20) (when max_range_size=10)
205
//   output ranges: [0,2), [4,7), [7,8), [10,11), [15,18), [18,20) (when max_range_size=3)
206
class SegmentIterator::BitmapRangeIterator {
207
public:
208
0
    BitmapRangeIterator() = default;
209
3.03k
    virtual ~BitmapRangeIterator() = default;
210
211
3.03k
    explicit BitmapRangeIterator(const roaring::Roaring& bitmap) {
212
3.03k
        roaring_init_iterator(&bitmap.roaring, &_iter);
213
3.03k
    }
214
215
0
    bool has_more_range() const { return !_eof; }
216
217
7.03k
    [[nodiscard]] static uint32_t get_batch_size() { return kBatchSize; }
218
219
    // read next range into [*from, *to) whose size <= max_range_size.
220
    // return false when there is no more range.
221
0
    virtual bool next_range(const uint32_t max_range_size, uint32_t* from, uint32_t* to) {
222
0
        if (_eof) {
223
0
            return false;
224
0
        }
225
226
0
        *from = _buf[_buf_pos];
227
0
        uint32_t range_size = 0;
228
0
        uint32_t expect_val = _buf[_buf_pos]; // this initial value just make first batch valid
229
230
        // if array is contiguous sequence then the following conditions need to be met :
231
        // a_0: x
232
        // a_1: x+1
233
        // a_2: x+2
234
        // ...
235
        // a_p: x+p
236
        // so we can just use (a_p-a_0)-p to check conditions
237
        // and should notice the previous batch needs to be continuous with the current batch
238
0
        while (!_eof && range_size + _buf_size - _buf_pos <= max_range_size &&
239
0
               expect_val == _buf[_buf_pos] &&
240
0
               _buf[_buf_size - 1] - _buf[_buf_pos] == _buf_size - 1 - _buf_pos) {
241
0
            range_size += _buf_size - _buf_pos;
242
0
            expect_val = _buf[_buf_size - 1] + 1;
243
0
            _read_next_batch();
244
0
        }
245
246
        // promise remain range not will reach next batch
247
0
        if (!_eof && range_size < max_range_size && expect_val == _buf[_buf_pos]) {
248
0
            do {
249
0
                _buf_pos++;
250
0
                range_size++;
251
0
            } while (range_size < max_range_size && _buf[_buf_pos] == _buf[_buf_pos - 1] + 1);
252
0
        }
253
0
        *to = *from + range_size;
254
0
        return true;
255
0
    }
256
257
    // read batch_size of rowids from roaring bitmap into buf array
258
13.3k
    virtual uint32_t read_batch_rowids(rowid_t* buf, uint32_t batch_size) {
259
13.3k
        return roaring::api::roaring_read_uint32_iterator(&_iter, buf, batch_size);
260
13.3k
    }
261
262
private:
263
0
    void _read_next_batch() {
264
0
        _buf_pos = 0;
265
0
        _buf_size = roaring::api::roaring_read_uint32_iterator(&_iter, _buf, kBatchSize);
266
0
        _eof = (_buf_size == 0);
267
0
    }
268
269
    static const uint32_t kBatchSize = 256;
270
    roaring::api::roaring_uint32_iterator_t _iter;
271
    uint32_t _buf[kBatchSize];
272
    uint32_t _buf_pos = 0;
273
    uint32_t _buf_size = 0;
274
    bool _eof = false;
275
};
276
277
// A backward range iterator for roaring bitmap. Output ranges use closed-open form, like [from, to).
278
// Example:
279
//   input bitmap:  [0 1 4 5 6 7 10 15 16 17 18 19]
280
//   output ranges: , [15,20), [10,11), [4,8), [0,2) (when max_range_size=10)
281
//   output ranges: [17,20), [15,17), [10,11), [5,8), [4, 5), [0,2) (when max_range_size=3)
282
class SegmentIterator::BackwardBitmapRangeIterator : public SegmentIterator::BitmapRangeIterator {
283
public:
284
0
    explicit BackwardBitmapRangeIterator(const roaring::Roaring& bitmap) {
285
0
        roaring_init_iterator_last(&bitmap.roaring, &_riter);
286
0
        _rowid_count = cast_set<uint32_t>(roaring_bitmap_get_cardinality(&bitmap.roaring));
287
0
        _rowid_left = _rowid_count;
288
0
    }
289
290
0
    bool has_more_range() const { return !_riter.has_value; }
291
292
    // read next range into [*from, *to) whose size <= max_range_size.
293
    // return false when there is no more range.
294
0
    bool next_range(const uint32_t max_range_size, uint32_t* from, uint32_t* to) override {
295
0
        if (!_riter.has_value) {
296
0
            return false;
297
0
        }
298
299
0
        uint32_t range_size = 0;
300
0
        *to = _riter.current_value + 1;
301
302
0
        do {
303
0
            *from = _riter.current_value;
304
0
            range_size++;
305
0
            roaring_previous_uint32_iterator(&_riter);
306
0
        } while (range_size < max_range_size && _riter.has_value &&
307
0
                 _riter.current_value + 1 == *from);
308
309
0
        return true;
310
0
    }
311
    /**
312
     * Reads a batch of row IDs from a roaring bitmap, starting from the end and moving backwards.
313
     * This function retrieves the last `batch_size` row IDs from the bitmap and stores them in the provided buffer.
314
     * It updates the internal state to track how many row IDs are left to read in subsequent calls.
315
     *
316
     * The row IDs are read in reverse order, but stored in the buffer maintaining their original order in the bitmap.
317
     *
318
     * Example:
319
     *   input bitmap: [0 1 4 5 6 7 10 15 16 17 18 19]
320
     *   If the bitmap has 12 elements and batch_size is set to 5, the function will first read [15, 16, 17, 18, 19]
321
     *   into the buffer, leaving 7 elements left. In the next call with batch_size 5, it will read [4, 5, 6, 7, 10].
322
     *
323
     */
324
0
    uint32_t read_batch_rowids(rowid_t* buf, uint32_t batch_size) override {
325
0
        if (!_riter.has_value || _rowid_left == 0) {
326
0
            return 0;
327
0
        }
328
329
0
        if (_rowid_count <= batch_size) {
330
0
            roaring_bitmap_to_uint32_array(_riter.parent,
331
0
                                           buf); // Fill 'buf' with '_rowid_count' elements.
332
0
            uint32_t num_read = _rowid_left;     // Save the number of row IDs read.
333
0
            _rowid_left = 0;                     // No row IDs left after this operation.
334
0
            return num_read;                     // Return the number of row IDs read.
335
0
        }
336
337
0
        uint32_t read_size = std::min(batch_size, _rowid_left);
338
0
        uint32_t num_read = 0; // Counter for the number of row IDs read.
339
340
        // Read row IDs into the buffer in reverse order.
341
0
        while (num_read < read_size && _riter.has_value) {
342
0
            buf[read_size - num_read - 1] = _riter.current_value;
343
0
            num_read++;
344
0
            _rowid_left--; // Decrement the count of remaining row IDs.
345
0
            roaring_previous_uint32_iterator(&_riter);
346
0
        }
347
348
        // Return the actual number of row IDs read.
349
0
        return num_read;
350
0
    }
351
352
private:
353
    roaring::api::roaring_uint32_iterator_t _riter;
354
    uint32_t _rowid_count;
355
    uint32_t _rowid_left;
356
};
357
358
SegmentIterator::SegmentIterator(std::shared_ptr<Segment> segment, SchemaSPtr schema)
359
3.04k
        : _segment(std::move(segment)),
360
3.04k
          _schema(schema),
361
3.04k
          _column_iterators(_schema->num_columns()),
362
3.04k
          _index_iterators(_schema->num_columns()),
363
3.04k
          _cur_rowid(0),
364
3.04k
          _lazy_materialization_read(false),
365
3.04k
          _lazy_inited(false),
366
3.04k
          _inited(false),
367
3.04k
          _pool(new ObjectPool) {}
368
369
5.92k
Status SegmentIterator::init(const StorageReadOptions& opts) {
370
5.92k
    auto status = _init_impl(opts);
371
5.92k
    if (!status.ok()) {
372
0
        _segment->update_healthy_status(status);
373
0
    }
374
5.92k
    return status;
375
5.92k
}
376
377
3.03k
std::unique_ptr<AdaptiveBlockSizePredictor> SegmentIterator::_make_block_size_predictor() const {
378
3.03k
    if (!config::enable_adaptive_batch_size || _opts.preferred_block_size_bytes == 0) {
379
0
        return nullptr;
380
0
    }
381
382
    // Collect per-column raw byte metadata from the segment footer for the columns
383
    // this iterator will actually output (defined by _schema, which is built from
384
    // _opts.return_columns).
385
3.03k
    std::vector<AdaptiveBlockSizePredictor::ColumnMetadata> col_metadata;
386
3.03k
    uint32_t seg_rows = _segment->num_rows();
387
3.03k
    uint64_t total_raw_bytes = 0;
388
3.03k
    double metadata_hint_bytes_per_row = 0.0;
389
3.03k
    if (seg_rows > 0) {
390
3.03k
        const auto& ts = _segment->tablet_schema();
391
3.03k
        if (ts) {
392
7.50k
            for (ColumnId cid : _schema->column_ids()) {
393
7.50k
                if (static_cast<size_t>(cid) < ts->num_columns()) {
394
7.37k
                    int32_t uid = ts->column(cid).unique_id();
395
7.37k
                    uint64_t raw_bytes = _segment->column_raw_data_bytes(uid);
396
7.37k
                    if (uid >= 0 && raw_bytes > 0) {
397
7.12k
                        total_raw_bytes += raw_bytes;
398
7.12k
                    }
399
7.37k
                }
400
7.50k
            }
401
3.03k
            metadata_hint_bytes_per_row = total_raw_bytes / static_cast<double>(seg_rows);
402
3.03k
        }
403
3.03k
    }
404
405
3.03k
    return std::make_unique<AdaptiveBlockSizePredictor>(
406
3.03k
            _opts.preferred_block_size_bytes, metadata_hint_bytes_per_row,
407
3.03k
            AdaptiveBlockSizePredictor::kDefaultProbeRows, _opts.block_row_max);
408
3.03k
}
409
410
5.92k
Status SegmentIterator::_init_impl(const StorageReadOptions& opts) {
411
    // get file handle from file descriptor of segment
412
5.92k
    if (_inited) {
413
2.89k
        return Status::OK();
414
2.89k
    }
415
3.03k
    _opts = opts;
416
3.03k
    SCOPED_RAW_TIMER(&_opts.stats->segment_iterator_init_timer_ns);
417
3.03k
    _inited = true;
418
3.03k
    _file_reader = _segment->_file_reader;
419
3.03k
    _col_predicates.clear();
420
421
3.03k
    for (const auto& predicate : opts.column_predicates) {
422
64
        if (!_segment->can_apply_predicate_safely(predicate->column_id(), *_schema,
423
64
                                                  _opts.target_cast_type_for_variants, _opts)) {
424
0
            continue;
425
0
        }
426
64
        _col_predicates.emplace_back(predicate);
427
64
    }
428
3.03k
    _tablet_id = opts.tablet_id;
429
    // Read options will not change, so that just resize here
430
3.03k
    _block_rowids.resize(_opts.block_row_max);
431
432
    // Adaptive batch size: snapshot the initial row limit and create predictor if enabled.
433
3.03k
    _initial_block_row_max = _opts.block_row_max;
434
3.03k
    _block_size_predictor = _make_block_size_predictor();
435
436
3.03k
    if (_schema->rowid_col_idx() > 0) {
437
0
        _record_rowids = true;
438
0
    }
439
440
3.03k
    _virtual_column_exprs = _opts.virtual_column_exprs;
441
3.03k
    _score_runtime = _opts.score_runtime;
442
3.03k
    _ann_topn_runtime = _opts.ann_topn_runtime;
443
444
3.03k
    _enable_prune_nested_column = _opts.io_ctx.reader_type == ReaderType::READER_QUERY &&
445
3.03k
                                  _opts.runtime_state &&
446
3.03k
                                  _opts.runtime_state->enable_prune_nested_column();
447
448
3.03k
    if (opts.output_columns != nullptr) {
449
1.40k
        _output_columns = *(opts.output_columns);
450
1.40k
    }
451
452
3.03k
    _storage_name_and_type.resize(_schema->columns().size());
453
3.03k
    auto storage_format = _opts.tablet_schema->get_inverted_index_storage_format();
454
26.8k
    for (int i = 0; i < _schema->columns().size(); ++i) {
455
23.8k
        const TabletColumn* col = _schema->column(i);
456
23.8k
        if (col) {
457
7.50k
            auto storage_type = _segment->get_data_type_of(*col, _opts);
458
7.50k
            if (storage_type == nullptr) {
459
0
                storage_type =
460
0
                        DataTypeFactory::instance().create_data_type(*col, col->is_nullable());
461
0
            }
462
            // Currently, when writing a lucene index, the field of the document is column_name, and the column name is
463
            // bound to the index field. Since version 1.2, the data file storage has been changed from column_name to
464
            // column_unique_id, allowing the column name to be changed. Due to current limitations, previous inverted
465
            // index data cannot be used after Doris changes the column name. Column names also support Unicode
466
            // characters, which may cause other problems with indexing in non-ASCII characters.
467
            // After consideration, it was decided to change the field name from column_name to column_unique_id in
468
            // format V2, while format V1 continues to use column_name.
469
7.50k
            std::string field_name;
470
7.50k
            if (storage_format == InvertedIndexStorageFormatPB::V1) {
471
4.37k
                field_name = col->name();
472
4.37k
            } else {
473
3.12k
                if (col->is_extracted_column()) {
474
                    // variant sub col
475
                    // field_name format: parent_unique_id.sub_col_name
476
198
                    field_name = std::to_string(col->parent_unique_id()) + "." + col->name();
477
2.93k
                } else {
478
2.93k
                    field_name = std::to_string(col->unique_id());
479
2.93k
                }
480
3.12k
            }
481
7.50k
            _storage_name_and_type[i] = std::make_pair(field_name, storage_type);
482
7.50k
            if (int32_t uid =
483
7.50k
                        col->is_extracted_column() ? col->parent_unique_id() : col->unique_id();
484
7.50k
                !_variant_sparse_column_cache.contains(uid)) {
485
7.37k
                DCHECK(uid >= 0);
486
7.37k
                _variant_sparse_column_cache.emplace(uid,
487
7.37k
                                                     std::make_unique<PathToBinaryColumnCache>());
488
7.37k
            }
489
7.50k
        }
490
23.8k
    }
491
492
3.03k
    RETURN_IF_ERROR(init_iterators());
493
494
3.03k
    RETURN_IF_ERROR(_construct_compound_expr_context());
495
3.03k
    VLOG_DEBUG << fmt::format(
496
0
            "Segment iterator init, virtual_column_exprs size: {}, common_expr_pushdown size: {}",
497
0
            _opts.virtual_column_exprs.size(), _common_expr_ctxs_push_down.size());
498
3.03k
    _initialize_predicate_results();
499
3.03k
    return Status::OK();
500
3.03k
}
501
502
3.03k
void SegmentIterator::_initialize_predicate_results() {
503
    // Initialize from _col_predicates
504
3.03k
    for (auto pred : _col_predicates) {
505
64
        int cid = pred->column_id();
506
64
        _column_predicate_index_exec_status[cid][pred] = false;
507
64
    }
508
509
3.03k
    _calculate_common_expr_index_exec_status();
510
3.03k
}
511
512
3.03k
Status SegmentIterator::init_iterators() {
513
3.03k
    RETURN_IF_ERROR(_init_return_column_iterators());
514
3.03k
    RETURN_IF_ERROR(_init_index_iterators());
515
3.03k
    return Status::OK();
516
3.03k
}
517
518
13.3k
Status SegmentIterator::_lazy_init(Block* block) {
519
13.3k
    if (_lazy_inited) {
520
10.3k
        return Status::OK();
521
10.3k
    }
522
3.03k
    SCOPED_RAW_TIMER(&_opts.stats->block_init_ns);
523
3.03k
    DorisMetrics::instance()->segment_read_total->increment(1);
524
3.03k
    _row_bitmap.addRange(0, _segment->num_rows());
525
3.03k
    _init_row_bitmap_by_condition_cache();
526
527
    // z-order can not use prefix index
528
3.03k
    if (_segment->_tablet_schema->sort_type() != SortType::ZORDER &&
529
3.03k
        _segment->_tablet_schema->cluster_key_uids().empty()) {
530
3.02k
        RETURN_IF_ERROR(_get_row_ranges_by_keys());
531
3.02k
    }
532
3.03k
    RETURN_IF_ERROR(_get_row_ranges_by_column_conditions());
533
3.03k
    RETURN_IF_ERROR(_vec_init_lazy_materialization());
534
    // Remove rows that have been marked deleted
535
3.03k
    if (_opts.delete_bitmap.count(segment_id()) > 0 &&
536
3.03k
        _opts.delete_bitmap.at(segment_id()) != nullptr) {
537
25
        size_t pre_size = _row_bitmap.cardinality();
538
25
        _row_bitmap -= *(_opts.delete_bitmap.at(segment_id()));
539
25
        _opts.stats->rows_del_by_bitmap += (pre_size - _row_bitmap.cardinality());
540
25
        VLOG_DEBUG << "read on segment: " << segment_id() << ", delete bitmap cardinality: "
541
0
                   << _opts.delete_bitmap.at(segment_id())->cardinality() << ", "
542
0
                   << _opts.stats->rows_del_by_bitmap << " rows deleted by bitmap";
543
25
    }
544
545
3.03k
    if (!_opts.row_ranges.is_empty()) {
546
0
        _row_bitmap &= RowRanges::ranges_to_roaring(_opts.row_ranges);
547
0
    }
548
549
3.03k
    _prepare_score_column_materialization();
550
551
3.03k
    RETURN_IF_ERROR(_apply_ann_topn_predicate());
552
553
3.03k
    if (_opts.read_orderby_key_reverse) {
554
0
        _range_iter.reset(new BackwardBitmapRangeIterator(_row_bitmap));
555
3.03k
    } else {
556
3.03k
        _range_iter.reset(new BitmapRangeIterator(_row_bitmap));
557
3.03k
    }
558
559
    // Reserve columns for _initial_block_row_max (the original max before any adaptive
560
    // prediction) because the predictor may increase block_row_max on subsequent batches
561
    // up to this ceiling. Using the current (possibly reduced) _opts.block_row_max would
562
    // cause heap-buffer-overflow if a later prediction is larger.
563
3.03k
    auto nrows_reserve_limit =
564
3.03k
            std::min(_row_bitmap.cardinality(), uint64_t(_initial_block_row_max));
565
3.03k
    if (_lazy_materialization_read || _opts.record_rowids || _is_need_expr_eval) {
566
921
        _block_rowids.resize(_initial_block_row_max);
567
921
    }
568
3.03k
    _current_return_columns.resize(_schema->columns().size());
569
570
10.5k
    for (size_t i = 0; i < _schema->column_ids().size(); i++) {
571
7.50k
        ColumnId cid = _schema->column_ids()[i];
572
7.50k
        const auto* column_desc = _schema->column(cid);
573
7.50k
        if (_is_pred_column[cid]) {
574
486
            auto storage_column_type = _storage_name_and_type[cid].second;
575
486
            RETURN_IF_CATCH_EXCEPTION(
576
                    // Here, cid will not go out of bounds
577
                    // because the size of _current_return_columns equals _schema->tablet_columns().size()
578
486
                    _current_return_columns[cid] = Schema::get_predicate_column_ptr(
579
486
                            storage_column_type, _opts.io_ctx.reader_type));
580
486
            _current_return_columns[cid]->set_rowset_segment_id(
581
486
                    {_segment->rowset_id(), _segment->id()});
582
486
            _current_return_columns[cid]->reserve(nrows_reserve_limit);
583
7.01k
        } else if (i >= block->columns()) {
584
            // This column needs to be scanned, but doesn't need to be returned upward. (delete sign)
585
            // if i >= block->columns means the column and not the pred_column means `column i` is
586
            // a delete condition column. but the column is not effective in the segment. so we just
587
            // create a column to hold the data.
588
            // a. origin data -> b. delete condition -> c. new load data
589
            // the segment of c do not effective delete condition, but it still need read the column
590
            // to match the schema.
591
            // TODO: skip read the not effective delete column to speed up segment read.
592
0
            _current_return_columns[cid] = Schema::get_data_type_ptr(*column_desc)->create_column();
593
0
            _current_return_columns[cid]->reserve(nrows_reserve_limit);
594
0
        }
595
7.50k
    }
596
597
    // Additional deleted filter condition will be materialized column be at the end of the block,
598
    // after _output_column_by_sel_idx  will be erase, we not need to filter it,
599
    // so erase it from _columns_to_filter in the first next_batch.
600
    // Eg:
601
    //      `delete from table where a = 10;`
602
    //      `select b from table;`
603
    // a column only effective in segment iterator, the block from query engine only contain the b column,
604
    // so no need to filter a column by expr.
605
3.03k
    for (auto it = _columns_to_filter.begin(); it != _columns_to_filter.end();) {
606
6
        if (*it >= block->columns()) {
607
0
            it = _columns_to_filter.erase(it);
608
6
        } else {
609
6
            ++it;
610
6
        }
611
6
    }
612
613
3.03k
    _lazy_inited = true;
614
615
3.03k
    _init_segment_prefetchers();
616
617
3.03k
    return Status::OK();
618
3.03k
}
619
620
3.03k
void SegmentIterator::_init_segment_prefetchers() {
621
3.03k
    SCOPED_RAW_TIMER(&_opts.stats->segment_iterator_init_segment_prefetchers_timer_ns);
622
3.03k
    if (!config::is_cloud_mode()) {
623
3.03k
        return;
624
3.03k
    }
625
0
    static std::vector<ReaderType> supported_reader_types {
626
0
            ReaderType::READER_QUERY, ReaderType::READER_BASE_COMPACTION,
627
0
            ReaderType::READER_CUMULATIVE_COMPACTION, ReaderType::READER_FULL_COMPACTION};
628
0
    if (std::ranges::none_of(supported_reader_types,
629
0
                             [&](ReaderType t) { return _opts.io_ctx.reader_type == t; })) {
630
0
        return;
631
0
    }
632
    // Initialize segment prefetcher for predicate and non-predicate columns
633
0
    bool is_query = (_opts.io_ctx.reader_type == ReaderType::READER_QUERY);
634
0
    bool enable_prefetch = is_query ? config::enable_query_segment_file_cache_prefetch
635
0
                                    : config::enable_compaction_segment_file_cache_prefetch;
636
0
    LOG_IF(INFO, config::enable_segment_prefetch_verbose_log) << fmt::format(
637
0
            "[verbose] SegmentIterator _init_segment_prefetchers, is_query={}, "
638
0
            "enable_prefetch={}, "
639
0
            "_row_bitmap.isEmpty()={}, row_bitmap.cardinality()={}, tablet={}, rowset={}, "
640
0
            "segment={}, predicate_column_ids={}, common_expr_column_ids={}",
641
0
            is_query, enable_prefetch, _row_bitmap.isEmpty(), _row_bitmap.cardinality(),
642
0
            _opts.tablet_id, _opts.rowset_id.to_string(), segment_id(),
643
0
            fmt::join(_predicate_column_ids, ","), fmt::join(_common_expr_column_ids, ","));
644
0
    if (enable_prefetch && !_row_bitmap.isEmpty()) {
645
0
        int window_size =
646
0
                1 + (is_query ? config::query_segment_file_cache_prefetch_block_size
647
0
                              : config::compaction_segment_file_cache_prefetch_block_size);
648
0
        LOG_IF(INFO, config::enable_segment_prefetch_verbose_log) << fmt::format(
649
0
                "[verbose] SegmentIterator prefetch config: window_size={}", window_size);
650
0
        if (window_size > 0 &&
651
0
            !_column_iterators.empty()) { // ensure init_iterators has been called
652
0
            SegmentPrefetcherConfig prefetch_config(window_size,
653
0
                                                    config::file_cache_each_block_size);
654
0
            for (auto cid : _schema->column_ids()) {
655
0
                auto& column_iter = _column_iterators[cid];
656
0
                if (column_iter == nullptr) {
657
0
                    continue;
658
0
                }
659
0
                const auto* tablet_column = _schema->column(cid);
660
0
                SegmentPrefetchParams params {
661
0
                        .config = prefetch_config,
662
0
                        .read_options = _opts,
663
0
                };
664
0
                LOG_IF(INFO, config::enable_segment_prefetch_verbose_log) << fmt::format(
665
0
                        "[verbose] SegmentIterator init_segment_prefetchers, "
666
0
                        "tablet={}, rowset={}, segment={}, column_id={}, col_name={}, type={}",
667
0
                        _opts.tablet_id, _opts.rowset_id.to_string(), segment_id(), cid,
668
0
                        tablet_column->name(), tablet_column->type());
669
0
                Status st = column_iter->init_prefetcher(params);
670
0
                if (!st.ok()) {
671
0
                    LOG_IF(WARNING, config::enable_segment_prefetch_verbose_log) << fmt::format(
672
0
                            "[verbose] failed to init prefetcher for column_id={}, "
673
0
                            "tablet={}, rowset={}, segment={}, error={}",
674
0
                            cid, _opts.tablet_id, _opts.rowset_id.to_string(), segment_id(),
675
0
                            st.to_string());
676
0
                }
677
0
            }
678
679
            // for compaction, it's guaranteed that all rows are read, so we can prefetch all data blocks
680
0
            PrefetcherInitMethod init_method = (is_query && _row_bitmap.cardinality() < num_rows())
681
0
                                                       ? PrefetcherInitMethod::FROM_ROWIDS
682
0
                                                       : PrefetcherInitMethod::ALL_DATA_BLOCKS;
683
0
            std::map<PrefetcherInitMethod, std::vector<SegmentPrefetcher*>> prefetchers;
684
0
            for (size_t idx = 0; idx < _column_iterators.size(); ++idx) {
685
0
                auto cid = cast_set<ColumnId>(idx);
686
0
                auto* column_iter = _column_iterators[cid].get();
687
0
                if (column_iter != nullptr) {
688
0
                    ScopedColumnIteratorReadPhase scoped_read_phase {
689
0
                            column_iter, _support_lazy_read_pruned_columns.contains(cid)
690
0
                                                 ? ColumnIterator::ReadPhase::PREDICATE
691
0
                                                 : ColumnIterator::ReadPhase::NORMAL};
692
0
                    column_iter->collect_prefetchers(prefetchers, init_method);
693
0
                }
694
0
            }
695
0
            for (auto& [method, prefetcher_vec] : prefetchers) {
696
0
                if (method == PrefetcherInitMethod::ALL_DATA_BLOCKS) {
697
0
                    for (auto* prefetcher : prefetcher_vec) {
698
0
                        prefetcher->build_all_data_blocks();
699
0
                    }
700
0
                } else if (method == PrefetcherInitMethod::FROM_ROWIDS && !prefetcher_vec.empty()) {
701
0
                    SegmentPrefetcher::build_blocks_by_rowids(_row_bitmap, prefetcher_vec);
702
0
                }
703
0
            }
704
0
        }
705
0
    }
706
0
}
707
708
3.02k
Status SegmentIterator::_get_row_ranges_by_keys() {
709
3.02k
    SCOPED_RAW_TIMER(&_opts.stats->generate_row_ranges_by_keys_ns);
710
3.02k
    DorisMetrics::instance()->segment_row_total->increment(num_rows());
711
712
    // fast path for empty segment or empty key ranges
713
3.02k
    if (_row_bitmap.isEmpty() || _opts.key_ranges.empty()) {
714
3.02k
        return Status::OK();
715
3.02k
    }
716
717
    // Read & seek key columns is a waste of time when no key column in _schema
718
0
    if (std::none_of(_schema->columns().begin(), _schema->columns().end(),
719
0
                     [&](const TabletColumnPtr& col) {
720
0
                         return col &&
721
0
                                _opts.tablet_schema->column_by_uid(col->unique_id()).is_key();
722
0
                     })) {
723
0
        return Status::OK();
724
0
    }
725
726
0
    RowRanges result_ranges;
727
0
    for (auto& key_range : _opts.key_ranges) {
728
0
        rowid_t lower_rowid = 0;
729
0
        rowid_t upper_rowid = num_rows();
730
0
        RETURN_IF_ERROR(_prepare_seek(key_range));
731
0
        if (key_range.upper_key != nullptr) {
732
            // If client want to read upper_bound, the include_upper is true. So we
733
            // should get the first ordinal at which key is larger than upper_bound.
734
            // So we call _lookup_ordinal with include_upper's negate
735
0
            RETURN_IF_ERROR(_lookup_ordinal(*key_range.upper_key, !key_range.include_upper,
736
0
                                            num_rows(), &upper_rowid));
737
0
        }
738
0
        if (upper_rowid > 0 && key_range.lower_key != nullptr) {
739
0
            RETURN_IF_ERROR(_lookup_ordinal(*key_range.lower_key, key_range.include_lower,
740
0
                                            upper_rowid, &lower_rowid));
741
0
        }
742
0
        auto row_range = RowRanges::create_single(lower_rowid, upper_rowid);
743
0
        RowRanges::ranges_union(result_ranges, row_range, &result_ranges);
744
0
    }
745
0
    size_t pre_size = _row_bitmap.cardinality();
746
0
    _row_bitmap &= RowRanges::ranges_to_roaring(result_ranges);
747
0
    _opts.stats->rows_key_range_filtered += (pre_size - _row_bitmap.cardinality());
748
749
0
    return Status::OK();
750
0
}
751
752
// Set up environment for the following seek.
753
0
Status SegmentIterator::_prepare_seek(const StorageReadOptions::KeyRange& key_range) {
754
0
    std::vector<const TabletColumn*> key_columns;
755
0
    std::set<uint32_t> column_set;
756
0
    if (key_range.lower_key != nullptr) {
757
0
        for (auto cid : key_range.lower_key->schema()->column_ids()) {
758
0
            column_set.emplace(cid);
759
0
            key_columns.emplace_back(key_range.lower_key->column(cid));
760
0
        }
761
0
    }
762
0
    if (key_range.upper_key != nullptr) {
763
0
        for (auto cid : key_range.upper_key->schema()->column_ids()) {
764
0
            if (column_set.count(cid) == 0) {
765
0
                key_columns.emplace_back(key_range.upper_key->column(cid));
766
0
                column_set.emplace(cid);
767
0
            }
768
0
        }
769
0
    }
770
0
    if (!_seek_schema) {
771
0
        std::vector<TabletColumnPtr> cols;
772
0
        cols.reserve(key_columns.size());
773
0
        for (const TabletColumn* col : key_columns) {
774
0
            cols.emplace_back(std::make_shared<TabletColumn>(*col));
775
0
        }
776
0
        std::vector<uint32_t> column_ids(cols.size());
777
0
        std::iota(column_ids.begin(), column_ids.end(), 0);
778
0
        _seek_schema = std::make_unique<Schema>(cols, column_ids);
779
0
    }
780
    // todo(wb) need refactor here, when using pk to search, _seek_block is useless
781
0
    if (_seek_block.size() == 0) {
782
0
        _seek_block.resize(_seek_schema->num_column_ids());
783
0
        int i = 0;
784
0
        for (auto cid : _seek_schema->column_ids()) {
785
0
            auto column_desc = _seek_schema->column(cid);
786
0
            _seek_block[i] = Schema::get_data_type_ptr(*column_desc)->create_column();
787
0
            i++;
788
0
        }
789
0
    }
790
791
    // create used column iterator
792
0
    for (auto cid : _seek_schema->column_ids()) {
793
0
        if (_column_iterators[cid] == nullptr) {
794
            // TODO: Do we need this?
795
0
            if (_virtual_column_exprs.contains(cid)) {
796
0
                _column_iterators[cid] = std::make_unique<VirtualColumnIterator>();
797
0
                continue;
798
0
            }
799
800
0
            RETURN_IF_ERROR(_segment->new_column_iterator(_opts.tablet_schema->column(cid),
801
0
                                                          &_column_iterators[cid], &_opts,
802
0
                                                          &_variant_sparse_column_cache));
803
0
            ColumnIteratorOptions iter_opts {
804
0
                    .use_page_cache = _opts.use_page_cache,
805
0
                    .file_reader = _file_reader.get(),
806
0
                    .stats = _opts.stats,
807
0
                    .io_ctx = _opts.io_ctx,
808
0
            };
809
0
            RETURN_IF_ERROR(_column_iterators[cid]->init(iter_opts));
810
0
        }
811
0
    }
812
813
0
    return Status::OK();
814
0
}
815
816
3.03k
Status SegmentIterator::_get_row_ranges_by_column_conditions() {
817
3.03k
    SCOPED_RAW_TIMER(&_opts.stats->generate_row_ranges_by_column_conditions_ns);
818
3.03k
    if (_row_bitmap.isEmpty()) {
819
0
        return Status::OK();
820
0
    }
821
822
3.03k
    {
823
3.03k
        if (_opts.runtime_state &&
824
3.03k
            _opts.runtime_state->query_options().enable_inverted_index_query &&
825
3.03k
            (has_index_in_iterators() || !_common_expr_ctxs_push_down.empty())) {
826
67
            SCOPED_RAW_TIMER(&_opts.stats->inverted_index_filter_timer);
827
67
            size_t input_rows = _row_bitmap.cardinality();
828
            // Only apply column-level inverted index if we have iterators
829
67
            if (has_index_in_iterators()) {
830
63
                RETURN_IF_ERROR(_apply_inverted_index());
831
63
            }
832
            // Always apply expr-level index (e.g., search expressions) if we have common_expr_pushdown
833
            // This allows search expressions with variant subcolumns to be evaluated even when
834
            // the segment doesn't have all subcolumns
835
67
            RETURN_IF_ERROR(_apply_index_expr());
836
67
            for (auto it = _common_expr_ctxs_push_down.begin();
837
85
                 it != _common_expr_ctxs_push_down.end();) {
838
18
                if ((*it)->all_expr_inverted_index_evaluated()) {
839
14
                    const auto* result = (*it)->get_index_context()->get_index_result_for_expr(
840
14
                            (*it)->root().get());
841
14
                    if (result != nullptr) {
842
14
                        _row_bitmap &= *result->get_data_bitmap();
843
14
                        it = _common_expr_ctxs_push_down.erase(it);
844
14
                    }
845
14
                } else {
846
4
                    ++it;
847
4
                }
848
18
            }
849
67
            _opts.condition_cache_digest =
850
67
                    _common_expr_ctxs_push_down.empty() ? 0 : _opts.condition_cache_digest;
851
67
            _opts.stats->rows_inverted_index_filtered += (input_rows - _row_bitmap.cardinality());
852
136
            for (auto cid : _schema->column_ids()) {
853
136
                bool result_true = _check_all_conditions_passed_inverted_index_for_column(cid);
854
136
                if (result_true) {
855
59
                    _need_read_data_indices[cid] = false;
856
59
                }
857
136
            }
858
67
        }
859
3.03k
    }
860
861
3.03k
    DBUG_EXECUTE_IF("segment_iterator.inverted_index.filtered_rows", {
862
3.03k
        LOG(INFO) << "Debug Point: segment_iterator.inverted_index.filtered_rows: "
863
3.03k
                  << _opts.stats->rows_inverted_index_filtered;
864
3.03k
        auto filtered_rows = DebugPoints::instance()->get_debug_param_or_default<int32_t>(
865
3.03k
                "segment_iterator.inverted_index.filtered_rows", "filtered_rows", -1);
866
3.03k
        if (filtered_rows != _opts.stats->rows_inverted_index_filtered) {
867
3.03k
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
868
3.03k
                    "filtered_rows: {} not equal to expected: {}",
869
3.03k
                    _opts.stats->rows_inverted_index_filtered, filtered_rows);
870
3.03k
        }
871
3.03k
    })
872
873
3.03k
    DBUG_EXECUTE_IF("segment_iterator.apply_inverted_index", {
874
3.03k
        LOG(INFO) << "Debug Point: segment_iterator.apply_inverted_index";
875
3.03k
        if (!_common_expr_ctxs_push_down.empty() || !_col_predicates.empty()) {
876
3.03k
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
877
3.03k
                    "it is failed to apply inverted index, common_expr_ctxs_push_down: {}, "
878
3.03k
                    "col_predicates: {}",
879
3.03k
                    _common_expr_ctxs_push_down.size(), _col_predicates.size());
880
3.03k
        }
881
3.03k
    })
882
883
3.03k
    if (!_row_bitmap.isEmpty() &&
884
3.03k
        (!_opts.topn_filter_source_node_ids.empty() || !_opts.col_id_to_predicates.empty() ||
885
3.02k
         _opts.delete_condition_predicates->num_of_column_predicate() > 0 ||
886
3.02k
         !_common_expr_ctxs_push_down.empty())) {
887
537
        RowRanges condition_row_ranges = RowRanges::create_single(_segment->num_rows());
888
537
        RETURN_IF_ERROR(_get_row_ranges_from_conditions(&condition_row_ranges));
889
537
        size_t pre_size = _row_bitmap.cardinality();
890
537
        _row_bitmap &= RowRanges::ranges_to_roaring(condition_row_ranges);
891
537
        _opts.stats->rows_conditions_filtered += (pre_size - _row_bitmap.cardinality());
892
537
    }
893
894
3.03k
    DBUG_EXECUTE_IF("bloom_filter_must_filter_data", {
895
3.03k
        if (_opts.stats->rows_bf_filtered == 0) {
896
3.03k
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
897
3.03k
                    "Bloom filter did not filter the data.");
898
3.03k
        }
899
3.03k
    })
900
901
    // TODO(hkp): calculate filter rate to decide whether to
902
    // use zone map/bloom filter/secondary index or not.
903
3.03k
    return Status::OK();
904
3.03k
}
905
906
0
bool SegmentIterator::_column_has_ann_index(int32_t cid) {
907
0
    bool has_ann_index = _index_iterators[cid] != nullptr &&
908
0
                         _index_iterators[cid]->get_reader(AnnIndexReaderType::ANN);
909
910
0
    return has_ann_index;
911
0
}
912
913
3.03k
Status SegmentIterator::_apply_ann_topn_predicate() {
914
3.03k
    if (_ann_topn_runtime == nullptr) {
915
3.03k
        return Status::OK();
916
3.03k
    }
917
918
0
    VLOG_DEBUG << fmt::format("Try apply ann topn: {}", _ann_topn_runtime->debug_string());
919
0
    size_t src_col_idx = _ann_topn_runtime->get_src_column_idx();
920
    // AnnTopNRuntime keeps VSlotRef::column_id(), which is the scan schema ordinal.
921
0
    ColumnId src_cid = _schema->column_id(src_col_idx);
922
0
    IndexIterator* ann_index_iterator = _index_iterators[src_cid].get();
923
0
    bool has_ann_index = _column_has_ann_index(src_cid);
924
0
    bool has_common_expr_push_down = !_common_expr_ctxs_push_down.empty();
925
0
    bool has_column_predicate = std::any_of(_is_pred_column.begin(), _is_pred_column.end(),
926
0
                                            [](bool is_pred) { return is_pred; });
927
0
    if (!has_ann_index || has_common_expr_push_down || has_column_predicate) {
928
0
        VLOG_DEBUG << fmt::format(
929
0
                "Ann topn can not be evaluated by ann index, has_ann_index: {}, "
930
0
                "has_common_expr_push_down: {}, has_column_predicate: {}",
931
0
                has_ann_index, has_common_expr_push_down, has_column_predicate);
932
        // Disable index-only scan on ann indexed column.
933
0
        _need_read_data_indices[src_cid] = true;
934
0
        _opts.stats->ann_fall_back_brute_force_cnt += 1;
935
0
        return Status::OK();
936
0
    }
937
938
    // Process asc & desc according to the type of metric
939
0
    auto index_reader = ann_index_iterator->get_reader(AnnIndexReaderType::ANN);
940
0
    auto ann_index_reader = dynamic_cast<AnnIndexReader*>(index_reader.get());
941
0
    DCHECK(ann_index_reader != nullptr);
942
0
    if (ann_index_reader->get_metric_type() == AnnIndexMetric::IP) {
943
0
        if (_ann_topn_runtime->is_asc()) {
944
0
            VLOG_DEBUG << fmt::format(
945
0
                    "Asc topn for inner product can not be evaluated by ann index");
946
            // Disable index-only scan on ann indexed column.
947
0
            _need_read_data_indices[src_cid] = true;
948
0
            _opts.stats->ann_fall_back_brute_force_cnt += 1;
949
0
            return Status::OK();
950
0
        }
951
0
    } else {
952
0
        if (!_ann_topn_runtime->is_asc()) {
953
0
            VLOG_DEBUG << fmt::format("Desc topn for l2/cosine can not be evaluated by ann index");
954
            // Disable index-only scan on ann indexed column.
955
0
            _need_read_data_indices[src_cid] = true;
956
0
            _opts.stats->ann_fall_back_brute_force_cnt += 1;
957
0
            return Status::OK();
958
0
        }
959
0
    }
960
961
0
    if (ann_index_reader->get_metric_type() != _ann_topn_runtime->get_metric_type()) {
962
0
        VLOG_DEBUG << fmt::format(
963
0
                "Ann topn metric type {} not match index metric type {}, can not be evaluated "
964
0
                "by "
965
0
                "ann index",
966
0
                metric_to_string(_ann_topn_runtime->get_metric_type()),
967
0
                metric_to_string(ann_index_reader->get_metric_type()));
968
        // Disable index-only scan on ann indexed column.
969
0
        _need_read_data_indices[src_cid] = true;
970
0
        _opts.stats->ann_fall_back_brute_force_cnt += 1;
971
0
        return Status::OK();
972
0
    }
973
974
0
    size_t pre_size = _row_bitmap.cardinality();
975
0
    size_t rows_of_segment = _segment->num_rows();
976
0
    const auto& user_params = _ann_topn_runtime->user_params();
977
0
    if (user_params.should_fallback_ann_index_by_small_candidate(pre_size, rows_of_segment)) {
978
0
        VLOG_DEBUG << fmt::format(
979
0
                "Ann topn predicate input rows {} reach small candidate threshold, "
980
0
                "rows_of_segment: {}, absolute_threshold: {}, percent_threshold: {}, "
981
0
                "will not use ann index to filter",
982
0
                pre_size, rows_of_segment, user_params.ann_index_candidate_rows_threshold,
983
0
                user_params.ann_index_candidate_rows_percent_threshold);
984
        // Disable index-only scan on ann indexed column.
985
0
        _need_read_data_indices[src_cid] = true;
986
0
        _opts.stats->ann_fall_back_brute_force_cnt += 1;
987
0
        _opts.stats->ann_topn_fallback_by_small_candidate_cnt += 1;
988
0
        _opts.stats->ann_topn_fallback_small_candidate_rows += pre_size;
989
0
        return Status::OK();
990
0
    }
991
0
    IColumn::MutablePtr result_column;
992
0
    std::shared_ptr<std::vector<uint64_t>> result_row_ids;
993
0
    segment_v2::AnnIndexStats ann_index_stats;
994
995
    // Try to load ANN index before search
996
0
    auto ann_index_iterator_casted =
997
0
            dynamic_cast<segment_v2::AnnIndexIterator*>(ann_index_iterator);
998
0
    if (ann_index_iterator_casted == nullptr) {
999
0
        VLOG_DEBUG << "Failed to cast index iterator to AnnIndexIterator, fallback to brute force";
1000
0
        _need_read_data_indices[src_cid] = true;
1001
0
        _opts.stats->ann_fall_back_brute_force_cnt += 1;
1002
0
        return Status::OK();
1003
0
    }
1004
1005
    // Track load index timing
1006
0
    {
1007
0
        SCOPED_TIMER(&(ann_index_stats.load_index_costs_ns));
1008
0
        if (!ann_index_iterator_casted->try_load_index()) {
1009
0
            VLOG_DEBUG << "Failed to load ANN index, fallback to brute force search";
1010
0
            _need_read_data_indices[src_cid] = true;
1011
0
            _opts.stats->ann_fall_back_brute_force_cnt += 1;
1012
0
            return Status::OK();
1013
0
        }
1014
0
        double load_costs_ms =
1015
0
                static_cast<double>(ann_index_stats.load_index_costs_ns.value()) / 1000000.0;
1016
0
        DorisMetrics::instance()->ann_index_load_costs_ms->increment(
1017
0
                static_cast<int64_t>(load_costs_ms));
1018
0
    }
1019
1020
0
    bool enable_ann_index_result_cache =
1021
0
            !_opts.runtime_state ||
1022
0
            !_opts.runtime_state->query_options().__isset.enable_ann_index_result_cache ||
1023
0
            _opts.runtime_state->query_options().enable_ann_index_result_cache;
1024
0
    RETURN_IF_ERROR(_ann_topn_runtime->evaluate_vector_ann_search(
1025
0
            ann_index_iterator_casted, &_row_bitmap, rows_of_segment, enable_ann_index_result_cache,
1026
0
            result_column, result_row_ids, ann_index_stats));
1027
1028
0
    VLOG_DEBUG << fmt::format("Ann topn filtered {} - {} = {} rows", pre_size,
1029
0
                              _row_bitmap.cardinality(), pre_size - _row_bitmap.cardinality());
1030
1031
0
    int64_t rows_filterd = pre_size - _row_bitmap.cardinality();
1032
0
    _opts.stats->rows_ann_index_topn_filtered += rows_filterd;
1033
0
    _opts.stats->ann_index_load_ns += ann_index_stats.load_index_costs_ns.value();
1034
0
    _opts.stats->ann_topn_search_ns += ann_index_stats.search_costs_ns.value();
1035
0
    _opts.stats->ann_ivf_on_disk_load_ns += ann_index_stats.ivf_on_disk_load_costs_ns.value();
1036
0
    _opts.stats->ann_ivf_on_disk_cache_hit_cnt += ann_index_stats.ivf_on_disk_cache_hit_cnt.value();
1037
0
    _opts.stats->ann_ivf_on_disk_cache_miss_cnt +=
1038
0
            ann_index_stats.ivf_on_disk_cache_miss_cnt.value();
1039
0
    _opts.stats->ann_index_topn_engine_search_ns += ann_index_stats.engine_search_ns.value();
1040
0
    _opts.stats->ann_index_topn_result_process_ns +=
1041
0
            ann_index_stats.result_process_costs_ns.value();
1042
0
    _opts.stats->ann_index_topn_engine_convert_ns += ann_index_stats.engine_convert_ns.value();
1043
0
    _opts.stats->ann_index_topn_engine_prepare_ns += ann_index_stats.engine_prepare_ns.value();
1044
0
    _opts.stats->ann_index_topn_search_cnt += 1;
1045
0
    _opts.stats->ann_index_cache_hits += ann_index_stats.topn_cache_hits.value();
1046
0
    const size_t dst_col_idx = _ann_topn_runtime->get_dest_column_idx();
1047
0
    ColumnIterator* column_iter = _column_iterators[_schema->column_id(dst_col_idx)].get();
1048
0
    DCHECK(column_iter != nullptr);
1049
0
    VirtualColumnIterator* virtual_column_iter = dynamic_cast<VirtualColumnIterator*>(column_iter);
1050
0
    DCHECK(virtual_column_iter != nullptr);
1051
0
    VLOG_DEBUG << fmt::format(
1052
0
            "Virtual column iterator, column_idx {}, is materialized with {} rows", dst_col_idx,
1053
0
            result_row_ids->size());
1054
    // reference count of result_column should be 1, so move will not issue any data copy.
1055
0
    virtual_column_iter->prepare_materialization(std::move(result_column), result_row_ids);
1056
1057
0
    _need_read_data_indices[src_cid] = false;
1058
0
    VLOG_DEBUG << fmt::format(
1059
0
            "Enable ANN index-only scan for src column cid {} (skip reading data pages)", src_cid);
1060
1061
0
    return Status::OK();
1062
0
}
1063
1064
537
Status SegmentIterator::_get_row_ranges_from_conditions(RowRanges* condition_row_ranges) {
1065
537
    std::set<int32_t> cids;
1066
537
    for (auto& entry : _opts.col_id_to_predicates) {
1067
64
        cids.insert(entry.first);
1068
64
    }
1069
1070
537
    {
1071
537
        SCOPED_RAW_TIMER(&_opts.stats->generate_row_ranges_by_dict_ns);
1072
        /// Low cardinality optimization is currently not very stable, so to prevent data corruption,
1073
        /// we are temporarily disabling its use in data compaction.
1074
        // TODO: enable it in not only ReaderTyper::READER_QUERY but also other reader types.
1075
537
        if (_opts.io_ctx.reader_type == ReaderType::READER_QUERY) {
1076
70
            RowRanges dict_row_ranges = RowRanges::create_single(num_rows());
1077
70
            for (auto cid : cids) {
1078
64
                if (!_segment->can_apply_predicate_safely(
1079
64
                            cid, *_schema, _opts.target_cast_type_for_variants, _opts)) {
1080
0
                    continue;
1081
0
                }
1082
64
                DCHECK(_opts.col_id_to_predicates.count(cid) > 0);
1083
64
                RETURN_IF_ERROR(_column_iterators[cid]->get_row_ranges_by_dict(
1084
64
                        _opts.col_id_to_predicates.at(cid).get(), &dict_row_ranges));
1085
1086
64
                if (dict_row_ranges.is_empty()) {
1087
0
                    break;
1088
0
                }
1089
64
            }
1090
1091
70
            if (dict_row_ranges.is_empty()) {
1092
0
                RowRanges::ranges_intersection(*condition_row_ranges, dict_row_ranges,
1093
0
                                               condition_row_ranges);
1094
0
                _opts.stats->segment_dict_filtered++;
1095
0
                _opts.stats->filtered_segment_number++;
1096
0
                return Status::OK();
1097
0
            }
1098
70
        }
1099
537
    }
1100
1101
537
    size_t pre_size = 0;
1102
537
    {
1103
537
        SCOPED_RAW_TIMER(&_opts.stats->generate_row_ranges_by_bf_ns);
1104
        // first filter data by bloom filter index
1105
        // bloom filter index only use CondColumn
1106
537
        RowRanges bf_row_ranges = RowRanges::create_single(num_rows());
1107
537
        for (auto& cid : cids) {
1108
64
            DCHECK(_opts.col_id_to_predicates.count(cid) > 0);
1109
64
            if (!_segment->can_apply_predicate_safely(cid, *_schema,
1110
64
                                                      _opts.target_cast_type_for_variants, _opts)) {
1111
0
                continue;
1112
0
            }
1113
            // get row ranges by bf index of this column,
1114
64
            RowRanges column_bf_row_ranges = RowRanges::create_single(num_rows());
1115
64
            RETURN_IF_ERROR(_column_iterators[cid]->get_row_ranges_by_bloom_filter(
1116
64
                    _opts.col_id_to_predicates.at(cid).get(), &column_bf_row_ranges));
1117
64
            RowRanges::ranges_intersection(bf_row_ranges, column_bf_row_ranges, &bf_row_ranges);
1118
64
        }
1119
1120
537
        pre_size = condition_row_ranges->count();
1121
537
        RowRanges::ranges_intersection(*condition_row_ranges, bf_row_ranges, condition_row_ranges);
1122
537
        _opts.stats->rows_bf_filtered += (pre_size - condition_row_ranges->count());
1123
537
    }
1124
1125
0
    {
1126
537
        SCOPED_RAW_TIMER(&_opts.stats->generate_row_ranges_by_zonemap_ns);
1127
537
        RowRanges zone_map_row_ranges = RowRanges::create_single(num_rows());
1128
        // second filter data by zone map
1129
537
        for (const auto& cid : cids) {
1130
64
            DCHECK(_opts.col_id_to_predicates.count(cid) > 0);
1131
64
            if (!_segment->can_apply_predicate_safely(cid, *_schema,
1132
64
                                                      _opts.target_cast_type_for_variants, _opts)) {
1133
0
                continue;
1134
0
            }
1135
64
            if (_segment->is_tso_placeholder_col(cid, *_schema, _opts)) {
1136
                // skip untrustworthy tso placeholder zonemap
1137
                // if possible already be pruned as a whole before,
1138
                // so just skip
1139
0
                continue;
1140
0
            }
1141
            // do not check zonemap if predicate does not support zonemap
1142
64
            if (!_opts.col_id_to_predicates.at(cid)->support_zonemap()) {
1143
0
                VLOG_DEBUG << "skip zonemap for column " << cid;
1144
0
                continue;
1145
0
            }
1146
            // get row ranges by zone map of this column,
1147
64
            RowRanges column_row_ranges = RowRanges::create_single(num_rows());
1148
64
            RETURN_IF_ERROR(_column_iterators[cid]->get_row_ranges_by_zone_map(
1149
64
                    _opts.col_id_to_predicates.at(cid).get(),
1150
64
                    _opts.del_predicates_for_zone_map.count(cid) > 0
1151
64
                            ? &(_opts.del_predicates_for_zone_map.at(cid))
1152
64
                            : nullptr,
1153
64
                    &column_row_ranges));
1154
            // intersect different columns's row ranges to get final row ranges by zone map
1155
64
            RowRanges::ranges_intersection(zone_map_row_ranges, column_row_ranges,
1156
64
                                           &zone_map_row_ranges);
1157
64
        }
1158
1159
537
        pre_size = condition_row_ranges->count();
1160
537
        RowRanges::ranges_intersection(*condition_row_ranges, zone_map_row_ranges,
1161
537
                                       condition_row_ranges);
1162
1163
537
        size_t pre_size2 = condition_row_ranges->count();
1164
537
        RowRanges::ranges_intersection(*condition_row_ranges, zone_map_row_ranges,
1165
537
                                       condition_row_ranges);
1166
537
        _opts.stats->rows_stats_rp_filtered += (pre_size2 - condition_row_ranges->count());
1167
537
        _opts.stats->rows_stats_filtered += (pre_size - condition_row_ranges->count());
1168
537
    }
1169
1170
0
    {
1171
537
        SCOPED_RAW_TIMER(&_opts.stats->generate_row_ranges_by_zonemap_ns);
1172
537
        if (!_common_expr_ctxs_push_down.empty()) {
1173
6
            const auto pre_expr_zonemap_size = condition_row_ranges->count();
1174
6
            RETURN_IF_ERROR(_apply_expr_zonemap_to_row_ranges(_common_expr_ctxs_push_down, 0,
1175
6
                                                              condition_row_ranges));
1176
6
            _opts.stats->rows_stats_filtered +=
1177
6
                    (pre_expr_zonemap_size - condition_row_ranges->count());
1178
6
        }
1179
537
    }
1180
1181
537
    return Status::OK();
1182
537
}
1183
1184
0
bool SegmentIterator::_is_literal_node(const TExprNodeType::type& node_type) {
1185
0
    switch (node_type) {
1186
0
    case TExprNodeType::BOOL_LITERAL:
1187
0
    case TExprNodeType::INT_LITERAL:
1188
0
    case TExprNodeType::LARGE_INT_LITERAL:
1189
0
    case TExprNodeType::FLOAT_LITERAL:
1190
0
    case TExprNodeType::DECIMAL_LITERAL:
1191
0
    case TExprNodeType::STRING_LITERAL:
1192
0
    case TExprNodeType::DATE_LITERAL:
1193
0
    case TExprNodeType::TIMEV2_LITERAL:
1194
0
        return true;
1195
0
    default:
1196
0
        return false;
1197
0
    }
1198
0
}
1199
1200
14
Status SegmentIterator::_extract_common_expr_columns(const VExprSPtr& expr) {
1201
14
    auto& children = expr->children();
1202
22
    for (int i = 0; i < children.size(); ++i) {
1203
8
        RETURN_IF_ERROR(_extract_common_expr_columns(children[i]));
1204
8
    }
1205
1206
14
    auto node_type = expr->node_type();
1207
14
    if (node_type == TExprNodeType::SLOT_REF) {
1208
6
        auto slot_expr = std::dynamic_pointer_cast<doris::VSlotRef>(expr);
1209
6
        auto cid = _schema->column_id(slot_expr->column_id());
1210
6
        _is_common_expr_column[cid] = true;
1211
6
        _common_expr_columns.insert(cid);
1212
8
    } else if (node_type == TExprNodeType::VIRTUAL_SLOT_REF) {
1213
0
        std::shared_ptr<VirtualSlotRef> virtual_slot_ref =
1214
0
                std::dynamic_pointer_cast<VirtualSlotRef>(expr);
1215
0
        RETURN_IF_ERROR(_extract_common_expr_columns(virtual_slot_ref->get_virtual_column_expr()));
1216
0
    }
1217
1218
14
    return Status::OK();
1219
14
}
1220
1221
45
bool SegmentIterator::_check_apply_by_inverted_index(std::shared_ptr<ColumnPredicate> pred) {
1222
45
    if (_opts.runtime_state && !_opts.runtime_state->query_options().enable_inverted_index_query) {
1223
0
        return false;
1224
0
    }
1225
45
    auto pred_column_id = pred->column_id();
1226
45
    if (_index_iterators[pred_column_id] == nullptr) {
1227
        //this column without inverted index
1228
0
        return false;
1229
0
    }
1230
1231
45
    if (_inverted_index_not_support_pred_type(pred->type())) {
1232
0
        return false;
1233
0
    }
1234
1235
45
    if (pred->type() == PredicateType::IN_LIST || pred->type() == PredicateType::NOT_IN_LIST) {
1236
        // in_list or not_in_list predicate produced by runtime filter
1237
0
        if (pred->is_runtime_filter()) {
1238
0
            return false;
1239
0
        }
1240
0
    }
1241
1242
    // UNTOKENIZED strings exceed ignore_above, they are written as null, causing range query errors
1243
45
    if (PredicateTypeTraits::is_range(pred->type()) &&
1244
45
        !IndexReaderHelper::has_bkd_index(_index_iterators[pred_column_id].get())) {
1245
0
        return false;
1246
0
    }
1247
1248
    // Function filter no apply inverted index
1249
45
    if (dynamic_cast<LikeColumnPredicate*>(pred.get()) != nullptr) {
1250
0
        return false;
1251
0
    }
1252
1253
45
    bool handle_by_fulltext = _column_has_fulltext_index(pred_column_id);
1254
45
    if (handle_by_fulltext) {
1255
        // when predicate is leafNode of andNode,
1256
        // can apply 'match query' and 'equal query' and 'list query' for fulltext index.
1257
0
        return pred->type() == PredicateType::MATCH || pred->type() == PredicateType::IS_NULL ||
1258
0
               pred->type() == PredicateType::IS_NOT_NULL ||
1259
0
               PredicateTypeTraits::is_equal_or_list(pred->type());
1260
0
    }
1261
1262
45
    return true;
1263
45
}
1264
1265
// TODO: optimization when all expr can not evaluate by inverted/ann index,
1266
67
Status SegmentIterator::_apply_index_expr() {
1267
67
    bool enable_ann_index_result_cache =
1268
67
            !_opts.runtime_state ||
1269
67
            !_opts.runtime_state->query_options().__isset.enable_ann_index_result_cache ||
1270
67
            _opts.runtime_state->query_options().enable_ann_index_result_cache;
1271
1272
67
    for (const auto& expr_ctx : _common_expr_ctxs_push_down) {
1273
18
        if (Status st = expr_ctx->evaluate_inverted_index(num_rows()); !st.ok()) {
1274
0
            if (_downgrade_without_index(st) || st.code() == ErrorCode::NOT_IMPLEMENTED_ERROR) {
1275
0
                continue;
1276
0
            } else {
1277
                // other code is not to be handled, we should just break
1278
0
                LOG(WARNING) << "failed to evaluate inverted index for expr_ctx: "
1279
0
                             << expr_ctx->root()->debug_string()
1280
0
                             << ", error msg: " << st.to_string();
1281
0
                return st;
1282
0
            }
1283
0
        }
1284
18
    }
1285
1286
    // Evaluate inverted index for virtual column MATCH expressions (projections).
1287
    // Unlike common exprs which filter rows, these only compute index result bitmaps
1288
    // for later materialization via fast_execute().
1289
67
    for (auto& [cid, expr_ctx] : _virtual_column_exprs) {
1290
0
        if (expr_ctx->get_index_context() == nullptr) {
1291
0
            continue;
1292
0
        }
1293
0
        if (Status st = expr_ctx->evaluate_inverted_index(num_rows()); !st.ok()) {
1294
0
            if (_downgrade_without_index(st) || st.code() == ErrorCode::NOT_IMPLEMENTED_ERROR) {
1295
0
                continue;
1296
0
            } else {
1297
0
                LOG(WARNING) << "failed to evaluate inverted index for virtual column expr: "
1298
0
                             << expr_ctx->root()->debug_string()
1299
0
                             << ", error msg: " << st.to_string();
1300
0
                return st;
1301
0
            }
1302
0
        }
1303
0
    }
1304
1305
    // Apply ann range search
1306
67
    for (const auto& expr_ctx : _common_expr_ctxs_push_down) {
1307
18
        segment_v2::AnnIndexStats ann_index_stats;
1308
18
        size_t origin_rows = _row_bitmap.cardinality();
1309
18
        bool ann_range_search_executed = false;
1310
18
        RETURN_IF_ERROR(expr_ctx->evaluate_ann_range_search(
1311
18
                _index_iterators, _schema->column_ids(), _column_iterators,
1312
18
                _common_expr_to_slotref_map, num_rows(), _row_bitmap, ann_index_stats,
1313
18
                enable_ann_index_result_cache, &ann_range_search_executed));
1314
18
        if (ann_range_search_executed) {
1315
0
            _opts.stats->ann_index_range_search_cnt++;
1316
0
        }
1317
18
        _opts.stats->rows_ann_index_range_filtered += (origin_rows - _row_bitmap.cardinality());
1318
18
        _opts.stats->ann_index_load_ns += ann_index_stats.load_index_costs_ns.value();
1319
18
        _opts.stats->ann_index_range_search_ns += ann_index_stats.search_costs_ns.value();
1320
18
        _opts.stats->ann_ivf_on_disk_load_ns += ann_index_stats.ivf_on_disk_load_costs_ns.value();
1321
18
        _opts.stats->ann_ivf_on_disk_cache_hit_cnt +=
1322
18
                ann_index_stats.ivf_on_disk_cache_hit_cnt.value();
1323
18
        _opts.stats->ann_ivf_on_disk_cache_miss_cnt +=
1324
18
                ann_index_stats.ivf_on_disk_cache_miss_cnt.value();
1325
18
        _opts.stats->ann_range_engine_search_ns += ann_index_stats.engine_search_ns.value();
1326
18
        _opts.stats->ann_range_result_convert_ns += ann_index_stats.result_process_costs_ns.value();
1327
18
        _opts.stats->ann_range_engine_convert_ns += ann_index_stats.engine_convert_ns.value();
1328
18
        _opts.stats->ann_range_pre_process_ns += ann_index_stats.engine_prepare_ns.value();
1329
18
        _opts.stats->ann_fall_back_brute_force_cnt += ann_index_stats.fall_back_brute_force_cnt;
1330
18
        _opts.stats->ann_range_fallback_by_small_candidate_cnt +=
1331
18
                ann_index_stats.range_fallback_by_small_candidate_cnt;
1332
18
        _opts.stats->ann_range_fallback_small_candidate_rows +=
1333
18
                ann_index_stats.range_fallback_small_candidate_rows;
1334
18
        _opts.stats->ann_index_range_cache_hits += ann_index_stats.range_cache_hits.value();
1335
18
    }
1336
1337
67
    return Status::OK();
1338
67
}
1339
1340
0
bool SegmentIterator::_downgrade_without_index(Status res, bool need_remaining) {
1341
0
    bool is_fallback =
1342
0
            _opts.runtime_state->query_options().enable_fallback_on_missing_inverted_index;
1343
0
    if ((res.code() == ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND && is_fallback) ||
1344
0
        res.code() == ErrorCode::INVERTED_INDEX_BYPASS ||
1345
0
        res.code() == ErrorCode::INVERTED_INDEX_EVALUATE_SKIPPED ||
1346
0
        (res.code() == ErrorCode::INVERTED_INDEX_NO_TERMS && need_remaining) ||
1347
0
        res.code() == ErrorCode::INVERTED_INDEX_FILE_CORRUPTED) {
1348
        // 1. INVERTED_INDEX_FILE_NOT_FOUND means index file has not been built,
1349
        //    usually occurs when creating a new index, queries can be downgraded
1350
        //    without index.
1351
        // 2. INVERTED_INDEX_BYPASS means the hit of condition by index
1352
        //    has reached the optimal limit, downgrade without index query can
1353
        //    improve query performance.
1354
        // 3. INVERTED_INDEX_EVALUATE_SKIPPED means the inverted index is not
1355
        //    suitable for executing this predicate, skipped it and filter data
1356
        //    by function later.
1357
        // 4. INVERTED_INDEX_NO_TERMS means the column has fulltext index,
1358
        //    but the column condition value no terms in specified parser,
1359
        //    such as: where A = '' and B = ','
1360
        //    the predicate of A and B need downgrade without index query.
1361
        // 5. INVERTED_INDEX_FILE_CORRUPTED means the index file is corrupted,
1362
        //    such as when index segment files are not generated
1363
        // above case can downgrade without index query
1364
0
        _opts.stats->inverted_index_downgrade_count++;
1365
0
        if (!res.is<ErrorCode::INVERTED_INDEX_BYPASS>()) {
1366
0
            LOG(INFO) << "will downgrade without index to evaluate predicate, because of res: "
1367
0
                      << res;
1368
0
        } else {
1369
0
            VLOG_DEBUG << "will downgrade without index to evaluate predicate, because of res: "
1370
0
                       << res;
1371
0
        }
1372
0
        return true;
1373
0
    }
1374
0
    return false;
1375
0
}
1376
1377
90
bool SegmentIterator::_column_has_fulltext_index(int32_t cid) {
1378
90
    bool has_fulltext_index =
1379
90
            _index_iterators[cid] != nullptr &&
1380
90
            _index_iterators[cid]->get_reader(InvertedIndexReaderType::FULLTEXT) &&
1381
90
            _index_iterators[cid]->get_reader(InvertedIndexReaderType::STRING_TYPE) == nullptr;
1382
1383
90
    return has_fulltext_index;
1384
90
}
1385
1386
45
inline bool SegmentIterator::_inverted_index_not_support_pred_type(const PredicateType& type) {
1387
45
    return type == PredicateType::BF;
1388
45
}
1389
1390
Status SegmentIterator::_apply_inverted_index_on_column_predicate(
1391
        std::shared_ptr<ColumnPredicate> pred,
1392
45
        std::vector<std::shared_ptr<ColumnPredicate>>& remaining_predicates, bool* continue_apply) {
1393
45
    if (!_check_apply_by_inverted_index(pred)) {
1394
0
        remaining_predicates.emplace_back(pred);
1395
45
    } else {
1396
45
        bool need_remaining_after_evaluate = _column_has_fulltext_index(pred->column_id()) &&
1397
45
                                             PredicateTypeTraits::is_equal_or_list(pred->type());
1398
45
        Status res =
1399
45
                pred->evaluate(_storage_name_and_type[pred->column_id()],
1400
45
                               _index_iterators[pred->column_id()].get(), num_rows(), &_row_bitmap);
1401
45
        if (!res.ok()) {
1402
0
            if (_downgrade_without_index(res, need_remaining_after_evaluate)) {
1403
0
                remaining_predicates.emplace_back(pred);
1404
0
                return Status::OK();
1405
0
            }
1406
0
            LOG(WARNING) << "failed to evaluate index"
1407
0
                         << ", column predicate type: " << pred->pred_type_string(pred->type())
1408
0
                         << ", error msg: " << res;
1409
0
            return res;
1410
0
        }
1411
1412
45
        if (_row_bitmap.isEmpty()) {
1413
            // all rows have been pruned, no need to process further predicates
1414
0
            *continue_apply = false;
1415
0
        }
1416
1417
45
        if (need_remaining_after_evaluate) {
1418
0
            remaining_predicates.emplace_back(pred);
1419
0
            return Status::OK();
1420
0
        }
1421
45
        if (!pred->is_runtime_filter()) {
1422
45
            _column_predicate_index_exec_status[pred->column_id()][pred] = true;
1423
45
        }
1424
45
    }
1425
45
    return Status::OK();
1426
45
}
1427
1428
28.0k
bool SegmentIterator::_need_read_data(ColumnId cid) {
1429
28.0k
    if (_opts.runtime_state && !_opts.runtime_state->query_options().enable_no_need_read_data_opt) {
1430
0
        return true;
1431
0
    }
1432
28.0k
    if (_can_skip_reading_extra_column(cid)) {
1433
0
        return false;
1434
0
    }
1435
    // only support DUP_KEYS and UNIQUE_KEYS with MOW
1436
28.0k
    if (!((_opts.tablet_schema->keys_type() == KeysType::DUP_KEYS ||
1437
28.0k
           (_opts.tablet_schema->keys_type() == KeysType::UNIQUE_KEYS &&
1438
11.1k
            _opts.enable_unique_key_merge_on_write)))) {
1439
7.58k
        return true;
1440
7.58k
    }
1441
    // this is a virtual column, we always need to read data
1442
20.4k
    if (_virtual_column_exprs.contains(cid)) {
1443
0
        return true;
1444
0
    }
1445
1446
    // if there is a delete predicate, we always need to read data
1447
20.4k
    if (_has_delete_predicate(cid)) {
1448
1.74k
        return true;
1449
1.74k
    }
1450
18.7k
    if (_output_columns.count(-1)) {
1451
        // if _output_columns contains -1, it means that the light
1452
        // weight schema change may not be enabled or other reasons
1453
        // caused the column unique_id not be set, to prevent errors
1454
        // occurring, return true here that column data needs to be read
1455
0
        return true;
1456
0
    }
1457
18.7k
    const auto& column = _opts.tablet_schema->column(cid);
1458
    // Different subcolumns may share the same parent_unique_id, so we choose to abandon this optimization.
1459
18.7k
    if (column.is_extracted_column() &&
1460
18.7k
        _opts.push_down_agg_type_opt != TPushAggOp::COUNT_ON_INDEX) {
1461
406
        return true;
1462
406
    }
1463
18.3k
    int32_t unique_id = column.unique_id();
1464
18.3k
    if (unique_id < 0) {
1465
9
        unique_id = column.parent_unique_id();
1466
9
    }
1467
    // A column can skip data reads when its predicates have already been fully resolved.
1468
    // zonemap_always_true_pred_cols is produced only for non-key columns because key columns
1469
    // must remain readable for short-key range seeks.
1470
18.3k
    const bool used_by_common_expr =
1471
18.3k
            cid < _is_common_expr_column.size() && _is_common_expr_column[cid];
1472
18.3k
    const bool zonemap_always_true_filter_column =
1473
18.3k
            _opts.zonemap_always_true_pred_cols.contains(cid);
1474
18.3k
    DCHECK(!zonemap_always_true_filter_column || !column.is_key());
1475
18.3k
    const bool no_need_read_filter_column =
1476
18.3k
            (_need_read_data_indices.contains(cid) && !_need_read_data_indices[cid]) ||
1477
18.3k
            (zonemap_always_true_filter_column && !used_by_common_expr);
1478
18.3k
    if ((no_need_read_filter_column && !_output_columns.contains(unique_id)) ||
1479
18.3k
        (no_need_read_filter_column && _output_columns.count(unique_id) == 1 &&
1480
18.2k
         _opts.push_down_agg_type_opt == TPushAggOp::COUNT_ON_INDEX)) {
1481
41
        VLOG_DEBUG << "SegmentIterator no need read data for column: "
1482
0
                   << _opts.tablet_schema->column_by_uid(unique_id).name();
1483
41
        return false;
1484
41
    }
1485
18.2k
    return true;
1486
18.3k
}
1487
1488
63
Status SegmentIterator::_apply_inverted_index() {
1489
63
    std::vector<std::shared_ptr<ColumnPredicate>> remaining_predicates;
1490
63
    std::set<std::shared_ptr<ColumnPredicate>> no_need_to_pass_column_predicate_set;
1491
1492
63
    for (auto pred : _col_predicates) {
1493
45
        if (no_need_to_pass_column_predicate_set.count(pred) > 0) {
1494
0
            continue;
1495
45
        } else {
1496
45
            bool continue_apply = true;
1497
45
            RETURN_IF_ERROR(_apply_inverted_index_on_column_predicate(pred, remaining_predicates,
1498
45
                                                                      &continue_apply));
1499
45
            if (!continue_apply) {
1500
0
                break;
1501
0
            }
1502
45
        }
1503
45
    }
1504
1505
63
    _col_predicates = std::move(remaining_predicates);
1506
63
    return Status::OK();
1507
63
}
1508
1509
/**
1510
 * @brief Checks if all conditions related to a specific column have passed in both
1511
 * `_column_predicate_inverted_index_status` and `_common_expr_inverted_index_status`.
1512
 *
1513
 * This function first checks the conditions in `_column_predicate_inverted_index_status`
1514
 * for the given `ColumnId`. If all conditions pass, it sets `default_return` to `true`.
1515
 * It then checks the conditions in `_common_expr_inverted_index_status` for the same column.
1516
 *
1517
 * The function returns `true` if all conditions in both maps pass. If any condition fails
1518
 * in either map, the function immediately returns `false`. If the column does not exist
1519
 * in one of the maps, the function returns `default_return`.
1520
 *
1521
 * @param cid The ColumnId of the column to check.
1522
 * @param default_return The default value to return if the column is not found in the status maps.
1523
 * @return true if all conditions in both status maps pass, or if the column is not found
1524
 *         and `default_return` is true.
1525
 * @return false if any condition in either status map fails, or if the column is not found
1526
 *         and `default_return` is false.
1527
 */
1528
bool SegmentIterator::_check_all_conditions_passed_inverted_index_for_column(ColumnId cid,
1529
145
                                                                             bool default_return) {
1530
145
    auto pred_it = _column_predicate_index_exec_status.find(cid);
1531
145
    if (pred_it != _column_predicate_index_exec_status.end()) {
1532
47
        const auto& pred_map = pred_it->second;
1533
47
        bool pred_passed = std::all_of(pred_map.begin(), pred_map.end(),
1534
47
                                       [](const auto& pred_entry) { return pred_entry.second; });
1535
47
        if (!pred_passed) {
1536
1
            return false;
1537
46
        } else {
1538
46
            default_return = true;
1539
46
        }
1540
47
    }
1541
1542
144
    auto expr_it = _common_expr_index_exec_status.find(cid);
1543
144
    if (expr_it != _common_expr_index_exec_status.end()) {
1544
18
        const auto& expr_map = expr_it->second;
1545
18
        return std::all_of(expr_map.begin(), expr_map.end(),
1546
18
                           [](const auto& expr_entry) { return expr_entry.second; });
1547
18
    }
1548
126
    return default_return;
1549
144
}
1550
1551
3.03k
Status SegmentIterator::_init_return_column_iterators() {
1552
3.03k
    SCOPED_RAW_TIMER(&_opts.stats->segment_iterator_init_return_column_iterators_timer_ns);
1553
3.03k
    if (_cur_rowid >= num_rows()) {
1554
0
        return Status::OK();
1555
0
    }
1556
1557
7.50k
    for (auto cid : _schema->column_ids()) {
1558
7.50k
        if (_schema->column(cid)->name() == BeConsts::ROWID_COL) {
1559
0
            _column_iterators[cid].reset(
1560
0
                    new RowIdColumnIterator(_opts.tablet_id, _opts.rowset_id, _segment->id()));
1561
0
            continue;
1562
0
        }
1563
1564
7.50k
        if (_schema->column(cid)->name().starts_with(BeConsts::GLOBAL_ROWID_COL)) {
1565
0
            auto& id_file_map = _opts.runtime_state->get_id_file_map();
1566
0
            uint32_t file_id = id_file_map->get_file_mapping_id(std::make_shared<FileMapping>(
1567
0
                    _opts.tablet_id, _opts.rowset_id, _segment->id()));
1568
0
            _column_iterators[cid].reset(new RowIdColumnIteratorV2(
1569
0
                    IdManager::ID_VERSION, BackendOptions::get_backend_id(), file_id));
1570
0
            continue;
1571
0
        }
1572
1573
7.50k
        if (_schema->column(cid)->name().starts_with(BeConsts::VIRTUAL_COLUMN_PREFIX)) {
1574
0
            _column_iterators[cid] = std::make_unique<VirtualColumnIterator>();
1575
0
            continue;
1576
0
        }
1577
1578
7.50k
        std::set<ColumnId> del_cond_id_set;
1579
7.50k
        _opts.delete_condition_predicates->get_all_column_ids(del_cond_id_set);
1580
7.50k
        std::vector<bool> tmp_is_pred_column;
1581
7.50k
        tmp_is_pred_column.resize(_schema->columns().size(), false);
1582
7.50k
        for (auto predicate : _col_predicates) {
1583
130
            auto p_cid = predicate->column_id();
1584
130
            tmp_is_pred_column[p_cid] = true;
1585
130
        }
1586
        // handle delete_condition
1587
7.50k
        for (auto d_cid : del_cond_id_set) {
1588
1.32k
            tmp_is_pred_column[d_cid] = true;
1589
1.32k
        }
1590
1591
7.50k
        if (_column_iterators[cid] == nullptr) {
1592
7.50k
            RETURN_IF_ERROR(_segment->new_column_iterator(_opts.tablet_schema->column(cid),
1593
7.50k
                                                          &_column_iterators[cid], &_opts,
1594
7.50k
                                                          &_variant_sparse_column_cache));
1595
7.50k
            ColumnIteratorOptions iter_opts {
1596
7.50k
                    .use_page_cache = _opts.use_page_cache,
1597
                    // If the col is predicate column, then should read the last page to check
1598
                    // if the column is full dict encoding
1599
7.50k
                    .is_predicate_column = tmp_is_pred_column[cid],
1600
7.50k
                    .file_reader = _file_reader.get(),
1601
7.50k
                    .stats = _opts.stats,
1602
7.50k
                    .io_ctx = _opts.io_ctx,
1603
7.50k
            };
1604
7.50k
            RETURN_IF_ERROR(_column_iterators[cid]->init(iter_opts));
1605
7.50k
        }
1606
7.50k
    }
1607
1608
3.03k
#ifndef NDEBUG
1609
3.03k
    for (const auto& entry : _virtual_column_exprs) {
1610
0
        ColumnId vir_col_cid = entry.first;
1611
0
        DCHECK(_column_iterators[vir_col_cid] != nullptr)
1612
0
                << "Virtual column iterator for " << vir_col_cid << " should not be null";
1613
0
        ColumnIterator* column_iter = _column_iterators[vir_col_cid].get();
1614
0
        DCHECK(dynamic_cast<VirtualColumnIterator*>(column_iter) != nullptr)
1615
0
                << "Virtual column iterator for " << vir_col_cid
1616
0
                << " should be VirtualColumnIterator";
1617
0
    }
1618
3.03k
#endif
1619
3.03k
    return Status::OK();
1620
3.03k
}
1621
1622
3.03k
Status SegmentIterator::_init_index_iterators() {
1623
3.03k
    SCOPED_RAW_TIMER(&_opts.stats->segment_iterator_init_index_iterators_timer_ns);
1624
3.03k
    if (_cur_rowid >= num_rows()) {
1625
0
        return Status::OK();
1626
0
    }
1627
1628
3.03k
    _index_query_context = std::make_shared<IndexQueryContext>();
1629
3.03k
    _index_query_context->io_ctx = &_opts.io_ctx;
1630
3.03k
    _index_query_context->stats = _opts.stats;
1631
3.03k
    _index_query_context->runtime_state = _opts.runtime_state;
1632
1633
3.03k
    if (_score_runtime) {
1634
0
        _index_query_context->collection_statistics = _opts.collection_statistics;
1635
0
        _index_query_context->collection_similarity = std::make_shared<CollectionSimilarity>();
1636
0
        _index_query_context->query_limit = _score_runtime->get_limit();
1637
0
        _index_query_context->is_asc = _score_runtime->is_asc();
1638
0
    }
1639
1640
    // Inverted index iterators
1641
7.50k
    for (auto cid : _schema->column_ids()) {
1642
        // Use segment’s own index_meta, for compatibility with future indexing needs to default to lowercase.
1643
7.50k
        if (_index_iterators[cid] == nullptr) {
1644
            // In the _opts.tablet_schema, the sub-column type information for the variant is FieldType::OLAP_FIELD_TYPE_VARIANT.
1645
            // This is because the sub-column is created in create_materialized_variant_column.
1646
            // We use this column to locate the metadata for the inverted index, which requires a unique_id and path.
1647
7.50k
            const auto& column = _opts.tablet_schema->column(cid);
1648
7.50k
            std::vector<const TabletIndex*> inverted_indexs;
1649
            // Keep shared_ptr alive to prevent use-after-free when accessing raw pointers
1650
7.50k
            TabletIndexes inverted_indexs_holder;
1651
            // If the column is an extracted column, we need to find the sub-column in the parent column reader.
1652
7.50k
            std::shared_ptr<ColumnReader> column_reader;
1653
7.50k
            if (column.is_extracted_column()) {
1654
204
                if (!_segment->_column_reader_cache->get_column_reader(
1655
204
                            column.parent_unique_id(), &column_reader, _opts.stats) ||
1656
204
                    column_reader == nullptr) {
1657
0
                    continue;
1658
0
                }
1659
204
                auto* variant_reader = assert_cast<VariantColumnReader*>(column_reader.get());
1660
204
                DataTypePtr data_type = _storage_name_and_type[cid].second;
1661
204
                if (data_type != nullptr &&
1662
204
                    data_type->get_primitive_type() == PrimitiveType::TYPE_VARIANT) {
1663
19
                    DataTypePtr inferred_type;
1664
19
                    Status st = variant_reader->infer_data_type_for_path(
1665
19
                            &inferred_type, column, _opts, _segment->_column_reader_cache.get());
1666
19
                    if (st.ok() && inferred_type != nullptr) {
1667
19
                        data_type = inferred_type;
1668
19
                    }
1669
19
                }
1670
204
                inverted_indexs_holder = variant_reader->find_subcolumn_tablet_indexes(
1671
204
                        column, data_type, _opts.stats);
1672
                // Extract raw pointers from shared_ptr for iteration
1673
204
                for (const auto& index_ptr : inverted_indexs_holder) {
1674
87
                    inverted_indexs.push_back(index_ptr.get());
1675
87
                }
1676
204
            }
1677
            // If the column is not an extracted column, we can directly get the inverted index metadata from the tablet schema.
1678
7.29k
            else {
1679
7.29k
                inverted_indexs = _segment->_tablet_schema->inverted_indexs(column);
1680
7.29k
            }
1681
7.50k
            if (column.is_extracted_column() && inverted_indexs.empty() && _opts.stats != nullptr) {
1682
127
                const auto relative_path = column.path_info_ptr()->copy_pop_front().get_path();
1683
127
                const auto diagnostic = fmt::format(
1684
127
                        "[VariantSearchBinding] phase=init_index_iterators "
1685
127
                        "result=no_candidate tablet_id={} rowset_id={} segment_id={} cid={} "
1686
127
                        "logical_path={} relative_path={} materialized_column={}",
1687
127
                        _tablet_id, _segment->rowset_id().to_string(), _segment->id(), cid,
1688
127
                        column.path_info_ptr()->get_path(), relative_path, column.name());
1689
127
                VLOG_DEBUG << diagnostic;
1690
127
                _opts.stats->inverted_index_stats.add_binding_diagnostic(diagnostic);
1691
127
            }
1692
7.50k
            for (const auto& inverted_index : inverted_indexs) {
1693
2.67k
                const bool had_iterator = _index_iterators[cid] != nullptr;
1694
2.67k
                RETURN_IF_ERROR(_segment->new_index_iterator(column, inverted_index, _opts,
1695
2.67k
                                                             &_index_iterators[cid]));
1696
2.67k
                if ((column.is_extracted_column() || column.is_variant_type()) &&
1697
2.67k
                    _opts.stats != nullptr) {
1698
87
                    const auto diagnostic = fmt::format(
1699
87
                            "[VariantSearchBinding] phase=init_index_iterators "
1700
87
                            "result={} tablet_id={} rowset_id={} segment_id={} cid={} "
1701
87
                            "logical_path={} materialized_column={} index_id={} suffix={} "
1702
87
                            "field_pattern={} iterator_state={}",
1703
87
                            _index_iterators[cid] == nullptr ? "no_iterator" : "accepted",
1704
87
                            _tablet_id, _segment->rowset_id().to_string(), _segment->id(), cid,
1705
87
                            column.has_path_info() ? column.path_info_ptr()->get_path()
1706
87
                                                   : column.name(),
1707
87
                            column.name(), inverted_index->index_id(),
1708
87
                            inverted_index->get_index_suffix(), inverted_index->field_pattern(),
1709
87
                            had_iterator ? "preserved" : "created");
1710
87
                    VLOG_DEBUG << diagnostic;
1711
87
                    _opts.stats->inverted_index_stats.add_binding_diagnostic(diagnostic);
1712
87
                }
1713
2.67k
            }
1714
7.50k
            if (_index_iterators[cid] != nullptr) {
1715
2.65k
                _index_iterators[cid]->set_context(_index_query_context);
1716
2.65k
            }
1717
7.50k
        }
1718
7.50k
    }
1719
1720
    // Ann index iterators
1721
7.50k
    for (auto cid : _schema->column_ids()) {
1722
7.50k
        if (_index_iterators[cid] == nullptr) {
1723
4.85k
            const auto& column = _opts.tablet_schema->column(cid);
1724
4.85k
            const auto* index_meta = _segment->_tablet_schema->ann_index(column);
1725
4.85k
            if (index_meta) {
1726
1
                RETURN_IF_ERROR(_segment->new_index_iterator(column, index_meta, _opts,
1727
1
                                                             &_index_iterators[cid]));
1728
1729
1
                if (_index_iterators[cid] != nullptr) {
1730
1
                    _index_iterators[cid]->set_context(_index_query_context);
1731
1
                }
1732
1
            }
1733
4.85k
        }
1734
7.50k
    }
1735
1736
3.03k
    return Status::OK();
1737
3.03k
}
1738
1739
Status SegmentIterator::_lookup_ordinal(const RowCursor& key, bool is_include, rowid_t upper_bound,
1740
0
                                        rowid_t* rowid) {
1741
0
    if (_segment->_tablet_schema->keys_type() == UNIQUE_KEYS &&
1742
0
        _segment->get_primary_key_index() != nullptr) {
1743
0
        return _lookup_ordinal_from_pk_index(key, is_include, rowid);
1744
0
    }
1745
0
    return _lookup_ordinal_from_sk_index(key, is_include, upper_bound, rowid);
1746
0
}
1747
1748
// look up one key to get its ordinal at which can get data by using short key index.
1749
// 'upper_bound' is defined the max ordinal the function will search.
1750
// We use upper_bound to reduce search times.
1751
// If we find a valid ordinal, it will be set in rowid and with Status::OK()
1752
// If we can not find a valid key in this segment, we will set rowid to upper_bound
1753
// Otherwise return error.
1754
// 1. get [start, end) ordinal through short key index
1755
// 2. binary search to find exact ordinal that match the input condition
1756
// Make is_include template to reduce branch
1757
Status SegmentIterator::_lookup_ordinal_from_sk_index(const RowCursor& key, bool is_include,
1758
0
                                                      rowid_t upper_bound, rowid_t* rowid) {
1759
0
    const ShortKeyIndexDecoder* sk_index_decoder = _segment->get_short_key_index();
1760
0
    DCHECK(sk_index_decoder != nullptr);
1761
1762
0
    std::string index_key;
1763
0
    key.encode_key_with_padding(&index_key, _segment->_tablet_schema->num_short_key_columns(),
1764
0
                                is_include);
1765
1766
0
    const auto& key_col_ids = key.schema()->column_ids();
1767
1768
0
    ssize_t start_block_id = 0;
1769
0
    auto start_iter = sk_index_decoder->lower_bound(index_key);
1770
0
    if (start_iter.valid()) {
1771
        // Because previous block may contain this key, so we should set rowid to
1772
        // last block's first row.
1773
0
        start_block_id = start_iter.ordinal();
1774
0
        if (start_block_id > 0) {
1775
0
            start_block_id--;
1776
0
        }
1777
0
    } else {
1778
        // When we don't find a valid index item, which means all short key is
1779
        // smaller than input key, this means that this key may exist in the last
1780
        // row block. so we set the rowid to first row of last row block.
1781
0
        start_block_id = sk_index_decoder->num_items() - 1;
1782
0
    }
1783
0
    rowid_t start = cast_set<rowid_t>(start_block_id) * sk_index_decoder->num_rows_per_block();
1784
1785
0
    rowid_t end = upper_bound;
1786
0
    auto end_iter = sk_index_decoder->upper_bound(index_key);
1787
0
    if (end_iter.valid()) {
1788
0
        end = cast_set<rowid_t>(end_iter.ordinal()) * sk_index_decoder->num_rows_per_block();
1789
0
    }
1790
1791
    // binary search to find the exact key
1792
0
    while (start < end) {
1793
0
        rowid_t mid = (start + end) / 2;
1794
0
        RETURN_IF_ERROR(_seek_and_peek(mid));
1795
0
        int cmp = _compare_short_key_with_seek_block(key, key_col_ids);
1796
0
        if (cmp > 0) {
1797
0
            start = mid + 1;
1798
0
        } else if (cmp == 0) {
1799
0
            if (is_include) {
1800
                // lower bound
1801
0
                end = mid;
1802
0
            } else {
1803
                // upper bound
1804
0
                start = mid + 1;
1805
0
            }
1806
0
        } else {
1807
0
            end = mid;
1808
0
        }
1809
0
    }
1810
1811
0
    *rowid = start;
1812
0
    return Status::OK();
1813
0
}
1814
1815
Status SegmentIterator::_lookup_ordinal_from_pk_index(const RowCursor& key, bool is_include,
1816
0
                                                      rowid_t* rowid) {
1817
0
    DCHECK(_segment->_tablet_schema->keys_type() == UNIQUE_KEYS);
1818
0
    const PrimaryKeyIndexReader* pk_index_reader = _segment->get_primary_key_index();
1819
0
    DCHECK(pk_index_reader != nullptr);
1820
1821
0
    std::string index_key;
1822
0
    key.encode_key_with_padding<true>(&index_key, _segment->_tablet_schema->num_key_columns(),
1823
0
                                      is_include);
1824
0
    if (index_key < _segment->min_key()) {
1825
0
        *rowid = 0;
1826
0
        return Status::OK();
1827
0
    } else if (index_key > _segment->max_key()) {
1828
0
        *rowid = num_rows();
1829
0
        return Status::OK();
1830
0
    }
1831
0
    bool exact_match = false;
1832
1833
0
    std::unique_ptr<segment_v2::IndexedColumnIterator> index_iterator;
1834
0
    RETURN_IF_ERROR(pk_index_reader->new_iterator(&index_iterator, _opts.stats, &_opts.io_ctx));
1835
1836
0
    Status status = index_iterator->seek_at_or_after(&index_key, &exact_match);
1837
0
    if (UNLIKELY(!status.ok())) {
1838
0
        *rowid = num_rows();
1839
0
        if (status.is<ENTRY_NOT_FOUND>()) {
1840
0
            return Status::OK();
1841
0
        }
1842
0
        return status;
1843
0
    }
1844
0
    *rowid = cast_set<rowid_t>(index_iterator->get_current_ordinal());
1845
1846
    // The sequence column needs to be removed from primary key index when comparing key
1847
0
    bool has_seq_col = _segment->_tablet_schema->has_sequence_col();
1848
    // Used to get key range from primary key index,
1849
    // for mow with cluster key table, we should get key range from short key index.
1850
0
    DCHECK(_segment->_tablet_schema->cluster_key_uids().empty());
1851
1852
    // if full key is exact_match, the primary key without sequence column should also the same
1853
0
    if (has_seq_col && !exact_match) {
1854
0
        size_t seq_col_length =
1855
0
                _segment->_tablet_schema->column(_segment->_tablet_schema->sequence_col_idx())
1856
0
                        .length() +
1857
0
                1;
1858
0
        auto index_type = DataTypeFactory::instance().create_data_type(
1859
0
                _segment->_pk_index_reader->type(), 1, 0);
1860
0
        auto index_column = index_type->create_column();
1861
0
        size_t num_to_read = 1;
1862
0
        size_t num_read = num_to_read;
1863
0
        RETURN_IF_ERROR(index_iterator->next_batch(&num_read, index_column));
1864
0
        DCHECK(num_to_read == num_read);
1865
1866
0
        Slice sought_key =
1867
0
                Slice(index_column->get_data_at(0).data, index_column->get_data_at(0).size);
1868
0
        Slice sought_key_without_seq =
1869
0
                Slice(sought_key.get_data(), sought_key.get_size() - seq_col_length);
1870
1871
        // compare key
1872
0
        if (Slice(index_key).compare(sought_key_without_seq) == 0) {
1873
0
            exact_match = true;
1874
0
        }
1875
0
    }
1876
1877
    // find the key in primary key index, and the is_include is false, so move
1878
    // to the next row.
1879
0
    if (exact_match && !is_include) {
1880
0
        *rowid += 1;
1881
0
    }
1882
0
    return Status::OK();
1883
0
}
1884
1885
// seek to the row and load that row to _key_cursor
1886
0
Status SegmentIterator::_seek_and_peek(rowid_t rowid) {
1887
0
    {
1888
0
        _opts.stats->block_init_seek_num += 1;
1889
0
        SCOPED_RAW_TIMER(&_opts.stats->block_init_seek_ns);
1890
0
        RETURN_IF_ERROR(_seek_columns(_seek_schema->column_ids(), rowid));
1891
0
    }
1892
0
    size_t num_rows = 1;
1893
1894
    //note(wb) reset _seek_block for memory reuse
1895
    // it is easier to use row based memory layout for clear memory
1896
0
    for (int i = 0; i < _seek_block.size(); i++) {
1897
0
        _seek_block[i]->clear();
1898
0
    }
1899
0
    RETURN_IF_ERROR(_read_columns(_seek_schema->column_ids(), _seek_block, num_rows));
1900
0
    return Status::OK();
1901
0
}
1902
1903
0
Status SegmentIterator::_seek_columns(const std::vector<ColumnId>& column_ids, rowid_t pos) {
1904
0
    for (auto cid : column_ids) {
1905
0
        if (!_need_read_data(cid)) {
1906
0
            continue;
1907
0
        }
1908
0
        RETURN_IF_ERROR(_column_iterators[cid]->seek_to_ordinal(pos));
1909
0
    }
1910
0
    return Status::OK();
1911
0
}
1912
1913
/* ---------------------- for vectorization implementation  ---------------------- */
1914
1915
/**
1916
 *  For storage layer data type, can be measured from two perspectives:
1917
 *  1 Whether the type can be read in a fast way(batch read using SIMD)
1918
 *    Such as integer type and float type, this type can be read in SIMD way.
1919
 *    For the type string/bitmap/hll, they can not be read in batch way, so read this type data is slow.
1920
 *   If a type can be read fast, we can try to eliminate Lazy Materialization, because we think for this type, seek cost > read cost.
1921
 *   This is an estimate, if we want more precise cost, statistics collection is necessary(this is a todo).
1922
 *   In short, when returned non-pred columns contains string/hll/bitmap, we using Lazy Materialization.
1923
 *   Otherwise, we disable it.
1924
 *
1925
 *   When Lazy Materialization enable, we need to read column at least two times.
1926
 *   First time to read Pred col, second time to read non-pred.
1927
 *   Here's an interesting question to research, whether read Pred col once is the best plan.
1928
 *   (why not read Pred col twice or more?)
1929
 *
1930
 *   When Lazy Materialization disable, we just need to read once.
1931
 *
1932
 *
1933
 *  2 Whether the predicate type can be evaluate in a fast way(using SIMD to eval pred)
1934
 *    Such as integer type and float type, they can be eval fast.
1935
 *    But for BloomFilter/string/date, they eval slow.
1936
 *    If a type can be eval fast, we use vectorization to eval it.
1937
 *    Otherwise, we use short-circuit to eval it.
1938
 *
1939
 *
1940
 */
1941
1942
// todo(wb) need a UT here
1943
3.03k
Status SegmentIterator::_vec_init_lazy_materialization() {
1944
3.03k
    _is_pred_column.resize(_schema->columns().size(), false);
1945
1946
    // including short/vec/delete pred
1947
3.03k
    std::set<ColumnId> pred_column_ids;
1948
3.03k
    _lazy_materialization_read = false;
1949
1950
3.03k
    std::set<ColumnId> del_cond_id_set;
1951
3.03k
    _opts.delete_condition_predicates->get_all_column_ids(del_cond_id_set);
1952
1953
3.03k
    std::set<std::shared_ptr<const ColumnPredicate>> delete_predicate_set {};
1954
3.03k
    _opts.delete_condition_predicates->get_all_column_predicate(delete_predicate_set);
1955
3.03k
    for (auto predicate : delete_predicate_set) {
1956
467
        if (PredicateTypeTraits::is_range(predicate->type())) {
1957
327
            _delete_range_column_ids.push_back(predicate->column_id());
1958
327
        } else if (PredicateTypeTraits::is_bloom_filter(predicate->type())) {
1959
0
            _delete_bloom_filter_column_ids.push_back(predicate->column_id());
1960
0
        }
1961
467
    }
1962
1963
    // Step1: extract columns that can be lazy materialization
1964
3.03k
    if (!_col_predicates.empty() || !del_cond_id_set.empty()) {
1965
486
        std::set<ColumnId> short_cir_pred_col_id_set; // using set for distinct cid
1966
486
        std::set<ColumnId> vec_pred_col_id_set;
1967
1968
486
        for (auto predicate : _col_predicates) {
1969
19
            auto cid = predicate->column_id();
1970
19
            _is_pred_column[cid] = true;
1971
19
            pred_column_ids.insert(cid);
1972
1973
            // check pred using short eval or vec eval
1974
19
            if (_can_evaluated_by_vectorized(predicate)) {
1975
19
                vec_pred_col_id_set.insert(cid);
1976
19
                _pre_eval_block_predicate.push_back(predicate);
1977
19
            } else {
1978
0
                short_cir_pred_col_id_set.insert(cid);
1979
0
                _short_cir_eval_predicate.push_back(predicate);
1980
0
            }
1981
19
            if (predicate->is_runtime_filter()) {
1982
0
                _filter_info_id.push_back(predicate);
1983
0
            }
1984
19
        }
1985
1986
        // handle delete_condition
1987
486
        if (!del_cond_id_set.empty()) {
1988
467
            short_cir_pred_col_id_set.insert(del_cond_id_set.begin(), del_cond_id_set.end());
1989
467
            pred_column_ids.insert(del_cond_id_set.begin(), del_cond_id_set.end());
1990
1991
467
            for (auto cid : del_cond_id_set) {
1992
467
                _is_pred_column[cid] = true;
1993
467
            }
1994
467
        }
1995
1996
486
        _vec_pred_column_ids.assign(vec_pred_col_id_set.cbegin(), vec_pred_col_id_set.cend());
1997
486
        _short_cir_pred_column_ids.assign(short_cir_pred_col_id_set.cbegin(),
1998
486
                                          short_cir_pred_col_id_set.cend());
1999
486
    }
2000
2001
3.03k
    if (!_vec_pred_column_ids.empty()) {
2002
19
        _is_need_vec_eval = true;
2003
19
    }
2004
3.03k
    if (!_short_cir_pred_column_ids.empty()) {
2005
467
        _is_need_short_eval = true;
2006
467
    }
2007
2008
    // Step2: extract columns that can execute expr context
2009
3.03k
    _is_common_expr_column.resize(_schema->columns().size(), false);
2010
3.03k
    if (!_common_expr_ctxs_push_down.empty()) {
2011
6
        for (const auto& expr_ctx : _common_expr_ctxs_push_down) {
2012
6
            RETURN_IF_ERROR(_extract_common_expr_columns(expr_ctx->root()));
2013
6
        }
2014
6
        if (!_common_expr_columns.empty()) {
2015
6
            _is_need_expr_eval = true;
2016
12
            for (auto cid : _schema->column_ids()) {
2017
                // pred column also needs to be filtered by expr, exclude additional delete condition column.
2018
                // if delete condition column not in the block, no filter is needed
2019
                // and will be removed from _columns_to_filter in the first next_batch.
2020
12
                if (_is_common_expr_column[cid] || _is_pred_column[cid]) {
2021
6
                    auto loc = _schema->column_index(cid);
2022
6
                    _columns_to_filter.push_back(loc);
2023
2024
6
                    const auto field_type = _schema->column(cid)->type();
2025
6
                    if (_is_common_expr_column[cid] && _enable_prune_nested_column &&
2026
6
                        (field_type == FieldType::OLAP_FIELD_TYPE_STRUCT ||
2027
0
                         field_type == FieldType::OLAP_FIELD_TYPE_ARRAY ||
2028
0
                         field_type == FieldType::OLAP_FIELD_TYPE_MAP)) {
2029
0
                        DCHECK(_column_iterators[cid]);
2030
0
                        if (_column_iterators[cid]->read_requirement() ==
2031
0
                                    ColumnIterator::ReadRequirement::PREDICATE &&
2032
0
                            _column_iterators[cid]->has_lazy_read_target()) {
2033
                            // Only split lazy recovery for complex common expr columns that have
2034
                            // both predicate-only and non-predicate nested targets. The two requirement
2035
                            // checks already imply that nested-column pruning happened: without an
2036
                            // explicit predicate sub-path the parent would not be
2037
                            // PREDICATE, and without a pruned non-predicate child there
2038
                            // would be no lazy target to recover after filtering.
2039
0
                            _support_lazy_read_pruned_columns.emplace(cid);
2040
0
                        }
2041
0
                    }
2042
6
                }
2043
12
            }
2044
2045
6
            for (const auto& entry : _virtual_column_exprs) {
2046
0
                _columns_to_filter.push_back(_schema->column_index(entry.first));
2047
0
            }
2048
6
        }
2049
6
    }
2050
2051
    // Step 3: fill non predicate columns and second read column
2052
    // if _schema columns size equal to pred_column_ids size, lazy_materialization_read is false,
2053
    // all columns are lazy materialization columns without non predicte column.
2054
    // If common expr pushdown exists, and expr column is not contained in lazy materialization columns,
2055
    // add to second read column, which will be read after lazy materialization
2056
3.03k
    if (_schema->column_ids().size() > pred_column_ids.size()) {
2057
        // pred_column_ids maybe empty, so that could not set _lazy_materialization_read = true here
2058
        // has to check there is at least one predicate column
2059
7.44k
        for (auto cid : _schema->column_ids()) {
2060
7.44k
            if (!_is_pred_column[cid]) {
2061
7.01k
                if (_is_need_vec_eval || _is_need_short_eval) {
2062
881
                    _lazy_materialization_read = true;
2063
881
                }
2064
7.01k
                if (_is_common_expr_column[cid]) {
2065
6
                    _common_expr_column_ids.push_back(cid);
2066
7.00k
                } else {
2067
7.00k
                    _non_predicate_columns.push_back(cid);
2068
7.00k
                }
2069
7.01k
            }
2070
7.44k
        }
2071
2.97k
    }
2072
2073
    // Step 4: fill first read columns
2074
3.03k
    if (_lazy_materialization_read) {
2075
        // insert pred cid to first_read_columns
2076
429
        for (auto cid : pred_column_ids) {
2077
429
            _predicate_column_ids.push_back(cid);
2078
429
        }
2079
2.60k
    } else if (!_is_need_vec_eval && !_is_need_short_eval && !_is_need_expr_eval) {
2080
8.65k
        for (int i = 0; i < _schema->num_column_ids(); i++) {
2081
6.12k
            auto cid = _schema->column_id(i);
2082
6.12k
            _predicate_column_ids.push_back(cid);
2083
6.12k
        }
2084
2.53k
    } else {
2085
63
        if (_is_need_vec_eval || _is_need_short_eval) {
2086
            // TODO To refactor, because we suppose lazy materialization is better performance.
2087
            // pred exits, but we can eliminate lazy materialization
2088
            // insert pred/non-pred cid to first read columns
2089
57
            std::set<ColumnId> pred_id_set;
2090
57
            pred_id_set.insert(_short_cir_pred_column_ids.begin(),
2091
57
                               _short_cir_pred_column_ids.end());
2092
57
            pred_id_set.insert(_vec_pred_column_ids.begin(), _vec_pred_column_ids.end());
2093
2094
57
            DCHECK(_common_expr_column_ids.empty());
2095
            // _non_predicate_column_ids must be empty. Otherwise _lazy_materialization_read must not false.
2096
114
            for (int i = 0; i < _schema->num_column_ids(); i++) {
2097
57
                auto cid = _schema->column_id(i);
2098
57
                if (pred_id_set.find(cid) != pred_id_set.end()) {
2099
57
                    _predicate_column_ids.push_back(cid);
2100
57
                }
2101
57
            }
2102
57
        } else if (_is_need_expr_eval) {
2103
6
            DCHECK(!_is_need_vec_eval && !_is_need_short_eval);
2104
6
            for (auto cid : _common_expr_columns) {
2105
6
                _predicate_column_ids.push_back(cid);
2106
6
            }
2107
6
        }
2108
63
    }
2109
2110
3.03k
    VLOG_DEBUG << fmt::format(
2111
0
            "Laze materialization init end. "
2112
0
            "lazy_materialization_read: {}, "
2113
0
            "_col_predicates size: {}, "
2114
0
            "_cols_read_by_column_predicate: [{}], "
2115
0
            "_non_predicate_columns: [{}], "
2116
0
            "_cols_read_by_common_expr: [{}], "
2117
0
            "columns_to_filter: [{}], "
2118
0
            "schema_column_id_to_index: [{}]",
2119
0
            _lazy_materialization_read, _col_predicates.size(),
2120
0
            fmt::join(_predicate_column_ids, ","), fmt::join(_non_predicate_columns, ","),
2121
0
            fmt::join(_common_expr_column_ids, ","), fmt::join(_columns_to_filter, ","),
2122
0
            fmt::join(_schema->column_id_to_index(), ","));
2123
3.03k
    return Status::OK();
2124
3.03k
}
2125
2126
19
bool SegmentIterator::_can_evaluated_by_vectorized(std::shared_ptr<ColumnPredicate> predicate) {
2127
19
    auto cid = predicate->column_id();
2128
19
    FieldType field_type = _schema->column(cid)->type();
2129
19
    if (field_type == FieldType::OLAP_FIELD_TYPE_VARIANT) {
2130
        // Use variant cast dst type
2131
0
        field_type = _opts.target_cast_type_for_variants[_schema->column(cid)->name()]
2132
0
                             ->get_storage_field_type();
2133
0
    }
2134
19
    switch (predicate->type()) {
2135
9
    case PredicateType::EQ:
2136
9
    case PredicateType::NE:
2137
9
    case PredicateType::LE:
2138
9
    case PredicateType::LT:
2139
9
    case PredicateType::GE:
2140
19
    case PredicateType::GT: {
2141
19
        if (field_type == FieldType::OLAP_FIELD_TYPE_VARCHAR ||
2142
19
            field_type == FieldType::OLAP_FIELD_TYPE_CHAR ||
2143
19
            field_type == FieldType::OLAP_FIELD_TYPE_STRING) {
2144
5
            return config::enable_low_cardinality_optimize &&
2145
5
                   _opts.io_ctx.reader_type == ReaderType::READER_QUERY &&
2146
5
                   _column_iterators[cid]->is_all_dict_encoding();
2147
14
        } else if (field_type == FieldType::OLAP_FIELD_TYPE_DECIMAL) {
2148
0
            return false;
2149
0
        }
2150
14
        return true;
2151
19
    }
2152
0
    default:
2153
0
        return false;
2154
19
    }
2155
19
}
2156
2157
// These placeholders are used only when the real column data is skipped after
2158
// index/count pushdown has already identified the matching rows. The value is
2159
// irrelevant, but nullable columns must stay non-NULL so COUNT(col) can count
2160
// the matched rows instead of treating every placeholder as NULL.
2161
38
static void insert_many_not_null_defaults(MutableColumnPtr& column, size_t num) {
2162
38
    if (auto* nullable_column = check_and_get_column<ColumnNullable>(column.get())) {
2163
8
        nullable_column->insert_not_null_elements(num);
2164
8
        return;
2165
8
    }
2166
30
    column->insert_many_defaults(num);
2167
30
}
2168
2169
bool SegmentIterator::_prune_column(ColumnId cid, MutableColumnPtr& column,
2170
28.0k
                                    size_t num_of_defaults) {
2171
28.0k
    if (_need_read_data(cid)) {
2172
28.0k
        return false;
2173
28.0k
    }
2174
38
    insert_many_not_null_defaults(column, num_of_defaults);
2175
38
    return true;
2176
28.0k
}
2177
2178
28.0k
bool SegmentIterator::_can_skip_reading_extra_column(ColumnId cid) {
2179
28.0k
    if (!_opts.extra_columns.contains(cid) || _is_pred_column.empty()) {
2180
28.0k
        return false;
2181
28.0k
    }
2182
28.0k
    DCHECK_EQ(_is_pred_column.size(), _is_common_expr_column.size());
2183
0
    DCHECK_LT(cid, _is_pred_column.size());
2184
2185
    // extra_columns is only an optimization hint. The real value is still
2186
    // required when the column participates in expression materialization or
2187
    // any predicate path.
2188
0
    return !_virtual_column_exprs.contains(cid) && !_has_delete_predicate(cid) &&
2189
0
           !_is_pred_column[cid] && !_is_common_expr_column[cid];
2190
28.0k
}
2191
2192
Status SegmentIterator::_read_columns(const std::vector<ColumnId>& column_ids,
2193
0
                                      MutableColumns& column_block, size_t nrows) {
2194
0
    for (auto cid : column_ids) {
2195
0
        auto& column = column_block[cid];
2196
0
        size_t rows_read = nrows;
2197
0
        if (_prune_column(cid, column, rows_read)) {
2198
0
            continue;
2199
0
        }
2200
0
        RETURN_IF_ERROR(_column_iterators[cid]->next_batch(&rows_read, column));
2201
0
        if (nrows != rows_read) {
2202
0
            return Status::Error<ErrorCode::INTERNAL_ERROR>("nrows({}) != rows_read({})", nrows,
2203
0
                                                            rows_read);
2204
0
        }
2205
0
    }
2206
0
    return Status::OK();
2207
0
}
2208
2209
Status SegmentIterator::_init_current_block(Block* block,
2210
                                            std::vector<MutableColumnPtr>& current_columns,
2211
13.3k
                                            uint32_t nrows_read_limit) {
2212
13.3k
    block->clear_column_data(_schema->num_column_ids());
2213
2214
42.3k
    for (size_t i = 0; i < _schema->num_column_ids(); i++) {
2215
28.9k
        auto cid = _schema->column_id(i);
2216
28.9k
        const auto* column_desc = _schema->column(cid);
2217
2218
28.9k
        auto file_column_type = _storage_name_and_type[cid].second;
2219
28.9k
        auto expected_type = Schema::get_data_type_ptr(*column_desc);
2220
28.9k
        if (!_is_pred_column[cid] && !file_column_type->equals(*expected_type)) {
2221
            // The storage layer type is different from schema needed type, so we use storage
2222
            // type to read columns instead of schema type for safety
2223
36
            VLOG_DEBUG << fmt::format(
2224
0
                    "Recreate column with expected type {}, file column type {}, col_name {}, "
2225
0
                    "col_path {}",
2226
0
                    block->get_by_position(i).type->get_name(), file_column_type->get_name(),
2227
0
                    column_desc->name(),
2228
0
                    column_desc->path_info_ptr() == nullptr
2229
0
                            ? ""
2230
0
                            : column_desc->path_info_ptr()->get_path());
2231
            // TODO reuse
2232
36
            current_columns[cid] = file_column_type->create_column();
2233
36
            current_columns[cid]->reserve(nrows_read_limit);
2234
28.9k
        } else {
2235
            // the column in block must clear() here to insert new data
2236
28.9k
            if (_is_pred_column[cid] ||
2237
28.9k
                i >= block->columns()) { //todo(wb) maybe we can release it after output block
2238
2.21k
                if (current_columns[cid].get() == nullptr) {
2239
0
                    return Status::InternalError(
2240
0
                            "SegmentIterator meet invalid column, id={}, name={}", cid,
2241
0
                            _schema->column(cid)->name());
2242
0
                }
2243
2.21k
                current_columns[cid]->clear();
2244
26.7k
            } else { // non-predicate column
2245
26.7k
                current_columns[cid] = std::move(*block->get_by_position(i).column).mutate();
2246
26.7k
                current_columns[cid]->reserve(nrows_read_limit);
2247
26.7k
            }
2248
28.9k
        }
2249
28.9k
    }
2250
2251
13.3k
    for (const auto& entry : _virtual_column_exprs) {
2252
0
        auto cid = entry.first;
2253
0
        current_columns[cid] = ColumnNothing::create(0);
2254
0
        current_columns[cid]->reserve(nrows_read_limit);
2255
0
    }
2256
2257
13.3k
    return Status::OK();
2258
13.3k
}
2259
2260
10.3k
Status SegmentIterator::_output_non_pred_columns(Block* block) {
2261
10.3k
    SCOPED_RAW_TIMER(&_opts.stats->output_col_ns);
2262
10.3k
    VLOG_DEBUG << fmt::format(
2263
0
            "Output non-predicate columns, _non_predicate_columns: [{}], "
2264
0
            "schema_column_id_to_index: [{}]",
2265
0
            fmt::join(_non_predicate_columns, ","), fmt::join(_schema->column_id_to_index(), ","));
2266
10.3k
    RETURN_IF_ERROR(_convert_to_expected_type(_non_predicate_columns));
2267
19.7k
    for (auto cid : _non_predicate_columns) {
2268
19.7k
        auto loc = _schema->column_index(cid);
2269
        // Whether a delete predicate column gets output depends on how the caller builds
2270
        // the block passed to next_batch(). Both calling paths now build the block with
2271
        // only the output schema (return_columns), so delete predicate columns are skipped:
2272
        //
2273
        // 1) VMergeIterator path: block_reset() builds _block using the output schema
2274
        //    (return_columns only), e.g. block has 2 columns {c1, c2}.
2275
        //    Here loc=2 for delete predicate c3, block->columns()=2, so loc < block->columns()
2276
        //    is false, and c3 is skipped.
2277
        //
2278
        // 2) VUnionIterator path: the caller's block is built with only return_columns
2279
        //    (output schema), e.g. block has 2 columns {c1, c2}.
2280
        //    Here loc=2 for c3, block->columns()=2, so loc < block->columns() is false,
2281
        //    and c3 is skipped — same behavior as the VMergeIterator path.
2282
19.7k
        if (loc < block->columns()) {
2283
19.7k
            bool column_in_block_is_nothing = check_and_get_column<const ColumnNothing>(
2284
19.7k
                    block->get_by_position(loc).column.get());
2285
19.7k
            bool column_is_normal = !_virtual_column_exprs.contains(cid);
2286
19.7k
            bool return_column_is_nothing =
2287
19.7k
                    check_and_get_column<const ColumnNothing>(_current_return_columns[cid].get());
2288
19.7k
            VLOG_DEBUG << fmt::format(
2289
0
                    "Cid {} loc {}, column_in_block_is_nothing {}, column_is_normal {}, "
2290
0
                    "return_column_is_nothing {}",
2291
0
                    cid, loc, column_in_block_is_nothing, column_is_normal,
2292
0
                    return_column_is_nothing);
2293
2294
19.7k
            if (column_in_block_is_nothing || column_is_normal) {
2295
19.7k
                block->replace_by_position(loc, std::move(_current_return_columns[cid]));
2296
19.7k
                VLOG_DEBUG << fmt::format(
2297
0
                        "Output non-predicate column, cid: {}, loc: {}, col_name: {}, rows {}", cid,
2298
0
                        loc, _schema->column(cid)->name(),
2299
0
                        block->get_by_position(loc).column->size());
2300
19.7k
            }
2301
            // Means virtual column in block has been materialized(maybe by common expr).
2302
            // so do nothing here.
2303
19.7k
        }
2304
19.7k
    }
2305
10.3k
    return Status::OK();
2306
10.3k
}
2307
2308
/**
2309
 * Reads columns by their index, handling both continuous and discontinuous rowid scenarios.
2310
 *
2311
 * This function is designed to read a specified number of rows (up to nrows_read_limit)
2312
 * from the segment iterator, dealing with both continuous and discontinuous rowid arrays.
2313
 * It operates as follows:
2314
 *
2315
 * 1. Reads a batch of rowids (up to the specified limit), and checks if they are continuous.
2316
 *    Continuous here means that the rowids form an unbroken sequence (e.g., 1, 2, 3, 4...).
2317
 *
2318
 * 2. For each column that needs to be read (identified by _predicate_column_ids):
2319
 *    - If the rowids are continuous, the function uses seek_to_ordinal and next_batch
2320
 *      for efficient reading.
2321
 *    - If the rowids are not continuous, the function processes them in smaller batches
2322
 *      (each of size up to 256). Each batch is checked for internal continuity:
2323
 *        a. If a batch is continuous, uses seek_to_ordinal and next_batch for that batch.
2324
 *        b. If a batch is not continuous, uses read_by_rowids for individual rowids in the batch.
2325
 *
2326
 * This approach optimizes reading performance by leveraging batch processing for continuous
2327
 * rowid sequences and handling discontinuities gracefully in smaller chunks.
2328
 */
2329
13.3k
Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16_t& nrows_read) {
2330
13.3k
    SCOPED_RAW_TIMER(&_opts.stats->predicate_column_read_ns);
2331
2332
13.3k
    nrows_read = (uint16_t)_range_iter->read_batch_rowids(_block_rowids.data(), nrows_read_limit);
2333
13.3k
    bool is_continuous = (nrows_read > 1) &&
2334
13.3k
                         (_block_rowids[nrows_read - 1] - _block_rowids[0] == nrows_read - 1);
2335
13.3k
    VLOG_DEBUG << fmt::format(
2336
0
            "nrows_read from range iterator: {}, is_continus {}, "
2337
0
            "_cols_read_by_column_predicate "
2338
0
            "[{}]",
2339
0
            nrows_read, is_continuous, fmt::join(_predicate_column_ids, ","));
2340
2341
13.3k
    LOG_IF(INFO, config::enable_segment_prefetch_verbose_log) << fmt::format(
2342
0
            "[verbose] SegmentIterator::_read_columns_by_index read {} rowids, continuous: {}, "
2343
0
            "rowids: [{}...{}]",
2344
0
            nrows_read, is_continuous, nrows_read > 0 ? _block_rowids[0] : 0,
2345
0
            nrows_read > 0 ? _block_rowids[nrows_read - 1] : 0);
2346
25.9k
    for (auto cid : _predicate_column_ids) {
2347
25.9k
        auto& column = _current_return_columns[cid];
2348
25.9k
        VLOG_DEBUG << fmt::format("Reading column {}, col_name {}", cid,
2349
0
                                  _schema->column(cid)->name());
2350
25.9k
        if (!_virtual_column_exprs.contains(cid)) {
2351
25.9k
            if (_no_need_read_key_data(cid, column, nrows_read)) {
2352
0
                VLOG_DEBUG << fmt::format("Column {} no need to read.", cid);
2353
0
                continue;
2354
0
            }
2355
25.9k
            if (_prune_column(cid, column, nrows_read)) {
2356
38
                VLOG_DEBUG << fmt::format("Column {} is pruned. No need to read data.", cid);
2357
38
                continue;
2358
38
            }
2359
25.8k
            DBUG_EXECUTE_IF("segment_iterator._read_columns_by_index", {
2360
25.8k
                auto col_name = _opts.tablet_schema->column(cid).name();
2361
25.8k
                auto debug_col_name =
2362
25.8k
                        DebugPoints::instance()->get_debug_param_or_default<std::string>(
2363
25.8k
                                "segment_iterator._read_columns_by_index", "column_name", "");
2364
25.8k
                if (debug_col_name.empty() && col_name != "__DORIS_DELETE_SIGN__") {
2365
25.8k
                    return Status::Error<ErrorCode::INTERNAL_ERROR>(
2366
25.8k
                            "does not need to read data, {}", col_name);
2367
25.8k
                }
2368
25.8k
                if (debug_col_name.find(col_name) != std::string::npos) {
2369
25.8k
                    return Status::Error<ErrorCode::INTERNAL_ERROR>(
2370
25.8k
                            "does not need to read data, {}", col_name);
2371
25.8k
                }
2372
25.8k
            })
2373
25.8k
        }
2374
2375
25.8k
        auto* column_iter = _column_iterators[cid].get();
2376
25.8k
        ScopedColumnIteratorReadPhase scoped_read_phase {
2377
25.8k
                column_iter, _support_lazy_read_pruned_columns.contains(cid)
2378
25.8k
                                     ? ColumnIterator::ReadPhase::PREDICATE
2379
25.8k
                                     : ColumnIterator::ReadPhase::NORMAL};
2380
2381
25.8k
        if (is_continuous) {
2382
18.8k
            size_t rows_read = nrows_read;
2383
18.8k
            _opts.stats->predicate_column_read_seek_num += 1;
2384
18.8k
            if (_opts.runtime_state && _opts.runtime_state->enable_profile()) {
2385
0
                SCOPED_RAW_TIMER(&_opts.stats->predicate_column_read_seek_ns);
2386
0
                RETURN_IF_ERROR(column_iter->seek_to_ordinal(_block_rowids[0]));
2387
18.8k
            } else {
2388
18.8k
                RETURN_IF_ERROR(column_iter->seek_to_ordinal(_block_rowids[0]));
2389
18.8k
            }
2390
18.8k
            RETURN_IF_ERROR(column_iter->next_batch(&rows_read, column));
2391
18.8k
            if (rows_read != nrows_read) {
2392
0
                return Status::Error<ErrorCode::INTERNAL_ERROR>("nrows({}) != rows_read({})",
2393
0
                                                                nrows_read, rows_read);
2394
0
            }
2395
18.8k
        } else {
2396
7.03k
            const uint32_t batch_size = _range_iter->get_batch_size();
2397
7.03k
            uint32_t processed = 0;
2398
8.34k
            while (processed < nrows_read) {
2399
1.31k
                uint32_t current_batch_size = std::min(batch_size, nrows_read - processed);
2400
1.31k
                bool batch_continuous = (current_batch_size > 1) &&
2401
1.31k
                                        (_block_rowids[processed + current_batch_size - 1] -
2402
1.20k
                                                 _block_rowids[processed] ==
2403
1.20k
                                         current_batch_size - 1);
2404
2405
1.31k
                if (batch_continuous) {
2406
0
                    size_t rows_read = current_batch_size;
2407
0
                    _opts.stats->predicate_column_read_seek_num += 1;
2408
0
                    if (_opts.runtime_state && _opts.runtime_state->enable_profile()) {
2409
0
                        SCOPED_RAW_TIMER(&_opts.stats->predicate_column_read_seek_ns);
2410
0
                        RETURN_IF_ERROR(column_iter->seek_to_ordinal(_block_rowids[processed]));
2411
0
                    } else {
2412
0
                        RETURN_IF_ERROR(column_iter->seek_to_ordinal(_block_rowids[processed]));
2413
0
                    }
2414
0
                    RETURN_IF_ERROR(column_iter->next_batch(&rows_read, column));
2415
0
                    if (rows_read != current_batch_size) {
2416
0
                        return Status::Error<ErrorCode::INTERNAL_ERROR>(
2417
0
                                "batch nrows({}) != rows_read({})", current_batch_size, rows_read);
2418
0
                    }
2419
1.31k
                } else {
2420
1.31k
                    RETURN_IF_ERROR(column_iter->read_by_rowids(&_block_rowids[processed],
2421
1.31k
                                                                current_batch_size, column));
2422
1.31k
                }
2423
1.31k
                processed += current_batch_size;
2424
1.31k
            }
2425
7.03k
        }
2426
25.8k
    }
2427
2428
13.3k
    return Status::OK();
2429
13.3k
}
2430
void SegmentIterator::_replace_version_col_if_needed(const std::vector<ColumnId>& column_ids,
2431
14.7k
                                                     size_t num_rows) {
2432
    // Only the rowset with single version need to replace the version column.
2433
    // Doris can't determine the version before publish_version finished, so
2434
    // we can't write data to __DORIS_VERSION_COL__ in segment writer, the value
2435
    // is 0 by default.
2436
    // So we need to replace the value to real version while reading.
2437
14.7k
    if (_opts.version.first != _opts.version.second) {
2438
6.54k
        return;
2439
6.54k
    }
2440
8.21k
    int32_t version_idx = _schema->version_col_idx();
2441
8.21k
    if (std::ranges::find(column_ids, version_idx) == column_ids.end()) {
2442
8.21k
        return;
2443
8.21k
    }
2444
2445
0
    const auto* column_desc = _schema->column(version_idx);
2446
0
    auto column = Schema::get_data_type_ptr(*column_desc)->create_column();
2447
0
    DCHECK(_schema->column(version_idx)->type() == FieldType::OLAP_FIELD_TYPE_BIGINT);
2448
0
    auto* col_ptr = assert_cast<ColumnInt64*>(column.get());
2449
0
    for (size_t j = 0; j < num_rows; j++) {
2450
0
        col_ptr->insert_value(_opts.version.second);
2451
0
    }
2452
0
    _current_return_columns[version_idx] = std::move(column);
2453
0
    VLOG_DEBUG << "replaced version column in segment iterator, version_col_idx:" << version_idx;
2454
0
}
2455
2456
void SegmentIterator::_update_tso_col_if_needed(const std::vector<ColumnId>& column_ids,
2457
14.7k
                                                size_t num_rows) {
2458
    // use physical time part of commit tso to replace timestamp col
2459
14.7k
    if (_opts.version.first != _opts.version.second) {
2460
6.54k
        return;
2461
6.54k
    }
2462
2463
8.21k
    if (_opts.io_ctx.reader_type != ReaderType::READER_BINLOG &&
2464
8.21k
        _opts.io_ctx.reader_type != ReaderType::READER_BINLOG_COMPACTION) {
2465
8.21k
        return;
2466
8.21k
    }
2467
2468
0
    int32_t tso_col_idx = _schema->tso_col_idx();
2469
0
    if (tso_col_idx < 0 || std::ranges::find(column_ids, tso_col_idx) == column_ids.end()) {
2470
0
        return;
2471
0
    }
2472
2473
0
    DCHECK_EQ(_opts.commit_tso.start_tso(), _opts.commit_tso.end_tso());
2474
0
    Int64 commit_tso = _opts.commit_tso.end_tso() == -1 ? 0 : _opts.commit_tso.end_tso();
2475
2476
0
    if (_is_pred_column[tso_col_idx]) {
2477
        // Nullable predicate column is represented as ColumnNullable(predicate_col)
2478
0
        if (auto* tso_nullable = check_and_get_column<ColumnNullable>(
2479
0
                    _current_return_columns[tso_col_idx].get())) {
2480
0
            _current_return_columns[tso_col_idx]->clear();
2481
0
            auto value = commit_tso;
2482
0
            for (size_t j = 0; j < num_rows; j++) {
2483
0
                tso_nullable->get_nested_column_ptr()->insert_data(
2484
0
                        reinterpret_cast<const char*>(&value), 0);
2485
0
                tso_nullable->get_null_map_data().emplace_back(0);
2486
0
            }
2487
0
            return;
2488
0
        }
2489
2490
0
        auto* tso_column = assert_cast<ColumnInt64*>(_current_return_columns[tso_col_idx].get());
2491
0
        tso_column->clear();
2492
0
        auto value = commit_tso;
2493
0
        for (size_t j = 0; j < num_rows; j++) {
2494
0
            tso_column->insert_data(reinterpret_cast<const char*>(&value), 0);
2495
0
        }
2496
0
        return;
2497
0
    }
2498
2499
0
    const auto* column_desc = _schema->column(tso_col_idx);
2500
0
    auto column = Schema::get_data_type_ptr(*column_desc)->create_column();
2501
0
    DCHECK(column_desc->type() == FieldType::OLAP_FIELD_TYPE_BIGINT);
2502
2503
0
    if (auto* tso_nullable = check_and_get_column<ColumnNullable>(column.get())) {
2504
0
        auto* col_ptr = assert_cast<ColumnInt64*>(&tso_nullable->get_nested_column());
2505
0
        for (size_t j = 0; j < num_rows; j++) {
2506
0
            col_ptr->insert_value(commit_tso);
2507
0
            tso_nullable->get_null_map_data().emplace_back(0);
2508
0
        }
2509
0
    } else {
2510
0
        auto* col_ptr = assert_cast<ColumnInt64*>(column.get());
2511
0
        for (size_t j = 0; j < num_rows; j++) {
2512
0
            col_ptr->insert_value(commit_tso);
2513
0
        }
2514
0
    }
2515
0
    _current_return_columns[tso_col_idx] = std::move(column);
2516
0
}
2517
2518
uint16_t SegmentIterator::_evaluate_vectorization_predicate(uint16_t* sel_rowid_idx,
2519
1.72k
                                                            uint16_t selected_size) {
2520
1.72k
    SCOPED_RAW_TIMER(&_opts.stats->vec_cond_ns);
2521
1.72k
    bool all_pred_always_true = true;
2522
1.72k
    for (const auto& pred : _pre_eval_block_predicate) {
2523
28
        if (!pred->always_true()) {
2524
28
            all_pred_always_true = false;
2525
28
        } else {
2526
0
            pred->update_filter_info(0, 0, selected_size);
2527
0
        }
2528
28
    }
2529
2530
1.72k
    const uint16_t original_size = selected_size;
2531
    //If all predicates are always_true, then return directly.
2532
1.72k
    if (all_pred_always_true || !_is_need_vec_eval) {
2533
3.90M
        for (uint16_t i = 0; i < original_size; ++i) {
2534
3.90M
            sel_rowid_idx[i] = i;
2535
3.90M
        }
2536
        // All preds are always_true, so return immediately and update the profile statistics here.
2537
1.69k
        _opts.stats->vec_cond_input_rows += original_size;
2538
1.69k
        return original_size;
2539
1.69k
    }
2540
2541
28
    _ret_flags.resize(original_size);
2542
28
    DCHECK(!_pre_eval_block_predicate.empty());
2543
28
    bool is_first = true;
2544
28
    for (auto& pred : _pre_eval_block_predicate) {
2545
28
        if (pred->always_true()) {
2546
0
            continue;
2547
0
        }
2548
28
        auto column_id = pred->column_id();
2549
28
        auto& column = _current_return_columns[column_id];
2550
28
        if (is_first) {
2551
28
            pred->evaluate_vec(*column, original_size, (bool*)_ret_flags.data());
2552
28
            is_first = false;
2553
28
        } else {
2554
0
            pred->evaluate_and_vec(*column, original_size, (bool*)_ret_flags.data());
2555
0
        }
2556
28
    }
2557
2558
28
    uint16_t new_size = 0;
2559
2560
28
    uint16_t sel_pos = 0;
2561
28
    const uint16_t sel_end = sel_pos + selected_size;
2562
28
    static constexpr size_t SIMD_BYTES = simd::bits_mask_length();
2563
28
    const uint16_t sel_end_simd = sel_pos + selected_size / SIMD_BYTES * SIMD_BYTES;
2564
2565
540
    while (sel_pos < sel_end_simd) {
2566
512
        auto mask = simd::bytes_mask_to_bits_mask(_ret_flags.data() + sel_pos);
2567
512
        if (0 == mask) {
2568
            //pass
2569
256
        } else if (simd::bits_mask_all() == mask) {
2570
8.44k
            for (uint16_t i = 0; i < SIMD_BYTES; i++) {
2571
8.19k
                sel_rowid_idx[new_size++] = sel_pos + i;
2572
8.19k
            }
2573
256
        } else {
2574
0
            simd::iterate_through_bits_mask(
2575
0
                    [&](const int bit_pos) {
2576
0
                        sel_rowid_idx[new_size++] = sel_pos + (uint16_t)bit_pos;
2577
0
                    },
2578
0
                    mask);
2579
0
        }
2580
512
        sel_pos += SIMD_BYTES;
2581
512
    }
2582
2583
55
    for (; sel_pos < sel_end; sel_pos++) {
2584
27
        if (_ret_flags[sel_pos]) {
2585
20
            sel_rowid_idx[new_size++] = sel_pos;
2586
20
        }
2587
27
    }
2588
2589
28
    _opts.stats->vec_cond_input_rows += original_size;
2590
28
    _opts.stats->rows_vec_cond_filtered += original_size - new_size;
2591
28
    return new_size;
2592
1.72k
}
2593
2594
uint16_t SegmentIterator::_evaluate_short_circuit_predicate(uint16_t* vec_sel_rowid_idx,
2595
1.72k
                                                            uint16_t selected_size) {
2596
1.72k
    SCOPED_RAW_TIMER(&_opts.stats->short_cond_ns);
2597
1.72k
    if (!_is_need_short_eval) {
2598
28
        return selected_size;
2599
28
    }
2600
2601
1.69k
    uint16_t original_size = selected_size;
2602
1.69k
    for (auto predicate : _short_cir_eval_predicate) {
2603
0
        auto column_id = predicate->column_id();
2604
0
        auto& short_cir_column = _current_return_columns[column_id];
2605
0
        selected_size = predicate->evaluate(*short_cir_column, vec_sel_rowid_idx, selected_size);
2606
0
    }
2607
2608
1.69k
    _opts.stats->short_circuit_cond_input_rows += original_size;
2609
1.69k
    _opts.stats->rows_short_circuit_cond_filtered += original_size - selected_size;
2610
2611
    // evaluate delete condition
2612
1.69k
    original_size = selected_size;
2613
1.69k
    selected_size = _opts.delete_condition_predicates->evaluate(_current_return_columns,
2614
1.69k
                                                                vec_sel_rowid_idx, selected_size);
2615
1.69k
    _opts.stats->rows_vec_del_cond_filtered += original_size - selected_size;
2616
1.69k
    return selected_size;
2617
1.72k
}
2618
2619
1
static void shrink_materialized_block_columns(Block* block, size_t rows) {
2620
2
    for (auto& entry : *block) {
2621
2
        if (entry.column && entry.column->size() > rows) {
2622
1
            entry.column = entry.column->shrink(rows);
2623
1
        }
2624
2
    }
2625
1
}
2626
2627
static void slice_materialized_block_columns(Block* block, size_t offset, size_t rows,
2628
1
                                             size_t original_rows) {
2629
1
    for (auto& entry : *block) {
2630
1
        if (!entry.column || entry.column->size() == 0) {
2631
0
            continue;
2632
0
        }
2633
1
        DORIS_CHECK(entry.column->size() == original_rows);
2634
1
        entry.column = entry.column->cut(offset, rows);
2635
1
    }
2636
1
}
2637
2638
1.73k
Status SegmentIterator::_apply_read_limit_to_selected_rows(Block* block, uint16_t& selected_size) {
2639
1.73k
    if (_opts.read_limit == 0) {
2640
1.73k
        return Status::OK();
2641
1.73k
    }
2642
2
    DORIS_CHECK(_rows_returned <= _opts.read_limit);
2643
2
    size_t remaining = _opts.read_limit - _rows_returned;
2644
2
    if (remaining == 0) {
2645
0
        selected_size = 0;
2646
0
        shrink_materialized_block_columns(block, 0);
2647
0
        return Status::OK();
2648
0
    }
2649
2
    if (selected_size > remaining) {
2650
2
        if (_opts.read_orderby_key_reverse) {
2651
1
            const auto original_size = selected_size;
2652
1
            const auto offset = original_size - remaining;
2653
21
            for (size_t i = 0; i < remaining; ++i) {
2654
20
                _sel_rowid_idx[i] = _sel_rowid_idx[offset + i];
2655
20
            }
2656
1
            selected_size = cast_set<uint16_t>(remaining);
2657
1
            slice_materialized_block_columns(block, offset, remaining, original_size);
2658
1
            return Status::OK();
2659
1
        }
2660
1
        selected_size = cast_set<uint16_t>(remaining);
2661
1
        shrink_materialized_block_columns(block, selected_size);
2662
1
    }
2663
1
    return Status::OK();
2664
2
}
2665
2666
Status SegmentIterator::_read_columns_by_rowids(std::vector<ColumnId>& read_column_ids,
2667
                                                std::vector<rowid_t>& rowid_vector,
2668
                                                uint16_t* sel_rowid_idx, size_t select_size,
2669
                                                MutableColumns* mutable_columns,
2670
                                                bool init_condition_cache,
2671
1.40k
                                                bool read_for_predicate) {
2672
1.40k
    SCOPED_RAW_TIMER(&_opts.stats->lazy_read_ns);
2673
1.40k
    std::vector<rowid_t> rowids(select_size);
2674
2675
1.40k
    if (init_condition_cache) {
2676
0
        DCHECK(_condition_cache);
2677
0
        auto& condition_cache = *_condition_cache;
2678
0
        for (size_t i = 0; i < select_size; ++i) {
2679
0
            rowids[i] = rowid_vector[sel_rowid_idx[i]];
2680
0
            condition_cache[rowids[i] / SegmentIterator::CONDITION_CACHE_OFFSET] = true;
2681
0
        }
2682
1.40k
    } else {
2683
2.75M
        for (size_t i = 0; i < select_size; ++i) {
2684
2.74M
            rowids[i] = rowid_vector[sel_rowid_idx[i]];
2685
2.74M
        }
2686
1.40k
    }
2687
2688
2.12k
    for (auto cid : read_column_ids) {
2689
2.12k
        auto& colunm = (*mutable_columns)[cid];
2690
2.12k
        if (_no_need_read_key_data(cid, colunm, select_size)) {
2691
0
            continue;
2692
0
        }
2693
2.12k
        if (_prune_column(cid, colunm, select_size)) {
2694
0
            continue;
2695
0
        }
2696
2697
2.12k
        DBUG_EXECUTE_IF("segment_iterator._read_columns_by_index", {
2698
2.12k
            auto debug_col_name = DebugPoints::instance()->get_debug_param_or_default<std::string>(
2699
2.12k
                    "segment_iterator._read_columns_by_index", "column_name", "");
2700
2.12k
            if (debug_col_name.empty()) {
2701
2.12k
                return Status::Error<ErrorCode::INTERNAL_ERROR>("does not need to read data");
2702
2.12k
            }
2703
2.12k
            auto col_name = _opts.tablet_schema->column(cid).name();
2704
2.12k
            if (debug_col_name.find(col_name) != std::string::npos) {
2705
2.12k
                return Status::Error<ErrorCode::INTERNAL_ERROR>("does not need to read data, {}",
2706
2.12k
                                                                debug_col_name);
2707
2.12k
            }
2708
2.12k
        })
2709
2710
2.12k
        if (_current_return_columns[cid].get() == nullptr) {
2711
0
            return Status::InternalError(
2712
0
                    "SegmentIterator meet invalid column, return columns size {}, cid {}",
2713
0
                    _current_return_columns.size(), cid);
2714
0
        }
2715
2716
2.12k
        auto* column_iter = _column_iterators[cid].get();
2717
2.12k
        ScopedColumnIteratorReadPhase scoped_read_phase {
2718
2.12k
                column_iter, read_for_predicate && _support_lazy_read_pruned_columns.contains(cid)
2719
2.12k
                                     ? ColumnIterator::ReadPhase::PREDICATE
2720
2.12k
                                     : ColumnIterator::ReadPhase::NORMAL};
2721
2722
2.12k
        RETURN_IF_ERROR(column_iter->read_by_rowids(rowids.data(), select_size,
2723
2.12k
                                                    _current_return_columns[cid]));
2724
2.12k
    }
2725
2726
1.40k
    return Status::OK();
2727
1.40k
}
2728
2729
1.73k
Status SegmentIterator::_read_lazy_pruned_columns(Block* block) {
2730
1.73k
    if (_support_lazy_read_pruned_columns.empty()) {
2731
1.73k
        return Status::OK();
2732
1.73k
    }
2733
2734
2
    SCOPED_RAW_TIMER(&_opts.stats->lazy_read_pruned_ns);
2735
2
    DorisVector<rowid_t> rowids(_selected_size);
2736
4
    for (size_t i = 0; i < _selected_size; ++i) {
2737
2
        rowids[i] = _block_rowids[_sel_rowid_idx[i]];
2738
2
    }
2739
2740
2
    for (auto cid : _support_lazy_read_pruned_columns) {
2741
2
        auto loc = _schema->column_index(cid);
2742
2
        auto column = IColumn::mutate(std::move(block->get_by_position(loc).column));
2743
2
        auto* column_iter = _column_iterators[cid].get();
2744
2
        ScopedColumnIteratorReadPhase scoped_read_phase {column_iter,
2745
2
                                                         ColumnIterator::ReadPhase::LAZY};
2746
2
        if (_selected_size > 0) {
2747
1
            RETURN_IF_ERROR(column_iter->read_by_rowids(rowids.data(), _selected_size, column));
2748
1
        }
2749
2
        column_iter->finalize_lazy_phase(column);
2750
2
        block->get_by_position(loc).column = std::move(column);
2751
2
    }
2752
2
    return Status::OK();
2753
2
}
2754
2755
13.3k
Status SegmentIterator::next_batch(Block* block) {
2756
    // Replace virtual columns with ColumnNothing at the begining of each next_batch call.
2757
13.3k
    _init_virtual_columns(block);
2758
13.3k
    auto status = [&]() {
2759
13.3k
        RETURN_IF_CATCH_EXCEPTION({
2760
            // Adaptive batch size: predict how many rows this batch should read.
2761
13.3k
            if (_block_size_predictor) {
2762
13.3k
                auto predicted = static_cast<uint32_t>(_block_size_predictor->predict_next_rows());
2763
13.3k
                _opts.block_row_max = std::min(predicted, _initial_block_row_max);
2764
13.3k
                _opts.stats->adaptive_batch_size_predict_min_rows =
2765
13.3k
                        std::min(_opts.stats->adaptive_batch_size_predict_min_rows,
2766
13.3k
                                 static_cast<int64_t>(predicted));
2767
13.3k
                _opts.stats->adaptive_batch_size_predict_max_rows =
2768
13.3k
                        std::max(_opts.stats->adaptive_batch_size_predict_max_rows,
2769
13.3k
                                 static_cast<int64_t>(predicted));
2770
13.3k
            } else {
2771
                // No predictor — record the fixed batch size using min/max so we don't
2772
                // clobber values already accumulated by other segment iterators that
2773
                // share the same OlapReaderStatistics.
2774
13.3k
                _opts.stats->adaptive_batch_size_predict_min_rows =
2775
13.3k
                        std::min(_opts.stats->adaptive_batch_size_predict_min_rows,
2776
13.3k
                                 static_cast<int64_t>(_opts.block_row_max));
2777
13.3k
                _opts.stats->adaptive_batch_size_predict_max_rows =
2778
13.3k
                        std::max(_opts.stats->adaptive_batch_size_predict_max_rows,
2779
13.3k
                                 static_cast<int64_t>(_opts.block_row_max));
2780
13.3k
            }
2781
2782
13.3k
            auto res = _next_batch_internal(block);
2783
2784
13.3k
            if (res.is<END_OF_FILE>()) {
2785
                // Since we have a type check at the caller.
2786
                // So a replacement of nothing column with real column is needed.
2787
13.3k
                for (const auto& [cid, expr_ctx] : _virtual_column_exprs) {
2788
13.3k
                    auto idx = _schema->column_index(cid);
2789
13.3k
                    auto type = expr_ctx->root()->data_type();
2790
13.3k
                    block->replace_by_position(idx, type->create_column());
2791
13.3k
                }
2792
2793
13.3k
                if (_opts.condition_cache_digest && !_find_condition_cache) {
2794
13.3k
                    auto* condition_cache = ConditionCache::instance();
2795
13.3k
                    ConditionCache::CacheKey cache_key(_opts.rowset_id, _segment->id(),
2796
13.3k
                                                       _opts.condition_cache_digest);
2797
13.3k
                    VLOG_DEBUG << "Condition cache insert, query id: "
2798
13.3k
                               << print_id(_opts.runtime_state->query_id())
2799
13.3k
                               << ", rowset id: " << _opts.rowset_id.to_string()
2800
13.3k
                               << ", segment id: " << _segment->id()
2801
13.3k
                               << ", cache digest: " << _opts.condition_cache_digest;
2802
13.3k
                    condition_cache->insert(cache_key, std::move(_condition_cache));
2803
13.3k
                }
2804
13.3k
                return res;
2805
13.3k
            }
2806
2807
13.3k
            RETURN_IF_ERROR(res);
2808
            // reverse block row order if read_orderby_key_reverse is true for key topn
2809
            // it should be processed for all success _next_batch_internal
2810
13.3k
            if (_opts.read_orderby_key_reverse) {
2811
13.3k
                size_t num_rows = block->rows();
2812
13.3k
                if (num_rows == 0) {
2813
13.3k
                    return Status::OK();
2814
13.3k
                }
2815
13.3k
                size_t num_columns = block->columns();
2816
13.3k
                IColumn::Permutation permutation;
2817
13.3k
                for (size_t i = 0; i < num_rows; ++i) permutation.emplace_back(num_rows - 1 - i);
2818
2819
13.3k
                for (size_t i = 0; i < num_columns; ++i)
2820
13.3k
                    block->get_by_position(i).column =
2821
13.3k
                            block->get_by_position(i).column->permute(permutation, num_rows);
2822
13.3k
            }
2823
2824
13.3k
            RETURN_IF_ERROR(block->check_type_and_column());
2825
2826
            // Adaptive batch size: update EWMA estimate from the completed batch.
2827
            // block->bytes() is accurate here: predicates have been applied and non-predicate
2828
            // columns have been filled for surviving rows by _next_batch_internal.
2829
13.3k
            if (_block_size_predictor && block->rows() > 0) {
2830
13.3k
                _block_size_predictor->update(*block);
2831
13.3k
            }
2832
2833
13.3k
            return Status::OK();
2834
13.3k
        });
2835
13.3k
    }();
2836
2837
    // if rows read by batch is 0, will return end of file, we should not remove segment cache in this situation.
2838
13.3k
    if (!status.ok() && !status.is<END_OF_FILE>()) {
2839
0
        _segment->update_healthy_status(status);
2840
0
    }
2841
13.3k
    return status;
2842
13.3k
}
2843
2844
13.3k
Status SegmentIterator::_convert_to_expected_type(const std::vector<ColumnId>& col_ids) {
2845
27.2k
    for (ColumnId i : col_ids) {
2846
27.2k
        if (!_current_return_columns[i] || _converted_column_ids[i] || _is_pred_column[i]) {
2847
486
            continue;
2848
486
        }
2849
26.7k
        const TabletColumn* column_desc = _schema->column(i);
2850
26.7k
        DataTypePtr expected_type = Schema::get_data_type_ptr(*column_desc);
2851
26.7k
        DataTypePtr file_column_type = _storage_name_and_type[i].second;
2852
26.7k
        if (!file_column_type->equals(*expected_type)) {
2853
36
            ColumnPtr expected;
2854
36
            ColumnPtr original = _current_return_columns[i]->assert_mutable()->get_ptr();
2855
36
            RETURN_IF_ERROR(variant_util::cast_column({original, file_column_type, ""},
2856
36
                                                      expected_type, &expected));
2857
36
            _current_return_columns[i] = expected->assert_mutable();
2858
36
            _converted_column_ids[i] = true;
2859
36
            VLOG_DEBUG << fmt::format("Convert {} fom file column type {} to {}, num_rows {}",
2860
0
                                      column_desc->path_info_ptr() == nullptr
2861
0
                                              ? ""
2862
0
                                              : column_desc->path_info_ptr()->get_path(),
2863
0
                                      file_column_type->get_name(), expected_type->get_name(),
2864
0
                                      _current_return_columns[i]->size());
2865
36
        }
2866
26.7k
    }
2867
13.3k
    return Status::OK();
2868
13.3k
}
2869
2870
Status SegmentIterator::copy_column_data_by_selector(IColumn* input_col_ptr,
2871
                                                     MutableColumnPtr& output_col,
2872
                                                     uint16_t* sel_rowid_idx, uint16_t select_size,
2873
1.41k
                                                     size_t batch_size) {
2874
1.41k
    if (is_column_nullable(*output_col) != is_column_nullable(*input_col_ptr)) {
2875
0
        LOG(WARNING) << "nullable mismatch for output_column: " << output_col->dump_structure()
2876
0
                     << " input_column: " << input_col_ptr->dump_structure()
2877
0
                     << " select_size: " << select_size;
2878
0
        return Status::RuntimeError("copy_column_data_by_selector nullable mismatch");
2879
0
    }
2880
1.41k
    output_col->reserve(select_size);
2881
1.41k
    return input_col_ptr->filter_by_selector(sel_rowid_idx, select_size, output_col.get());
2882
1.41k
}
2883
2884
13.3k
Status SegmentIterator::_next_batch_internal(Block* block) {
2885
13.3k
    SCOPED_CONCURRENCY_COUNT(ConcurrencyStatsManager::instance().segment_iterator_next_batch);
2886
2887
13.3k
    bool is_mem_reuse = block->mem_reuse();
2888
13.3k
    DCHECK(is_mem_reuse);
2889
2890
13.3k
    RETURN_IF_ERROR(_lazy_init(block));
2891
2892
13.3k
    SCOPED_RAW_TIMER(&_opts.stats->block_load_ns);
2893
2894
13.3k
    if (_opts.read_limit > 0 && _rows_returned >= _opts.read_limit) {
2895
0
        return _process_eof(block);
2896
0
    }
2897
2898
    // If the row bitmap size is smaller than nrows_read_limit, there's no need to reserve that many column rows.
2899
13.3k
    uint32_t nrows_read_limit =
2900
13.3k
            std::min(cast_set<uint32_t>(_row_bitmap.cardinality()), _opts.block_row_max);
2901
13.3k
    if (_can_opt_limit_reads()) {
2902
        // No SegmentIterator-side conjunct remains to be evaluated, so LIMIT is equivalent before
2903
        // and after filtering. Cap the first read directly; this is the no-conjunct fast path that
2904
        // avoids reading rows past the pushed-down local LIMIT.
2905
0
        size_t cap = (_opts.read_limit > _rows_returned) ? (_opts.read_limit - _rows_returned) : 0;
2906
0
        if (cap < nrows_read_limit) {
2907
0
            nrows_read_limit = static_cast<uint32_t>(cap);
2908
0
        }
2909
0
    }
2910
13.3k
    DBUG_EXECUTE_IF("segment_iterator.topn_opt_1", {
2911
13.3k
        if (nrows_read_limit != 1) {
2912
13.3k
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
2913
13.3k
                    "topn opt 1 execute failed: nrows_read_limit={}, "
2914
13.3k
                    "_opts.read_limit={}",
2915
13.3k
                    nrows_read_limit, _opts.read_limit);
2916
13.3k
        }
2917
13.3k
    })
2918
2919
13.3k
    RETURN_IF_ERROR(_init_current_block(block, _current_return_columns, nrows_read_limit));
2920
13.3k
    _converted_column_ids.assign(_schema->columns().size(), false);
2921
2922
13.3k
    _selected_size = 0;
2923
13.3k
    RETURN_IF_ERROR(_read_columns_by_index(nrows_read_limit, _selected_size));
2924
13.3k
    _replace_version_col_if_needed(_predicate_column_ids, _selected_size);
2925
13.3k
    _update_tso_col_if_needed(_predicate_column_ids, _selected_size);
2926
2927
13.3k
    _opts.stats->blocks_load += 1;
2928
13.3k
    _opts.stats->raw_rows_read += _selected_size;
2929
2930
13.3k
    if (_selected_size == 0) {
2931
3.02k
        return _process_eof(block);
2932
3.02k
    }
2933
2934
10.3k
    if (_is_need_vec_eval || _is_need_short_eval || _is_need_expr_eval) {
2935
1.73k
        _sel_rowid_idx.resize(_selected_size);
2936
2937
1.73k
        if (_is_need_vec_eval || _is_need_short_eval) {
2938
1.72k
            _convert_dict_code_for_predicate_if_necessary();
2939
2940
            // step 1: evaluate vectorization predicate
2941
1.72k
            _selected_size =
2942
1.72k
                    _evaluate_vectorization_predicate(_sel_rowid_idx.data(), _selected_size);
2943
2944
            // step 2: evaluate short circuit predicate
2945
            // todo(wb) research whether need to read short predicate after vectorization evaluation
2946
            //          to reduce cost of read short circuit columns.
2947
            //          In SSB test, it make no difference; So need more scenarios to test
2948
1.72k
            _selected_size =
2949
1.72k
                    _evaluate_short_circuit_predicate(_sel_rowid_idx.data(), _selected_size);
2950
1.72k
            VLOG_DEBUG << fmt::format("After evaluate predicates, selected size: {} ",
2951
0
                                      _selected_size);
2952
1.72k
            if (_selected_size > 0) {
2953
                // step 3.1: output short circuit and predicate column
2954
                // when lazy materialization enables, _predicate_column_ids = distinct(_short_cir_pred_column_ids + _vec_pred_column_ids)
2955
                // see _vec_init_lazy_materialization
2956
                // todo(wb) need to tell input columnids from output columnids
2957
1.68k
                RETURN_IF_ERROR(_output_column_by_sel_idx(block, _predicate_column_ids,
2958
1.68k
                                                          _sel_rowid_idx.data(), _selected_size));
2959
2960
                // step 3.2: read remaining expr column and evaluate it.
2961
1.68k
                if (_is_need_expr_eval) {
2962
                    // The predicate column contains the remaining expr column, no need second read.
2963
0
                    if (_common_expr_column_ids.size() > 0) {
2964
0
                        SCOPED_RAW_TIMER(&_opts.stats->non_predicate_read_ns);
2965
0
                        RETURN_IF_ERROR(_read_columns_by_rowids(
2966
0
                                _common_expr_column_ids, _block_rowids, _sel_rowid_idx.data(),
2967
0
                                _selected_size, &_current_return_columns, false, true));
2968
0
                        _replace_version_col_if_needed(_common_expr_column_ids, _selected_size);
2969
0
                        _update_tso_col_if_needed(_common_expr_column_ids, _selected_size);
2970
0
                        RETURN_IF_ERROR(_process_columns(_common_expr_column_ids, block));
2971
0
                    }
2972
2973
0
                    DCHECK(block->columns() > _schema->column_index(*_common_expr_columns.begin()));
2974
0
                    RETURN_IF_ERROR(
2975
0
                            _process_common_expr(_sel_rowid_idx.data(), _selected_size, block));
2976
0
                }
2977
1.68k
            } else {
2978
38
                _fill_column_nothing();
2979
38
                if (_is_need_expr_eval) {
2980
0
                    RETURN_IF_ERROR(_process_columns(_common_expr_column_ids, block));
2981
0
                }
2982
38
            }
2983
1.72k
        } else if (_is_need_expr_eval) {
2984
6
            DCHECK(!_predicate_column_ids.empty());
2985
6
            RETURN_IF_ERROR(_process_columns(_predicate_column_ids, block));
2986
            // first read all rows are insert block, initialize sel_rowid_idx to all rows.
2987
23
            for (uint16_t i = 0; i < _selected_size; ++i) {
2988
17
                _sel_rowid_idx[i] = i;
2989
17
            }
2990
6
            RETURN_IF_ERROR(_process_common_expr(_sel_rowid_idx.data(), _selected_size, block));
2991
6
        }
2992
2993
1.73k
        RETURN_IF_ERROR(_apply_read_limit_to_selected_rows(block, _selected_size));
2994
2995
        // step4: read non_predicate column
2996
1.73k
        if (_selected_size > 0) {
2997
1.69k
            if (!_non_predicate_columns.empty()) {
2998
1.40k
                RETURN_IF_ERROR(_read_columns_by_rowids(
2999
1.40k
                        _non_predicate_columns, _block_rowids, _sel_rowid_idx.data(),
3000
1.40k
                        _selected_size, &_current_return_columns,
3001
1.40k
                        _opts.condition_cache_digest && !_find_condition_cache, false));
3002
1.40k
                _replace_version_col_if_needed(_non_predicate_columns, _selected_size);
3003
1.40k
                _update_tso_col_if_needed(_non_predicate_columns, _selected_size);
3004
1.40k
            } else {
3005
286
                if (_opts.condition_cache_digest && !_find_condition_cache) {
3006
0
                    auto& condition_cache = *_condition_cache;
3007
0
                    for (size_t i = 0; i < _selected_size; ++i) {
3008
0
                        auto rowid = _block_rowids[_sel_rowid_idx[i]];
3009
0
                        condition_cache[rowid / SegmentIterator::CONDITION_CACHE_OFFSET] = true;
3010
0
                    }
3011
0
                }
3012
286
            }
3013
1.69k
        }
3014
3015
1.73k
        RETURN_IF_ERROR(_read_lazy_pruned_columns(block));
3016
1.73k
    }
3017
3018
    // step5: output columns
3019
10.3k
    RETURN_IF_ERROR(_output_non_pred_columns(block));
3020
    // Convert inverted index bitmaps to result columns for virtual column exprs
3021
    // (e.g., MATCH projections). This must run before _materialization_of_virtual_column
3022
    // so that fast_execute() can find the pre-computed result columns.
3023
10.3k
    if (!_virtual_column_exprs.empty()) {
3024
0
        bool use_sel = _is_need_vec_eval || _is_need_short_eval || _is_need_expr_eval;
3025
0
        uint16_t* sel_rowid_idx = use_sel ? _sel_rowid_idx.data() : nullptr;
3026
0
        VExprContextSPtrs vir_ctxs;
3027
0
        vir_ctxs.reserve(_virtual_column_exprs.size());
3028
0
        for (auto& [cid, ctx] : _virtual_column_exprs) {
3029
0
            vir_ctxs.push_back(ctx);
3030
0
        }
3031
0
        _output_index_result_column(vir_ctxs, sel_rowid_idx, _selected_size);
3032
0
    }
3033
10.3k
    RETURN_IF_ERROR(_materialization_of_virtual_column(block));
3034
10.3k
    if (_opts.read_limit > 0) {
3035
0
        _rows_returned += block->rows();
3036
0
    }
3037
10.3k
    return _check_output_block(block);
3038
10.3k
}
3039
3040
6
Status SegmentIterator::_process_columns(const std::vector<ColumnId>& column_ids, Block* block) {
3041
6
    RETURN_IF_ERROR(_convert_to_expected_type(column_ids));
3042
6
    for (auto cid : column_ids) {
3043
6
        auto loc = _schema->column_index(cid);
3044
6
        block->replace_by_position(loc, std::move(_current_return_columns[cid]));
3045
6
    }
3046
6
    return Status::OK();
3047
6
}
3048
3049
38
void SegmentIterator::_fill_column_nothing() {
3050
    // If column_predicate filters out all rows, the corresponding column in _current_return_columns[cid] must be a ColumnNothing.
3051
    // Because:
3052
    // 1. Before each batch, _init_return_columns is called to initialize _current_return_columns, and virtual columns in _current_return_columns are initialized as ColumnNothing.
3053
    // 2. When select_size == 0, the read method of VirtualColumnIterator will definitely not be called, so the corresponding Column remains a ColumnNothing
3054
38
    for (const auto& [cid, expr_ctx] : _virtual_column_exprs) {
3055
0
        [[maybe_unused]] const auto* nothing_col =
3056
0
                assert_cast<const ColumnNothing*>(_current_return_columns[cid].get());
3057
0
        _current_return_columns[cid] = expr_ctx->root()->data_type()->create_column();
3058
0
    }
3059
38
}
3060
3061
10.3k
Status SegmentIterator::_check_output_block(Block* block) {
3062
10.3k
#ifndef NDEBUG
3063
10.3k
    size_t rows = block->rows();
3064
10.3k
    size_t idx = 0;
3065
21.2k
    for (const auto& entry : *block) {
3066
21.2k
        if (!entry.column) {
3067
0
            return Status::InternalError(
3068
0
                    "Column in idx {} is null, block columns {}, normal_columns {}, "
3069
0
                    "virtual_columns {}",
3070
0
                    idx, block->columns(), _schema->num_column_ids(), _virtual_column_exprs.size());
3071
21.2k
        } else if (check_and_get_column<ColumnNothing>(entry.column.get())) {
3072
0
            if (rows > 0) {
3073
0
                std::vector<ColumnId> virtual_column_ids;
3074
0
                for (const auto& pair : _virtual_column_exprs) {
3075
0
                    virtual_column_ids.push_back(pair.first);
3076
0
                }
3077
0
                return Status::InternalError(
3078
0
                        "Column in idx {} is nothing, block columns {}, normal_columns {}, "
3079
0
                        "virtual_column_ids [{}]",
3080
0
                        idx, block->columns(), _schema->num_column_ids(),
3081
0
                        fmt::join(virtual_column_ids, ","));
3082
0
            }
3083
21.2k
        } else if (entry.column->size() != rows) {
3084
0
            return Status::InternalError(
3085
0
                    "Unmatched size {}, expected {}, column: {}, type: {}, idx_in_block: {}, "
3086
0
                    "block: {}",
3087
0
                    entry.column->size(), rows, entry.column->get_name(), entry.type->get_name(),
3088
0
                    idx, block->dump_structure());
3089
0
        }
3090
21.2k
        idx++;
3091
21.2k
    }
3092
10.3k
#endif
3093
10.3k
    return Status::OK();
3094
10.3k
}
3095
3096
3.02k
Status SegmentIterator::_process_eof(Block* block) {
3097
    // Convert all columns in _current_return_columns to schema column
3098
3.02k
    RETURN_IF_ERROR(_convert_to_expected_type(_schema->column_ids()));
3099
10.3k
    for (int i = 0; i < block->columns(); i++) {
3100
7.28k
        auto cid = _schema->column_id(i);
3101
7.28k
        if (!_is_pred_column[cid]) {
3102
6.98k
            block->replace_by_position(i, std::move(_current_return_columns[cid]));
3103
6.98k
        }
3104
7.28k
    }
3105
3.02k
    block->clear_column_data();
3106
    // clear and release iterators memory footprint in advance
3107
3.02k
    _column_iterators.clear();
3108
3.02k
    _index_iterators.clear();
3109
3.02k
    return Status::EndOfFile("no more data in segment");
3110
3.02k
}
3111
3112
Status SegmentIterator::_process_common_expr(uint16_t* sel_rowid_idx, uint16_t& selected_size,
3113
6
                                             Block* block) {
3114
6
    VLOG_DEBUG << fmt::format("Execute common expr. block rows {}, selected size {}", block->rows(),
3115
0
                              _selected_size);
3116
3117
6
    RETURN_IF_ERROR(_execute_common_expr(sel_rowid_idx, selected_size, block));
3118
3119
6
    VLOG_DEBUG << fmt::format("Execute common expr end. block rows {}, selected size {}",
3120
0
                              block->rows(), _selected_size);
3121
6
    return Status::OK();
3122
6
}
3123
3124
Status SegmentIterator::_execute_common_expr(uint16_t* sel_rowid_idx, uint16_t& selected_size,
3125
6
                                             Block* block) {
3126
6
    SCOPED_RAW_TIMER(&_opts.stats->expr_filter_ns);
3127
6
    DCHECK(!_common_expr_ctxs_push_down.empty());
3128
6
    _output_index_result_column(_common_expr_ctxs_push_down, sel_rowid_idx, selected_size);
3129
3130
6
    uint16_t original_size = selected_size;
3131
6
    _opts.stats->expr_cond_input_rows += original_size;
3132
3133
    // Some output columns may stay empty until after common expr filtering. Use the
3134
    // selected row count instead of Block::rows(), which is derived from the first column.
3135
6
    IColumn::Filter filter(selected_size, 1);
3136
6
    bool can_filter_all = false;
3137
6
    auto* __restrict filter_data = filter.data();
3138
6
    for (const auto& expr_ctx : _common_expr_ctxs_push_down) {
3139
6
        RETURN_IF_ERROR(expr_ctx->execute_filter(block, filter_data, selected_size, false,
3140
6
                                                 &can_filter_all));
3141
6
        if (can_filter_all) {
3142
0
            break;
3143
0
        }
3144
6
    }
3145
6
    RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(block, _columns_to_filter, filter));
3146
3147
6
    selected_size = _evaluate_common_expr_filter(sel_rowid_idx, selected_size, filter);
3148
6
    _opts.stats->rows_expr_cond_filtered += original_size - selected_size;
3149
6
    return Status::OK();
3150
6
}
3151
3152
uint16_t SegmentIterator::_evaluate_common_expr_filter(uint16_t* sel_rowid_idx,
3153
                                                       uint16_t selected_size,
3154
6
                                                       const IColumn::Filter& filter) {
3155
6
    size_t count = filter.size() - simd::count_zero_num((int8_t*)filter.data(), filter.size());
3156
6
    if (count == 0) {
3157
0
        return 0;
3158
6
    } else {
3159
6
        const UInt8* filt_pos = filter.data();
3160
3161
6
        uint16_t new_size = 0;
3162
6
        uint32_t sel_pos = 0;
3163
6
        const uint32_t sel_end = selected_size;
3164
6
        static constexpr size_t SIMD_BYTES = simd::bits_mask_length();
3165
6
        const uint32_t sel_end_simd = sel_pos + selected_size / SIMD_BYTES * SIMD_BYTES;
3166
3167
6
        while (sel_pos < sel_end_simd) {
3168
0
            auto mask = simd::bytes_mask_to_bits_mask(filt_pos + sel_pos);
3169
0
            if (0 == mask) {
3170
                //pass
3171
0
            } else if (simd::bits_mask_all() == mask) {
3172
0
                for (uint32_t i = 0; i < SIMD_BYTES; i++) {
3173
0
                    sel_rowid_idx[new_size++] = sel_rowid_idx[sel_pos + i];
3174
0
                }
3175
0
            } else {
3176
0
                simd::iterate_through_bits_mask(
3177
0
                        [&](const size_t bit_pos) {
3178
0
                            sel_rowid_idx[new_size++] = sel_rowid_idx[sel_pos + bit_pos];
3179
0
                        },
3180
0
                        mask);
3181
0
            }
3182
0
            sel_pos += SIMD_BYTES;
3183
0
        }
3184
3185
23
        for (; sel_pos < sel_end; sel_pos++) {
3186
17
            if (filt_pos[sel_pos]) {
3187
7
                sel_rowid_idx[new_size++] = sel_rowid_idx[sel_pos];
3188
7
            }
3189
17
        }
3190
6
        return new_size;
3191
6
    }
3192
6
}
3193
3194
void SegmentIterator::_output_index_result_column(const VExprContextSPtrs& expr_ctxs,
3195
6
                                                  uint16_t* sel_rowid_idx, uint16_t select_size) {
3196
6
    SCOPED_RAW_TIMER(&_opts.stats->output_index_result_column_timer);
3197
6
    if (select_size == 0) {
3198
0
        return;
3199
0
    }
3200
6
    for (const auto& expr_ctx : expr_ctxs) {
3201
6
        auto index_ctx = expr_ctx->get_index_context();
3202
6
        if (index_ctx == nullptr) {
3203
0
            continue;
3204
0
        }
3205
6
        for (auto& inverted_index_result_bitmap_for_expr : index_ctx->get_index_result_bitmap()) {
3206
0
            const auto* expr = inverted_index_result_bitmap_for_expr.first;
3207
0
            const auto& result_bitmap = inverted_index_result_bitmap_for_expr.second;
3208
0
            const auto& index_result_bitmap = result_bitmap.get_data_bitmap();
3209
0
            auto index_result_column = ColumnUInt8::create();
3210
0
            ColumnUInt8::Container& vec_match_pred = index_result_column->get_data();
3211
0
            vec_match_pred.resize(select_size);
3212
0
            std::fill(vec_match_pred.begin(), vec_match_pred.end(), 0);
3213
3214
0
            const auto& null_bitmap = result_bitmap.get_null_bitmap();
3215
0
            bool has_null_bitmap = null_bitmap != nullptr && !null_bitmap->isEmpty();
3216
0
            bool expr_returns_nullable = expr->data_type()->is_nullable();
3217
3218
0
            ColumnUInt8::MutablePtr null_map_column = nullptr;
3219
0
            ColumnUInt8::Container* null_map_data = nullptr;
3220
0
            if (has_null_bitmap && expr_returns_nullable) {
3221
0
                null_map_column = ColumnUInt8::create();
3222
0
                auto& null_map_vec = null_map_column->get_data();
3223
0
                null_map_vec.resize(select_size);
3224
0
                std::fill(null_map_vec.begin(), null_map_vec.end(), 0);
3225
0
                null_map_data = &null_map_column->get_data();
3226
0
            }
3227
3228
0
            roaring::BulkContext bulk_context;
3229
0
            for (uint32_t i = 0; i < select_size; i++) {
3230
0
                auto rowid = sel_rowid_idx ? _block_rowids[sel_rowid_idx[i]] : _block_rowids[i];
3231
0
                if (index_result_bitmap) {
3232
0
                    vec_match_pred[i] = index_result_bitmap->containsBulk(bulk_context, rowid);
3233
0
                }
3234
0
                if (null_map_data != nullptr && null_bitmap->contains(rowid)) {
3235
0
                    (*null_map_data)[i] = 1;
3236
0
                    vec_match_pred[i] = 0;
3237
0
                }
3238
0
            }
3239
3240
0
            DCHECK(select_size == vec_match_pred.size());
3241
3242
0
            if (null_map_column) {
3243
0
                index_ctx->set_index_result_column_for_expr(
3244
0
                        expr, ColumnNullable::create(std::move(index_result_column),
3245
0
                                                     std::move(null_map_column)));
3246
0
            } else {
3247
0
                index_ctx->set_index_result_column_for_expr(expr, std::move(index_result_column));
3248
0
            }
3249
0
        }
3250
6
    }
3251
6
}
3252
3253
1.72k
void SegmentIterator::_convert_dict_code_for_predicate_if_necessary() {
3254
1.72k
    for (auto predicate : _short_cir_eval_predicate) {
3255
0
        _convert_dict_code_for_predicate_if_necessary_impl(predicate);
3256
0
    }
3257
3258
1.72k
    for (auto predicate : _pre_eval_block_predicate) {
3259
28
        _convert_dict_code_for_predicate_if_necessary_impl(predicate);
3260
28
    }
3261
3262
1.72k
    for (auto column_id : _delete_range_column_ids) {
3263
1.55k
        _current_return_columns[column_id].get()->convert_dict_codes_if_necessary();
3264
1.55k
    }
3265
3266
1.72k
    for (auto column_id : _delete_bloom_filter_column_ids) {
3267
0
        _current_return_columns[column_id].get()->initialize_hash_values_for_runtime_filter();
3268
0
    }
3269
1.72k
}
3270
3271
void SegmentIterator::_convert_dict_code_for_predicate_if_necessary_impl(
3272
28
        std::shared_ptr<ColumnPredicate> predicate) {
3273
28
    auto& column = _current_return_columns[predicate->column_id()];
3274
28
    auto* col_ptr = column.get();
3275
3276
28
    if (PredicateTypeTraits::is_range(predicate->type())) {
3277
22
        col_ptr->convert_dict_codes_if_necessary();
3278
22
    } else if (PredicateTypeTraits::is_bloom_filter(predicate->type())) {
3279
0
        col_ptr->initialize_hash_values_for_runtime_filter();
3280
0
    }
3281
28
}
3282
3283
2.93k
Status SegmentIterator::current_block_row_locations(std::vector<RowLocation>* block_row_locations) {
3284
2.93k
    DCHECK(_opts.record_rowids);
3285
2.93k
    DCHECK_GE(_block_rowids.size(), _selected_size);
3286
2.93k
    block_row_locations->resize(_selected_size);
3287
2.93k
    uint32_t sid = segment_id();
3288
2.93k
    if (!_is_need_vec_eval && !_is_need_short_eval && !_is_need_expr_eval) {
3289
4.24M
        for (auto i = 0; i < _selected_size; i++) {
3290
4.23M
            (*block_row_locations)[i] = RowLocation(sid, _block_rowids[i]);
3291
4.23M
        }
3292
1.76k
    } else {
3293
2.46M
        for (auto i = 0; i < _selected_size; i++) {
3294
2.46M
            (*block_row_locations)[i] = RowLocation(sid, _block_rowids[_sel_rowid_idx[i]]);
3295
2.46M
        }
3296
1.16k
    }
3297
2.93k
    return Status::OK();
3298
2.93k
}
3299
3300
3.03k
Status SegmentIterator::_construct_compound_expr_context() {
3301
3.03k
    ColumnIteratorOptions iter_opts {
3302
3.03k
            .use_page_cache = _opts.use_page_cache,
3303
3.03k
            .file_reader = _file_reader.get(),
3304
3.03k
            .stats = _opts.stats,
3305
3.03k
            .io_ctx = _opts.io_ctx,
3306
3.03k
    };
3307
3.03k
    auto inverted_index_context = std::make_shared<IndexExecContext>(
3308
3.03k
            _schema->column_ids(), _index_iterators, _storage_name_and_type,
3309
3.03k
            _common_expr_index_exec_status, _score_runtime, _segment.get(), iter_opts);
3310
3.03k
    inverted_index_context->set_index_query_context(_index_query_context);
3311
3.03k
    for (const auto& expr_ctx : _opts.common_expr_ctxs_push_down) {
3312
21
        VExprContextSPtr context;
3313
        // _ann_range_search_runtime will do deep copy.
3314
21
        RETURN_IF_ERROR(expr_ctx->clone(_opts.runtime_state, context));
3315
21
        context->set_index_context(inverted_index_context);
3316
21
        _common_expr_ctxs_push_down.emplace_back(context);
3317
21
    }
3318
    // Clone virtual column exprs before setting IndexExecContext, because
3319
    // IndexExecContext holds segment-specific index iterator references.
3320
    // Without cloning, shared VExprContext would be overwritten per-segment
3321
    // and could point to the wrong segment's context.
3322
3.03k
    for (auto& [cid, expr_ctx] : _virtual_column_exprs) {
3323
0
        VExprContextSPtr context;
3324
0
        RETURN_IF_ERROR(expr_ctx->clone(_opts.runtime_state, context));
3325
0
        context->set_index_context(inverted_index_context);
3326
0
        expr_ctx = context;
3327
0
    }
3328
3.03k
    return Status::OK();
3329
3.03k
}
3330
3331
Status SegmentIterator::_apply_expr_zonemap_to_row_ranges(const VExprContextSPtrs& conjuncts,
3332
                                                          rowid_t min_rowid,
3333
7
                                                          RowRanges* row_ranges) {
3334
7
    DORIS_CHECK(row_ranges != nullptr);
3335
7
    if (!expr_zonemap::is_expr_zonemap_filter_enabled(_opts.runtime_state) || conjuncts.empty() ||
3336
7
        row_ranges->is_empty()) {
3337
0
        return Status::OK();
3338
0
    }
3339
3340
7
    std::unordered_map<int, VExprContextSPtrs> ctxs_by_slot;
3341
7
    for (const auto& conjunct : conjuncts) {
3342
7
        auto slot_index = expr_zonemap::single_slot_zonemap_index(conjunct);
3343
7
        if (slot_index >= 0) {
3344
5
            ctxs_by_slot[slot_index].emplace_back(conjunct);
3345
5
        }
3346
7
    }
3347
    // Page zone maps are stored per column. Multi-slot expressions need page alignment across
3348
    // multiple column readers and are therefore left to segment-level pruning for now.
3349
7
    if (ctxs_by_slot.empty()) {
3350
2
        return Status::OK();
3351
2
    }
3352
3353
5
    ColumnIteratorOptions iter_opts {
3354
5
            .use_page_cache = _opts.use_page_cache,
3355
5
            .file_reader = _file_reader.get(),
3356
5
            .stats = _opts.stats,
3357
5
            .io_ctx = _opts.io_ctx,
3358
5
    };
3359
5
    for (const auto& [slot_index, slot_conjuncts] : ctxs_by_slot) {
3360
5
        if (cast_set<size_t>(slot_index) >= _schema->num_column_ids()) {
3361
0
            continue;
3362
0
        }
3363
5
        const auto cid = _schema->column_id(cast_set<size_t>(slot_index));
3364
5
        if (!_segment->can_apply_predicate_safely(cid, *_schema,
3365
5
                                                  _opts.target_cast_type_for_variants, _opts)) {
3366
0
            continue;
3367
0
        }
3368
5
        const auto* tablet_column = _schema->column(cid);
3369
5
        if (tablet_column == nullptr) {
3370
0
            continue;
3371
0
        }
3372
5
        std::shared_ptr<ColumnReader> reader;
3373
5
        Status st =
3374
5
                _segment->get_column_reader(*tablet_column, &reader, _opts.stats, &_opts.io_ctx);
3375
5
        if (st.is<ErrorCode::NOT_FOUND>()) {
3376
4
            continue;
3377
4
        }
3378
1
        RETURN_IF_ERROR(st);
3379
1
        if (reader == nullptr || !reader->has_zone_map()) {
3380
0
            continue;
3381
0
        }
3382
1
        const std::vector<ZoneMapPB>* page_zone_maps = nullptr;
3383
1
        RETURN_IF_ERROR(reader->get_page_zone_maps(iter_opts, &page_zone_maps));
3384
1
        if (page_zone_maps == nullptr || page_zone_maps->empty()) {
3385
0
            continue;
3386
0
        }
3387
1
        auto data_type = _segment->get_data_type_of(*tablet_column, _opts);
3388
1
        if (data_type == nullptr) {
3389
0
            continue;
3390
0
        }
3391
3392
1
        RowRanges column_ranges;
3393
1
        ZoneMapEvalStats page_stats;
3394
9
        for (uint32_t page_index = 0; page_index < page_zone_maps->size(); ++page_index) {
3395
8
            RowRange page_range;
3396
8
            RETURN_IF_ERROR(reader->get_row_range_for_page(page_index, iter_opts, &page_range));
3397
8
            if (!page_range.is_valid() || page_range.to() <= min_rowid) {
3398
0
                continue;
3399
0
            }
3400
8
            ZoneMapEvalContext ctx;
3401
8
            ZoneMapEvalContext::SlotZoneMap slot_zone_map;
3402
8
            slot_zone_map.data_type = data_type;
3403
8
            ZoneMap zone_map;
3404
8
            RETURN_IF_ERROR(
3405
8
                    ZoneMap::from_proto((*page_zone_maps)[page_index], data_type, zone_map));
3406
8
            slot_zone_map.zone_map = std::make_shared<ZoneMap>(std::move(zone_map));
3407
8
            ctx.slots.emplace(slot_index, std::move(slot_zone_map));
3408
8
            const auto result = VExprContext::evaluate_zonemap_filter(slot_conjuncts, ctx);
3409
8
            page_stats.merge_page_eval_stats(ctx.stats);
3410
8
            if (result != ZoneMapFilterResult::kNoMatch) {
3411
4
                column_ranges.add(
3412
4
                        RowRange(std::max<int64_t>(page_range.from(), min_rowid), page_range.to()));
3413
4
            } else {
3414
4
                ++_opts.stats->expr_zonemap_filtered_pages;
3415
4
            }
3416
8
        }
3417
1
        page_stats.accumulate_to(_opts.stats);
3418
1
        RowRanges::ranges_intersection(*row_ranges, column_ranges, row_ranges);
3419
1
        if (row_ranges->is_empty()) {
3420
0
            return Status::OK();
3421
0
        }
3422
1
    }
3423
5
    return Status::OK();
3424
5
}
3425
3426
3.03k
void SegmentIterator::_calculate_common_expr_index_exec_status() {
3427
3.03k
    for (const auto& root_expr_ctx : _common_expr_ctxs_push_down) {
3428
21
        const auto& root_expr = root_expr_ctx->root();
3429
21
        if (root_expr == nullptr) {
3430
0
            continue;
3431
0
        }
3432
21
        _common_expr_to_slotref_map[root_expr_ctx.get()] = std::unordered_map<ColumnId, VExpr*>();
3433
3434
21
        std::stack<VExprSPtr> stack;
3435
21
        stack.emplace(root_expr);
3436
3437
42
        while (!stack.empty()) {
3438
21
            const auto& expr = stack.top();
3439
21
            stack.pop();
3440
3441
30
            for (const auto& child : expr->children()) {
3442
30
                if (child->is_virtual_slot_ref()) {
3443
                    // Expand virtual slot ref to its underlying expression tree and
3444
                    // collect real slot refs used inside. We still associate those
3445
                    // slot refs with the current parent expr node for inverted index
3446
                    // tracking, just like normal slot refs.
3447
0
                    auto* vir_slot_ref = assert_cast<VirtualSlotRef*>(child.get());
3448
0
                    auto vir_expr = vir_slot_ref->get_virtual_column_expr();
3449
0
                    if (vir_expr) {
3450
0
                        std::stack<VExprSPtr> vir_stack;
3451
0
                        vir_stack.emplace(vir_expr);
3452
3453
0
                        while (!vir_stack.empty()) {
3454
0
                            const auto& vir_node = vir_stack.top();
3455
0
                            vir_stack.pop();
3456
3457
0
                            for (const auto& vir_child : vir_node->children()) {
3458
0
                                if (vir_child->is_slot_ref()) {
3459
0
                                    auto* inner_slot_ref = assert_cast<VSlotRef*>(vir_child.get());
3460
0
                                    auto cid = _schema->column_id(inner_slot_ref->column_id());
3461
0
                                    _common_expr_index_exec_status[cid][expr.get()] = false;
3462
0
                                    _common_expr_to_slotref_map[root_expr_ctx.get()]
3463
0
                                                               [inner_slot_ref->column_id()] =
3464
0
                                                                       expr.get();
3465
0
                                }
3466
3467
0
                                if (!vir_child->children().empty()) {
3468
0
                                    vir_stack.emplace(vir_child);
3469
0
                                }
3470
0
                            }
3471
0
                        }
3472
0
                    }
3473
0
                }
3474
                // Example: CAST(v['a'] AS VARCHAR) MATCH 'hello', do not add CAST expr to index tracking.
3475
30
                auto expr_without_cast = VExpr::expr_without_cast(child);
3476
30
                if (expr_without_cast->is_slot_ref() && expr->op() != TExprOpcode::CAST) {
3477
20
                    auto* column_slot_ref = assert_cast<VSlotRef*>(expr_without_cast.get());
3478
20
                    auto cid = _schema->column_id(column_slot_ref->column_id());
3479
20
                    _common_expr_index_exec_status[cid][expr.get()] = false;
3480
20
                    _common_expr_to_slotref_map[root_expr_ctx.get()][column_slot_ref->column_id()] =
3481
20
                            expr.get();
3482
20
                }
3483
30
            }
3484
3485
21
            const auto& children = expr->children();
3486
51
            for (int i = cast_set<int>(children.size()) - 1; i >= 0; --i) {
3487
30
                if (!children[i]->children().empty()) {
3488
0
                    stack.emplace(children[i]);
3489
0
                }
3490
30
            }
3491
21
        }
3492
21
    }
3493
3.03k
}
3494
3495
bool SegmentIterator::_no_need_read_key_data(ColumnId cid, MutableColumnPtr& column,
3496
28.0k
                                             size_t nrows_read) {
3497
28.0k
    if (_opts.runtime_state && !_opts.runtime_state->query_options().enable_no_need_read_data_opt) {
3498
0
        return false;
3499
0
    }
3500
3501
28.0k
    if (!((_opts.tablet_schema->keys_type() == KeysType::DUP_KEYS ||
3502
28.0k
           (_opts.tablet_schema->keys_type() == KeysType::UNIQUE_KEYS &&
3503
11.1k
            _opts.enable_unique_key_merge_on_write)))) {
3504
7.58k
        return false;
3505
7.58k
    }
3506
3507
20.4k
    if (_opts.push_down_agg_type_opt != TPushAggOp::COUNT_ON_INDEX) {
3508
20.4k
        return false;
3509
20.4k
    }
3510
3511
20
    if (!_opts.tablet_schema->column(cid).is_key()) {
3512
14
        return false;
3513
14
    }
3514
3515
6
    if (_has_delete_predicate(cid)) {
3516
0
        return false;
3517
0
    }
3518
3519
6
    if (!_check_all_conditions_passed_inverted_index_for_column(cid)) {
3520
6
        return false;
3521
6
    }
3522
3523
0
    insert_many_not_null_defaults(column, nrows_read);
3524
0
    return true;
3525
6
}
3526
3527
20.4k
bool SegmentIterator::_has_delete_predicate(ColumnId cid) {
3528
20.4k
    std::set<uint32_t> delete_columns_set;
3529
20.4k
    _opts.delete_condition_predicates->get_all_column_ids(delete_columns_set);
3530
20.4k
    return delete_columns_set.contains(cid);
3531
20.4k
}
3532
3533
13.3k
bool SegmentIterator::_can_opt_limit_reads() {
3534
13.3k
    if (_opts.read_limit == 0) {
3535
13.3k
        return false;
3536
13.3k
    }
3537
3538
    // If SegmentIterator still needs to evaluate predicates/common exprs, LIMIT must be applied to
3539
    // post-filter rows by _apply_read_limit_to_selected_rows(); capping the raw read here could
3540
    // return fewer rows than the query LIMIT.
3541
7
    if (_is_need_vec_eval || _is_need_short_eval || _is_need_expr_eval) {
3542
3
        return false;
3543
3
    }
3544
3545
4
    if (_opts.delete_condition_predicates->num_of_column_predicate() > 0) {
3546
1
        return false;
3547
1
    }
3548
3549
3
    bool all_true = std::ranges::all_of(_schema->column_ids(), [this](auto cid) {
3550
3
        if (cid == _opts.tablet_schema->delete_sign_idx()) {
3551
0
            return true;
3552
0
        }
3553
3
        if (_check_all_conditions_passed_inverted_index_for_column(cid, true)) {
3554
2
            return true;
3555
2
        }
3556
1
        return false;
3557
3
    });
3558
3559
3
    DBUG_EXECUTE_IF("segment_iterator.topn_opt_1", {
3560
3
        LOG(INFO) << "col_predicates: " << _col_predicates.size() << ", all_true: " << all_true;
3561
3
    })
3562
3563
3
    DBUG_EXECUTE_IF("segment_iterator.topn_opt_2", {
3564
3
        if (all_true) {
3565
3
            return Status::Error<ErrorCode::INTERNAL_ERROR>("topn opt 2 execute failed");
3566
3
        }
3567
3
    })
3568
3569
3
    return all_true;
3570
3
}
3571
3572
// Before get next batch. make sure all virtual columns in block has type ColumnNothing.
3573
13.3k
void SegmentIterator::_init_virtual_columns(Block* block) {
3574
13.3k
    for (const auto& [cid, expr_ctx] : _virtual_column_exprs) {
3575
0
        auto idx = _schema->column_index(cid);
3576
0
        auto& col_with_type_and_name = block->get_by_position(idx);
3577
0
        col_with_type_and_name.column = ColumnNothing::create(0);
3578
0
        col_with_type_and_name.type = expr_ctx->root()->data_type();
3579
0
    }
3580
13.3k
}
3581
3582
10.3k
Status SegmentIterator::_materialization_of_virtual_column(Block* block) {
3583
    // Some expr can not process empty block, such as function `element_at`.
3584
    // So materialize virtual column in advance to avoid errors.
3585
10.3k
    if (_selected_size == 0) {
3586
38
        for (const auto& [cid, expr_ctx] : _virtual_column_exprs) {
3587
0
            auto idx = _schema->column_index(cid);
3588
0
            auto& col_with_type_and_name = block->get_by_position(idx);
3589
0
            col_with_type_and_name.column = expr_ctx->root()->data_type()->create_column();
3590
0
            col_with_type_and_name.type = expr_ctx->root()->data_type();
3591
0
        }
3592
38
        return Status::OK();
3593
38
    }
3594
10.2k
    if (_virtual_column_exprs.empty()) {
3595
10.2k
        return Status::OK();
3596
10.2k
    }
3597
3598
0
    for (const auto& cid_and_expr : _virtual_column_exprs) {
3599
0
        auto cid = cid_and_expr.first;
3600
0
        auto column_expr = cid_and_expr.second;
3601
0
        auto materialized_pos = _schema->column_index(cid);
3602
0
        auto& column = block->get_by_position(materialized_pos).column;
3603
0
        if (check_and_get_column<const ColumnNothing>(column.get())) {
3604
0
            VLOG_DEBUG << fmt::format("Virtual column is doing materialization, cid {}, col idx {}",
3605
0
                                      cid, materialized_pos);
3606
0
            ColumnPtr result_column;
3607
            // The first block column may still be ColumnNothing(0) for a virtual column, while
3608
            // predicates have already reduced _selected_size. Evaluate the expression over the
3609
            // selected row count instead of Block::rows().
3610
0
            RETURN_IF_ERROR(column_expr->root()->execute_column(column_expr.get(), block, nullptr,
3611
0
                                                                _selected_size, result_column));
3612
3613
0
            block->replace_by_position(materialized_pos, std::move(result_column));
3614
0
        }
3615
0
    }
3616
0
    return Status::OK();
3617
0
}
3618
3619
3.03k
void SegmentIterator::_prepare_score_column_materialization() {
3620
3.03k
    if (_score_runtime == nullptr) {
3621
3.03k
        return;
3622
3.03k
    }
3623
3624
0
    ScoreRangeFilterPtr filter;
3625
0
    if (_score_runtime->has_score_range_filter()) {
3626
0
        const auto& range_info = _score_runtime->get_score_range_info();
3627
0
        filter = std::make_shared<ScoreRangeFilter>(range_info->op, range_info->threshold);
3628
0
    }
3629
3630
0
    IColumn::MutablePtr result_column;
3631
0
    auto result_row_ids = std::make_unique<std::vector<uint64_t>>();
3632
0
    if (_score_runtime->get_limit() > 0 && _col_predicates.empty() &&
3633
0
        _common_expr_ctxs_push_down.empty()) {
3634
0
        OrderType order_type = _score_runtime->is_asc() ? OrderType::ASC : OrderType::DESC;
3635
0
        _index_query_context->collection_similarity->get_topn_bm25_scores(
3636
0
                &_row_bitmap, result_column, result_row_ids, order_type,
3637
0
                _score_runtime->get_limit(), filter);
3638
0
    } else {
3639
0
        _index_query_context->collection_similarity->get_bm25_scores(&_row_bitmap, result_column,
3640
0
                                                                     result_row_ids, filter);
3641
0
    }
3642
0
    const size_t dst_col_idx = _score_runtime->get_dest_column_idx();
3643
0
    auto* column_iter = _column_iterators[_schema->column_id(dst_col_idx)].get();
3644
0
    auto* virtual_column_iter = dynamic_cast<VirtualColumnIterator*>(column_iter);
3645
0
    virtual_column_iter->prepare_materialization(
3646
0
            std::move(result_column),
3647
0
            std::shared_ptr<std::vector<uint64_t>>(std::move(result_row_ids)));
3648
0
}
3649
3650
} // namespace segment_v2
3651
} // namespace doris