ClickHouse/dbms/include/DB/DataStreams/CollapsingSortedBlockInputStream.h

93 lines
4.7 KiB
C
Raw Normal View History

2012-08-14 20:35:44 +00:00
#pragma once
#include <Yandex/logger_useful.h>
2012-08-14 20:35:44 +00:00
#include <DB/DataStreams/MergingSortedBlockInputStream.h>
namespace DB
{
/** Соединяет несколько сортированных потоков в один.
* При этом, для каждой группы идущих подряд одинаковых значений первичного ключа (столбцов, по которым сортируются данные),
2012-08-16 22:11:53 +00:00
* оставляет не более одной строки со значением столбца sign_column = -1 ("отрицательной строки")
* и не более одиной строки со значением столбца sign_column = 1 ("положительной строки").
2012-08-16 17:02:31 +00:00
* То есть - производит схлопывание записей из лога изменений.
*
2013-10-30 08:50:58 +00:00
* Если количество положительных и отрицательных строк совпадает, и последняя строка положительная - то пишет первую отрицательную и последнюю положительную строку.
* Если количество положительных и отрицательных строк совпадает, и последняя строка отрицательная - то ничего не пишет.
2012-08-16 17:02:31 +00:00
* Если положительных на 1 больше, чем отрицательных - то пишет только последнюю положительную строку.
* Если отрицательных на 1 больше, чем положительных - то пишет только первую отрицательную строку.
* Иначе - логическая ошибка.
2012-08-14 20:35:44 +00:00
*/
class CollapsingSortedBlockInputStream : public MergingSortedBlockInputStream
{
public:
2014-03-14 17:03:52 +00:00
CollapsingSortedBlockInputStream(BlockInputStreams inputs_, const SortDescription & description_,
const String & sign_column_, size_t max_block_size_)
2012-08-16 17:02:31 +00:00
: MergingSortedBlockInputStream(inputs_, description_, max_block_size_),
2015-01-18 08:25:56 +00:00
sign_column(sign_column_)
2012-08-14 20:35:44 +00:00
{
}
String getName() const override { return "CollapsingSorted"; }
2012-08-14 20:35:44 +00:00
String getID() const override
{
std::stringstream res;
res << "CollapsingSorted(inputs";
2013-05-04 04:05:15 +00:00
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 << ", sign_column, " << sign_column << ")";
return res.str();
}
2012-10-20 02:10:47 +00:00
protected:
/// Может возвращаться на 1 больше записей, чем max_block_size.
Block readImpl() override;
2012-10-20 02:10:47 +00:00
2012-08-14 20:35:44 +00:00
private:
2012-08-16 17:02:31 +00:00
String sign_column;
2015-01-18 08:25:56 +00:00
size_t sign_column_number = 0;
2012-08-14 20:35:44 +00:00
2015-01-18 08:25:56 +00:00
Logger * log = &Logger::get("CollapsingSortedBlockInputStream");
/// Прочитали до конца.
bool finished = false;
2012-08-16 17:02:31 +00:00
Row current_key; /// Текущий первичный ключ.
Row next_key; /// Первичный ключ следующей строки.
Row first_negative; /// Первая отрицательная строка для текущего первичного ключа.
Row last_positive; /// Последняя положительная строка для текущего первичного ключа.
2013-10-30 08:50:58 +00:00
Row last_negative; /// Последняя отрицательная. Сорраняется только если ни одной строки в ответ еще не выписано.
2012-08-16 17:02:31 +00:00
2015-01-18 08:25:56 +00:00
size_t count_positive = 0; /// Количество положительных строк для текущего первичного ключа.
size_t count_negative = 0; /// Количество отрицательных строк для текущего первичного ключа.
bool last_is_positive = false; /// true, если последняя строка для текущего первичного ключа положительная.
2012-08-16 17:02:31 +00:00
2015-01-18 08:25:56 +00:00
size_t count_incorrect_data = 0; /// Чтобы не писать в лог слишком много сообщений об ошибке.
2015-01-18 08:25:56 +00:00
size_t blocks_written = 0;
2013-10-30 08:50:58 +00:00
2013-05-28 16:56:05 +00:00
/** Делаем поддержку двух разных курсоров - с Collation и без.
* Шаблоны используем вместо полиморфных SortCursor'ов и вызовов виртуальных функций.
*/
template<class TSortCursor>
void merge(ColumnPlainPtrs & merged_columns, std::priority_queue<TSortCursor> & queue);
2012-08-16 17:02:31 +00:00
/// Вставить в результат строки для текущего идентификатора "визита".
2013-10-30 08:50:58 +00:00
void insertRows(ColumnPlainPtrs & merged_columns, size_t & merged_rows, bool last_in_stream = false);
2012-08-22 16:30:41 +00:00
void reportIncorrectData();
2012-08-14 20:35:44 +00:00
};
}