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 |