2013-10-01 18:09:31 +00:00
|
|
|
|
#pragma once
|
|
|
|
|
|
|
|
|
|
#include <Yandex/logger_useful.h>
|
|
|
|
|
|
|
|
|
|
#include <DB/Core/Row.h>
|
|
|
|
|
#include <DB/Core/ColumnNumbers.h>
|
|
|
|
|
#include <DB/DataStreams/MergingSortedBlockInputStream.h>
|
2015-04-08 16:38:38 +00:00
|
|
|
|
#include <DB/Storages/MergeTree/PKCondition.h>
|
2015-04-09 17:01:02 +00:00
|
|
|
|
#include <statdaemons/ext/range.hpp>
|
|
|
|
|
#include <statdaemons/ext/map.hpp>
|
2013-10-01 18:09:31 +00:00
|
|
|
|
|
|
|
|
|
|
|
|
|
|
namespace DB
|
|
|
|
|
{
|
|
|
|
|
|
|
|
|
|
/** Соединяет несколько сортированных потоков в один.
|
|
|
|
|
* При этом, для каждой группы идущих подряд одинаковых значений первичного ключа (столбцов, по которым сортируются данные),
|
|
|
|
|
* схлопывает их в одну строку, суммируя все числовые столбцы кроме первичного ключа.
|
2013-11-30 18:43:59 +00:00
|
|
|
|
* Если во всех числовых столбцах кроме первичного ключа получился ноль, то удаляет строчку.
|
2013-10-01 18:09:31 +00:00
|
|
|
|
*/
|
|
|
|
|
class SummingSortedBlockInputStream : public MergingSortedBlockInputStream
|
|
|
|
|
{
|
|
|
|
|
public:
|
2014-11-22 02:22:30 +00:00
|
|
|
|
SummingSortedBlockInputStream(BlockInputStreams inputs_,
|
|
|
|
|
const SortDescription & description_,
|
|
|
|
|
/// Список столбцов, которых нужно суммировать. Если пустое - берутся все числовые столбцы, не входящие в description.
|
|
|
|
|
const Names & column_names_to_sum_,
|
|
|
|
|
size_t max_block_size_)
|
|
|
|
|
: MergingSortedBlockInputStream(inputs_, description_, max_block_size_), column_names_to_sum(column_names_to_sum_)
|
2013-10-01 18:09:31 +00:00
|
|
|
|
{
|
|
|
|
|
}
|
|
|
|
|
|
2014-11-08 23:52:18 +00:00
|
|
|
|
String getName() const override { return "SummingSortedBlockInputStream"; }
|
2013-10-01 18:09:31 +00:00
|
|
|
|
|
2014-11-08 23:52:18 +00:00
|
|
|
|
String getID() const override
|
2013-10-01 18:09:31 +00:00
|
|
|
|
{
|
|
|
|
|
std::stringstream res;
|
|
|
|
|
res << "SummingSorted(inputs";
|
|
|
|
|
|
|
|
|
|
for (size_t i = 0; i < children.size(); ++i)
|
|
|
|
|
res << ", " << children[i]->getID();
|
|
|
|
|
|
|
|
|
|
res << ", description";
|
|
|
|
|
|
|
|
|
|
for (size_t i = 0; i < description.size(); ++i)
|
|
|
|
|
res << ", " << description[i].getID();
|
|
|
|
|
|
|
|
|
|
res << ")";
|
|
|
|
|
return res.str();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
protected:
|
|
|
|
|
/// Может возвращаться на 1 больше записей, чем max_block_size.
|
2014-11-08 23:52:18 +00:00
|
|
|
|
Block readImpl() override;
|
2013-10-01 18:09:31 +00:00
|
|
|
|
|
|
|
|
|
private:
|
2014-11-22 02:22:30 +00:00
|
|
|
|
Logger * log = &Logger::get("SummingSortedBlockInputStream");
|
2013-10-01 18:09:31 +00:00
|
|
|
|
|
2015-01-18 08:25:56 +00:00
|
|
|
|
/// Прочитали до конца.
|
|
|
|
|
bool finished = false;
|
|
|
|
|
|
2013-10-01 18:09:31 +00:00
|
|
|
|
/// Столбцы с какими номерами надо суммировать.
|
2014-11-22 02:22:30 +00:00
|
|
|
|
Names column_names_to_sum; /// Если задано - преобразуется в column_numbers_to_sum при инициализации.
|
2013-10-01 18:09:31 +00:00
|
|
|
|
ColumnNumbers column_numbers_to_sum;
|
|
|
|
|
|
2015-04-10 10:02:57 +00:00
|
|
|
|
/** Таблица может иметь вложенные таблицы, обрабатываемые особым образом.
|
|
|
|
|
* Если название вложенной таблицы заканчинвается на `Map` и она содержит не менее двух столбцов,
|
2015-04-09 11:34:40 +00:00
|
|
|
|
* удовлетворяющих следующим критериям:
|
|
|
|
|
* - первый столбец - числовой ((U)IntN, Date, DateTime), назовем его условно key,
|
2015-04-10 10:02:57 +00:00
|
|
|
|
* - остальные столбцы - арифметические ((U)IntN, Float32/64), условно (values...).
|
|
|
|
|
* Такая вложенная таблица воспринимается как отображение key => (values...) и при слиянии
|
|
|
|
|
* ее строк выполняется слияние элементов двух множеств по key со сложением соответствующих (values...).
|
2015-04-09 11:34:40 +00:00
|
|
|
|
* Пример:
|
|
|
|
|
* [(1, 100)] + [(2, 150)] -> [(1, 100), (2, 150)]
|
|
|
|
|
* [(1, 100)] + [(1, 150)] -> [(1, 250)]
|
|
|
|
|
* [(1, 100)] + [(1, 150), (2, 150)] -> [(1, 250), (2, 150)]
|
|
|
|
|
* [(1, 100), (2, 150)] + [(1, -100)] -> [(2, 150)]
|
|
|
|
|
*/
|
|
|
|
|
|
2015-04-10 10:02:57 +00:00
|
|
|
|
/// Хранит номера столбца-ключа и столбцов-значений
|
2015-04-08 16:38:38 +00:00
|
|
|
|
struct map_description
|
|
|
|
|
{
|
|
|
|
|
std::size_t key_col_num;
|
2015-04-09 17:01:02 +00:00
|
|
|
|
std::vector<std::size_t> val_col_nums;
|
2015-04-08 16:38:38 +00:00
|
|
|
|
};
|
|
|
|
|
|
2015-04-09 11:34:40 +00:00
|
|
|
|
/// Найденные вложенные Map таблицы
|
2015-04-08 16:38:38 +00:00
|
|
|
|
std::vector<map_description> maps_to_sum;
|
|
|
|
|
|
2013-10-01 18:09:31 +00:00
|
|
|
|
Row current_key; /// Текущий первичный ключ.
|
|
|
|
|
Row next_key; /// Первичный ключ следующей строки.
|
2014-11-08 23:52:18 +00:00
|
|
|
|
|
2013-10-01 18:09:31 +00:00
|
|
|
|
Row current_row;
|
2014-11-22 02:22:30 +00:00
|
|
|
|
bool current_row_is_zero = false; /// Текущая строчка просуммировалась в ноль, и её следует удалить.
|
2013-10-01 18:09:31 +00:00
|
|
|
|
|
2014-11-22 02:22:30 +00:00
|
|
|
|
bool output_is_non_empty = false; /// Отдали ли мы наружу хоть одну строку.
|
2014-05-07 11:44:14 +00:00
|
|
|
|
|
2013-10-01 18:09:31 +00:00
|
|
|
|
/** Делаем поддержку двух разных курсоров - с Collation и без.
|
|
|
|
|
* Шаблоны используем вместо полиморфных SortCursor'ов и вызовов виртуальных функций.
|
|
|
|
|
*/
|
|
|
|
|
template<class TSortCursor>
|
|
|
|
|
void merge(Block & merged_block, ColumnPlainPtrs & merged_columns, std::priority_queue<TSortCursor> & queue);
|
|
|
|
|
|
|
|
|
|
/// Вставить в результат просуммированную строку для текущей группы.
|
|
|
|
|
void insertCurrentRow(ColumnPlainPtrs & merged_columns);
|
|
|
|
|
|
|
|
|
|
|
2013-11-30 18:43:59 +00:00
|
|
|
|
/** Реализует операцию +=.
|
|
|
|
|
* Возвращает false, если результат получился нулевым.
|
|
|
|
|
*/
|
|
|
|
|
class FieldVisitorSum : public StaticVisitor<bool>
|
2013-10-01 18:09:31 +00:00
|
|
|
|
{
|
|
|
|
|
private:
|
|
|
|
|
const Field & rhs;
|
|
|
|
|
public:
|
|
|
|
|
FieldVisitorSum(const Field & rhs_) : rhs(rhs_) {}
|
|
|
|
|
|
2013-11-30 18:43:59 +00:00
|
|
|
|
bool operator() (UInt64 & x) const { x += get<UInt64>(rhs); return x != 0; }
|
|
|
|
|
bool operator() (Int64 & x) const { x += get<Int64>(rhs); return x != 0; }
|
|
|
|
|
bool operator() (Float64 & x) const { x += get<Float64>(rhs); return x != 0; }
|
2013-10-01 18:09:31 +00:00
|
|
|
|
|
2013-11-30 18:43:59 +00:00
|
|
|
|
bool operator() (Null & x) const { throw Exception("Cannot sum Nulls", ErrorCodes::LOGICAL_ERROR); }
|
|
|
|
|
bool operator() (String & x) const { throw Exception("Cannot sum Strings", ErrorCodes::LOGICAL_ERROR); }
|
|
|
|
|
bool operator() (Array & x) const { throw Exception("Cannot sum Arrays", ErrorCodes::LOGICAL_ERROR); }
|
2013-10-01 18:09:31 +00:00
|
|
|
|
};
|
|
|
|
|
|
2013-11-30 18:43:59 +00:00
|
|
|
|
/** Прибавить строчку под курсором к row.
|
2015-04-09 11:34:40 +00:00
|
|
|
|
* Для вложенных Map выполняется слияние по ключу с выбрасыванием нулевых элементов.
|
2013-11-30 18:43:59 +00:00
|
|
|
|
* Возвращает false, если результат получился нулевым.
|
|
|
|
|
*/
|
2013-10-01 18:09:31 +00:00
|
|
|
|
template<class TSortCursor>
|
2013-11-30 18:43:59 +00:00
|
|
|
|
bool addRow(Row & row, TSortCursor & cursor)
|
2013-10-01 18:09:31 +00:00
|
|
|
|
{
|
2015-04-09 11:34:40 +00:00
|
|
|
|
/// merge nested maps
|
2015-04-08 16:38:38 +00:00
|
|
|
|
for (const auto & map : maps_to_sum)
|
|
|
|
|
{
|
2015-04-09 17:01:02 +00:00
|
|
|
|
const auto value_count = map.val_col_nums.size();
|
|
|
|
|
|
2015-04-09 11:34:40 +00:00
|
|
|
|
/// fetch key and val array references from accumulator-row
|
2015-04-08 16:38:38 +00:00
|
|
|
|
auto & key_array = row[map.key_col_num].get<Array>();
|
2015-04-09 17:01:02 +00:00
|
|
|
|
|
|
|
|
|
std::map<Field, std::vector<Field>> result;
|
|
|
|
|
|
|
|
|
|
/// populate map from current row
|
|
|
|
|
for (const auto i : ext::range(0, key_array.size()))
|
|
|
|
|
{
|
|
|
|
|
const auto it = result.find(key_array[i]);
|
|
|
|
|
/// row for such key is not key present in the map, emplace
|
|
|
|
|
if (it == std::end(result))
|
|
|
|
|
{
|
|
|
|
|
/// compose a vector of i-th elements of each one of `map.val_col_nums`
|
|
|
|
|
result.emplace(key_array[i],
|
|
|
|
|
ext::map<std::vector>(map.val_col_nums, [&] (const auto col_num) -> decltype(auto) {
|
|
|
|
|
return row[col_num].get<Array>()[i];
|
|
|
|
|
}));
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
|
|
|
|
/// row for requested key found, merge corresponding values
|
|
|
|
|
for (const auto val_col_index : ext::range(0, value_count))
|
|
|
|
|
{
|
|
|
|
|
const auto col_num = map.val_col_nums[val_col_index];
|
|
|
|
|
apply_visitor(FieldVisitorSum{row[col_num].get<Array>()[i]}, it->second[val_col_index]);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
2015-04-08 16:38:38 +00:00
|
|
|
|
|
2015-04-09 11:34:40 +00:00
|
|
|
|
/// copy key and value fields from current row under cursor
|
2015-04-08 16:38:38 +00:00
|
|
|
|
const auto key_field_rhs = (*cursor->all_columns[map.key_col_num])[cursor->pos];
|
2015-04-09 17:01:02 +00:00
|
|
|
|
/// for each element of `map.val_col_nums` get corresponding array under cursor and put into vector
|
|
|
|
|
const auto val_fields_rhs = ext::map<std::vector>(map.val_col_nums,
|
|
|
|
|
[&] (const auto col_num) -> decltype(auto) {
|
|
|
|
|
return (*cursor->all_columns[col_num])[cursor->pos];
|
|
|
|
|
});
|
2015-04-08 16:38:38 +00:00
|
|
|
|
|
2015-04-09 17:01:02 +00:00
|
|
|
|
/// fetch key array reference from current row
|
2015-04-08 16:57:01 +00:00
|
|
|
|
const auto & key_array_rhs = key_field_rhs.get<Array>();
|
2015-04-08 16:38:38 +00:00
|
|
|
|
|
2015-04-09 11:34:40 +00:00
|
|
|
|
/// merge current row into map
|
2015-04-08 16:38:38 +00:00
|
|
|
|
for (const auto i : ext::range(0, key_array_rhs.size()))
|
2015-04-09 17:01:02 +00:00
|
|
|
|
{
|
|
|
|
|
const auto it = result.find(key_array_rhs[i]);
|
|
|
|
|
/// row for such key is not key present in the map, emplace
|
|
|
|
|
if (it == std::end(result))
|
|
|
|
|
{
|
|
|
|
|
/// compose a vector of i-th elements of each one of `map.val_col_nums`
|
|
|
|
|
result.emplace(key_array_rhs[i],
|
|
|
|
|
ext::map<std::vector>(val_fields_rhs, [&] (const auto & val_field) -> decltype(auto) {
|
|
|
|
|
return val_field.get<Array>()[i];
|
|
|
|
|
}));
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
|
|
|
|
/// row for requested key found, merge corresponding values
|
|
|
|
|
for (const auto val_col_index : ext::range(0, value_count))
|
|
|
|
|
{
|
|
|
|
|
apply_visitor(FieldVisitorSum{val_fields_rhs[val_col_index].get<Array>()[i]},
|
|
|
|
|
it->second[val_col_index]);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
2015-04-08 16:38:38 +00:00
|
|
|
|
|
|
|
|
|
Array key_array_result;
|
2015-04-09 17:01:02 +00:00
|
|
|
|
std::vector<Array> val_arrays_result(value_count);
|
2015-04-08 16:38:38 +00:00
|
|
|
|
|
2015-04-09 11:34:40 +00:00
|
|
|
|
/// serialize result key-value pairs back into separate arrays of keys and values
|
2015-04-09 17:01:02 +00:00
|
|
|
|
const auto zero_field = nearestFieldType(0);
|
|
|
|
|
for (const auto & key_values_pair : result)
|
2015-04-08 16:38:38 +00:00
|
|
|
|
{
|
2015-04-09 17:01:02 +00:00
|
|
|
|
auto only_zeroes = true;
|
|
|
|
|
|
|
|
|
|
for (const auto & value : key_values_pair.second)
|
|
|
|
|
if (!apply_visitor(FieldVisitorAccurateEquals{}, value, zero_field))
|
|
|
|
|
only_zeroes = false;
|
|
|
|
|
|
|
|
|
|
if (only_zeroes)
|
|
|
|
|
continue;
|
|
|
|
|
|
|
|
|
|
key_array_result.emplace_back(key_values_pair.first);
|
|
|
|
|
for (const auto val_col_index : ext::range(0, value_count))
|
|
|
|
|
val_arrays_result[val_col_index].emplace_back(key_values_pair.second[val_col_index]);
|
2015-04-08 16:38:38 +00:00
|
|
|
|
}
|
|
|
|
|
|
2015-04-09 11:34:40 +00:00
|
|
|
|
/// replace accumulator row key and value arrays with the merged ones
|
2015-04-08 16:38:38 +00:00
|
|
|
|
key_array = std::move(key_array_result);
|
2015-04-09 17:01:02 +00:00
|
|
|
|
for (const auto val_col_index : ext::range(0, value_count))
|
|
|
|
|
row[map.val_col_nums[val_col_index]].get<Array>() = std::move(val_arrays_result[val_col_index]);
|
2015-04-08 16:38:38 +00:00
|
|
|
|
}
|
|
|
|
|
|
2013-11-30 18:43:59 +00:00
|
|
|
|
bool res = false; /// Есть ли хотя бы одно ненулевое число.
|
2014-11-08 23:52:18 +00:00
|
|
|
|
|
2013-10-01 18:09:31 +00:00
|
|
|
|
for (size_t i = 0, size = column_numbers_to_sum.size(); i < size; ++i)
|
|
|
|
|
{
|
|
|
|
|
size_t j = column_numbers_to_sum[i];
|
2013-11-30 18:43:59 +00:00
|
|
|
|
if (apply_visitor(FieldVisitorSum((*cursor->all_columns[j])[cursor->pos]), row[j]))
|
|
|
|
|
res = true;
|
2013-10-01 18:09:31 +00:00
|
|
|
|
}
|
2013-11-30 18:43:59 +00:00
|
|
|
|
|
|
|
|
|
return res;
|
2013-10-01 18:09:31 +00:00
|
|
|
|
}
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
}
|