1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
#include <Processors/Transforms/DistinctTransform.h>
namespace DB
{
namespace ErrorCodes
{
extern const int SET_SIZE_LIMIT_EXCEEDED;
}
DistinctTransform::DistinctTransform(
const Block & header_,
const SizeLimits & set_size_limits_,
const UInt64 limit_hint_,
const Names & columns_)
: ISimpleTransform(header_, header_, true)
, limit_hint(limit_hint_)
, set_size_limits(set_size_limits_)
{
const size_t num_columns = columns_.empty() ? header_.columns() : columns_.size();
key_columns_pos.reserve(num_columns);
for (size_t i = 0; i < num_columns; ++i)
{
const auto pos = columns_.empty() ? i : header_.getPositionByName(columns_[i]);
const auto & col = header_.getByPosition(pos).column;
if (col && !isColumnConst(*col))
key_columns_pos.emplace_back(pos);
}
}
template <typename Method>
void DistinctTransform::buildFilter(
Method & method,
const ColumnRawPtrs & columns,
IColumn::Filter & filter,
const size_t rows,
SetVariants & variants) const
{
typename Method::State state(columns, key_sizes, nullptr);
for (size_t i = 0; i < rows; ++i)
{
auto emplace_result = state.emplaceKey(method.data, i, variants.string_pool);
/// Emit the record if there is no such key in the current set yet.
/// Skip it otherwise.
filter[i] = emplace_result.isInserted();
}
}
void DistinctTransform::transform(Chunk & chunk)
{
if (unlikely(!chunk.hasRows()))
return;
/// Convert to full column, because SetVariant for sparse column is not implemented.
convertToFullIfSparse(chunk);
const auto num_rows = chunk.getNumRows();
auto columns = chunk.detachColumns();
/// Special case, - only const columns, return single row
if (unlikely(key_columns_pos.empty()))
{
for (auto & column : columns)
column = column->cut(0, 1);
chunk.setColumns(std::move(columns), 1);
stopReading();
return;
}
ColumnRawPtrs column_ptrs;
column_ptrs.reserve(key_columns_pos.size());
for (auto pos : key_columns_pos)
column_ptrs.emplace_back(columns[pos].get());
if (data.empty())
data.init(SetVariants::chooseMethod(column_ptrs, key_sizes));
const auto old_set_size = data.getTotalRowCount();
IColumn::Filter filter(num_rows);
switch (data.type)
{
case SetVariants::Type::EMPTY:
break;
#define M(NAME) \
case SetVariants::Type::NAME: \
buildFilter(*data.NAME, column_ptrs, filter, num_rows, data); \
break;
APPLY_FOR_SET_VARIANTS(M)
#undef M
}
/// Just go to the next chunk if there isn't any new record in the current one.
size_t new_set_size = data.getTotalRowCount();
if (new_set_size == old_set_size)
return;
if (!set_size_limits.check(new_set_size, data.getTotalByteCount(), "DISTINCT", ErrorCodes::SET_SIZE_LIMIT_EXCEEDED))
return;
for (auto & column : columns)
column = column->filter(filter, -1);
chunk.setColumns(std::move(columns), new_set_size - old_set_size);
/// Stop reading if we already reach the limit
if (limit_hint && new_set_size >= limit_hint)
{
stopReading();
return;
}
}
}
|
