2015-01-21 04:17:02 +00:00
|
|
|
#include <DB/DataStreams/DistinctBlockInputStream.h>
|
|
|
|
|
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
|
2016-01-11 21:46:36 +00:00
|
|
|
namespace ErrorCodes
|
|
|
|
{
|
|
|
|
extern const int SET_SIZE_LIMIT_EXCEEDED;
|
|
|
|
}
|
|
|
|
|
2015-01-21 04:17:02 +00:00
|
|
|
DistinctBlockInputStream::DistinctBlockInputStream(BlockInputStreamPtr input_, const Limits & limits, size_t limit_, Names columns_)
|
|
|
|
: columns_names(columns_),
|
|
|
|
limit(limit_),
|
|
|
|
max_rows(limits.max_rows_in_distinct),
|
|
|
|
max_bytes(limits.max_bytes_in_distinct),
|
|
|
|
overflow_mode(limits.distinct_overflow_mode)
|
|
|
|
{
|
|
|
|
children.push_back(input_);
|
|
|
|
}
|
|
|
|
|
2016-11-10 21:24:40 +00:00
|
|
|
String DistinctBlockInputStream::getID() const
|
|
|
|
{
|
|
|
|
std::stringstream res;
|
|
|
|
res << "Distinct(" << children.back()->getID() << ")";
|
|
|
|
return res.str();
|
|
|
|
}
|
2015-01-21 04:17:02 +00:00
|
|
|
|
|
|
|
Block DistinctBlockInputStream::readImpl()
|
|
|
|
{
|
|
|
|
/// Пока не встретится блок, после фильтрации которого что-нибудь останется, или поток не закончится.
|
|
|
|
while (1)
|
|
|
|
{
|
|
|
|
/// Если уже прочитали достаточно строк - то больше читать не будем.
|
2016-12-22 08:51:34 +00:00
|
|
|
if (limit && data.getTotalRowCount() >= limit)
|
2015-01-21 04:17:02 +00:00
|
|
|
return Block();
|
|
|
|
|
|
|
|
Block block = children[0]->read();
|
|
|
|
|
|
|
|
if (!block)
|
|
|
|
return Block();
|
|
|
|
|
|
|
|
size_t rows = block.rows();
|
|
|
|
size_t columns = columns_names.empty() ? block.columns() : columns_names.size();
|
|
|
|
|
2015-05-15 00:20:25 +00:00
|
|
|
ConstColumnPlainPtrs column_ptrs;
|
|
|
|
column_ptrs.reserve(columns);
|
2015-01-21 04:17:02 +00:00
|
|
|
|
|
|
|
for (size_t i = 0; i < columns; ++i)
|
|
|
|
{
|
2015-05-15 00:20:25 +00:00
|
|
|
auto & column = columns_names.empty()
|
|
|
|
? block.getByPosition(i).column
|
|
|
|
: block.getByName(columns_names[i]).column;
|
|
|
|
|
|
|
|
/// Игнорируем все константные столбцы.
|
|
|
|
if (!column->isConst())
|
|
|
|
column_ptrs.emplace_back(column.get());
|
2015-01-21 04:17:02 +00:00
|
|
|
}
|
|
|
|
|
2015-05-15 00:20:25 +00:00
|
|
|
columns = column_ptrs.size();
|
|
|
|
|
2016-12-22 08:51:34 +00:00
|
|
|
if (data.empty())
|
|
|
|
data.init(SetVariants::chooseMethod(column_ptrs, key_sizes));
|
|
|
|
|
2015-01-21 04:17:02 +00:00
|
|
|
/// Будем фильтровать блок, оставляя там только строки, которых мы ещё не видели.
|
|
|
|
IColumn::Filter filter(rows);
|
|
|
|
|
2016-12-22 08:51:34 +00:00
|
|
|
size_t old_set_size = data.getTotalRowCount();
|
2015-01-21 04:17:02 +00:00
|
|
|
|
2016-12-22 08:51:34 +00:00
|
|
|
switch (data.type)
|
2015-01-21 04:17:02 +00:00
|
|
|
{
|
2016-12-22 08:51:34 +00:00
|
|
|
case SetVariants::Type::EMPTY:
|
2015-01-21 04:17:02 +00:00
|
|
|
break;
|
2016-12-22 08:51:34 +00:00
|
|
|
#define M(NAME) \
|
|
|
|
case SetVariants::Type::NAME: \
|
|
|
|
{ executeImpl(*data.NAME, column_ptrs, filter, rows); } \
|
|
|
|
break;
|
|
|
|
APPLY_FOR_SET_VARIANTS(M)
|
|
|
|
#undef M
|
2015-01-21 04:17:02 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
/// Если ни одной новой строки не было в блоке - перейдём к следующему блоку.
|
2016-12-22 08:51:34 +00:00
|
|
|
if (data.getTotalRowCount() == old_set_size)
|
2015-01-21 04:17:02 +00:00
|
|
|
continue;
|
|
|
|
|
|
|
|
if (!checkLimits())
|
|
|
|
{
|
|
|
|
if (overflow_mode == OverflowMode::THROW)
|
|
|
|
throw Exception("DISTINCT-Set size limit exceeded."
|
2016-12-22 08:51:34 +00:00
|
|
|
" Rows: " + toString(data.getTotalRowCount()) +
|
2015-01-21 04:17:02 +00:00
|
|
|
", limit: " + toString(max_rows) +
|
2016-12-22 08:51:34 +00:00
|
|
|
". Bytes: " + toString(data.getTotalByteCount()) +
|
2015-01-21 04:17:02 +00:00
|
|
|
", limit: " + toString(max_bytes) + ".",
|
|
|
|
ErrorCodes::SET_SIZE_LIMIT_EXCEEDED);
|
|
|
|
|
|
|
|
if (overflow_mode == OverflowMode::BREAK)
|
|
|
|
return Block();
|
|
|
|
|
|
|
|
throw Exception("Logical error: unknown overflow mode", ErrorCodes::LOGICAL_ERROR);
|
|
|
|
}
|
|
|
|
|
|
|
|
size_t all_columns = block.columns();
|
|
|
|
for (size_t i = 0; i < all_columns; ++i)
|
2015-12-05 07:01:18 +00:00
|
|
|
block.getByPosition(i).column = block.getByPosition(i).column->filter(filter, -1);
|
2015-01-21 04:17:02 +00:00
|
|
|
|
|
|
|
return block;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-12-22 08:51:34 +00:00
|
|
|
template <typename Method>
|
|
|
|
void DistinctBlockInputStream::executeImpl(
|
|
|
|
Method & method,
|
|
|
|
const ConstColumnPlainPtrs & columns,
|
|
|
|
IColumn::Filter & filter,
|
|
|
|
size_t rows) const
|
|
|
|
{
|
|
|
|
typename Method::State state;
|
|
|
|
state.init(columns);
|
|
|
|
size_t keys_size = columns.size();
|
|
|
|
|
|
|
|
/// Для всех строчек
|
|
|
|
for (size_t i = 0; i < rows; ++i)
|
|
|
|
{
|
|
|
|
/// Строим ключ
|
|
|
|
typename Method::Key key = state.getKey(columns, keys_size, i, key_sizes);
|
|
|
|
|
|
|
|
/// Если вставилось в множество - строчку оставляем, иначе - удаляем.
|
|
|
|
filter[i] = method.data.insert(key).second;
|
|
|
|
|
|
|
|
if (limit && data.getTotalRowCount() == limit)
|
|
|
|
{
|
|
|
|
memset(&filter[i + 1], 0, (rows - (i + 1)) * sizeof(IColumn::Filter::value_type));
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-11-10 21:24:40 +00:00
|
|
|
bool DistinctBlockInputStream::checkLimits() const
|
|
|
|
{
|
2016-12-22 08:51:34 +00:00
|
|
|
if (max_rows && data.getTotalRowCount() > max_rows)
|
2016-11-10 21:24:40 +00:00
|
|
|
return false;
|
2016-12-22 08:51:34 +00:00
|
|
|
if (max_bytes && data.getTotalByteCount() > max_bytes)
|
2016-11-10 21:24:40 +00:00
|
|
|
return false;
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
|
2015-01-21 04:17:02 +00:00
|
|
|
}
|