1#include <Columns/ColumnConst.h>
2#include <Columns/ColumnVector.h>
3#include <Common/typeid_cast.h>
4#include <Common/assert_cast.h>
5
6#include <type_traits>
7
8#ifdef __SSE2__
9 #define LIBDIVIDE_USE_SSE2 1
10#endif
11
12#include <libdivide.h>
13
14
15namespace DB
16{
17
18template <typename T>
19IColumn::Selector createBlockSelector(
20 const IColumn & column,
21 const std::vector<UInt64> & slots)
22{
23 const auto total_weight = slots.size();
24 size_t num_rows = column.size();
25 IColumn::Selector selector(num_rows);
26
27 /** Modulo of division of negative numbers to positive number in C++11 is negative (so called truncated division).
28 * This is not suitable for our task. So we will process signed numbers as unsigned.
29 * It is not near like remainder of division, but is suitable for our task.
30 */
31 using UnsignedT = std::make_unsigned_t<T>;
32
33 /// const columns contain only one value, therefore we do not need to read it at every iteration
34 if (isColumnConst(column))
35 {
36 const auto data = assert_cast<const ColumnConst &>(column).getValue<T>();
37 const auto shard_num = slots[static_cast<UnsignedT>(data) % total_weight];
38 selector.assign(num_rows, shard_num);
39 }
40 else
41 {
42 /// libdivide support only UInt32 and UInt64.
43 using TUInt32Or64 = std::conditional_t<sizeof(UnsignedT) <= 4, UInt32, UInt64>;
44
45 libdivide::divider<TUInt32Or64> divider(total_weight);
46
47 const auto & data = typeid_cast<const ColumnVector<T> &>(column).getData();
48
49 for (size_t i = 0; i < num_rows; ++i)
50 selector[i] = slots[static_cast<TUInt32Or64>(data[i]) - (static_cast<TUInt32Or64>(data[i]) / divider) * total_weight];
51 }
52
53 return selector;
54}
55
56
57/// Explicit instantiations to avoid code bloat in headers.
58template IColumn::Selector createBlockSelector<UInt8>(const IColumn & column, const std::vector<UInt64> & slots);
59template IColumn::Selector createBlockSelector<UInt16>(const IColumn & column, const std::vector<UInt64> & slots);
60template IColumn::Selector createBlockSelector<UInt32>(const IColumn & column, const std::vector<UInt64> & slots);
61template IColumn::Selector createBlockSelector<UInt64>(const IColumn & column, const std::vector<UInt64> & slots);
62template IColumn::Selector createBlockSelector<Int8>(const IColumn & column, const std::vector<UInt64> & slots);
63template IColumn::Selector createBlockSelector<Int16>(const IColumn & column, const std::vector<UInt64> & slots);
64template IColumn::Selector createBlockSelector<Int32>(const IColumn & column, const std::vector<UInt64> & slots);
65template IColumn::Selector createBlockSelector<Int64>(const IColumn & column, const std::vector<UInt64> & slots);
66
67}
68