1 | #include "ComplexKeyHashedDictionary.h" |
2 | #include <ext/map.h> |
3 | #include <ext/range.h> |
4 | #include "DictionaryBlockInputStream.h" |
5 | #include "DictionaryFactory.h" |
6 | |
7 | namespace DB |
8 | { |
9 | namespace ErrorCodes |
10 | { |
11 | extern const int TYPE_MISMATCH; |
12 | extern const int ARGUMENT_OUT_OF_BOUND; |
13 | extern const int BAD_ARGUMENTS; |
14 | extern const int DICTIONARY_IS_EMPTY; |
15 | } |
16 | |
17 | ComplexKeyHashedDictionary::ComplexKeyHashedDictionary( |
18 | const std::string & database_, |
19 | const std::string & name_, |
20 | const DictionaryStructure & dict_struct_, |
21 | DictionarySourcePtr source_ptr_, |
22 | const DictionaryLifetime dict_lifetime_, |
23 | bool require_nonempty_, |
24 | BlockPtr saved_block_) |
25 | : database(database_) |
26 | , name(name_) |
27 | , full_name{database_.empty() ? name_ : (database_ + "." + name_)} |
28 | , dict_struct(dict_struct_) |
29 | , source_ptr{std::move(source_ptr_)} |
30 | , dict_lifetime(dict_lifetime_) |
31 | , require_nonempty(require_nonempty_) |
32 | , saved_block{std::move(saved_block_)} |
33 | { |
34 | createAttributes(); |
35 | loadData(); |
36 | calculateBytesAllocated(); |
37 | } |
38 | |
39 | #define DECLARE(TYPE) \ |
40 | void ComplexKeyHashedDictionary::get##TYPE( \ |
41 | const std::string & attribute_name, const Columns & key_columns, const DataTypes & key_types, ResultArrayType<TYPE> & out) const \ |
42 | { \ |
43 | dict_struct.validateKeyTypes(key_types); \ |
44 | \ |
45 | const auto & attribute = getAttribute(attribute_name); \ |
46 | checkAttributeType(full_name, attribute_name, attribute.type, AttributeUnderlyingType::ut##TYPE); \ |
47 | \ |
48 | const auto null_value = std::get<TYPE>(attribute.null_values); \ |
49 | \ |
50 | getItemsImpl<TYPE, TYPE>( \ |
51 | attribute, \ |
52 | key_columns, \ |
53 | [&](const size_t row, const auto value) { out[row] = value; }, \ |
54 | [&](const size_t) { return null_value; }); \ |
55 | } |
56 | DECLARE(UInt8) |
57 | DECLARE(UInt16) |
58 | DECLARE(UInt32) |
59 | DECLARE(UInt64) |
60 | DECLARE(UInt128) |
61 | DECLARE(Int8) |
62 | DECLARE(Int16) |
63 | DECLARE(Int32) |
64 | DECLARE(Int64) |
65 | DECLARE(Float32) |
66 | DECLARE(Float64) |
67 | DECLARE(Decimal32) |
68 | DECLARE(Decimal64) |
69 | DECLARE(Decimal128) |
70 | #undef DECLARE |
71 | |
72 | void ComplexKeyHashedDictionary::getString( |
73 | const std::string & attribute_name, const Columns & key_columns, const DataTypes & key_types, ColumnString * out) const |
74 | { |
75 | dict_struct.validateKeyTypes(key_types); |
76 | |
77 | const auto & attribute = getAttribute(attribute_name); |
78 | checkAttributeType(full_name, attribute_name, attribute.type, AttributeUnderlyingType::utString); |
79 | |
80 | const auto & null_value = StringRef{std::get<String>(attribute.null_values)}; |
81 | |
82 | getItemsImpl<StringRef, StringRef>( |
83 | attribute, |
84 | key_columns, |
85 | [&](const size_t, const StringRef value) { out->insertData(value.data, value.size); }, |
86 | [&](const size_t) { return null_value; }); |
87 | } |
88 | |
89 | #define DECLARE(TYPE) \ |
90 | void ComplexKeyHashedDictionary::get##TYPE( \ |
91 | const std::string & attribute_name, \ |
92 | const Columns & key_columns, \ |
93 | const DataTypes & key_types, \ |
94 | const PaddedPODArray<TYPE> & def, \ |
95 | ResultArrayType<TYPE> & out) const \ |
96 | { \ |
97 | dict_struct.validateKeyTypes(key_types); \ |
98 | \ |
99 | const auto & attribute = getAttribute(attribute_name); \ |
100 | checkAttributeType(full_name, attribute_name, attribute.type, AttributeUnderlyingType::ut##TYPE); \ |
101 | \ |
102 | getItemsImpl<TYPE, TYPE>( \ |
103 | attribute, \ |
104 | key_columns, \ |
105 | [&](const size_t row, const auto value) { out[row] = value; }, \ |
106 | [&](const size_t row) { return def[row]; }); \ |
107 | } |
108 | DECLARE(UInt8) |
109 | DECLARE(UInt16) |
110 | DECLARE(UInt32) |
111 | DECLARE(UInt64) |
112 | DECLARE(UInt128) |
113 | DECLARE(Int8) |
114 | DECLARE(Int16) |
115 | DECLARE(Int32) |
116 | DECLARE(Int64) |
117 | DECLARE(Float32) |
118 | DECLARE(Float64) |
119 | DECLARE(Decimal32) |
120 | DECLARE(Decimal64) |
121 | DECLARE(Decimal128) |
122 | #undef DECLARE |
123 | |
124 | void ComplexKeyHashedDictionary::getString( |
125 | const std::string & attribute_name, |
126 | const Columns & key_columns, |
127 | const DataTypes & key_types, |
128 | const ColumnString * const def, |
129 | ColumnString * const out) const |
130 | { |
131 | dict_struct.validateKeyTypes(key_types); |
132 | |
133 | const auto & attribute = getAttribute(attribute_name); |
134 | checkAttributeType(full_name, attribute_name, attribute.type, AttributeUnderlyingType::utString); |
135 | |
136 | getItemsImpl<StringRef, StringRef>( |
137 | attribute, |
138 | key_columns, |
139 | [&](const size_t, const StringRef value) { out->insertData(value.data, value.size); }, |
140 | [&](const size_t row) { return def->getDataAt(row); }); |
141 | } |
142 | |
143 | #define DECLARE(TYPE) \ |
144 | void ComplexKeyHashedDictionary::get##TYPE( \ |
145 | const std::string & attribute_name, \ |
146 | const Columns & key_columns, \ |
147 | const DataTypes & key_types, \ |
148 | const TYPE def, \ |
149 | ResultArrayType<TYPE> & out) const \ |
150 | { \ |
151 | dict_struct.validateKeyTypes(key_types); \ |
152 | \ |
153 | const auto & attribute = getAttribute(attribute_name); \ |
154 | checkAttributeType(full_name, attribute_name, attribute.type, AttributeUnderlyingType::ut##TYPE); \ |
155 | \ |
156 | getItemsImpl<TYPE, TYPE>( \ |
157 | attribute, key_columns, [&](const size_t row, const auto value) { out[row] = value; }, [&](const size_t) { return def; }); \ |
158 | } |
159 | DECLARE(UInt8) |
160 | DECLARE(UInt16) |
161 | DECLARE(UInt32) |
162 | DECLARE(UInt64) |
163 | DECLARE(UInt128) |
164 | DECLARE(Int8) |
165 | DECLARE(Int16) |
166 | DECLARE(Int32) |
167 | DECLARE(Int64) |
168 | DECLARE(Float32) |
169 | DECLARE(Float64) |
170 | DECLARE(Decimal32) |
171 | DECLARE(Decimal64) |
172 | DECLARE(Decimal128) |
173 | #undef DECLARE |
174 | |
175 | void ComplexKeyHashedDictionary::getString( |
176 | const std::string & attribute_name, |
177 | const Columns & key_columns, |
178 | const DataTypes & key_types, |
179 | const String & def, |
180 | ColumnString * const out) const |
181 | { |
182 | dict_struct.validateKeyTypes(key_types); |
183 | |
184 | const auto & attribute = getAttribute(attribute_name); |
185 | checkAttributeType(full_name, attribute_name, attribute.type, AttributeUnderlyingType::utString); |
186 | |
187 | getItemsImpl<StringRef, StringRef>( |
188 | attribute, |
189 | key_columns, |
190 | [&](const size_t, const StringRef value) { out->insertData(value.data, value.size); }, |
191 | [&](const size_t) { return StringRef{def}; }); |
192 | } |
193 | |
194 | void ComplexKeyHashedDictionary::has(const Columns & key_columns, const DataTypes & key_types, PaddedPODArray<UInt8> & out) const |
195 | { |
196 | dict_struct.validateKeyTypes(key_types); |
197 | |
198 | const auto & attribute = attributes.front(); |
199 | |
200 | switch (attribute.type) |
201 | { |
202 | case AttributeUnderlyingType::utUInt8: |
203 | has<UInt8>(attribute, key_columns, out); |
204 | break; |
205 | case AttributeUnderlyingType::utUInt16: |
206 | has<UInt16>(attribute, key_columns, out); |
207 | break; |
208 | case AttributeUnderlyingType::utUInt32: |
209 | has<UInt32>(attribute, key_columns, out); |
210 | break; |
211 | case AttributeUnderlyingType::utUInt64: |
212 | has<UInt64>(attribute, key_columns, out); |
213 | break; |
214 | case AttributeUnderlyingType::utUInt128: |
215 | has<UInt128>(attribute, key_columns, out); |
216 | break; |
217 | case AttributeUnderlyingType::utInt8: |
218 | has<Int8>(attribute, key_columns, out); |
219 | break; |
220 | case AttributeUnderlyingType::utInt16: |
221 | has<Int16>(attribute, key_columns, out); |
222 | break; |
223 | case AttributeUnderlyingType::utInt32: |
224 | has<Int32>(attribute, key_columns, out); |
225 | break; |
226 | case AttributeUnderlyingType::utInt64: |
227 | has<Int64>(attribute, key_columns, out); |
228 | break; |
229 | case AttributeUnderlyingType::utFloat32: |
230 | has<Float32>(attribute, key_columns, out); |
231 | break; |
232 | case AttributeUnderlyingType::utFloat64: |
233 | has<Float64>(attribute, key_columns, out); |
234 | break; |
235 | case AttributeUnderlyingType::utString: |
236 | has<StringRef>(attribute, key_columns, out); |
237 | break; |
238 | |
239 | case AttributeUnderlyingType::utDecimal32: |
240 | has<Decimal32>(attribute, key_columns, out); |
241 | break; |
242 | case AttributeUnderlyingType::utDecimal64: |
243 | has<Decimal64>(attribute, key_columns, out); |
244 | break; |
245 | case AttributeUnderlyingType::utDecimal128: |
246 | has<Decimal128>(attribute, key_columns, out); |
247 | break; |
248 | } |
249 | } |
250 | |
251 | void ComplexKeyHashedDictionary::createAttributes() |
252 | { |
253 | const auto size = dict_struct.attributes.size(); |
254 | attributes.reserve(size); |
255 | |
256 | for (const auto & attribute : dict_struct.attributes) |
257 | { |
258 | attribute_index_by_name.emplace(attribute.name, attributes.size()); |
259 | attributes.push_back(createAttributeWithType(attribute.underlying_type, attribute.null_value)); |
260 | |
261 | if (attribute.hierarchical) |
262 | throw Exception{full_name + ": hierarchical attributes not supported for dictionary of type " + getTypeName(), |
263 | ErrorCodes::TYPE_MISMATCH}; |
264 | } |
265 | } |
266 | |
267 | void ComplexKeyHashedDictionary::blockToAttributes(const Block & block) |
268 | { |
269 | /// created upfront to avoid excess allocations |
270 | const auto keys_size = dict_struct.key->size(); |
271 | StringRefs keys(keys_size); |
272 | |
273 | const auto attributes_size = attributes.size(); |
274 | const auto rows = block.rows(); |
275 | element_count += rows; |
276 | |
277 | const auto key_column_ptrs = ext::map<Columns>( |
278 | ext::range(0, keys_size), [&](const size_t attribute_idx) { return block.safeGetByPosition(attribute_idx).column; }); |
279 | |
280 | const auto attribute_column_ptrs = ext::map<Columns>(ext::range(0, attributes_size), [&](const size_t attribute_idx) |
281 | { |
282 | return block.safeGetByPosition(keys_size + attribute_idx).column; |
283 | }); |
284 | |
285 | for (const auto row_idx : ext::range(0, rows)) |
286 | { |
287 | /// calculate key once per row |
288 | const auto key = placeKeysInPool(row_idx, key_column_ptrs, keys, keys_pool); |
289 | |
290 | auto should_rollback = false; |
291 | |
292 | for (const auto attribute_idx : ext::range(0, attributes_size)) |
293 | { |
294 | const auto & attribute_column = *attribute_column_ptrs[attribute_idx]; |
295 | auto & attribute = attributes[attribute_idx]; |
296 | const auto inserted = setAttributeValue(attribute, key, attribute_column[row_idx]); |
297 | if (!inserted) |
298 | should_rollback = true; |
299 | } |
300 | |
301 | /// @note on multiple equal keys the mapped value for the first one is stored |
302 | if (should_rollback) |
303 | keys_pool.rollback(key.size); |
304 | } |
305 | } |
306 | |
307 | void ComplexKeyHashedDictionary::updateData() |
308 | { |
309 | /// created upfront to avoid excess allocations |
310 | const auto keys_size = dict_struct.key->size(); |
311 | StringRefs keys(keys_size); |
312 | |
313 | const auto attributes_size = attributes.size(); |
314 | |
315 | if (!saved_block || saved_block->rows() == 0) |
316 | { |
317 | auto stream = source_ptr->loadUpdatedAll(); |
318 | stream->readPrefix(); |
319 | |
320 | while (const auto block = stream->read()) |
321 | { |
322 | /// We are using this method to keep saved data if input stream consists of multiple blocks |
323 | if (!saved_block) |
324 | saved_block = std::make_shared<DB::Block>(block.cloneEmpty()); |
325 | for (const auto attribute_idx : ext::range(0, keys_size + attributes_size)) |
326 | { |
327 | const IColumn & update_column = *block.getByPosition(attribute_idx).column.get(); |
328 | MutableColumnPtr saved_column = saved_block->getByPosition(attribute_idx).column->assumeMutable(); |
329 | saved_column->insertRangeFrom(update_column, 0, update_column.size()); |
330 | } |
331 | } |
332 | stream->readSuffix(); |
333 | } |
334 | else |
335 | { |
336 | auto stream = source_ptr->loadUpdatedAll(); |
337 | |
338 | stream->readPrefix(); |
339 | while (Block block = stream->read()) |
340 | { |
341 | const auto saved_key_column_ptrs = ext::map<Columns>( |
342 | ext::range(0, keys_size), [&](const size_t key_idx) { return saved_block->safeGetByPosition(key_idx).column; }); |
343 | |
344 | const auto update_key_column_ptrs = ext::map<Columns>( |
345 | ext::range(0, keys_size), [&](const size_t key_idx) { return block.safeGetByPosition(key_idx).column; }); |
346 | |
347 | Arena temp_key_pool; |
348 | ContainerType<std::vector<size_t>> update_key_hash; |
349 | |
350 | for (size_t i = 0; i < block.rows(); ++i) |
351 | { |
352 | const auto u_key = placeKeysInPool(i, update_key_column_ptrs, keys, temp_key_pool); |
353 | update_key_hash[u_key].push_back(i); |
354 | } |
355 | |
356 | const size_t rows = saved_block->rows(); |
357 | IColumn::Filter filter(rows); |
358 | |
359 | for (size_t i = 0; i < saved_block->rows(); ++i) |
360 | { |
361 | const auto s_key = placeKeysInPool(i, saved_key_column_ptrs, keys, temp_key_pool); |
362 | auto it = update_key_hash.find(s_key); |
363 | if (it) |
364 | filter[i] = 0; |
365 | else |
366 | filter[i] = 1; |
367 | } |
368 | |
369 | auto block_columns = block.mutateColumns(); |
370 | for (const auto attribute_idx : ext::range(0, keys_size + attributes_size)) |
371 | { |
372 | auto & column = saved_block->safeGetByPosition(attribute_idx).column; |
373 | const auto & filtered_column = column->filter(filter, -1); |
374 | |
375 | block_columns[attribute_idx]->insertRangeFrom(*filtered_column.get(), 0, filtered_column->size()); |
376 | } |
377 | |
378 | saved_block->setColumns(std::move(block_columns)); |
379 | } |
380 | stream->readSuffix(); |
381 | } |
382 | |
383 | if (saved_block) |
384 | blockToAttributes(*saved_block.get()); |
385 | } |
386 | |
387 | void ComplexKeyHashedDictionary::loadData() |
388 | { |
389 | if (!source_ptr->hasUpdateField()) |
390 | { |
391 | auto stream = source_ptr->loadAll(); |
392 | stream->readPrefix(); |
393 | |
394 | while (const auto block = stream->read()) |
395 | blockToAttributes(block); |
396 | |
397 | stream->readSuffix(); |
398 | } |
399 | else |
400 | updateData(); |
401 | |
402 | if (require_nonempty && 0 == element_count) |
403 | throw Exception{full_name + ": dictionary source is empty and 'require_nonempty' property is set." , ErrorCodes::DICTIONARY_IS_EMPTY}; |
404 | } |
405 | |
406 | template <typename T> |
407 | void ComplexKeyHashedDictionary::addAttributeSize(const Attribute & attribute) |
408 | { |
409 | const auto & map_ref = std::get<ContainerType<T>>(attribute.maps); |
410 | bytes_allocated += sizeof(ContainerType<T>) + map_ref.getBufferSizeInBytes(); |
411 | bucket_count = map_ref.getBufferSizeInCells(); |
412 | } |
413 | |
414 | void ComplexKeyHashedDictionary::calculateBytesAllocated() |
415 | { |
416 | bytes_allocated += attributes.size() * sizeof(attributes.front()); |
417 | |
418 | for (const auto & attribute : attributes) |
419 | { |
420 | switch (attribute.type) |
421 | { |
422 | case AttributeUnderlyingType::utUInt8: |
423 | addAttributeSize<UInt8>(attribute); |
424 | break; |
425 | case AttributeUnderlyingType::utUInt16: |
426 | addAttributeSize<UInt16>(attribute); |
427 | break; |
428 | case AttributeUnderlyingType::utUInt32: |
429 | addAttributeSize<UInt32>(attribute); |
430 | break; |
431 | case AttributeUnderlyingType::utUInt64: |
432 | addAttributeSize<UInt64>(attribute); |
433 | break; |
434 | case AttributeUnderlyingType::utUInt128: |
435 | addAttributeSize<UInt128>(attribute); |
436 | break; |
437 | case AttributeUnderlyingType::utInt8: |
438 | addAttributeSize<Int8>(attribute); |
439 | break; |
440 | case AttributeUnderlyingType::utInt16: |
441 | addAttributeSize<Int16>(attribute); |
442 | break; |
443 | case AttributeUnderlyingType::utInt32: |
444 | addAttributeSize<Int32>(attribute); |
445 | break; |
446 | case AttributeUnderlyingType::utInt64: |
447 | addAttributeSize<Int64>(attribute); |
448 | break; |
449 | case AttributeUnderlyingType::utFloat32: |
450 | addAttributeSize<Float32>(attribute); |
451 | break; |
452 | case AttributeUnderlyingType::utFloat64: |
453 | addAttributeSize<Float64>(attribute); |
454 | break; |
455 | |
456 | case AttributeUnderlyingType::utDecimal32: |
457 | addAttributeSize<Decimal32>(attribute); |
458 | break; |
459 | case AttributeUnderlyingType::utDecimal64: |
460 | addAttributeSize<Decimal64>(attribute); |
461 | break; |
462 | case AttributeUnderlyingType::utDecimal128: |
463 | addAttributeSize<Decimal128>(attribute); |
464 | break; |
465 | |
466 | case AttributeUnderlyingType::utString: |
467 | { |
468 | addAttributeSize<StringRef>(attribute); |
469 | bytes_allocated += sizeof(Arena) + attribute.string_arena->size(); |
470 | |
471 | break; |
472 | } |
473 | } |
474 | } |
475 | |
476 | bytes_allocated += keys_pool.size(); |
477 | } |
478 | |
479 | template <typename T> |
480 | void ComplexKeyHashedDictionary::createAttributeImpl(Attribute & attribute, const Field & null_value) |
481 | { |
482 | attribute.null_values = T(null_value.get<NearestFieldType<T>>()); |
483 | attribute.maps.emplace<ContainerType<T>>(); |
484 | } |
485 | |
486 | ComplexKeyHashedDictionary::Attribute |
487 | ComplexKeyHashedDictionary::createAttributeWithType(const AttributeUnderlyingType type, const Field & null_value) |
488 | { |
489 | Attribute attr{type, {}, {}, {}}; |
490 | |
491 | switch (type) |
492 | { |
493 | case AttributeUnderlyingType::utUInt8: |
494 | createAttributeImpl<UInt8>(attr, null_value); |
495 | break; |
496 | case AttributeUnderlyingType::utUInt16: |
497 | createAttributeImpl<UInt16>(attr, null_value); |
498 | break; |
499 | case AttributeUnderlyingType::utUInt32: |
500 | createAttributeImpl<UInt32>(attr, null_value); |
501 | break; |
502 | case AttributeUnderlyingType::utUInt64: |
503 | createAttributeImpl<UInt64>(attr, null_value); |
504 | break; |
505 | case AttributeUnderlyingType::utUInt128: |
506 | createAttributeImpl<UInt128>(attr, null_value); |
507 | break; |
508 | case AttributeUnderlyingType::utInt8: |
509 | createAttributeImpl<Int8>(attr, null_value); |
510 | break; |
511 | case AttributeUnderlyingType::utInt16: |
512 | createAttributeImpl<Int16>(attr, null_value); |
513 | break; |
514 | case AttributeUnderlyingType::utInt32: |
515 | createAttributeImpl<Int32>(attr, null_value); |
516 | break; |
517 | case AttributeUnderlyingType::utInt64: |
518 | createAttributeImpl<Int64>(attr, null_value); |
519 | break; |
520 | case AttributeUnderlyingType::utFloat32: |
521 | createAttributeImpl<Float32>(attr, null_value); |
522 | break; |
523 | case AttributeUnderlyingType::utFloat64: |
524 | createAttributeImpl<Float64>(attr, null_value); |
525 | break; |
526 | |
527 | case AttributeUnderlyingType::utDecimal32: |
528 | createAttributeImpl<Decimal32>(attr, null_value); |
529 | break; |
530 | case AttributeUnderlyingType::utDecimal64: |
531 | createAttributeImpl<Decimal64>(attr, null_value); |
532 | break; |
533 | case AttributeUnderlyingType::utDecimal128: |
534 | createAttributeImpl<Decimal128>(attr, null_value); |
535 | break; |
536 | |
537 | case AttributeUnderlyingType::utString: |
538 | { |
539 | attr.null_values = null_value.get<String>(); |
540 | attr.maps.emplace<ContainerType<StringRef>>(); |
541 | attr.string_arena = std::make_unique<Arena>(); |
542 | break; |
543 | } |
544 | } |
545 | |
546 | return attr; |
547 | } |
548 | |
549 | |
550 | template <typename AttributeType, typename OutputType, typename ValueSetter, typename DefaultGetter> |
551 | void ComplexKeyHashedDictionary::getItemsImpl( |
552 | const Attribute & attribute, const Columns & key_columns, ValueSetter && set_value, DefaultGetter && get_default) const |
553 | { |
554 | const auto & attr = std::get<ContainerType<AttributeType>>(attribute.maps); |
555 | |
556 | const auto keys_size = key_columns.size(); |
557 | StringRefs keys(keys_size); |
558 | Arena temporary_keys_pool; |
559 | |
560 | const auto rows = key_columns.front()->size(); |
561 | for (const auto i : ext::range(0, rows)) |
562 | { |
563 | /// copy key data to arena so it is contiguous and return StringRef to it |
564 | const auto key = placeKeysInPool(i, key_columns, keys, temporary_keys_pool); |
565 | |
566 | const auto it = attr.find(key); |
567 | set_value(i, it ? static_cast<OutputType>(it->getMapped()) : get_default(i)); |
568 | |
569 | /// free memory allocated for the key |
570 | temporary_keys_pool.rollback(key.size); |
571 | } |
572 | |
573 | query_count.fetch_add(rows, std::memory_order_relaxed); |
574 | } |
575 | |
576 | |
577 | template <typename T> |
578 | bool ComplexKeyHashedDictionary::setAttributeValueImpl(Attribute & attribute, const StringRef key, const T value) |
579 | { |
580 | auto & map = std::get<ContainerType<T>>(attribute.maps); |
581 | const auto pair = map.insert({key, value}); |
582 | return pair.second; |
583 | } |
584 | |
585 | bool ComplexKeyHashedDictionary::setAttributeValue(Attribute & attribute, const StringRef key, const Field & value) |
586 | { |
587 | switch (attribute.type) |
588 | { |
589 | case AttributeUnderlyingType::utUInt8: |
590 | return setAttributeValueImpl<UInt8>(attribute, key, value.get<UInt64>()); |
591 | case AttributeUnderlyingType::utUInt16: |
592 | return setAttributeValueImpl<UInt16>(attribute, key, value.get<UInt64>()); |
593 | case AttributeUnderlyingType::utUInt32: |
594 | return setAttributeValueImpl<UInt32>(attribute, key, value.get<UInt64>()); |
595 | case AttributeUnderlyingType::utUInt64: |
596 | return setAttributeValueImpl<UInt64>(attribute, key, value.get<UInt64>()); |
597 | case AttributeUnderlyingType::utUInt128: |
598 | return setAttributeValueImpl<UInt128>(attribute, key, value.get<UInt128>()); |
599 | case AttributeUnderlyingType::utInt8: |
600 | return setAttributeValueImpl<Int8>(attribute, key, value.get<Int64>()); |
601 | case AttributeUnderlyingType::utInt16: |
602 | return setAttributeValueImpl<Int16>(attribute, key, value.get<Int64>()); |
603 | case AttributeUnderlyingType::utInt32: |
604 | return setAttributeValueImpl<Int32>(attribute, key, value.get<Int64>()); |
605 | case AttributeUnderlyingType::utInt64: |
606 | return setAttributeValueImpl<Int64>(attribute, key, value.get<Int64>()); |
607 | case AttributeUnderlyingType::utFloat32: |
608 | return setAttributeValueImpl<Float32>(attribute, key, value.get<Float64>()); |
609 | case AttributeUnderlyingType::utFloat64: |
610 | return setAttributeValueImpl<Float64>(attribute, key, value.get<Float64>()); |
611 | |
612 | case AttributeUnderlyingType::utDecimal32: |
613 | return setAttributeValueImpl<Decimal32>(attribute, key, value.get<Decimal32>()); |
614 | case AttributeUnderlyingType::utDecimal64: |
615 | return setAttributeValueImpl<Decimal64>(attribute, key, value.get<Decimal64>()); |
616 | case AttributeUnderlyingType::utDecimal128: |
617 | return setAttributeValueImpl<Decimal128>(attribute, key, value.get<Decimal128>()); |
618 | |
619 | case AttributeUnderlyingType::utString: |
620 | { |
621 | auto & map = std::get<ContainerType<StringRef>>(attribute.maps); |
622 | const auto & string = value.get<String>(); |
623 | const auto string_in_arena = attribute.string_arena->insert(string.data(), string.size()); |
624 | const auto pair = map.insert({key, StringRef{string_in_arena, string.size()}}); |
625 | return pair.second; |
626 | } |
627 | } |
628 | |
629 | return {}; |
630 | } |
631 | |
632 | const ComplexKeyHashedDictionary::Attribute & ComplexKeyHashedDictionary::getAttribute(const std::string & attribute_name) const |
633 | { |
634 | const auto it = attribute_index_by_name.find(attribute_name); |
635 | if (it == std::end(attribute_index_by_name)) |
636 | throw Exception{full_name + ": no such attribute '" + attribute_name + "'" , ErrorCodes::BAD_ARGUMENTS}; |
637 | |
638 | return attributes[it->second]; |
639 | } |
640 | |
641 | StringRef ComplexKeyHashedDictionary::placeKeysInPool(const size_t row, const Columns & key_columns, StringRefs & keys, Arena & pool) |
642 | { |
643 | const auto keys_size = key_columns.size(); |
644 | size_t sum_keys_size{}; |
645 | |
646 | const char * block_start = nullptr; |
647 | for (size_t j = 0; j < keys_size; ++j) |
648 | { |
649 | keys[j] = key_columns[j]->serializeValueIntoArena(row, pool, block_start); |
650 | sum_keys_size += keys[j].size; |
651 | } |
652 | |
653 | auto key_start = block_start; |
654 | for (size_t j = 0; j < keys_size; ++j) |
655 | { |
656 | keys[j].data = key_start; |
657 | key_start += keys[j].size; |
658 | } |
659 | |
660 | return {block_start, sum_keys_size}; |
661 | } |
662 | |
663 | template <typename T> |
664 | void ComplexKeyHashedDictionary::has(const Attribute & attribute, const Columns & key_columns, PaddedPODArray<UInt8> & out) const |
665 | { |
666 | const auto & attr = std::get<ContainerType<T>>(attribute.maps); |
667 | const auto keys_size = key_columns.size(); |
668 | StringRefs keys(keys_size); |
669 | Arena temporary_keys_pool; |
670 | const auto rows = key_columns.front()->size(); |
671 | |
672 | for (const auto i : ext::range(0, rows)) |
673 | { |
674 | /// copy key data to arena so it is contiguous and return StringRef to it |
675 | const auto key = placeKeysInPool(i, key_columns, keys, temporary_keys_pool); |
676 | |
677 | const auto it = attr.find(key); |
678 | out[i] = static_cast<bool>(it); |
679 | |
680 | /// free memory allocated for the key |
681 | temporary_keys_pool.rollback(key.size); |
682 | } |
683 | |
684 | query_count.fetch_add(rows, std::memory_order_relaxed); |
685 | } |
686 | |
687 | std::vector<StringRef> ComplexKeyHashedDictionary::getKeys() const |
688 | { |
689 | const Attribute & attribute = attributes.front(); |
690 | |
691 | switch (attribute.type) |
692 | { |
693 | case AttributeUnderlyingType::utUInt8: |
694 | return getKeys<UInt8>(attribute); |
695 | case AttributeUnderlyingType::utUInt16: |
696 | return getKeys<UInt16>(attribute); |
697 | case AttributeUnderlyingType::utUInt32: |
698 | return getKeys<UInt32>(attribute); |
699 | case AttributeUnderlyingType::utUInt64: |
700 | return getKeys<UInt64>(attribute); |
701 | case AttributeUnderlyingType::utUInt128: |
702 | return getKeys<UInt128>(attribute); |
703 | case AttributeUnderlyingType::utInt8: |
704 | return getKeys<Int8>(attribute); |
705 | case AttributeUnderlyingType::utInt16: |
706 | return getKeys<Int16>(attribute); |
707 | case AttributeUnderlyingType::utInt32: |
708 | return getKeys<Int32>(attribute); |
709 | case AttributeUnderlyingType::utInt64: |
710 | return getKeys<Int64>(attribute); |
711 | case AttributeUnderlyingType::utFloat32: |
712 | return getKeys<Float32>(attribute); |
713 | case AttributeUnderlyingType::utFloat64: |
714 | return getKeys<Float64>(attribute); |
715 | case AttributeUnderlyingType::utString: |
716 | return getKeys<StringRef>(attribute); |
717 | |
718 | case AttributeUnderlyingType::utDecimal32: |
719 | return getKeys<Decimal32>(attribute); |
720 | case AttributeUnderlyingType::utDecimal64: |
721 | return getKeys<Decimal64>(attribute); |
722 | case AttributeUnderlyingType::utDecimal128: |
723 | return getKeys<Decimal128>(attribute); |
724 | } |
725 | return {}; |
726 | } |
727 | |
728 | template <typename T> |
729 | std::vector<StringRef> ComplexKeyHashedDictionary::getKeys(const Attribute & attribute) const |
730 | { |
731 | const ContainerType<T> & attr = std::get<ContainerType<T>>(attribute.maps); |
732 | std::vector<StringRef> keys; |
733 | keys.reserve(attr.size()); |
734 | for (const auto & key : attr) |
735 | keys.push_back(key.getKey()); |
736 | |
737 | return keys; |
738 | } |
739 | |
740 | BlockInputStreamPtr ComplexKeyHashedDictionary::getBlockInputStream(const Names & column_names, size_t max_block_size) const |
741 | { |
742 | using BlockInputStreamType = DictionaryBlockInputStream<ComplexKeyHashedDictionary, UInt64>; |
743 | return std::make_shared<BlockInputStreamType>(shared_from_this(), max_block_size, getKeys(), column_names); |
744 | } |
745 | |
746 | void registerDictionaryComplexKeyHashed(DictionaryFactory & factory) |
747 | { |
748 | auto create_layout = [=](const std::string &, |
749 | const DictionaryStructure & dict_struct, |
750 | const Poco::Util::AbstractConfiguration & config, |
751 | const std::string & config_prefix, |
752 | DictionarySourcePtr source_ptr) -> DictionaryPtr |
753 | { |
754 | if (!dict_struct.key) |
755 | throw Exception{"'key' is required for dictionary of layout 'complex_key_hashed'" , ErrorCodes::BAD_ARGUMENTS}; |
756 | |
757 | const String database = config.getString(config_prefix + ".database" , "" ); |
758 | const String name = config.getString(config_prefix + ".name" ); |
759 | const DictionaryLifetime dict_lifetime{config, config_prefix + ".lifetime" }; |
760 | const bool require_nonempty = config.getBool(config_prefix + ".require_nonempty" , false); |
761 | return std::make_unique<ComplexKeyHashedDictionary>(database, name, dict_struct, std::move(source_ptr), dict_lifetime, require_nonempty); |
762 | }; |
763 | factory.registerLayout("complex_key_hashed" , create_layout, true); |
764 | } |
765 | |
766 | } |
767 | |