Coverage Report

Created: 2026-03-17 00:16

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format/parquet/vparquet_group_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
#pragma once
18
#include <stddef.h>
19
#include <stdint.h>
20
21
#include <limits>
22
#include <memory>
23
#include <string>
24
#include <tuple>
25
#include <unordered_map>
26
#include <utility>
27
#include <vector>
28
29
#include "common/status.h"
30
#include "core/column/column.h"
31
#include "exprs/vexpr_fwd.h"
32
#include "format/parquet/parquet_common.h"
33
#include "format/parquet/vparquet_column_reader.h"
34
#include "format/table/table_format_reader.h"
35
#include "io/fs/file_reader_writer_fwd.h"
36
#include "storage/id_manager.h"
37
#include "storage/utils.h"
38
39
namespace cctz {
40
class time_zone;
41
} // namespace cctz
42
namespace doris {
43
class ObjectPool;
44
class RowDescriptor;
45
class RuntimeState;
46
class SlotDescriptor;
47
class TupleDescriptor;
48
49
namespace io {
50
struct IOContext;
51
} // namespace io
52
class Block;
53
class FieldDescriptor;
54
} // namespace doris
55
namespace tparquet {
56
class ColumnMetaData;
57
class OffsetIndex;
58
class RowGroup;
59
} // namespace tparquet
60
61
namespace doris::segment_v2 {
62
class RowIdColumnIteratorV2;
63
}
64
65
namespace doris {
66
#include "common/compile_check_begin.h"
67
// TODO: we need to determine it by test.
68
69
class RowGroupReader : public ProfileCollector {
70
public:
71
    std::shared_ptr<TableSchemaChangeHelper::Node> _table_info_node_ptr;
72
    static const std::vector<int64_t> NO_DELETE;
73
74
    struct RowGroupIndex {
75
        int32_t row_group_id;
76
        int64_t first_row;
77
        int64_t last_row;
78
        RowGroupIndex(int32_t id, int64_t first, int64_t last)
79
177
                : row_group_id(id), first_row(first), last_row(last) {}
80
    };
81
82
    // table name
83
    struct LazyReadContext {
84
        // all conjuncts: in sql, join runtime filter, topn runtime filter.
85
        VExprContextSPtrs conjuncts;
86
87
        // ParquetReader::set_fill_columns(xxx, xxx) will set these two members
88
        std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
89
                fill_partition_columns;
90
        std::unordered_map<std::string, VExprContextSPtr> fill_missing_columns;
91
92
        phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>
93
                slot_id_to_predicates;
94
        bool can_lazy_read = false;
95
        // block->rows() returns the number of rows of the first column,
96
        // so we should check and resize the first column
97
        bool resize_first_column = true;
98
        std::vector<std::string> all_read_columns;
99
        // include predicate_partition_columns & predicate_missing_columns
100
        std::vector<uint32_t> all_predicate_col_ids;
101
        // save slot_id to find dict filter column name, because expr column name may
102
        // be different with parquet column name
103
        // std::pair<std::vector<col_name>, std::vector<slot_id>>
104
        std::pair<std::vector<std::string>, std::vector<int>> predicate_columns;
105
        std::vector<std::string> lazy_read_columns;
106
        std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
107
                predicate_partition_columns;
108
        // lazy read partition columns or all partition columns
109
        std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
110
                partition_columns;
111
        std::unordered_map<std::string, VExprContextSPtr> predicate_missing_columns;
112
        VExprContextSPtrs missing_columns_conjuncts;
113
        // lazy read missing columns or all missing columns
114
        std::unordered_map<std::string, VExprContextSPtr> missing_columns;
115
        // should turn off filtering by page index, lazy read and dict filter if having complex type
116
        bool has_complex_type = false;
117
    };
118
119
    /**
120
     * Support row-level delete in iceberg:
121
     * https://iceberg.apache.org/spec/#position-delete-files
122
     */
123
    struct PositionDeleteContext {
124
        // the filtered rows in current row group
125
        const std::vector<int64_t>& delete_rows;
126
        // the first row id of current row group in parquet file
127
        const int64_t first_row_id;
128
        // the number of rows in current row group
129
        const int64_t num_rows;
130
        const int64_t last_row_id;
131
        // current row id to read in the row group
132
        int64_t current_row_id;
133
        // start index in delete_rows
134
        const int64_t start_index;
135
        // end index in delete_rows
136
        const int64_t end_index;
137
        // current index in delete_rows
138
        int64_t index;
139
        const bool has_filter;
140
141
        PositionDeleteContext(const std::vector<int64_t>& delete_rows, const int64_t num_rows,
142
                              const int64_t first_row_id, const int64_t start_index,
143
                              const int64_t end_index)
144
35
                : delete_rows(delete_rows),
145
35
                  first_row_id(first_row_id),
146
35
                  num_rows(num_rows),
147
35
                  last_row_id(first_row_id + num_rows),
148
35
                  current_row_id(first_row_id),
149
35
                  start_index(start_index),
150
35
                  end_index(end_index),
151
35
                  index(start_index),
152
35
                  has_filter(end_index > start_index) {}
153
154
        PositionDeleteContext(const int64_t num_rows, const int64_t first_row)
155
35
                : PositionDeleteContext(NO_DELETE, num_rows, first_row, 0, 0) {}
156
157
        PositionDeleteContext(const PositionDeleteContext& filter) = default;
158
    };
159
160
    RowGroupReader(io::FileReaderSPtr file_reader, const std::vector<std::string>& read_columns,
161
                   const int32_t row_group_id, const tparquet::RowGroup& row_group,
162
                   const cctz::time_zone* ctz, io::IOContext* io_ctx,
163
                   const PositionDeleteContext& position_delete_ctx,
164
                   const LazyReadContext& lazy_read_ctx, RuntimeState* state,
165
                   const std::set<uint64_t>& column_ids,
166
                   const std::set<uint64_t>& filter_column_ids);
167
168
    ~RowGroupReader();
169
    Status init(const FieldDescriptor& schema, RowRanges& row_ranges,
170
                std::unordered_map<int, tparquet::OffsetIndex>& col_offsets,
171
                const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
172
                const std::unordered_map<std::string, int>* colname_to_slot_id,
173
                const VExprContextSPtrs* not_single_slot_filter_conjuncts,
174
                const std::unordered_map<int, VExprContextSPtrs>* slot_id_to_filter_conjuncts);
175
    Status next_batch(Block* block, size_t batch_size, size_t* read_rows, bool* batch_eof);
176
35
    int64_t lazy_read_filtered_rows() const { return _lazy_read_filtered_rows; }
177
35
    int64_t predicate_filter_time() const { return _predicate_filter_time; }
178
35
    int64_t dict_filter_rewrite_time() const { return _dict_filter_rewrite_time; }
179
180
    ParquetColumnReader::ColumnStatistics merged_column_statistics();
181
0
    void set_remaining_rows(int64_t rows) { _remaining_rows = rows; }
182
0
    int64_t get_remaining_rows() { return _remaining_rows; }
183
184
    void set_row_id_column_iterator(
185
            const std::pair<std::shared_ptr<segment_v2::RowIdColumnIteratorV2>, int>&
186
35
                    iterator_pair) {
187
35
        _row_id_column_iterator_pair = iterator_pair;
188
35
    }
189
190
35
    void set_current_row_group_idx(RowGroupIndex row_group_idx) {
191
35
        _current_row_group_idx = row_group_idx;
192
35
    }
193
194
    void set_col_name_to_block_idx(
195
35
            std::unordered_map<std::string, uint32_t>* col_name_to_block_idx) {
196
35
        _col_name_to_block_idx = col_name_to_block_idx;
197
35
    }
198
199
protected:
200
14
    void _collect_profile_before_close() override {
201
14
        if (_file_reader != nullptr) {
202
14
            _file_reader->collect_profile_before_close();
203
14
        }
204
14
    }
205
206
private:
207
    Status _read_empty_batch(size_t batch_size, size_t* read_rows, bool* batch_eof,
208
                             bool* modify_row_ids);
209
210
    Status _read_column_data(Block* block, const std::vector<std::string>& columns,
211
                             size_t batch_size, size_t* read_rows, bool* batch_eof,
212
                             FilterMap& filter_map);
213
214
    Status _do_lazy_read(Block* block, size_t batch_size, size_t* read_rows, bool* batch_eof);
215
    Status _rebuild_filter_map(FilterMap& filter_map,
216
                               DorisUniqueBufferPtr<uint8_t>& filter_map_data,
217
                               size_t pre_read_rows) const;
218
    Status _fill_partition_columns(
219
            Block* block, size_t rows,
220
            const std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>&
221
                    partition_columns);
222
    Status _fill_missing_columns(
223
            Block* block, size_t rows,
224
            const std::unordered_map<std::string, VExprContextSPtr>& missing_columns);
225
    Status _build_pos_delete_filter(size_t read_rows);
226
    Status _filter_block(Block* block, int column_to_keep,
227
                         const std::vector<uint32_t>& columns_to_filter);
228
    Status _filter_block_internal(Block* block, const std::vector<uint32_t>& columns_to_filter,
229
                                  const IColumn::Filter& filter);
230
231
    bool _can_filter_by_dict(int slot_id, const tparquet::ColumnMetaData& column_metadata);
232
    bool is_dictionary_encoded(const tparquet::ColumnMetaData& column_metadata);
233
    Status _rewrite_dict_predicates();
234
    Status _rewrite_dict_conjuncts(std::vector<int32_t>& dict_codes, int slot_id, bool is_nullable);
235
    Status _convert_dict_cols_to_string_cols(Block* block);
236
237
    Status _get_current_batch_row_id(size_t read_rows);
238
    Status _fill_row_id_columns(Block* block, size_t read_rows, bool is_current_row_ids);
239
240
    io::FileReaderSPtr _file_reader;
241
    std::unordered_map<std::string, std::unique_ptr<ParquetColumnReader>>
242
            _column_readers; // table_column_name
243
    const std::vector<std::string>& _read_table_columns;
244
245
    const int32_t _row_group_id;
246
    const tparquet::RowGroup& _row_group_meta;
247
    int64_t _remaining_rows;
248
    const cctz::time_zone* _ctz = nullptr;
249
    io::IOContext* _io_ctx = nullptr;
250
    PositionDeleteContext _position_delete_ctx;
251
    // merge the row ranges generated from page index and position delete.
252
    RowRanges _read_ranges;
253
254
    const LazyReadContext& _lazy_read_ctx;
255
    int64_t _lazy_read_filtered_rows = 0;
256
    int64_t _predicate_filter_time = 0;
257
    int64_t _dict_filter_rewrite_time = 0;
258
    // If continuous batches are skipped, we can cache them to skip a whole page
259
    size_t _cached_filtered_rows = 0;
260
    std::unique_ptr<IColumn::Filter> _pos_delete_filter_ptr;
261
    int64_t _total_read_rows = 0;
262
    const TupleDescriptor* _tuple_descriptor = nullptr;
263
    const RowDescriptor* _row_descriptor = nullptr;
264
    const std::unordered_map<std::string, int>* _col_name_to_slot_id = nullptr;
265
    const std::unordered_map<int, VExprContextSPtrs>* _slot_id_to_filter_conjuncts = nullptr;
266
    VExprContextSPtrs _dict_filter_conjuncts;
267
    VExprContextSPtrs _filter_conjuncts;
268
    // std::pair<col_name, slot_id>
269
    std::vector<std::pair<std::string, int>> _dict_filter_cols;
270
    RuntimeState* _state = nullptr;
271
    std::shared_ptr<ObjectPool> _obj_pool;
272
    const std::set<uint64_t>& _column_ids;
273
    const std::set<uint64_t>& _filter_column_ids;
274
    bool _is_row_group_filtered = false;
275
276
    RowGroupIndex _current_row_group_idx {0, 0, 0};
277
    std::pair<std::shared_ptr<segment_v2::RowIdColumnIteratorV2>, int>
278
            _row_id_column_iterator_pair = {nullptr, -1};
279
    std::vector<rowid_t> _current_batch_row_ids;
280
281
    std::unordered_map<std::string, uint32_t>* _col_name_to_block_idx = nullptr;
282
};
283
#include "common/compile_check_end.h"
284
285
} // namespace doris