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

270 lines
7.3 KiB
C
Raw Normal View History

#pragma once
#include <Yandex/logger_useful.h>
#include <DB/DataStreams/IProfilingBlockInputStream.h>
#include <DB/Core/SortDescription.h>
#include <DB/Columns/ColumnsNumber.h>
#include <queue>
namespace DB
{
/// Схлопывает одинаковые строки с противоположным знаком примерно как CollapsingSortedBlockInputStream.
/// Выдает строки в произвольном порядке (входные потоки по-прежнему должны быть упорядочены).
/// Выдает только строки с положительным знаком.
class CollapsingFinalBlockInputStream : public IProfilingBlockInputStream
{
public:
2014-03-14 17:03:52 +00:00
CollapsingFinalBlockInputStream(BlockInputStreams inputs_, const SortDescription & description_,
const String & sign_column_)
2013-05-04 04:05:15 +00:00
: description(description_), sign_column(sign_column_),
2014-04-04 17:20:45 +00:00
log(&Logger::get("CollapsingFinalBlockInputStream")),
first(true), count_positive(0), count_negative(0), count_incorrect_data(0), blocks_fetched(0), blocks_output(0)
2013-05-04 04:05:15 +00:00
{
children.insert(children.end(), inputs_.begin(), inputs_.end());
}
~CollapsingFinalBlockInputStream();
String getName() const override { return "CollapsingFinalBlockInputStream"; }
String getID() const override
{
std::stringstream res;
res << "CollapsingFinal(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();
}
protected:
Block readImpl() override;
private:
struct MergingBlock;
typedef std::vector<MergingBlock*> BlockPlainPtrs;
struct MergingBlock : boost::noncopyable
{
MergingBlock(Block block_,
size_t stream_index_,
SortDescription desc_,
String sign_column_name,
BlockPlainPtrs * output_blocks)
: block(block_), stream_index(stream_index_), desc(desc_), refcount(0), output_blocks(output_blocks)
{
sort_columns.resize(desc.size());
for (size_t i = 0; i < desc.size(); ++i)
{
size_t column_number = !desc[i].column_name.empty()
? block.getPositionByName(desc[i].column_name)
: desc[i].column_number;
sort_columns[i] = &*block.getByPosition(column_number).column;
}
const IColumn * sign_icolumn = &*block.getByName(sign_column_name).column;
sign_column = typeid_cast<const ColumnInt8 *>(sign_icolumn);
if (!sign_column)
throw Exception("Sign column must have type Int8", ErrorCodes::BAD_TYPE_OF_FIELD);
rows = sign_column->size();
/// Заполняется целиком нулями. Потом выставляются единички в позициях строчек, которых нужно оставить.
filter.resize_fill(rows);
}
Block block;
/// Строки с одинаковым ключом будут упорядочены по возрастанию stream_index.
size_t stream_index;
SortDescription desc;
size_t rows;
/// Какие строки нужно оставить. Заполняется при слиянии потоков.
IColumn::Filter filter;
/// Указывают в block.
ConstColumnPlainPtrs sort_columns;
const ColumnInt8 * sign_column;
/// Когда достигает нуля, блок можно выдавать в ответ.
int refcount;
/// Куда положить блок, когда он готов попасть в ответ.
BlockPlainPtrs * output_blocks;
};
/// При удалении последней ссылки на блок, добавляет блок в output_blocks.
class MergingBlockPtr
{
public:
MergingBlockPtr() : ptr() {}
explicit MergingBlockPtr(MergingBlock * ptr_) : ptr(ptr_)
{
if (ptr)
++ptr->refcount;
}
MergingBlockPtr(const MergingBlockPtr & rhs) : ptr(rhs.ptr)
{
if (ptr)
++ptr->refcount;
}
MergingBlockPtr & operator=(const MergingBlockPtr & rhs)
{
destroy();
ptr = rhs.ptr;
if (ptr)
++ptr->refcount;
return *this;
}
~MergingBlockPtr()
{
destroy();
}
/// Обнулить указатель и не добавлять блок в output_blocks.
void cancel()
{
if (ptr)
{
--ptr->refcount;
if (!ptr->refcount)
delete ptr;
2014-04-08 07:31:51 +00:00
ptr = nullptr;
}
}
MergingBlock & operator*() const { return *ptr; }
MergingBlock * operator->() const { return ptr; }
operator bool() const { return !!ptr; }
bool operator!() const { return !ptr; }
private:
MergingBlock * ptr;
void destroy()
{
if (ptr)
{
--ptr->refcount;
if (!ptr->refcount)
{
if (std::uncaught_exception())
delete ptr;
else
ptr->output_blocks->push_back(ptr);
}
2014-04-08 07:31:51 +00:00
ptr = nullptr;
}
}
};
struct Cursor
{
MergingBlockPtr block;
size_t pos;
Cursor() {}
explicit Cursor(MergingBlockPtr block_, size_t pos_ = 0) : block(block_), pos(pos_) {}
bool operator<(const Cursor & rhs) const
{
for (size_t i = 0; i < block->sort_columns.size(); ++i)
{
int direction = block->desc[i].direction;
int res = direction * block->sort_columns[i]->compareAt(pos, rhs.pos, *(rhs.block->sort_columns[i]), direction);
if (res > 0)
return true;
if (res < 0)
return false;
}
return block->stream_index > rhs.block->stream_index;
}
/// Не согласован с operator< : не учитывает order.
bool equal(const Cursor & rhs) const
{
if (!block || !rhs.block)
return false;
for (size_t i = 0; i < block->sort_columns.size(); ++i)
{
int res = block->desc[i].direction * block->sort_columns[i]->compareAt(pos, rhs.pos, *(rhs.block->sort_columns[i]), 1);
if (res != 0)
return false;
}
return true;
}
Int8 getSign()
{
return block->sign_column->getData()[pos];
}
/// Помечает, что эту строку нужно взять в ответ.
void addToFilter()
{
block->filter[pos] = 1;
}
bool isLast()
{
return pos + 1 == block->rows;
}
void next()
{
++pos;
}
};
typedef std::priority_queue<Cursor> Queue;
SortDescription description;
String sign_column;
Logger * log;
bool first;
BlockPlainPtrs output_blocks;
Queue queue;
Cursor previous; /// Текущий первичный ключ.
Cursor last_positive; /// Последняя положительная строка для текущего первичного ключа.
size_t count_positive; /// Количество положительных строк для текущего первичного ключа.
size_t count_negative; /// Количество отрицательных строк для текущего первичного ключа.
bool last_is_positive; /// true, если последняя строка для текущего первичного ключа положительная.
size_t count_incorrect_data; /// Чтобы не писать в лог слишком много сообщений об ошибке.
/// Посчитаем, сколько блоков получили на вход и отдали на выход.
size_t blocks_fetched;
size_t blocks_output;
void fetchNextBlock(size_t input_index);
void commitCurrent();
void reportBadCounts();
void reportBadSign(Int8 sign);
};
}