1#include "CacheDictionary.h"
2
3#include <functional>
4#include <memory>
5#include <Columns/ColumnString.h>
6#include <Common/BitHelpers.h>
7#include <Common/CurrentMetrics.h>
8#include <Common/HashTable/Hash.h>
9#include <Common/ProfileEvents.h>
10#include <Common/ProfilingScopedRWLock.h>
11#include <Common/randomSeed.h>
12#include <Common/typeid_cast.h>
13#include <ext/range.h>
14#include <ext/size.h>
15#include "CacheDictionary.inc.h"
16#include "DictionaryBlockInputStream.h"
17#include "DictionaryFactory.h"
18
19namespace ProfileEvents
20{
21extern const Event DictCacheKeysRequested;
22extern const Event DictCacheKeysRequestedMiss;
23extern const Event DictCacheKeysRequestedFound;
24extern const Event DictCacheKeysExpired;
25extern const Event DictCacheKeysNotFound;
26extern const Event DictCacheKeysHit;
27extern const Event DictCacheRequestTimeNs;
28extern const Event DictCacheRequests;
29extern const Event DictCacheLockWriteNs;
30extern const Event DictCacheLockReadNs;
31}
32
33namespace CurrentMetrics
34{
35extern const Metric DictCacheRequests;
36}
37
38
39namespace DB
40{
41namespace ErrorCodes
42{
43 extern const int TYPE_MISMATCH;
44 extern const int BAD_ARGUMENTS;
45 extern const int UNSUPPORTED_METHOD;
46 extern const int LOGICAL_ERROR;
47 extern const int TOO_SMALL_BUFFER_SIZE;
48}
49
50
51inline size_t CacheDictionary::getCellIdx(const Key id) const
52{
53 const auto hash = intHash64(id);
54 const auto idx = hash & size_overlap_mask;
55 return idx;
56}
57
58
59CacheDictionary::CacheDictionary(
60 const std::string & database_,
61 const std::string & name_,
62 const DictionaryStructure & dict_struct_,
63 DictionarySourcePtr source_ptr_,
64 const DictionaryLifetime dict_lifetime_,
65 const size_t size_)
66 : database(database_)
67 , name(name_)
68 , full_name{database_.empty() ? name_ : (database_ + "." + name_)}
69 , dict_struct(dict_struct_)
70 , source_ptr{std::move(source_ptr_)}
71 , dict_lifetime(dict_lifetime_)
72 , log(&Logger::get("ExternalDictionaries"))
73 , size{roundUpToPowerOfTwoOrZero(std::max(size_, size_t(max_collision_length)))}
74 , size_overlap_mask{this->size - 1}
75 , cells{this->size}
76 , rnd_engine(randomSeed())
77{
78 if (!this->source_ptr->supportsSelectiveLoad())
79 throw Exception{full_name + ": source cannot be used with CacheDictionary", ErrorCodes::UNSUPPORTED_METHOD};
80
81 createAttributes();
82}
83
84
85void CacheDictionary::toParent(const PaddedPODArray<Key> & ids, PaddedPODArray<Key> & out) const
86{
87 const auto null_value = std::get<UInt64>(hierarchical_attribute->null_values);
88
89 getItemsNumberImpl<UInt64, UInt64>(*hierarchical_attribute, ids, out, [&](const size_t) { return null_value; });
90}
91
92
93/// Allow to use single value in same way as array.
94static inline CacheDictionary::Key getAt(const PaddedPODArray<CacheDictionary::Key> & arr, const size_t idx)
95{
96 return arr[idx];
97}
98static inline CacheDictionary::Key getAt(const CacheDictionary::Key & value, const size_t)
99{
100 return value;
101}
102
103
104template <typename AncestorType>
105void CacheDictionary::isInImpl(const PaddedPODArray<Key> & child_ids, const AncestorType & ancestor_ids, PaddedPODArray<UInt8> & out) const
106{
107 /// Transform all children to parents until ancestor id or null_value will be reached.
108
109 size_t out_size = out.size();
110 memset(out.data(), 0xFF, out_size); /// 0xFF means "not calculated"
111
112 const auto null_value = std::get<UInt64>(hierarchical_attribute->null_values);
113
114 PaddedPODArray<Key> children(out_size, 0);
115 PaddedPODArray<Key> parents(child_ids.begin(), child_ids.end());
116
117 while (true)
118 {
119 size_t out_idx = 0;
120 size_t parents_idx = 0;
121 size_t new_children_idx = 0;
122
123 while (out_idx < out_size)
124 {
125 /// Already calculated
126 if (out[out_idx] != 0xFF)
127 {
128 ++out_idx;
129 continue;
130 }
131
132 /// No parent
133 if (parents[parents_idx] == null_value)
134 {
135 out[out_idx] = 0;
136 }
137 /// Found ancestor
138 else if (parents[parents_idx] == getAt(ancestor_ids, parents_idx))
139 {
140 out[out_idx] = 1;
141 }
142 /// Loop detected
143 else if (children[new_children_idx] == parents[parents_idx])
144 {
145 out[out_idx] = 1;
146 }
147 /// Found intermediate parent, add this value to search at next loop iteration
148 else
149 {
150 children[new_children_idx] = parents[parents_idx];
151 ++new_children_idx;
152 }
153
154 ++out_idx;
155 ++parents_idx;
156 }
157
158 if (new_children_idx == 0)
159 break;
160
161 /// Transform all children to its parents.
162 children.resize(new_children_idx);
163 parents.resize(new_children_idx);
164
165 toParent(children, parents);
166 }
167}
168
169void CacheDictionary::isInVectorVector(
170 const PaddedPODArray<Key> & child_ids, const PaddedPODArray<Key> & ancestor_ids, PaddedPODArray<UInt8> & out) const
171{
172 isInImpl(child_ids, ancestor_ids, out);
173}
174
175void CacheDictionary::isInVectorConstant(const PaddedPODArray<Key> & child_ids, const Key ancestor_id, PaddedPODArray<UInt8> & out) const
176{
177 isInImpl(child_ids, ancestor_id, out);
178}
179
180void CacheDictionary::isInConstantVector(const Key child_id, const PaddedPODArray<Key> & ancestor_ids, PaddedPODArray<UInt8> & out) const
181{
182 /// Special case with single child value.
183
184 const auto null_value = std::get<UInt64>(hierarchical_attribute->null_values);
185
186 PaddedPODArray<Key> child(1, child_id);
187 PaddedPODArray<Key> parent(1);
188 std::vector<Key> ancestors(1, child_id);
189
190 /// Iteratively find all ancestors for child.
191 while (true)
192 {
193 toParent(child, parent);
194
195 if (parent[0] == null_value)
196 break;
197
198 child[0] = parent[0];
199 ancestors.push_back(parent[0]);
200 }
201
202 /// Assuming short hierarchy, so linear search is Ok.
203 for (size_t i = 0, out_size = out.size(); i < out_size; ++i)
204 out[i] = std::find(ancestors.begin(), ancestors.end(), ancestor_ids[i]) != ancestors.end();
205}
206
207void CacheDictionary::getString(const std::string & attribute_name, const PaddedPODArray<Key> & ids, ColumnString * out) const
208{
209 auto & attribute = getAttribute(attribute_name);
210 checkAttributeType(full_name, attribute_name, attribute.type, AttributeUnderlyingType::utString);
211
212 const auto null_value = StringRef{std::get<String>(attribute.null_values)};
213
214 getItemsString(attribute, ids, out, [&](const size_t) { return null_value; });
215}
216
217void CacheDictionary::getString(
218 const std::string & attribute_name, const PaddedPODArray<Key> & ids, const ColumnString * const def, ColumnString * const out) const
219{
220 auto & attribute = getAttribute(attribute_name);
221 checkAttributeType(full_name, attribute_name, attribute.type, AttributeUnderlyingType::utString);
222
223 getItemsString(attribute, ids, out, [&](const size_t row) { return def->getDataAt(row); });
224}
225
226void CacheDictionary::getString(
227 const std::string & attribute_name, const PaddedPODArray<Key> & ids, const String & def, ColumnString * const out) const
228{
229 auto & attribute = getAttribute(attribute_name);
230 checkAttributeType(full_name, attribute_name, attribute.type, AttributeUnderlyingType::utString);
231
232 getItemsString(attribute, ids, out, [&](const size_t) { return StringRef{def}; });
233}
234
235
236/// returns cell_idx (always valid for replacing), 'cell is valid' flag, 'cell is outdated' flag
237/// true false found and valid
238/// false true not found (something outdated, maybe our cell)
239/// false false not found (other id stored with valid data)
240/// true true impossible
241///
242/// todo: split this func to two: find_for_get and find_for_set
243CacheDictionary::FindResult CacheDictionary::findCellIdx(const Key & id, const CellMetadata::time_point_t now) const
244{
245 auto pos = getCellIdx(id);
246 auto oldest_id = pos;
247 auto oldest_time = CellMetadata::time_point_t::max();
248 const auto stop = pos + max_collision_length;
249 for (; pos < stop; ++pos)
250 {
251 const auto cell_idx = pos & size_overlap_mask;
252 const auto & cell = cells[cell_idx];
253
254 if (cell.id != id)
255 {
256 /// maybe we already found nearest expired cell (try minimize collision_length on insert)
257 if (oldest_time > now && oldest_time > cell.expiresAt())
258 {
259 oldest_time = cell.expiresAt();
260 oldest_id = cell_idx;
261 }
262 continue;
263 }
264
265 if (cell.expiresAt() < now)
266 {
267 return {cell_idx, false, true};
268 }
269
270 return {cell_idx, true, false};
271 }
272
273 return {oldest_id, false, false};
274}
275
276void CacheDictionary::has(const PaddedPODArray<Key> & ids, PaddedPODArray<UInt8> & out) const
277{
278 /// Mapping: <id> -> { all indices `i` of `ids` such that `ids[i]` = <id> }
279 std::unordered_map<Key, std::vector<size_t>> outdated_ids;
280
281 size_t cache_expired = 0, cache_not_found = 0, cache_hit = 0;
282
283 const auto rows = ext::size(ids);
284 {
285 const ProfilingScopedReadRWLock read_lock{rw_lock, ProfileEvents::DictCacheLockReadNs};
286
287 const auto now = std::chrono::system_clock::now();
288 /// fetch up-to-date values, decide which ones require update
289 for (const auto row : ext::range(0, rows))
290 {
291 const auto id = ids[row];
292 const auto find_result = findCellIdx(id, now);
293 const auto & cell_idx = find_result.cell_idx;
294 if (!find_result.valid)
295 {
296 outdated_ids[id].push_back(row);
297 if (find_result.outdated)
298 ++cache_expired;
299 else
300 ++cache_not_found;
301 }
302 else
303 {
304 ++cache_hit;
305 const auto & cell = cells[cell_idx];
306 out[row] = !cell.isDefault();
307 }
308 }
309 }
310
311 ProfileEvents::increment(ProfileEvents::DictCacheKeysExpired, cache_expired);
312 ProfileEvents::increment(ProfileEvents::DictCacheKeysNotFound, cache_not_found);
313 ProfileEvents::increment(ProfileEvents::DictCacheKeysHit, cache_hit);
314
315 query_count.fetch_add(rows, std::memory_order_relaxed);
316 hit_count.fetch_add(rows - outdated_ids.size(), std::memory_order_release);
317
318 if (outdated_ids.empty())
319 return;
320
321 std::vector<Key> required_ids(outdated_ids.size());
322 std::transform(std::begin(outdated_ids), std::end(outdated_ids), std::begin(required_ids), [](auto & pair) { return pair.first; });
323
324 /// request new values
325 update(
326 required_ids,
327 [&](const auto id, const auto)
328 {
329 for (const auto row : outdated_ids[id])
330 out[row] = true;
331 },
332 [&](const auto id, const auto)
333 {
334 for (const auto row : outdated_ids[id])
335 out[row] = false;
336 });
337}
338
339
340void CacheDictionary::createAttributes()
341{
342 const auto attributes_size = dict_struct.attributes.size();
343 attributes.reserve(attributes_size);
344
345 bytes_allocated += size * sizeof(CellMetadata);
346 bytes_allocated += attributes_size * sizeof(attributes.front());
347
348 for (const auto & attribute : dict_struct.attributes)
349 {
350 attribute_index_by_name.emplace(attribute.name, attributes.size());
351 attributes.push_back(createAttributeWithType(attribute.underlying_type, attribute.null_value));
352
353 if (attribute.hierarchical)
354 {
355 hierarchical_attribute = &attributes.back();
356
357 if (hierarchical_attribute->type != AttributeUnderlyingType::utUInt64)
358 throw Exception{full_name + ": hierarchical attribute must be UInt64.", ErrorCodes::TYPE_MISMATCH};
359 }
360 }
361}
362
363CacheDictionary::Attribute CacheDictionary::createAttributeWithType(const AttributeUnderlyingType type, const Field & null_value)
364{
365 Attribute attr{type, {}, {}};
366
367 switch (type)
368 {
369#define DISPATCH(TYPE) \
370 case AttributeUnderlyingType::ut##TYPE: \
371 attr.null_values = TYPE(null_value.get<NearestFieldType<TYPE>>()); \
372 attr.arrays = std::make_unique<ContainerType<TYPE>>(size); \
373 bytes_allocated += size * sizeof(TYPE); \
374 break;
375 DISPATCH(UInt8)
376 DISPATCH(UInt16)
377 DISPATCH(UInt32)
378 DISPATCH(UInt64)
379 DISPATCH(UInt128)
380 DISPATCH(Int8)
381 DISPATCH(Int16)
382 DISPATCH(Int32)
383 DISPATCH(Int64)
384 DISPATCH(Decimal32)
385 DISPATCH(Decimal64)
386 DISPATCH(Decimal128)
387 DISPATCH(Float32)
388 DISPATCH(Float64)
389#undef DISPATCH
390 case AttributeUnderlyingType::utString:
391 attr.null_values = null_value.get<String>();
392 attr.arrays = std::make_unique<ContainerType<StringRef>>(size);
393 bytes_allocated += size * sizeof(StringRef);
394 if (!string_arena)
395 string_arena = std::make_unique<ArenaWithFreeLists>();
396 break;
397 }
398
399 return attr;
400}
401
402void CacheDictionary::setDefaultAttributeValue(Attribute & attribute, const Key idx) const
403{
404 switch (attribute.type)
405 {
406 case AttributeUnderlyingType::utUInt8:
407 std::get<ContainerPtrType<UInt8>>(attribute.arrays)[idx] = std::get<UInt8>(attribute.null_values);
408 break;
409 case AttributeUnderlyingType::utUInt16:
410 std::get<ContainerPtrType<UInt16>>(attribute.arrays)[idx] = std::get<UInt16>(attribute.null_values);
411 break;
412 case AttributeUnderlyingType::utUInt32:
413 std::get<ContainerPtrType<UInt32>>(attribute.arrays)[idx] = std::get<UInt32>(attribute.null_values);
414 break;
415 case AttributeUnderlyingType::utUInt64:
416 std::get<ContainerPtrType<UInt64>>(attribute.arrays)[idx] = std::get<UInt64>(attribute.null_values);
417 break;
418 case AttributeUnderlyingType::utUInt128:
419 std::get<ContainerPtrType<UInt128>>(attribute.arrays)[idx] = std::get<UInt128>(attribute.null_values);
420 break;
421 case AttributeUnderlyingType::utInt8:
422 std::get<ContainerPtrType<Int8>>(attribute.arrays)[idx] = std::get<Int8>(attribute.null_values);
423 break;
424 case AttributeUnderlyingType::utInt16:
425 std::get<ContainerPtrType<Int16>>(attribute.arrays)[idx] = std::get<Int16>(attribute.null_values);
426 break;
427 case AttributeUnderlyingType::utInt32:
428 std::get<ContainerPtrType<Int32>>(attribute.arrays)[idx] = std::get<Int32>(attribute.null_values);
429 break;
430 case AttributeUnderlyingType::utInt64:
431 std::get<ContainerPtrType<Int64>>(attribute.arrays)[idx] = std::get<Int64>(attribute.null_values);
432 break;
433 case AttributeUnderlyingType::utFloat32:
434 std::get<ContainerPtrType<Float32>>(attribute.arrays)[idx] = std::get<Float32>(attribute.null_values);
435 break;
436 case AttributeUnderlyingType::utFloat64:
437 std::get<ContainerPtrType<Float64>>(attribute.arrays)[idx] = std::get<Float64>(attribute.null_values);
438 break;
439
440 case AttributeUnderlyingType::utDecimal32:
441 std::get<ContainerPtrType<Decimal32>>(attribute.arrays)[idx] = std::get<Decimal32>(attribute.null_values);
442 break;
443 case AttributeUnderlyingType::utDecimal64:
444 std::get<ContainerPtrType<Decimal64>>(attribute.arrays)[idx] = std::get<Decimal64>(attribute.null_values);
445 break;
446 case AttributeUnderlyingType::utDecimal128:
447 std::get<ContainerPtrType<Decimal128>>(attribute.arrays)[idx] = std::get<Decimal128>(attribute.null_values);
448 break;
449
450 case AttributeUnderlyingType::utString:
451 {
452 const auto & null_value_ref = std::get<String>(attribute.null_values);
453 auto & string_ref = std::get<ContainerPtrType<StringRef>>(attribute.arrays)[idx];
454
455 if (string_ref.data != null_value_ref.data())
456 {
457 if (string_ref.data)
458 string_arena->free(const_cast<char *>(string_ref.data), string_ref.size);
459
460 string_ref = StringRef{null_value_ref};
461 }
462
463 break;
464 }
465 }
466}
467
468void CacheDictionary::setAttributeValue(Attribute & attribute, const Key idx, const Field & value) const
469{
470 switch (attribute.type)
471 {
472 case AttributeUnderlyingType::utUInt8:
473 std::get<ContainerPtrType<UInt8>>(attribute.arrays)[idx] = value.get<UInt64>();
474 break;
475 case AttributeUnderlyingType::utUInt16:
476 std::get<ContainerPtrType<UInt16>>(attribute.arrays)[idx] = value.get<UInt64>();
477 break;
478 case AttributeUnderlyingType::utUInt32:
479 std::get<ContainerPtrType<UInt32>>(attribute.arrays)[idx] = value.get<UInt64>();
480 break;
481 case AttributeUnderlyingType::utUInt64:
482 std::get<ContainerPtrType<UInt64>>(attribute.arrays)[idx] = value.get<UInt64>();
483 break;
484 case AttributeUnderlyingType::utUInt128:
485 std::get<ContainerPtrType<UInt128>>(attribute.arrays)[idx] = value.get<UInt128>();
486 break;
487 case AttributeUnderlyingType::utInt8:
488 std::get<ContainerPtrType<Int8>>(attribute.arrays)[idx] = value.get<Int64>();
489 break;
490 case AttributeUnderlyingType::utInt16:
491 std::get<ContainerPtrType<Int16>>(attribute.arrays)[idx] = value.get<Int64>();
492 break;
493 case AttributeUnderlyingType::utInt32:
494 std::get<ContainerPtrType<Int32>>(attribute.arrays)[idx] = value.get<Int64>();
495 break;
496 case AttributeUnderlyingType::utInt64:
497 std::get<ContainerPtrType<Int64>>(attribute.arrays)[idx] = value.get<Int64>();
498 break;
499 case AttributeUnderlyingType::utFloat32:
500 std::get<ContainerPtrType<Float32>>(attribute.arrays)[idx] = value.get<Float64>();
501 break;
502 case AttributeUnderlyingType::utFloat64:
503 std::get<ContainerPtrType<Float64>>(attribute.arrays)[idx] = value.get<Float64>();
504 break;
505
506 case AttributeUnderlyingType::utDecimal32:
507 std::get<ContainerPtrType<Decimal32>>(attribute.arrays)[idx] = value.get<Decimal32>();
508 break;
509 case AttributeUnderlyingType::utDecimal64:
510 std::get<ContainerPtrType<Decimal64>>(attribute.arrays)[idx] = value.get<Decimal64>();
511 break;
512 case AttributeUnderlyingType::utDecimal128:
513 std::get<ContainerPtrType<Decimal128>>(attribute.arrays)[idx] = value.get<Decimal128>();
514 break;
515
516 case AttributeUnderlyingType::utString:
517 {
518 const auto & string = value.get<String>();
519 auto & string_ref = std::get<ContainerPtrType<StringRef>>(attribute.arrays)[idx];
520 const auto & null_value_ref = std::get<String>(attribute.null_values);
521
522 /// free memory unless it points to a null_value
523 if (string_ref.data && string_ref.data != null_value_ref.data())
524 string_arena->free(const_cast<char *>(string_ref.data), string_ref.size);
525
526 const auto str_size = string.size();
527 if (str_size != 0)
528 {
529 auto string_ptr = string_arena->alloc(str_size + 1);
530 std::copy(string.data(), string.data() + str_size + 1, string_ptr);
531 string_ref = StringRef{string_ptr, str_size};
532 }
533 else
534 string_ref = {};
535
536 break;
537 }
538 }
539}
540
541CacheDictionary::Attribute & CacheDictionary::getAttribute(const std::string & attribute_name) const
542{
543 const auto it = attribute_index_by_name.find(attribute_name);
544 if (it == std::end(attribute_index_by_name))
545 throw Exception{full_name + ": no such attribute '" + attribute_name + "'", ErrorCodes::BAD_ARGUMENTS};
546
547 return attributes[it->second];
548}
549
550bool CacheDictionary::isEmptyCell(const UInt64 idx) const
551{
552 return (idx != zero_cell_idx && cells[idx].id == 0)
553 || (cells[idx].data == ext::safe_bit_cast<CellMetadata::time_point_urep_t>(CellMetadata::time_point_t()));
554}
555
556PaddedPODArray<CacheDictionary::Key> CacheDictionary::getCachedIds() const
557{
558 const ProfilingScopedReadRWLock read_lock{rw_lock, ProfileEvents::DictCacheLockReadNs};
559
560 PaddedPODArray<Key> array;
561 for (size_t idx = 0; idx < cells.size(); ++idx)
562 {
563 auto & cell = cells[idx];
564 if (!isEmptyCell(idx) && !cells[idx].isDefault())
565 {
566 array.push_back(cell.id);
567 }
568 }
569 return array;
570}
571
572BlockInputStreamPtr CacheDictionary::getBlockInputStream(const Names & column_names, size_t max_block_size) const
573{
574 using BlockInputStreamType = DictionaryBlockInputStream<CacheDictionary, Key>;
575 return std::make_shared<BlockInputStreamType>(shared_from_this(), max_block_size, getCachedIds(), column_names);
576}
577
578std::exception_ptr CacheDictionary::getLastException() const
579{
580 const ProfilingScopedReadRWLock read_lock{rw_lock, ProfileEvents::DictCacheLockReadNs};
581 return last_exception;
582}
583
584void registerDictionaryCache(DictionaryFactory & factory)
585{
586 auto create_layout = [=](const std::string & full_name,
587 const DictionaryStructure & dict_struct,
588 const Poco::Util::AbstractConfiguration & config,
589 const std::string & config_prefix,
590 DictionarySourcePtr source_ptr) -> DictionaryPtr
591 {
592 if (dict_struct.key)
593 throw Exception{"'key' is not supported for dictionary of layout 'cache'", ErrorCodes::UNSUPPORTED_METHOD};
594
595 if (dict_struct.range_min || dict_struct.range_max)
596 throw Exception{full_name
597 + ": elements .structure.range_min and .structure.range_max should be defined only "
598 "for a dictionary of layout 'range_hashed'",
599 ErrorCodes::BAD_ARGUMENTS};
600 const auto & layout_prefix = config_prefix + ".layout";
601 const auto size = config.getInt(layout_prefix + ".cache.size_in_cells");
602 if (size == 0)
603 throw Exception{full_name + ": dictionary of layout 'cache' cannot have 0 cells", ErrorCodes::TOO_SMALL_BUFFER_SIZE};
604
605 const bool require_nonempty = config.getBool(config_prefix + ".require_nonempty", false);
606 if (require_nonempty)
607 throw Exception{full_name + ": dictionary of layout 'cache' cannot have 'require_nonempty' attribute set",
608 ErrorCodes::BAD_ARGUMENTS};
609
610 const String database = config.getString(config_prefix + ".database", "");
611 const String name = config.getString(config_prefix + ".name");
612 const DictionaryLifetime dict_lifetime{config, config_prefix + ".lifetime"};
613 return std::make_unique<CacheDictionary>(database, name, dict_struct, std::move(source_ptr), dict_lifetime, size);
614 };
615 factory.registerLayout("cache", create_layout, false);
616}
617
618
619}
620