Coverage Report

Created: 2026-09-21 12:29

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/tablet/tablet_reader.h
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
#pragma once
19
20
#include <gen_cpp/Descriptors_types.h>
21
#include <gen_cpp/PaloInternalService_types.h>
22
#include <gen_cpp/PlanNodes_types.h>
23
#include <stddef.h>
24
#include <stdint.h>
25
26
#include <memory>
27
#include <optional>
28
#include <set>
29
#include <string>
30
#include <unordered_set>
31
#include <utility>
32
#include <vector>
33
34
#include "agent/be_exec_version_manager.h"
35
#include "common/status.h"
36
#include "io/io_common.h"
37
#include "storage/delete/delete_handler.h"
38
#include "storage/iterators.h"
39
#include "storage/olap_common.h"
40
#include "storage/olap_tuple.h"
41
#include "storage/predicate/filter_olap_param.h"
42
#include "storage/row_cursor.h"
43
#include "storage/rowid_conversion.h"
44
#include "storage/rowset/rowset.h"
45
#include "storage/rowset/rowset_meta.h"
46
#include "storage/rowset/rowset_reader.h"
47
#include "storage/rowset/rowset_reader_context.h"
48
#include "storage/segment/variant/variant_compaction_paths.h"
49
#include "storage/tablet/base_tablet.h"
50
#include "storage/tablet/tablet_fwd.h"
51
52
namespace doris {
53
54
class RuntimeState;
55
class BloomFilterFuncBase;
56
class ColumnPredicate;
57
class DeleteBitmap;
58
class HybridSetBase;
59
class RuntimeProfile;
60
61
class VCollectIterator;
62
class Block;
63
class VExpr;
64
class VExprContext;
65
66
// Used to compare row with input scan key. Scan key only contains key columns,
67
// row contains all key columns, which is superset of key columns.
68
// So we should compare the common prefix columns of lhs and rhs.
69
//
70
// NOTE: if you are not sure if you can use it, please don't use this function.
71
1.33M
inline int compare_row_key(const RowCursor& lhs, const RowCursor& rhs) {
72
1.33M
    auto cmp_cids = std::min(lhs.field_count(), rhs.field_count());
73
2.61M
    for (uint32_t cid = 0; cid < cmp_cids; ++cid) {
74
2.58M
        const auto& lf = lhs.field(cid);
75
2.58M
        const auto& rf = rhs.field(cid);
76
        // Handle nulls: null < non-null
77
2.58M
        if (lf.is_null() != rf.is_null()) {
78
18.4E
            return lf.is_null() ? -1 : 1;
79
179k
        }
80
2.40M
        if (lf.is_null()) {
81
3.86k
            continue; // both null
82
3.86k
        }
83
2.39M
        auto cmp = lf <=> rf;
84
2.39M
        if (cmp < 0) return -1;
85
1.28M
        if (cmp > 0) return 1;
86
1.28M
    }
87
33.1k
    return 0;
88
1.33M
}
89
90
class TabletReader {
91
    struct KeysParam {
92
        std::vector<std::optional<RowCursor>> start_keys;
93
        std::vector<std::optional<RowCursor>> end_keys;
94
        bool start_key_include = false;
95
        bool end_key_include = false;
96
    };
97
98
public:
99
    // Params for Reader,
100
    // mainly include tablet, data version and fetch range.
101
    struct ReaderParams {
102
901k
        bool has_single_version() const {
103
901k
            return (rs_splits.size() == 1 &&
104
901k
                    rs_splits[0].rs_reader->rowset()->start_version() == 0 &&
105
901k
                    !rs_splits[0].rs_reader->rowset()->rowset_meta()->is_segments_overlapping()) ||
106
901k
                   (rs_splits.size() == 2 &&
107
894k
                    rs_splits[0].rs_reader->rowset()->rowset_meta()->num_rows() == 0 &&
108
894k
                    rs_splits[1].rs_reader->rowset()->start_version() == 2 &&
109
894k
                    !rs_splits[1].rs_reader->rowset()->rowset_meta()->is_segments_overlapping());
110
901k
        }
111
112
14.3k
        int get_be_exec_version() const {
113
14.3k
            if (runtime_state) {
114
10.4k
                return runtime_state->be_exec_version();
115
10.4k
            }
116
3.91k
            return BeExecVersionManager::get_newest_version();
117
14.3k
        }
118
119
934k
        void set_read_source(TabletReadSource read_source, bool skip_delete_bitmap = false) {
120
934k
            rs_splits = std::move(read_source.rs_splits);
121
934k
            delete_predicates = std::move(read_source.delete_predicates);
122
934k
#ifndef BE_TEST
123
935k
            if (!skip_delete_bitmap && tablet->need_read_delete_bitmap()) {
124
500k
                delete_bitmap = std::move(read_source.delete_bitmap);
125
500k
            }
126
934k
#endif
127
934k
        }
128
129
        BaseTabletSPtr tablet;
130
        TabletSchemaSPtr tablet_schema;
131
        // Set only by compaction, alongside its extended tablet_schema.
132
        VariantCompactionPathsSPtr variant_compaction_paths = nullptr;
133
        ReaderType reader_type = ReaderType::READER_QUERY;
134
        bool read_row_binlog = false;
135
        bool direct_mode = false;
136
        bool aggregation = false;
137
        // for compaction, schema_change, check_sum: we don't use page cache
138
        // for query, when the BE config disable_storage_page_cache is false, we use page cache
139
        bool use_page_cache = false;
140
        Version version = Version(-1, 0);
141
142
        // The vectors are range-aligned; nullopt represents an unbounded endpoint.
143
        std::vector<std::optional<OlapTuple>> start_key;
144
        std::vector<std::optional<OlapTuple>> end_key;
145
        bool start_key_include = false;
146
        bool end_key_include = false;
147
148
        std::vector<std::shared_ptr<ColumnPredicate>> predicates;
149
        std::vector<RowsetMetaSharedPtr> delete_predicates;
150
        // slots that cast may be eliminated in storage layer
151
        std::map<std::string, DataTypePtr> target_cast_type_for_variants;
152
153
        std::map<int32_t, TColumnAccessPaths> all_access_paths;
154
        std::map<int32_t, TColumnAccessPaths> predicate_access_paths;
155
156
        std::vector<RowSetSplits> rs_splits;
157
        // For unique key table with merge-on-write
158
        DeleteBitmapPtr delete_bitmap = nullptr;
159
160
        // The read schema of every Block at the TabletReader boundary.
161
        // Query scanners construct the FE-slot prefix in scan-tuple order;
162
        // TabletReader may append storage-only delete-predicate columns.
163
        ReadSchemaSPtr read_schema;
164
        // output_columns only contain columns in OrderByExprs and outputExprs
165
        // (column unique ids, not ordinals)
166
        std::set<int32_t> output_columns;
167
        // Ordinals (positions in read_schema) of extra storage key columns
168
        // that are present only for scan-schema alignment.
169
        // Example: for AGG keys (k1, k2), a query that returns k2 can scan
170
        // (k1, k2) and project away k1. Direct readers may avoid reading such
171
        // columns only if the lower iterator proves their real values are not
172
        // required by predicates, delete conditions, or expressions.
173
        std::set<ColumnId> extra_columns;
174
        RuntimeProfile* profile = nullptr;
175
        RuntimeState* runtime_state = nullptr;
176
177
        TPushAggOp::type push_down_agg_type_opt = TPushAggOp::NONE;
178
        VExprContextSPtrs common_expr_ctxs_push_down;
179
180
        // used for compaction to record row ids
181
        bool record_rowids = false;
182
        RowIdConversion* rowid_conversion = nullptr;
183
        std::vector<int> topn_filter_source_node_ids;
184
        // used for special optimization for query : ORDER BY key LIMIT n
185
        bool read_orderby_key = false;
186
        // used for special optimization for query : ORDER BY key DESC LIMIT n
187
        bool read_orderby_key_reverse = false;
188
        // For rows with the same key, use ascending order (small-to-large) for tie-breakers.
189
        // For example, use lower rowset version / segment id first.
190
        bool use_insert_order_when_same = false;
191
        // Force globally key-ordered reading for row-binlog scans (e.g. so MIN_DELTA can
192
        // group consecutive same-key changes across segments). Overlapping segments use the
193
        // merge iterator; segments proven non-overlapping use an ordered union.
194
        // See BetaRowsetReader::is_merge_iterator() in beta_rowset_reader.h:62.
195
        bool force_key_ordered_read = false;
196
        // num of columns for orderby key
197
        size_t read_orderby_key_num_prefix_columns = 0;
198
        // limit of rows for read_orderby_key
199
        size_t read_orderby_key_limit = 0;
200
        // for vertical compaction
201
        bool is_key_column_group = false;
202
        std::vector<uint32_t> key_group_cluster_key_idxes;
203
204
        // For sparse column compaction optimization
205
        // When true, use optimized path for sparse wide tables
206
        bool enable_sparse_optimization = false;
207
208
        bool is_segcompaction = false;
209
210
        // Enable value predicate pushdown for MOR tables
211
        bool enable_mor_value_predicate_pushdown = false;
212
213
        std::vector<RowwiseIteratorUPtr>* segment_iters_ptr = nullptr;
214
215
        void check_validation() const;
216
217
        int64_t batch_size = -1;
218
219
        // virtual column ordinal (position in read_schema) -> expression
220
        std::map<ColumnId, VExprContextSPtr> virtual_column_exprs;
221
222
        std::shared_ptr<ScoreRuntime> score_runtime;
223
        CollectionStatisticsPtr collection_statistics;
224
        std::shared_ptr<segment_v2::AnnTopNRuntime> ann_topn_runtime;
225
226
        uint64_t condition_cache_digest = 0;
227
228
        // General LIMIT budget forwarded to SegmentIterator. -1 means no limit.
229
        int64_t general_read_limit = -1;
230
        TBinlogScanType::type binlog_scan_type = TBinlogScanType::NONE;
231
    };
232
233
933k
    TabletReader() = default;
234
235
932k
    virtual ~TabletReader() = default;
236
237
    TabletReader(const TabletReader&) = delete;
238
    void operator=(const TabletReader&) = delete;
239
240
    // Initialize TabletReader with tablet, data version and fetch range.
241
    virtual Status init(const ReaderParams& read_params);
242
243
    // Read next block with aggregation.
244
    // Return OK and set `*eof` to false when next block is read
245
    // Return OK and set `*eof` to true when no more rows can be read.
246
    // Return others when unexpected error happens.
247
0
    virtual Status next_block_with_aggregation(Block* block, bool* eof) {
248
0
        return Status::Error<ErrorCode::READER_INITIALIZE_ERROR>(
249
0
                "TabletReader not support next_block_with_aggregation");
250
0
    }
251
252
1.36k
    virtual uint64_t merged_rows() const { return _merged_rows; }
253
254
11.9k
    uint64_t filtered_rows() const {
255
11.9k
        return _stats.rows_del_filtered + _stats.rows_del_by_bitmap +
256
11.9k
               _stats.rows_conditions_filtered + _stats.rows_vec_del_cond_filtered +
257
11.9k
               _stats.rows_vec_cond_filtered + _stats.rows_short_circuit_cond_filtered;
258
11.9k
    }
259
260
902k
    void set_batch_size(int batch_size) { _reader_context.batch_size = batch_size; }
261
262
10.3M
    size_t batch_max_rows() const { return _reader_context.batch_size; }
263
264
902k
    void set_preferred_block_size_bytes(size_t bytes) {
265
902k
        _reader_context.preferred_block_size_bytes = bytes;
266
902k
    }
267
268
    // Returns the preferred output block byte budget. Subclasses that support adaptive batch size
269
    // should override this; the base returns 0 (disabled) so VCollectIterator degrades safely
270
    // when called through a TabletReader* that has not been configured.
271
0
    virtual size_t preferred_block_size_bytes() const { return 0; }
272
273
3.17M
    const OlapReaderStatistics& stats() const { return _stats; }
274
4.68M
    OlapReaderStatistics* mutable_stats() { return &_stats; }
275
276
2
    virtual void update_profile(RuntimeProfile* profile) {}
277
278
    // Remove the delete-condition columns from `all_access_paths` so they fall back to a full
279
    // read (a meta-only read would make the storage delete predicate match nothing and leak
280
    // deleted rows).
281
    static void remove_delete_columns_from_access_paths(
282
            const DeleteHandler& delete_handler, const ReadSchema& read_schema,
283
            std::map<int32_t, TColumnAccessPaths>& all_access_paths);
284
285
protected:
286
    friend class VCollectIterator;
287
288
    Status _init_params(const ReaderParams& read_params);
289
290
    Status _capture_rs_readers(const ReaderParams& read_params);
291
292
    Status _init_keys_param(const ReaderParams& read_params);
293
294
    Status _init_orderby_keys_param(const ReaderParams& read_params);
295
296
    Status _init_column_predicates(const ReaderParams& read_params);
297
298
    Status _init_delete_condition(const ReaderParams& read_params);
299
300
1.85M
    const BaseTabletSPtr& tablet() { return _tablet; }
301
302
    // The read schema shared by the caller, TabletReader, VCollect and each rowset reader.
303
    ReadSchemaSPtr _read_schema;
304
305
    // used for special optimization for query : ORDER BY key [ASC|DESC] LIMIT n
306
    // columns for orderby keys
307
    std::vector<uint32_t> _orderby_key_columns;
308
    BaseTabletSPtr _tablet;
309
    RowsetReaderContext _reader_context;
310
    TabletSchemaSPtr _tablet_schema;
311
    KeysParam _keys_param;
312
    std::vector<bool> _is_lower_keys_included;
313
    std::vector<bool> _is_upper_keys_included;
314
    std::vector<std::shared_ptr<ColumnPredicate>> _col_predicates;
315
    std::vector<std::shared_ptr<ColumnPredicate>> _value_col_predicates;
316
    DeleteHandler _delete_handler;
317
318
    // Indicates whether the tablets has do a aggregation in storage engine.
319
    bool _aggregation = false;
320
    // for agg query, we don't need to finalize when scan agg object data
321
    ReaderType _reader_type = ReaderType::READER_QUERY;
322
    bool _delete_sign_available = false;
323
    bool _filter_delete = false;
324
    bool _direct_mode = false;
325
326
    uint64_t _merged_rows = 0;
327
    OlapReaderStatistics _stats;
328
};
329
330
} // namespace doris