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

80 lines
3.8 KiB
C
Raw Normal View History

2012-08-14 20:35:44 +00:00
#pragma once
#include <Yandex/logger_useful.h>
2012-08-16 17:02:31 +00:00
#include <DB/Core/Row.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
* То есть - производит схлопывание записей из лога изменений.
*
* Если количество положительных и отрицательных строк совпадает - то пишет первую отрицательную и последнюю положительную строку.
* Если положительных на 1 больше, чем отрицательных - то пишет только последнюю положительную строку.
* Если отрицательных на 1 больше, чем положительных - то пишет только первую отрицательную строку.
* Иначе - логическая ошибка.
2012-08-14 20:35:44 +00:00
*/
class CollapsingSortedBlockInputStream : public MergingSortedBlockInputStream
{
public:
2012-08-16 17:02:31 +00:00
CollapsingSortedBlockInputStream(BlockInputStreams inputs_, 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_),
sign_column(sign_column_), sign_column_number(0),
2012-08-16 17:02:31 +00:00
log(&Logger::get("CollapsingSortedBlockInputStream")),
count_positive(0), count_negative(0)
2012-08-14 20:35:44 +00:00
{
}
String getName() const { return "CollapsingSortedBlockInputStream"; }
BlockInputStreamPtr clone() { return new CollapsingSortedBlockInputStream(inputs, description, sign_column, max_block_size); }
2012-08-14 20:35:44 +00:00
2012-10-20 02:10:47 +00:00
protected:
/// Может возвращаться на 1 больше записей, чем max_block_size.
Block readImpl();
2012-08-14 20:35:44 +00:00
private:
2012-08-16 17:02:31 +00:00
String sign_column;
size_t sign_column_number;
2012-08-14 20:35:44 +00:00
Logger * log;
2012-08-16 17:02:31 +00:00
Row current_key; /// Текущий первичный ключ.
Row next_key; /// Первичный ключ следующей строки.
2012-08-16 17:02:31 +00:00
Row first_negative; /// Первая отрицательная строка для текущего первичного ключа.
Row last_positive; /// Последняя положительная строка для текущего первичного ключа.
2012-08-16 17:02:31 +00:00
size_t count_positive; /// Количество положительных строк для текущего первичного ключа.
size_t count_negative; /// Количество отрицательных строк для текущего первичного ключа.
2012-08-16 17:02:31 +00:00
/// Сохранить строчку, на которую указывает cursor в row.
void setRow(Row & row, SortCursor & cursor)
{
for (size_t i = 0; i < num_columns; ++i)
cursor->all_columns[i]->get(cursor->pos, row[i]);
2012-08-16 17:02:31 +00:00
}
/// Сохранить первичный ключ, на который указывает cursor в row.
void setPrimaryKey(Row & row, SortCursor & cursor)
{
for (size_t i = 0; i < cursor->sort_columns_size; ++i)
cursor->sort_columns[i]->get(cursor->pos, row[i]);
}
2012-08-16 17:02:31 +00:00
/// Вставить в результат строки для текущего идентификатора "визита".
2012-08-16 20:07:13 +00:00
void insertRows(ColumnPlainPtrs & merged_columns, size_t & merged_rows);
2012-08-22 16:30:41 +00:00
void reportIncorrectData();
2012-08-14 20:35:44 +00:00
};
}