be/src/core/block/block.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 | | // This file is copied from |
18 | | // https://github.com/ClickHouse/ClickHouse/blob/master/src/Core/Block.h |
19 | | // and modified by Doris |
20 | | |
21 | | #pragma once |
22 | | |
23 | | #include <glog/logging.h> |
24 | | #include <parallel_hashmap/phmap.h> |
25 | | |
26 | | #include <cstddef> |
27 | | #include <cstdint> |
28 | | #include <initializer_list> |
29 | | #include <list> |
30 | | #include <memory> |
31 | | #include <ostream> |
32 | | #include <set> |
33 | | #include <string> |
34 | | #include <utility> |
35 | | #include <vector> |
36 | | |
37 | | #include "common/be_mock_util.h" |
38 | | #include "common/exception.h" |
39 | | #include "common/factory_creator.h" |
40 | | #include "common/status.h" |
41 | | #include "core/block/column_with_type_and_name.h" |
42 | | #include "core/block/columns_with_type_and_name.h" |
43 | | #include "core/column/column.h" |
44 | | #include "core/column/column_nullable.h" |
45 | | #include "core/data_type/data_type.h" |
46 | | #include "core/data_type/data_type_nullable.h" |
47 | | #include "core/types.h" |
48 | | |
49 | | class SipHash; |
50 | | |
51 | | namespace doris { |
52 | | |
53 | | class TupleDescriptor; |
54 | | class PBlock; |
55 | | class SlotDescriptor; |
56 | | |
57 | | namespace segment_v2 { |
58 | | enum CompressionTypePB : int; |
59 | | } // namespace segment_v2 |
60 | | |
61 | | /** Container for set of columns for bunch of rows in memory. |
62 | | * This is unit of data processing. |
63 | | * Also contains metadata - data types of columns and their names |
64 | | * (either original names from a table, or generated names during temporary calculations). |
65 | | * Allows to insert, remove columns in arbitrary position, to change order of columns. |
66 | | */ |
67 | | class MutableBlock; |
68 | | |
69 | | class Block { |
70 | | ENABLE_FACTORY_CREATOR(Block); |
71 | | |
72 | | private: |
73 | | using Container = ColumnsWithTypeAndName; |
74 | | Container data; |
75 | | |
76 | | public: |
77 | 430k | Block() = default; |
78 | | Block(std::initializer_list<ColumnWithTypeAndName> il); |
79 | | Block(ColumnsWithTypeAndName data_); |
80 | | Block(const std::vector<SlotDescriptor*>& slots, size_t block_size); |
81 | | Block(const std::vector<SlotDescriptor>& slots, size_t block_size); |
82 | | |
83 | 814k | MOCK_FUNCTION ~Block() = default; |
84 | 848 | Block(const Block& block) = default; |
85 | 48.2k | Block& operator=(const Block& p) = default; |
86 | 63.7k | Block(Block&& block) = default; |
87 | 911 | Block& operator=(Block&& other) = default; |
88 | | |
89 | | void reserve(size_t count); |
90 | | // Make sure the nammes is useless when use block |
91 | | void clear_names(); |
92 | | |
93 | | /// insert the column at the specified position |
94 | | void insert(size_t position, const ColumnWithTypeAndName& elem); |
95 | | void insert(size_t position, ColumnWithTypeAndName&& elem); |
96 | | /// insert the column to the end |
97 | | void insert(const ColumnWithTypeAndName& elem); |
98 | | void insert(ColumnWithTypeAndName&& elem); |
99 | | /// remove the column at the specified position |
100 | | void erase(size_t position); |
101 | | /// remove the column at the [start, end) |
102 | | void erase_tail(size_t start); |
103 | | /// remove the columns at the specified positions |
104 | | void erase(const std::set<size_t>& positions); |
105 | | // T was std::set<int>, std::vector<int>, std::list<int> |
106 | | template <class T> |
107 | 0 | void erase_not_in(const T& container) { |
108 | 0 | Container new_data; |
109 | 0 | for (auto pos : container) { |
110 | 0 | new_data.emplace_back(std::move(data[pos])); |
111 | 0 | } |
112 | 0 | std::swap(data, new_data); |
113 | 0 | } |
114 | | |
115 | 19 | std::unordered_map<std::string, uint32_t> get_name_to_pos_map() const { |
116 | 19 | std::unordered_map<std::string, uint32_t> name_to_index_map; |
117 | 142 | for (uint32_t i = 0; i < data.size(); ++i) { |
118 | 123 | name_to_index_map[data[i].name] = i; |
119 | 123 | } |
120 | 19 | return name_to_index_map; |
121 | 19 | } |
122 | | |
123 | | /// References are invalidated after calling functions above. |
124 | 17.6M | ColumnWithTypeAndName& get_by_position(size_t position) { |
125 | 18.4E | DCHECK(data.size() > position) |
126 | 18.4E | << ", data.size()=" << data.size() << ", position=" << position; |
127 | 17.6M | return data[position]; |
128 | 17.6M | } |
129 | 29.1M | const ColumnWithTypeAndName& get_by_position(size_t position) const { return data[position]; } |
130 | | |
131 | 33.2k | void replace_by_position(size_t position, ColumnPtr&& res) { |
132 | 33.2k | this->get_by_position(position).column = std::move(res); |
133 | 33.2k | } |
134 | | |
135 | 6 | void replace_by_position(size_t position, const ColumnPtr& res) { |
136 | 6 | this->get_by_position(position).column = res; |
137 | 6 | } |
138 | | |
139 | 740 | void replace_by_position_if_const(size_t position) { |
140 | 740 | auto& element = this->get_by_position(position); |
141 | 740 | element.column = element.column->convert_to_full_column_if_const(); |
142 | 740 | } |
143 | | |
144 | | ColumnWithTypeAndName& safe_get_by_position(size_t position); |
145 | | const ColumnWithTypeAndName& safe_get_by_position(size_t position) const; |
146 | | |
147 | 72.1k | Container::iterator begin() { return data.begin(); } |
148 | 72.1k | Container::iterator end() { return data.end(); } |
149 | 9.64k | Container::const_iterator begin() const { return data.begin(); } |
150 | 9.61k | Container::const_iterator end() const { return data.end(); } |
151 | 0 | Container::const_iterator cbegin() const { return data.cbegin(); } |
152 | 0 | Container::const_iterator cend() const { return data.cend(); } |
153 | | |
154 | | // Get position of column by name. Returns -1 if there is no column with that name. |
155 | | // ATTN: this method is O(N). better maintain name -> position map in caller if you need to call it frequently. |
156 | | int get_position_by_name(const std::string& name) const; |
157 | | |
158 | | const ColumnsWithTypeAndName& get_columns_with_type_and_name() const; |
159 | | |
160 | | std::vector<std::string> get_names() const; |
161 | | DataTypes get_data_types() const; |
162 | | |
163 | 60 | DataTypePtr get_data_type(size_t index) const { |
164 | 60 | CHECK(index < data.size()); |
165 | 60 | return data[index].type; |
166 | 60 | } |
167 | | |
168 | | /// Returns number of rows from first column in block, not equal to nullptr. If no columns, returns 0. |
169 | | size_t rows() const; |
170 | | |
171 | | // Cut the rows in block, use in LIMIT operation |
172 | | void set_num_rows(size_t length); |
173 | | |
174 | | // Skip the rows in block, use in OFFSET, LIMIT operation |
175 | | void skip_num_rows(int64_t& offset); |
176 | | |
177 | | /// As the assumption we used around, the number of columns won't exceed int16 range. so no need to worry when we |
178 | | /// assign it to int32. |
179 | 18.2M | uint32_t columns() const { return static_cast<uint32_t>(data.size()); } |
180 | | |
181 | | /// Checks that every column in block is not nullptr and has same number of elements. |
182 | | void check_number_of_rows(bool allow_null_columns = false) const; |
183 | | |
184 | | Status check_type_and_column() const; |
185 | | |
186 | | Status check_column_and_type_not_null() const; |
187 | | |
188 | | Status check_no_column_string64() const; |
189 | | |
190 | | /// Approximate number of bytes used by column data in memory. |
191 | | /// This reflects the actual data footprint (e.g. string contents, numeric arrays) |
192 | | /// and is the metric used by adaptive batch size byte budgets. |
193 | | size_t bytes() const; |
194 | | |
195 | | /// Approximate number of allocated (reserved) bytes in memory. |
196 | | /// This may be larger than bytes() due to pre-allocated capacity in vectors/arenas. |
197 | | /// Used for memory tracking and profiling. |
198 | | MOCK_FUNCTION size_t allocated_bytes() const; |
199 | | |
200 | | /** Get a list of column names separated by commas. */ |
201 | | std::string dump_names() const; |
202 | | |
203 | | std::string dump_types() const; |
204 | | |
205 | | /** List of names, types and lengths of columns. Designed for debugging. */ |
206 | | std::string dump_structure() const; |
207 | | |
208 | | /** Get the same block, but empty. */ |
209 | | Block clone_empty() const; |
210 | | |
211 | | Columns get_columns() const; |
212 | | Columns get_columns_and_convert(); |
213 | | |
214 | | Block clone_without_columns(const std::vector<int>* column_offset = nullptr) const; |
215 | | |
216 | | /** Get empty columns with the same types as in block. */ |
217 | | MutableColumns clone_empty_columns() const; |
218 | | |
219 | | // RAII owner for mutating columns borrowed from a live Block. While the |
220 | | // guard is alive, the Block's column slots are moved out and column data |
221 | | // must be accessed through mutable_columns(). The guard restores columns on |
222 | | // destruction, so use it when the caller may exit early after detaching. |
223 | | class ScopedMutableColumns { |
224 | | public: |
225 | | explicit ScopedMutableColumns(Block& block); |
226 | | ~ScopedMutableColumns(); |
227 | | |
228 | | ScopedMutableColumns(const ScopedMutableColumns&) = delete; |
229 | | ScopedMutableColumns& operator=(const ScopedMutableColumns&) = delete; |
230 | | ScopedMutableColumns(ScopedMutableColumns&& other) noexcept; |
231 | | ScopedMutableColumns& operator=(ScopedMutableColumns&& other) noexcept; |
232 | | |
233 | 1.70k | MutableColumns& mutable_columns() { return _columns; } |
234 | 0 | const MutableColumns& mutable_columns() const { return _columns; } |
235 | | const DataTypePtr& get_datatype_by_position(size_t position) const; |
236 | | const std::string& get_name_by_position(size_t position) const; |
237 | | |
238 | | // Transfer the borrowed owners to another RAII object that will restore |
239 | | // them. After release(), the original Block remains without columns |
240 | | // until that owner restores them. Normal callers should let this guard |
241 | | // restore on destruction. |
242 | | MutableColumns release(); |
243 | | void restore(); |
244 | | |
245 | | private: |
246 | | Block* _block = nullptr; |
247 | | MutableColumns _columns; |
248 | | }; |
249 | | |
250 | | // Single-column variant for localized mutation of a live Block slot. The |
251 | | // selected slot is unavailable from the Block until this guard restores it. |
252 | | class ScopedMutableColumn { |
253 | | public: |
254 | | ScopedMutableColumn(Block& block, size_t position); |
255 | | ~ScopedMutableColumn(); |
256 | | |
257 | | ScopedMutableColumn(const ScopedMutableColumn&) = delete; |
258 | | ScopedMutableColumn& operator=(const ScopedMutableColumn&) = delete; |
259 | | ScopedMutableColumn(ScopedMutableColumn&& other) noexcept; |
260 | | ScopedMutableColumn& operator=(ScopedMutableColumn&& other) noexcept; |
261 | | |
262 | 64 | MutableColumnPtr& mutable_column() { return _column; } |
263 | 0 | const MutableColumnPtr& mutable_column() const { return _column; } |
264 | | |
265 | | void restore(); |
266 | | |
267 | | private: |
268 | | Block* _block = nullptr; |
269 | | size_t _position = 0; |
270 | | MutableColumnPtr _column; |
271 | | }; |
272 | | |
273 | | /** Get columns from a consumed block for mutation. Columns in block will be nullptr. */ |
274 | | MutableColumns mutate_columns() &&; |
275 | | MutableColumns mutate_columns() & = delete; |
276 | | |
277 | | /** Temporarily mutate a live Block's columns. The returned guard owns the columns and |
278 | | * restores them on destruction; prefer this over manual move/writeback. |
279 | | */ |
280 | | ScopedMutableColumns mutate_columns_scoped() &; |
281 | | ScopedMutableColumns mutate_columns_scoped() && = delete; |
282 | | |
283 | | /** Temporarily mutate one live Block column; use when only one slot needs ownership. */ |
284 | | ScopedMutableColumn mutate_column_scoped(size_t position) &; |
285 | | ScopedMutableColumn mutate_column_scoped(size_t position) && = delete; |
286 | | |
287 | | /** Replace columns in a block */ |
288 | | void set_columns(MutableColumns&& columns); |
289 | | void clear(); |
290 | | void swap(Block& other) noexcept; |
291 | | void swap(Block&& other) noexcept; |
292 | | |
293 | | // Shuffle columns in place based on the result_column_ids |
294 | | void shuffle_columns(const std::vector<int>& result_column_ids); |
295 | | |
296 | | // column_size == -1 clears all columns; otherwise clear [0, column_size) |
297 | | // and drop the rest. Shared columns are detached through clone_empty(), so |
298 | | // allocation or clone failures propagate. |
299 | | void clear_column_data(int64_t column_size = -1); |
300 | | void clear_column_data(const std::vector<uint32_t>& columns_to_clear); |
301 | | |
302 | 15.8k | MOCK_FUNCTION bool mem_reuse() { return !data.empty(); } |
303 | | |
304 | 12 | bool is_empty_column() { return data.empty(); } |
305 | | |
306 | 2.97M | bool empty() const { return rows() == 0; } |
307 | | |
308 | | /** |
309 | | * Updates SipHash of the Block, using update method of columns. |
310 | | * Returns hash for block, that could be used to differentiate blocks |
311 | | * with same structure, but different data. |
312 | | */ |
313 | | void update_hash(SipHash& hash) const; |
314 | | |
315 | | /** |
316 | | * Get block data in string. |
317 | | * If code is in default_implementation_for_nulls or something likely, type and column's nullity could |
318 | | * temporarily be not same. set allow_null_mismatch to true to dump it correctly. |
319 | | */ |
320 | | std::string dump_data(size_t begin = 0, size_t row_limit = 100, |
321 | | bool allow_null_mismatch = false) const; |
322 | | |
323 | | std::string dump_data_json(size_t begin = 0, size_t row_limit = 100, |
324 | | bool allow_null_mismatch = false) const; |
325 | | |
326 | | /** Get one line data from block, only use in load data */ |
327 | | std::string dump_one_line(size_t row, int column_end) const; |
328 | | |
329 | | Status append_to_block_by_selector(MutableBlock* dst, const IColumn::Selector& selector) const; |
330 | | |
331 | | // need exception safety |
332 | | static void filter_block_internal(Block* block, const std::vector<uint32_t>& columns_to_filter, |
333 | | const IColumn::Filter& filter); |
334 | | // need exception safety |
335 | | static void filter_block_internal(Block* block, const IColumn::Filter& filter, |
336 | | uint32_t column_to_keep); |
337 | | // need exception safety |
338 | | static void filter_block_internal(Block* block, const IColumn::Filter& filter); |
339 | | |
340 | | static Status filter_block(Block* block, const std::vector<uint32_t>& columns_to_filter, |
341 | | size_t filter_column_id, size_t column_to_keep); |
342 | | |
343 | | static Status filter_block(Block* block, size_t filter_column_id, size_t column_to_keep); |
344 | | |
345 | 708 | static void erase_useless_column(Block* block, size_t column_to_keep) { |
346 | 708 | block->erase_tail(column_to_keep); |
347 | 708 | } |
348 | | |
349 | | // serialize block to PBlock |
350 | | Status serialize(int be_exec_version, PBlock* pblock, size_t* uncompressed_bytes, |
351 | | size_t* compressed_bytes, int64_t* compress_time, |
352 | | segment_v2::CompressionTypePB compression_type, |
353 | | bool allow_transfer_large_data = false) const; |
354 | | |
355 | | Status deserialize(const PBlock& pblock, size_t* uncompressed_bytes, int64_t* decompress_time); |
356 | | |
357 | | std::unique_ptr<Block> create_same_struct_block(size_t size, bool is_reserve = false) const; |
358 | | |
359 | | /** Compares (*this) n-th row and rhs m-th row. |
360 | | * Returns negative number, 0, or positive number (*this) n-th row is less, equal, greater than rhs m-th row respectively. |
361 | | * Is used in sortings. |
362 | | * |
363 | | * If one of element's value is NaN or NULLs, then: |
364 | | * - if nan_direction_hint == -1, NaN and NULLs are considered as least than everything other; |
365 | | * - if nan_direction_hint == 1, NaN and NULLs are considered as greatest than everything other. |
366 | | * For example, if nan_direction_hint == -1 is used by descending sorting, NaNs will be at the end. |
367 | | * |
368 | | * For non Nullable and non floating point types, nan_direction_hint is ignored. |
369 | | */ |
370 | 3 | int compare_at(size_t n, size_t m, const Block& rhs, int nan_direction_hint) const { |
371 | 3 | DCHECK_EQ(columns(), rhs.columns()); |
372 | 3 | return compare_at(n, m, columns(), rhs, nan_direction_hint); |
373 | 3 | } |
374 | | |
375 | | int compare_at(size_t n, size_t m, size_t num_columns, const Block& rhs, |
376 | 7.15M | int nan_direction_hint) const { |
377 | 7.15M | DCHECK_GE(columns(), num_columns); |
378 | 7.15M | DCHECK_GE(rhs.columns(), num_columns); |
379 | | |
380 | 7.15M | DCHECK_LE(n, rows()); |
381 | 7.15M | DCHECK_LE(m, rhs.rows()); |
382 | 9.94M | for (size_t i = 0; i < num_columns; ++i) { |
383 | 7.19M | DCHECK(get_by_position(i).type->equals(*rhs.get_by_position(i).type)); |
384 | 7.19M | auto res = get_by_position(i).column->compare_at(n, m, *(rhs.get_by_position(i).column), |
385 | 7.19M | nan_direction_hint); |
386 | 7.19M | if (res) { |
387 | 4.40M | return res; |
388 | 4.40M | } |
389 | 7.19M | } |
390 | 2.74M | return 0; |
391 | 7.15M | } |
392 | | |
393 | | int compare_at(size_t n, size_t m, const std::vector<uint32_t>* compare_columns, |
394 | 2 | const Block& rhs, int nan_direction_hint) const { |
395 | 2 | DCHECK_GE(columns(), compare_columns->size()); |
396 | 2 | DCHECK_GE(rhs.columns(), compare_columns->size()); |
397 | | |
398 | 2 | DCHECK_LE(n, rows()); |
399 | 2 | DCHECK_LE(m, rhs.rows()); |
400 | 3 | for (auto i : *compare_columns) { |
401 | 3 | DCHECK(get_by_position(i).type->equals(*rhs.get_by_position(i).type)); |
402 | 3 | auto res = get_by_position(i).column->compare_at(n, m, *(rhs.get_by_position(i).column), |
403 | 3 | nan_direction_hint); |
404 | 3 | if (res) { |
405 | 2 | return res; |
406 | 2 | } |
407 | 3 | } |
408 | 0 | return 0; |
409 | 2 | } |
410 | | |
411 | | //note(wb) no DCHECK here, because this method is only used after compare_at now, so no need to repeat check here. |
412 | | // If this method is used in more places, you can add DCHECK case by case. |
413 | | int compare_column_at(size_t n, size_t m, size_t col_idx, const Block& rhs, |
414 | 27.0k | int nan_direction_hint) const { |
415 | 27.0k | auto res = get_by_position(col_idx).column->compare_at( |
416 | 27.0k | n, m, *(rhs.get_by_position(col_idx).column), nan_direction_hint); |
417 | 27.0k | return res; |
418 | 27.0k | } |
419 | | |
420 | | void clear_column_mem_not_keep(const std::vector<bool>& column_keep_flags, |
421 | | bool need_keep_first); |
422 | | |
423 | | // Helper: sum byte_size() of all mutable columns. |
424 | | // Unlike Block::bytes() which operates on immutable ColumnPtr, |
425 | | // this works on MutableColumns during block construction (e.g. in BlockReader). |
426 | 300k | static inline size_t columns_byte_size(const MutableColumns& cols) { |
427 | 300k | size_t total = 0; |
428 | 605k | for (const auto& col : cols) { |
429 | 605k | total += col->byte_size(); |
430 | 605k | } |
431 | 300k | return total; |
432 | 300k | } |
433 | | |
434 | | private: |
435 | | void erase_impl(size_t position); |
436 | | }; |
437 | | |
438 | | using Blocks = std::vector<Block>; |
439 | | using BlocksList = std::list<Block>; |
440 | | using BlocksPtr = std::shared_ptr<Blocks>; |
441 | | using BlocksPtrs = std::shared_ptr<std::vector<BlocksPtr>>; |
442 | | |
443 | | class MutableBlock { |
444 | | ENABLE_FACTORY_CREATOR(MutableBlock); |
445 | | |
446 | | private: |
447 | | MutableColumns _columns; |
448 | | DataTypes _data_types; |
449 | | std::vector<std::string> _names; |
450 | | |
451 | | public: |
452 | | // Build from a consumed Block. This has no restore contract: the source |
453 | | // Block is left without columns and must not be used as a live output block. |
454 | | // For caller-owned live Blocks, use ScopedMutableBlock or |
455 | | // mutate_columns_scoped() instead. |
456 | 48.0k | static MutableBlock build_mutable_block(Block&& block) { |
457 | 48.0k | return MutableBlock(std::move(block)); |
458 | 48.0k | } |
459 | 1 | static MutableBlock build_mutable_block(std::nullptr_t) { return MutableBlock(); } |
460 | | static MutableBlock build_mutable_block(Block* block) = delete; |
461 | 73.0k | MutableBlock() = default; |
462 | 217k | ~MutableBlock() = default; |
463 | | MutableBlock(const MutableBlock&) = delete; |
464 | | MutableBlock& operator=(const MutableBlock&) = delete; |
465 | | MutableBlock(MutableBlock&& m_block) noexcept |
466 | | : _columns(std::move(m_block._columns)), |
467 | | _data_types(std::move(m_block._data_types)), |
468 | 0 | _names(std::move(m_block._names)) {} |
469 | | |
470 | | // Consumes block columns and converts them to mutable columns recursively. |
471 | | // This constructor is for temporary/owned Blocks only. |
472 | | MutableBlock(Block&& block) |
473 | 144k | : _columns(std::move(block).mutate_columns()), |
474 | 144k | _data_types(block.get_data_types()), |
475 | 144k | _names(block.get_names()) {} |
476 | | |
477 | 96.0k | MutableBlock& operator=(MutableBlock&& m_block) noexcept { |
478 | 96.0k | _columns = std::move(m_block._columns); |
479 | 96.0k | _data_types = std::move(m_block._data_types); |
480 | 96.0k | _names = std::move(m_block._names); |
481 | 96.0k | return *this; |
482 | 96.0k | } |
483 | | |
484 | | size_t rows() const; |
485 | 454 | size_t columns() const { return _columns.size(); } |
486 | | |
487 | 144k | bool empty() const { return rows() == 0; } |
488 | | |
489 | 49.4k | MutableColumns& mutable_columns() { return _columns; } |
490 | 0 | const MutableColumns& mutable_columns() const { return _columns; } |
491 | | |
492 | 824 | void set_mutable_columns(MutableColumns&& columns) { _columns = std::move(columns); } |
493 | | |
494 | 823 | DataTypes& data_types() { return _data_types; } |
495 | | |
496 | 218 | MutableColumnPtr& get_column_by_position(size_t position) { return _columns[position]; } |
497 | 70 | const MutableColumnPtr& get_column_by_position(size_t position) const { |
498 | 70 | return _columns[position]; |
499 | 70 | } |
500 | | |
501 | 9 | DataTypePtr& get_datatype_by_position(size_t position) { return _data_types[position]; } |
502 | 26 | const DataTypePtr& get_datatype_by_position(size_t position) const { |
503 | 26 | return _data_types[position]; |
504 | 26 | } |
505 | | |
506 | 44 | int compare_one_column(size_t n, size_t m, size_t column_id, int nan_direction_hint) const { |
507 | 44 | DCHECK_LE(column_id, columns()); |
508 | 44 | DCHECK_LE(n, rows()); |
509 | 44 | DCHECK_LE(m, rows()); |
510 | 44 | auto& column = get_column_by_position(column_id); |
511 | 44 | return column->compare_at(n, m, *column, nan_direction_hint); |
512 | 44 | } |
513 | | |
514 | | int compare_at(size_t n, size_t m, size_t num_columns, const MutableBlock& rhs, |
515 | 6 | int nan_direction_hint) const { |
516 | 6 | DCHECK_GE(columns(), num_columns); |
517 | 6 | DCHECK_GE(rhs.columns(), num_columns); |
518 | | |
519 | 6 | DCHECK_LE(n, rows()); |
520 | 6 | DCHECK_LE(m, rhs.rows()); |
521 | 14 | for (size_t i = 0; i < num_columns; ++i) { |
522 | 11 | DCHECK(get_datatype_by_position(i)->equals(*rhs.get_datatype_by_position(i))); |
523 | 11 | auto res = get_column_by_position(i)->compare_at(n, m, *(rhs.get_column_by_position(i)), |
524 | 11 | nan_direction_hint); |
525 | 11 | if (res) { |
526 | 3 | return res; |
527 | 3 | } |
528 | 11 | } |
529 | 3 | return 0; |
530 | 6 | } |
531 | | |
532 | | int compare_at(size_t n, size_t m, const std::vector<uint32_t>* compare_columns, |
533 | 0 | const MutableBlock& rhs, int nan_direction_hint) const { |
534 | 0 | DCHECK_GE(columns(), compare_columns->size()); |
535 | 0 | DCHECK_GE(rhs.columns(), compare_columns->size()); |
536 | |
|
537 | 0 | DCHECK_LE(n, rows()); |
538 | 0 | DCHECK_LE(m, rhs.rows()); |
539 | 0 | for (auto i : *compare_columns) { |
540 | 0 | DCHECK(get_datatype_by_position(i)->equals(*rhs.get_datatype_by_position(i))); |
541 | 0 | auto res = get_column_by_position(i)->compare_at(n, m, *(rhs.get_column_by_position(i)), |
542 | 0 | nan_direction_hint); |
543 | 0 | if (res) { |
544 | 0 | return res; |
545 | 0 | } |
546 | 0 | } |
547 | 0 | return 0; |
548 | 0 | } |
549 | | |
550 | 4 | std::string dump_types() const { |
551 | 4 | std::string res; |
552 | 10 | for (auto type : _data_types) { |
553 | 10 | if (!res.empty()) { |
554 | 6 | res += ", "; |
555 | 6 | } |
556 | 10 | res += type->get_name(); |
557 | 10 | } |
558 | 4 | return res; |
559 | 4 | } |
560 | | |
561 | | template <typename T> |
562 | 153 | [[nodiscard]] Status merge(T&& block) { |
563 | 153 | RETURN_IF_CATCH_EXCEPTION(return merge_impl(block);); |
564 | 153 | } _ZN5doris12MutableBlock5mergeIRNS_5BlockEEENS_6StatusEOT_ Line | Count | Source | 562 | 75 | [[nodiscard]] Status merge(T&& block) { | 563 | 75 | RETURN_IF_CATCH_EXCEPTION(return merge_impl(block);); | 564 | 75 | } |
_ZN5doris12MutableBlock5mergeINS_5BlockEEENS_6StatusEOT_ Line | Count | Source | 562 | 78 | [[nodiscard]] Status merge(T&& block) { | 563 | 78 | RETURN_IF_CATCH_EXCEPTION(return merge_impl(block);); | 564 | 78 | } |
|
565 | | |
566 | | template <typename T> |
567 | 48.0k | [[nodiscard]] Status merge_ignore_overflow(T&& block) { |
568 | 48.0k | RETURN_IF_CATCH_EXCEPTION(return merge_impl_ignore_overflow(block);); |
569 | 48.0k | } _ZN5doris12MutableBlock21merge_ignore_overflowIRNS_5BlockEEENS_6StatusEOT_ Line | Count | Source | 567 | 1 | [[nodiscard]] Status merge_ignore_overflow(T&& block) { | 568 | 1 | RETURN_IF_CATCH_EXCEPTION(return merge_impl_ignore_overflow(block);); | 569 | 1 | } |
_ZN5doris12MutableBlock21merge_ignore_overflowINS_5BlockEEENS_6StatusEOT_ Line | Count | Source | 567 | 48.0k | [[nodiscard]] Status merge_ignore_overflow(T&& block) { | 568 | 48.0k | RETURN_IF_CATCH_EXCEPTION(return merge_impl_ignore_overflow(block);); | 569 | 48.0k | } |
|
570 | | |
571 | | // only use for join. call ignore_overflow to prevent from throw exception in join |
572 | | template <typename T> |
573 | 48.0k | [[nodiscard]] Status merge_impl_ignore_overflow(T&& block) { |
574 | 48.0k | if (_columns.size() != block.columns()) { |
575 | 1 | return Status::Error<ErrorCode::INTERNAL_ERROR>( |
576 | 1 | "Merge block not match, self column count: {}, [columns: {}, types: {}], " |
577 | 1 | "input column count: {}, [columns: {}, " |
578 | 1 | "types: {}], ", |
579 | 1 | _columns.size(), dump_names(), dump_types(), block.columns(), |
580 | 1 | block.dump_names(), block.dump_types()); |
581 | 1 | } |
582 | 142k | for (int i = 0; i < _columns.size(); ++i) { |
583 | 94.2k | if (!_data_types[i]->equals(*block.get_by_position(i).type)) { |
584 | 1 | throw doris::Exception(doris::ErrorCode::FATAL_ERROR, |
585 | 1 | "Merge block not match, self:[columns: {}, types: {}], " |
586 | 1 | "input:[columns: {}, types: {}], ", |
587 | 1 | dump_names(), dump_types(), block.dump_names(), |
588 | 1 | block.dump_types()); |
589 | 1 | } |
590 | 94.2k | _columns[i]->insert_range_from_ignore_overflow( |
591 | 94.2k | *block.get_by_position(i).column->convert_to_full_column_if_const().get(), 0, |
592 | 94.2k | block.rows()); |
593 | 94.2k | } |
594 | 48.0k | return Status::OK(); |
595 | 48.0k | } _ZN5doris12MutableBlock26merge_impl_ignore_overflowIRNS_5BlockEEENS_6StatusEOT_ Line | Count | Source | 573 | 48.0k | [[nodiscard]] Status merge_impl_ignore_overflow(T&& block) { | 574 | 48.0k | if (_columns.size() != block.columns()) { | 575 | 1 | return Status::Error<ErrorCode::INTERNAL_ERROR>( | 576 | 1 | "Merge block not match, self column count: {}, [columns: {}, types: {}], " | 577 | 1 | "input column count: {}, [columns: {}, " | 578 | 1 | "types: {}], ", | 579 | 1 | _columns.size(), dump_names(), dump_types(), block.columns(), | 580 | 1 | block.dump_names(), block.dump_types()); | 581 | 1 | } | 582 | 142k | for (int i = 0; i < _columns.size(); ++i) { | 583 | 94.2k | if (!_data_types[i]->equals(*block.get_by_position(i).type)) { | 584 | 0 | throw doris::Exception(doris::ErrorCode::FATAL_ERROR, | 585 | 0 | "Merge block not match, self:[columns: {}, types: {}], " | 586 | 0 | "input:[columns: {}, types: {}], ", | 587 | 0 | dump_names(), dump_types(), block.dump_names(), | 588 | 0 | block.dump_types()); | 589 | 0 | } | 590 | 94.2k | _columns[i]->insert_range_from_ignore_overflow( | 591 | 94.2k | *block.get_by_position(i).column->convert_to_full_column_if_const().get(), 0, | 592 | 94.2k | block.rows()); | 593 | 94.2k | } | 594 | 48.0k | return Status::OK(); | 595 | 48.0k | } |
_ZN5doris12MutableBlock26merge_impl_ignore_overflowINS_5BlockEEENS_6StatusEOT_ Line | Count | Source | 573 | 1 | [[nodiscard]] Status merge_impl_ignore_overflow(T&& block) { | 574 | 1 | if (_columns.size() != block.columns()) { | 575 | 0 | return Status::Error<ErrorCode::INTERNAL_ERROR>( | 576 | 0 | "Merge block not match, self column count: {}, [columns: {}, types: {}], " | 577 | 0 | "input column count: {}, [columns: {}, " | 578 | 0 | "types: {}], ", | 579 | 0 | _columns.size(), dump_names(), dump_types(), block.columns(), | 580 | 0 | block.dump_names(), block.dump_types()); | 581 | 0 | } | 582 | 3 | for (int i = 0; i < _columns.size(); ++i) { | 583 | 3 | if (!_data_types[i]->equals(*block.get_by_position(i).type)) { | 584 | 1 | throw doris::Exception(doris::ErrorCode::FATAL_ERROR, | 585 | 1 | "Merge block not match, self:[columns: {}, types: {}], " | 586 | 1 | "input:[columns: {}, types: {}], ", | 587 | 1 | dump_names(), dump_types(), block.dump_names(), | 588 | 1 | block.dump_types()); | 589 | 1 | } | 590 | 2 | _columns[i]->insert_range_from_ignore_overflow( | 591 | 2 | *block.get_by_position(i).column->convert_to_full_column_if_const().get(), 0, | 592 | 2 | block.rows()); | 593 | 2 | } | 594 | 0 | return Status::OK(); | 595 | 1 | } |
|
596 | | |
597 | | template <typename T> |
598 | 155 | [[nodiscard]] Status merge_impl(T&& block) { |
599 | | // merge is not supported in dynamic block |
600 | 155 | if (_columns.empty() && _data_types.empty()) { |
601 | 40 | _data_types = block.get_data_types(); |
602 | 40 | _names = block.get_names(); |
603 | 40 | _columns.resize(block.columns()); |
604 | 139 | for (size_t i = 0; i < block.columns(); ++i) { |
605 | 99 | if (block.get_by_position(i).column) { |
606 | 98 | _columns[i] = (*std::move(block.get_by_position(i) |
607 | 98 | .column->convert_to_full_column_if_const())) |
608 | 98 | .mutate(); |
609 | 98 | } else { |
610 | 1 | _columns[i] = _data_types[i]->create_column(); |
611 | 1 | } |
612 | 99 | } |
613 | 115 | } else { |
614 | 115 | if (_columns.size() != block.columns()) { |
615 | 2 | return Status::Error<ErrorCode::INTERNAL_ERROR>( |
616 | 2 | "Merge block not match, self column count: {}, [columns: {}, types: {}], " |
617 | 2 | "input column count: {}, [columns: {}, " |
618 | 2 | "types: {}], ", |
619 | 2 | _columns.size(), dump_names(), dump_types(), block.columns(), |
620 | 2 | block.dump_names(), block.dump_types()); |
621 | 2 | } |
622 | 286 | for (int i = 0; i < _columns.size(); ++i) { |
623 | 173 | if (!_data_types[i]->equals(*block.get_by_position(i).type)) { |
624 | 1 | DCHECK(_data_types[i]->is_nullable()) |
625 | 0 | << " target type: " << _data_types[i]->get_name() |
626 | 0 | << " src type: " << block.get_by_position(i).type->get_name(); |
627 | 1 | DCHECK(((DataTypeNullable*)_data_types[i].get()) |
628 | 1 | ->get_nested_type() |
629 | 1 | ->equals(*block.get_by_position(i).type)); |
630 | 1 | DCHECK(!block.get_by_position(i).type->is_nullable()); |
631 | 1 | _columns[i]->insert_range_from(*make_nullable(block.get_by_position(i).column) |
632 | 1 | ->convert_to_full_column_if_const(), |
633 | 1 | 0, block.rows()); |
634 | 172 | } else { |
635 | 172 | _columns[i]->insert_range_from( |
636 | 172 | *block.get_by_position(i) |
637 | 172 | .column->convert_to_full_column_if_const() |
638 | 172 | .get(), |
639 | 172 | 0, block.rows()); |
640 | 172 | } |
641 | 173 | } |
642 | 113 | } |
643 | 153 | return Status::OK(); |
644 | 155 | } _ZN5doris12MutableBlock10merge_implIRNS_5BlockEEENS_6StatusEOT_ Line | Count | Source | 598 | 153 | [[nodiscard]] Status merge_impl(T&& block) { | 599 | | // merge is not supported in dynamic block | 600 | 153 | if (_columns.empty() && _data_types.empty()) { | 601 | 40 | _data_types = block.get_data_types(); | 602 | 40 | _names = block.get_names(); | 603 | 40 | _columns.resize(block.columns()); | 604 | 139 | for (size_t i = 0; i < block.columns(); ++i) { | 605 | 99 | if (block.get_by_position(i).column) { | 606 | 98 | _columns[i] = (*std::move(block.get_by_position(i) | 607 | 98 | .column->convert_to_full_column_if_const())) | 608 | 98 | .mutate(); | 609 | 98 | } else { | 610 | 1 | _columns[i] = _data_types[i]->create_column(); | 611 | 1 | } | 612 | 99 | } | 613 | 113 | } else { | 614 | 113 | if (_columns.size() != block.columns()) { | 615 | 1 | return Status::Error<ErrorCode::INTERNAL_ERROR>( | 616 | 1 | "Merge block not match, self column count: {}, [columns: {}, types: {}], " | 617 | 1 | "input column count: {}, [columns: {}, " | 618 | 1 | "types: {}], ", | 619 | 1 | _columns.size(), dump_names(), dump_types(), block.columns(), | 620 | 1 | block.dump_names(), block.dump_types()); | 621 | 1 | } | 622 | 282 | for (int i = 0; i < _columns.size(); ++i) { | 623 | 170 | if (!_data_types[i]->equals(*block.get_by_position(i).type)) { | 624 | 0 | DCHECK(_data_types[i]->is_nullable()) | 625 | 0 | << " target type: " << _data_types[i]->get_name() | 626 | 0 | << " src type: " << block.get_by_position(i).type->get_name(); | 627 | 0 | DCHECK(((DataTypeNullable*)_data_types[i].get()) | 628 | 0 | ->get_nested_type() | 629 | 0 | ->equals(*block.get_by_position(i).type)); | 630 | 0 | DCHECK(!block.get_by_position(i).type->is_nullable()); | 631 | 0 | _columns[i]->insert_range_from(*make_nullable(block.get_by_position(i).column) | 632 | 0 | ->convert_to_full_column_if_const(), | 633 | 0 | 0, block.rows()); | 634 | 170 | } else { | 635 | 170 | _columns[i]->insert_range_from( | 636 | 170 | *block.get_by_position(i) | 637 | 170 | .column->convert_to_full_column_if_const() | 638 | 170 | .get(), | 639 | 170 | 0, block.rows()); | 640 | 170 | } | 641 | 170 | } | 642 | 112 | } | 643 | 152 | return Status::OK(); | 644 | 153 | } |
_ZN5doris12MutableBlock10merge_implINS_5BlockEEENS_6StatusEOT_ Line | Count | Source | 598 | 2 | [[nodiscard]] Status merge_impl(T&& block) { | 599 | | // merge is not supported in dynamic block | 600 | 2 | if (_columns.empty() && _data_types.empty()) { | 601 | 0 | _data_types = block.get_data_types(); | 602 | 0 | _names = block.get_names(); | 603 | 0 | _columns.resize(block.columns()); | 604 | 0 | for (size_t i = 0; i < block.columns(); ++i) { | 605 | 0 | if (block.get_by_position(i).column) { | 606 | 0 | _columns[i] = (*std::move(block.get_by_position(i) | 607 | 0 | .column->convert_to_full_column_if_const())) | 608 | 0 | .mutate(); | 609 | 0 | } else { | 610 | 0 | _columns[i] = _data_types[i]->create_column(); | 611 | 0 | } | 612 | 0 | } | 613 | 2 | } else { | 614 | 2 | if (_columns.size() != block.columns()) { | 615 | 1 | return Status::Error<ErrorCode::INTERNAL_ERROR>( | 616 | 1 | "Merge block not match, self column count: {}, [columns: {}, types: {}], " | 617 | 1 | "input column count: {}, [columns: {}, " | 618 | 1 | "types: {}], ", | 619 | 1 | _columns.size(), dump_names(), dump_types(), block.columns(), | 620 | 1 | block.dump_names(), block.dump_types()); | 621 | 1 | } | 622 | 4 | for (int i = 0; i < _columns.size(); ++i) { | 623 | 3 | if (!_data_types[i]->equals(*block.get_by_position(i).type)) { | 624 | 1 | DCHECK(_data_types[i]->is_nullable()) | 625 | 0 | << " target type: " << _data_types[i]->get_name() | 626 | 0 | << " src type: " << block.get_by_position(i).type->get_name(); | 627 | 1 | DCHECK(((DataTypeNullable*)_data_types[i].get()) | 628 | 1 | ->get_nested_type() | 629 | 1 | ->equals(*block.get_by_position(i).type)); | 630 | 1 | DCHECK(!block.get_by_position(i).type->is_nullable()); | 631 | 1 | _columns[i]->insert_range_from(*make_nullable(block.get_by_position(i).column) | 632 | 1 | ->convert_to_full_column_if_const(), | 633 | 1 | 0, block.rows()); | 634 | 2 | } else { | 635 | 2 | _columns[i]->insert_range_from( | 636 | 2 | *block.get_by_position(i) | 637 | 2 | .column->convert_to_full_column_if_const() | 638 | 2 | .get(), | 639 | 2 | 0, block.rows()); | 640 | 2 | } | 641 | 3 | } | 642 | 1 | } | 643 | 1 | return Status::OK(); | 644 | 2 | } |
|
645 | | |
646 | | // move to columns' data to a Block. this will invalidate |
647 | | Block to_block(int start_column = 0); |
648 | | Block to_block(int start_column, int end_column); |
649 | | |
650 | | void swap(MutableBlock& other) noexcept; |
651 | | |
652 | | void add_row(const Block* block, int row); |
653 | | // Batch add row should return error status if allocate memory failed. |
654 | | Status add_rows(const Block* block, const uint32_t* row_begin, const uint32_t* row_end, |
655 | | const std::vector<int>* column_offset = nullptr); |
656 | | Status add_rows(const Block* block, size_t row_begin, size_t length); |
657 | | |
658 | | std::string dump_data(size_t row_limit = 100) const; |
659 | | std::string dump_data_json(size_t row_limit = 100) const; |
660 | | |
661 | 66 | void clear() { |
662 | 66 | _columns.clear(); |
663 | 66 | _data_types.clear(); |
664 | 66 | _names.clear(); |
665 | 66 | } |
666 | | |
667 | | // Clear owned mutable columns in place. MutableBlock already owns its |
668 | | // columns exclusively, so this does not perform COW detaching or cloning. |
669 | | void clear_column_data() noexcept; |
670 | | |
671 | | size_t allocated_bytes() const; |
672 | | |
673 | 48.0k | size_t bytes() const { |
674 | 48.0k | size_t res = 0; |
675 | 94.0k | for (const auto& elem : _columns) { |
676 | 94.0k | res += elem->byte_size(); |
677 | 94.0k | } |
678 | | |
679 | 48.0k | return res; |
680 | 48.0k | } |
681 | | |
682 | 823 | std::vector<std::string>& get_names() { return _names; } |
683 | | |
684 | | /** Get a list of column names separated by commas. */ |
685 | | std::string dump_names() const; |
686 | | }; |
687 | | |
688 | | // RAII adapter for code that wants the MutableBlock API over a live Block. It |
689 | | // owns only the temporary mutable columns and restores them to the Block on |
690 | | // destruction. While the adapter is alive, read/write column data through |
691 | | // mutable_block()/mutable_columns(); the Block's column slots are moved out. |
692 | | class ScopedMutableBlock { |
693 | | public: |
694 | | ScopedMutableBlock() = delete; |
695 | | explicit ScopedMutableBlock(Block* block); |
696 | 823 | ~ScopedMutableBlock() { restore(); } |
697 | | |
698 | | ScopedMutableBlock(const ScopedMutableBlock&) = delete; |
699 | | ScopedMutableBlock& operator=(const ScopedMutableBlock&) = delete; |
700 | | |
701 | | ScopedMutableBlock(ScopedMutableBlock&& other) noexcept |
702 | | : _block(std::exchange(other._block, nullptr)), |
703 | 0 | _mutable_block(std::move(other._mutable_block)) {} |
704 | | |
705 | 0 | ScopedMutableBlock& operator=(ScopedMutableBlock&& other) noexcept { |
706 | 0 | if (this != &other) { |
707 | 0 | restore(); |
708 | 0 | _block = std::exchange(other._block, nullptr); |
709 | 0 | _mutable_block = std::move(other._mutable_block); |
710 | 0 | } |
711 | 0 | return *this; |
712 | 0 | } |
713 | | |
714 | 822 | MutableBlock& mutable_block() { return _mutable_block; } |
715 | 0 | const MutableBlock& mutable_block() const { return _mutable_block; } |
716 | 1 | MutableColumns& mutable_columns() { return _mutable_block.mutable_columns(); } |
717 | 0 | const MutableColumns& mutable_columns() const { return _mutable_block.mutable_columns(); } |
718 | | |
719 | 950 | void restore() { |
720 | 950 | if (_block != nullptr) { |
721 | 823 | _block->set_columns(std::move(_mutable_block.mutable_columns())); |
722 | 823 | _block = nullptr; |
723 | 823 | } |
724 | 950 | } |
725 | | |
726 | | private: |
727 | | Block* _block = nullptr; |
728 | | MutableBlock _mutable_block; |
729 | | }; |
730 | | |
731 | | struct IteratorRowRef { |
732 | | std::shared_ptr<Block> block; |
733 | | int row_pos; |
734 | | bool is_same; |
735 | | |
736 | | template <typename T> |
737 | 1.06M | int compare(const IteratorRowRef& rhs, const T& compare_arguments) const { |
738 | 1.06M | return block->compare_at(row_pos, rhs.row_pos, compare_arguments, *rhs.block, -1); |
739 | 1.06M | } Unexecuted instantiation: _ZNK5doris14IteratorRowRef7compareIPKSt6vectorIjSaIjEEEEiRKS0_RKT_ _ZNK5doris14IteratorRowRef7compareImEEiRKS0_RKT_ Line | Count | Source | 737 | 1.06M | int compare(const IteratorRowRef& rhs, const T& compare_arguments) const { | 738 | 1.06M | return block->compare_at(row_pos, rhs.row_pos, compare_arguments, *rhs.block, -1); | 739 | 1.06M | } |
|
740 | | |
741 | 512 | void reset() { |
742 | 512 | block = nullptr; |
743 | 512 | row_pos = -1; |
744 | 512 | is_same = false; |
745 | 512 | } |
746 | | }; |
747 | | |
748 | | using BlockView = std::vector<IteratorRowRef>; |
749 | | using BlockUPtr = std::unique_ptr<Block>; |
750 | | |
751 | | } // namespace doris |