be/src/format/table/equality_delete.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 "format/table/equality_delete.h" |
19 | | |
20 | | #include "core/column/column_const.h" |
21 | | #include "core/column/column_nullable.h" |
22 | | #include "exprs/create_predicate_function.h" |
23 | | #include "util/hash_util.hpp" |
24 | | |
25 | | namespace doris { |
26 | | |
27 | | namespace { |
28 | | |
29 | | // Iceberg BINARY / FIXED keys are read as VARBINARY and string keys as STRING, and the delete file and |
30 | | // the data file may carry the same key in different ones of these classes (a delete block holding |
31 | | // STRING bytes for a VARBINARY data column, say). Keys of these types therefore match by their bytes. |
32 | 208 | bool is_equality_delete_byte_type(const DataTypePtr& type) { |
33 | 208 | const auto primitive_type = remove_nullable(type)->get_primitive_type(); |
34 | 208 | return is_string_type(primitive_type) || is_varbinary(primitive_type); |
35 | 208 | } |
36 | | |
37 | | // The rows of a byte-key column. A const column repeats its single value, which can be NULL: when the FE does |
38 | | // not send Iceberg scan semantics v2, a key column that the data file lacks is filled from |
39 | | // Const(Nullable(String) NULL), and its rows must match a NULL delete key rather than an empty string. |
40 | | class ByteKeyColumn { |
41 | | public: |
42 | 18 | explicit ByteKeyColumn(const IColumn& column) : _values(&column) { |
43 | 18 | if (const auto* const_column = check_and_get_column<ColumnConst>(_values)) { |
44 | 2 | _values = &const_column->get_data_column(); |
45 | 2 | _is_const = true; |
46 | 2 | } |
47 | 18 | if (const auto* nullable = check_and_get_column<ColumnNullable>(_values)) { |
48 | 10 | _null_map = &nullable->get_null_map_data(); |
49 | 10 | _values = &nullable->get_nested_column(); |
50 | 10 | } |
51 | 18 | } |
52 | | |
53 | 37 | bool is_null(size_t row) const { |
54 | 37 | return _null_map != nullptr && (*_null_map)[_is_const ? 0 : row] != 0; |
55 | 37 | } |
56 | | |
57 | 24 | StringRef value(size_t row) const { return _values->get_data_at(_is_const ? 0 : row); } |
58 | | |
59 | | private: |
60 | | const IColumn* _values; |
61 | | const NullMap* _null_map = nullptr; |
62 | | bool _is_const = false; |
63 | | }; |
64 | | |
65 | 3 | void insert_byte_values(const IColumn& column, HybridSetBase* byte_set) { |
66 | 3 | const ByteKeyColumn keys(column); |
67 | 9 | for (size_t row = 0; row < column.size(); ++row) { |
68 | 6 | if (keys.is_null(row)) { |
69 | 3 | byte_set->insert(static_cast<const void*>(nullptr)); |
70 | 3 | continue; |
71 | 3 | } |
72 | 3 | const StringRef value = keys.value(row); |
73 | 3 | byte_set->insert(&value); |
74 | 3 | } |
75 | 3 | } |
76 | | |
77 | | void find_byte_values(const IColumn& column, const HybridSetBase& byte_set, |
78 | 3 | IColumn::Filter* matches) { |
79 | 3 | const ByteKeyColumn keys(column); |
80 | 12 | for (size_t row = 0; row < column.size(); ++row) { |
81 | 9 | if (keys.is_null(row)) { |
82 | 3 | (*matches)[row] = byte_set.contain_null(); |
83 | 3 | continue; |
84 | 3 | } |
85 | 6 | const StringRef value = keys.value(row); |
86 | 6 | (*matches)[row] = byte_set.find(&value); |
87 | 6 | } |
88 | 3 | } |
89 | | |
90 | | void update_equality_delete_hashes(const ColumnWithTypeAndName& column_and_type, |
91 | 58 | std::vector<uint64_t>* hashes) { |
92 | 58 | if (!is_equality_delete_byte_type(column_and_type.type)) { |
93 | 52 | column_and_type.column->update_hashes_with_value(hashes->data(), nullptr); |
94 | 52 | return; |
95 | 52 | } |
96 | | // Hash the bytes rather than call update_hashes_with_value: VARBINARY does not implement it, and a |
97 | | // STRING delete key must land on the same hash as the equal VARBINARY data key. |
98 | 6 | const ByteKeyColumn keys(*column_and_type.column); |
99 | 21 | for (size_t row = 0; row < column_and_type.column->size(); ++row) { |
100 | 15 | if (keys.is_null(row)) { |
101 | 4 | (*hashes)[row] = HashUtil::xxHash64NullWithSeed((*hashes)[row]); |
102 | 4 | continue; |
103 | 4 | } |
104 | 11 | const StringRef value = keys.value(row); |
105 | 11 | (*hashes)[row] = HashUtil::xxHash64WithSeed(value.data, value.size, (*hashes)[row]); |
106 | 11 | } |
107 | 6 | } |
108 | | |
109 | | bool equality_delete_values_equal(const ColumnWithTypeAndName& data_column, size_t data_row, |
110 | 30 | const ColumnWithTypeAndName& delete_column, size_t delete_row) { |
111 | 30 | if (!is_equality_delete_byte_type(data_column.type) || |
112 | 30 | !is_equality_delete_byte_type(delete_column.type)) { |
113 | 27 | return data_column.column->compare_at(data_row, delete_row, *delete_column.column, -1) == 0; |
114 | 27 | } |
115 | 3 | const ByteKeyColumn data_keys(*data_column.column); |
116 | 3 | const ByteKeyColumn delete_keys(*delete_column.column); |
117 | 3 | if (data_keys.is_null(data_row) || delete_keys.is_null(delete_row)) { |
118 | 1 | return data_keys.is_null(data_row) && delete_keys.is_null(delete_row); |
119 | 1 | } |
120 | 2 | return data_keys.value(data_row) == delete_keys.value(delete_row); |
121 | 3 | } |
122 | | |
123 | | } // namespace |
124 | | |
125 | | std::unique_ptr<EqualityDeleteBase> EqualityDeleteBase::get_delete_impl( |
126 | 52 | const Block* delete_block, const std::vector<int>& delete_col_ids) { |
127 | 52 | DCHECK_EQ(delete_block->columns(), delete_col_ids.size()); |
128 | 52 | if (delete_block->columns() == 1) { |
129 | 41 | return std::make_unique<SimpleEqualityDelete>(delete_block, delete_col_ids); |
130 | 41 | } else { |
131 | 11 | return std::make_unique<MultiEqualityDelete>(delete_block, delete_col_ids); |
132 | 11 | } |
133 | 52 | } |
134 | | |
135 | 41 | Status SimpleEqualityDelete::_build_set() { |
136 | 41 | COUNTER_UPDATE(num_delete_rows, _delete_block->rows()); |
137 | 41 | if (_delete_block->columns() != 1) [[unlikely]] { |
138 | 0 | return Status::InternalError("Simple equality delete can be only applied with one column"); |
139 | 0 | } |
140 | 41 | const auto& column_and_type = _delete_block->get_by_position(0); |
141 | 41 | auto delete_column_type = remove_nullable(column_and_type.type)->get_primitive_type(); |
142 | 41 | size_t non_null_rows = _delete_block->rows(); |
143 | 41 | if (const auto* nullable = check_and_get_column<ColumnNullable>(column_and_type.column.get()); |
144 | 41 | nullable != nullptr) { |
145 | 41 | non_null_rows = std::ranges::count(nullable->get_null_map_data(), UInt8(0)); |
146 | 41 | } |
147 | 41 | if (is_equality_delete_byte_type(column_and_type.type)) { |
148 | | // create_set has no VARBINARY set (it throws NOT_IMPLEMENTED), so every byte key goes into a |
149 | | // STRING set as bytes, whichever of STRING and VARBINARY the delete file was read as. |
150 | 3 | _hybrid_set.reset(create_set(TYPE_STRING, non_null_rows, true)); |
151 | 3 | insert_byte_values(*column_and_type.column, _hybrid_set.get()); |
152 | 38 | } else { |
153 | 38 | _hybrid_set.reset(create_set(delete_column_type, non_null_rows, true)); |
154 | 38 | _hybrid_set->insert_fixed_len(column_and_type.column, 0); |
155 | 38 | } |
156 | 41 | return Status::OK(); |
157 | 41 | } |
158 | | |
159 | | Status SimpleEqualityDelete::filter_data_block( |
160 | | Block* data_block, const std::unordered_map<std::string, uint32_t>* col_name_to_block_idx, |
161 | | const std::unordered_map<int, std::string>& id_to_block_column_name, |
162 | 41 | IColumn::Filter& filter) { |
163 | 41 | SCOPED_TIMER(equality_delete_time); |
164 | 41 | DCHECK(_delete_col_ids.size() == 1); |
165 | 41 | auto column_field_id = _delete_col_ids[0]; |
166 | | |
167 | 41 | const auto& block_col_name = id_to_block_column_name.at(column_field_id); |
168 | 41 | auto block_idx = col_name_to_block_idx->at(block_col_name); |
169 | | |
170 | 41 | auto column_and_type = data_block->get_by_position(block_idx); |
171 | 41 | const bool delete_is_byte = |
172 | 41 | is_equality_delete_byte_type(_delete_block->get_by_position(0).type); |
173 | 41 | const bool byte_compatible = |
174 | 41 | delete_is_byte && is_equality_delete_byte_type(column_and_type.type); |
175 | | // The byte set can only be probed with bytes: an INT data column would be read as StringRef values. |
176 | 41 | if (delete_is_byte && !byte_compatible) [[unlikely]] { |
177 | 0 | return Status::InternalError( |
178 | 0 | "Not support type change in column '{}', src type: {}, target type: {}", |
179 | 0 | column_and_type.name, _delete_block->get_by_position(0).type->get_name(), |
180 | 0 | column_and_type.type->get_name()); |
181 | 0 | } |
182 | | |
183 | 41 | size_t rows = data_block->rows(); |
184 | | // _filter: 1 => in _hybrid_set; 0 => not in _hybrid_set |
185 | 41 | if (_single_filter == nullptr) { |
186 | 41 | _single_filter = std::make_unique<IColumn::Filter>(rows, 0); |
187 | 41 | } else { |
188 | | // reset the array capacity and fill all elements using the 0 |
189 | 0 | _single_filter->assign(rows, UInt8(0)); |
190 | 0 | } |
191 | 41 | auto column = column_and_type.column->convert_to_full_column_if_const(); |
192 | 41 | if (byte_compatible) { |
193 | 3 | find_byte_values(*column, *_hybrid_set, _single_filter.get()); |
194 | 38 | } else if (const auto* nullable = check_and_get_column<ColumnNullable>(column.get())) { |
195 | 38 | const NullMap& null_map = nullable->get_null_map_data(); |
196 | 38 | _hybrid_set->find_batch_nullable(*remove_nullable(column), rows, null_map, *_single_filter); |
197 | 38 | if (_hybrid_set->contain_null()) { |
198 | 4 | auto* filter_data = _single_filter->data(); |
199 | 16 | for (size_t i = 0; i < rows; ++i) { |
200 | 12 | filter_data[i] = filter_data[i] || null_map[i]; |
201 | 12 | } |
202 | 4 | } |
203 | 38 | } else { |
204 | 0 | _hybrid_set->find_batch(*column, rows, *_single_filter); |
205 | 0 | } |
206 | | // should reverse _filter |
207 | 41 | auto* filter_data = filter.data(); |
208 | 162 | for (size_t i = 0; i < rows; ++i) { |
209 | 121 | filter_data[i] &= !_single_filter->data()[i]; |
210 | 121 | } |
211 | 41 | return Status::OK(); |
212 | 41 | } |
213 | | |
214 | 13 | Status MultiEqualityDelete::_build_set() { |
215 | 13 | COUNTER_UPDATE(num_delete_rows, _delete_block->rows()); |
216 | 13 | size_t rows = _delete_block->rows(); |
217 | 13 | _delete_hashes.clear(); |
218 | 13 | _delete_hashes.resize(rows, 0); |
219 | 29 | for (const auto& column : _delete_block->get_columns_with_type_and_name()) { |
220 | 29 | update_equality_delete_hashes(column, &_delete_hashes); |
221 | 29 | } |
222 | 29 | for (size_t i = 0; i < rows; ++i) { |
223 | 16 | _delete_hash_map.insert({_delete_hashes[i], i}); |
224 | 16 | } |
225 | 13 | _data_column_index.resize(_delete_block->columns()); |
226 | 13 | return Status::OK(); |
227 | 13 | } |
228 | | |
229 | | Status MultiEqualityDelete::filter_data_block( |
230 | | Block* data_block, const std::unordered_map<std::string, uint32_t>* col_name_to_block_idx, |
231 | | const std::unordered_map<int, std::string>& id_to_block_column_name, |
232 | 13 | IColumn::Filter& filter) { |
233 | 13 | SCOPED_TIMER(equality_delete_time); |
234 | 13 | DCHECK_EQ(_delete_block->get_columns_with_type_and_name().size(), _delete_col_ids.size()); |
235 | 13 | size_t column_index = 0; |
236 | | |
237 | 42 | for (size_t idx = 0; idx < _delete_block->get_columns_with_type_and_name().size(); ++idx) { |
238 | 29 | auto delete_col = _delete_block->get_columns_with_type_and_name()[idx]; |
239 | 29 | auto delete_col_id = _delete_col_ids[idx]; |
240 | | |
241 | 29 | DCHECK(id_to_block_column_name.contains(delete_col_id)); |
242 | 29 | const auto& block_column_name = id_to_block_column_name.at(delete_col_id); |
243 | 29 | if (!col_name_to_block_idx->contains(block_column_name)) [[unlikely]] { |
244 | 0 | return Status::InternalError("Column '{}' not found in data block: {}", |
245 | 0 | block_column_name, data_block->dump_structure()); |
246 | 0 | } |
247 | 29 | auto column_and_type = |
248 | 29 | data_block->safe_get_by_position(col_name_to_block_idx->at(block_column_name)); |
249 | | // A STRING delete key may meet a VARBINARY data column for the same Iceberg BINARY field: both are |
250 | | // hashed and compared as bytes, so only a change to or from a non-byte type is rejected. |
251 | 29 | const bool byte_compatible = is_equality_delete_byte_type(delete_col.type) && |
252 | 29 | is_equality_delete_byte_type(column_and_type.type); |
253 | 29 | if (!delete_col.type->equals(*column_and_type.type) && !byte_compatible) [[unlikely]] { |
254 | 0 | return Status::InternalError( |
255 | 0 | "Not support type change in column '{}', src type: {}, target type: {}", |
256 | 0 | block_column_name, delete_col.type->get_name(), |
257 | 0 | column_and_type.type->get_name()); |
258 | 0 | } |
259 | 29 | _data_column_index[column_index++] = col_name_to_block_idx->at(block_column_name); |
260 | 29 | } |
261 | 13 | size_t rows = data_block->rows(); |
262 | 13 | _data_hashes.clear(); |
263 | 13 | _data_hashes.resize(rows, 0); |
264 | 29 | for (size_t index : _data_column_index) { |
265 | 29 | update_equality_delete_hashes(data_block->get_by_position(index), &_data_hashes); |
266 | 29 | } |
267 | 13 | auto* filter_data = filter.data(); |
268 | 47 | for (size_t i = 0; i < rows; ++i) { |
269 | 34 | for (auto beg = _delete_hash_map.lower_bound(_data_hashes[i]), |
270 | 34 | end = _delete_hash_map.upper_bound(_data_hashes[i]); |
271 | 34 | beg != end; ++beg) { |
272 | 14 | if (filter[i] && _equal(data_block, i, beg->second)) { |
273 | 14 | filter_data[i] = 0; |
274 | 14 | break; |
275 | 14 | } |
276 | 14 | } |
277 | 34 | } |
278 | | |
279 | 13 | return Status::OK(); |
280 | 13 | } |
281 | | |
282 | | bool MultiEqualityDelete::_equal(Block* data_block, size_t data_row_index, |
283 | 14 | size_t delete_row_index) { |
284 | 44 | for (size_t i = 0; i < _delete_block->columns(); ++i) { |
285 | 30 | if (!equality_delete_values_equal(data_block->get_by_position(_data_column_index[i]), |
286 | 30 | data_row_index, _delete_block->get_by_position(i), |
287 | 30 | delete_row_index)) { |
288 | 0 | return false; |
289 | 0 | } |
290 | 30 | } |
291 | 14 | return true; |
292 | 14 | } |
293 | | |
294 | | } // namespace doris |