ClickHouse/dbms/include/DB/Columns/ColumnAggregateFunction.h

344 lines
11 KiB
C++
Raw Normal View History

2011-09-19 01:42:16 +00:00
#pragma once
#include <DB/Common/Arena.h>
2011-09-19 01:42:16 +00:00
#include <DB/AggregateFunctions/IAggregateFunction.h>
2014-06-05 23:52:28 +00:00
#include <DB/Columns/IColumn.h>
2011-09-19 01:42:16 +00:00
#include <DB/Core/Field.h>
#include <DB/IO/ReadBufferFromString.h>
#include <DB/IO/WriteBuffer.h>
#include <DB/IO/WriteHelpers.h>
2011-09-19 01:42:16 +00:00
namespace DB
{
namespace ErrorCodes
{
extern const int PARAMETER_OUT_OF_BOUND;
extern const int SIZES_OF_COLUMNS_DOESNT_MATCH;
}
2014-06-08 22:17:44 +00:00
/** Столбец состояний агрегатных функций.
* Представлен в виде массива указателей на состояния агрегатных функций (data).
* Сами состояния хранятся в одном из пулов (arenas).
*
* Может быть в двух вариантах:
*
* 1. Владеть своими значениями - то есть, отвечать за их уничтожение.
* Столбец состоит из значений, "отданных ему на попечение" после выполнения агрегации (см. Aggregator, функция convertToBlocks),
2014-06-08 22:17:44 +00:00
* или из значений, созданных им самим (см. метод insert).
* В этом случае, src будет равно nullptr, и столбец будет сам уничтожать (вызывать IAggregateFunction::destroy)
* состояния агрегатных функций в деструкторе.
*
2014-06-08 22:17:44 +00:00
* 2. Не владеть своими значениями, а использовать значения, взятые из другого столбца ColumnAggregateFunction.
* Например, это столбец, полученный перестановкой/фильтрацией или другими преобразованиями из другого столбца.
* В этом случае, в src будет shared ptr-ом на столбец-источник. Уничтожением значений будет заниматься этот столбец-источник.
*
* Это решение несколько ограничено:
* - не поддерживается вариант, в котором столбец содержит часть "своих" и часть "чужих" значений;
* - не поддерживается вариант наличия нескольких столбцов-источников, что может понадобиться для более оптимального слияния двух столбцов.
*
* Эти ограничения можно снять, если добавить массив флагов или даже refcount-ов,
* определяющий, какие отдельные значения надо уничтожать, а какие - нет.
* Ясно, что этот метод имел бы существенно ненулевую цену.
*/
class ColumnAggregateFunction final : public IColumn, public std::enable_shared_from_this<ColumnAggregateFunction>
{
public:
using Container_t = PaddedPODArray<AggregateDataPtr>;
2014-06-08 22:17:44 +00:00
private:
/// Пулы, в которых выделены состояния агрегатных функций.
Arenas arenas;
2014-06-05 23:52:28 +00:00
/// Используется для уничтожения состояний и для финализации значений.
AggregateFunctionPtr func;
/// Источник. Используется (удерживает источник от уничтожения),
/// если данный столбец создан из другого и использует все или часть его значений.
std::shared_ptr<const ColumnAggregateFunction> src;
2014-06-05 23:52:28 +00:00
/// Массив указателей на состояния агрегатных функций, расположенных в пулах.
Container_t data;
2014-06-05 23:52:28 +00:00
public:
2014-06-08 22:17:44 +00:00
/// Создать столбец на основе другого.
ColumnAggregateFunction(const ColumnAggregateFunction & other)
: arenas(other.arenas), func(other.func), src(other.shared_from_this())
2014-06-08 22:17:44 +00:00
{
}
2014-06-05 23:52:28 +00:00
2014-06-08 22:17:44 +00:00
ColumnAggregateFunction(const AggregateFunctionPtr & func_)
: func(func_)
2014-06-06 19:35:41 +00:00
{
}
2014-06-08 22:17:44 +00:00
ColumnAggregateFunction(const AggregateFunctionPtr & func_, const Arenas & arenas_)
: arenas(arenas_), func(func_)
{
}
~ColumnAggregateFunction()
{
2016-06-09 04:54:30 +00:00
if (!func->hasTrivialDestructor() && !src)
for (auto val : data)
func->destroy(val);
}
2014-06-08 22:17:44 +00:00
void set(const AggregateFunctionPtr & func_)
{
func = func_;
}
AggregateFunctionPtr getAggregateFunction() { return func; }
AggregateFunctionPtr getAggregateFunction() const { return func; }
2014-02-27 12:49:21 +00:00
/// Захватить владение ареной.
void addArena(ArenaPtr arena_)
{
2014-06-08 22:17:44 +00:00
arenas.push_back(arena_);
}
/** Преобразовать столбец состояний агрегатной функции в столбец с готовыми значениями результатов.
*/
ColumnPtr convertToValues() const;
2014-02-27 12:49:21 +00:00
std::string getName() const override { return "ColumnAggregateFunction"; }
size_t sizeOfField() const override { return sizeof(getData()[0]); }
2014-06-05 19:52:13 +00:00
size_t size() const override
2014-06-05 23:52:28 +00:00
{
2014-06-08 22:17:44 +00:00
return getData().size();
}
2011-09-19 01:42:16 +00:00
ColumnPtr cloneEmpty() const override
{
return std::make_shared<ColumnAggregateFunction>(func, Arenas(1, std::make_shared<Arena>()));
2014-06-05 23:52:28 +00:00
};
2011-09-19 01:42:16 +00:00
Field operator[](size_t n) const override
2011-09-19 01:42:16 +00:00
{
Field field = String();
{
WriteBufferFromString buffer(field.get<String &>());
func->serialize(getData()[n], buffer);
}
return field;
2011-09-19 01:42:16 +00:00
}
void get(size_t n, Field & res) const override
{
2014-06-05 23:52:28 +00:00
res = String();
{
WriteBufferFromString buffer(res.get<String &>());
func->serialize(getData()[n], buffer);
}
}
StringRef getDataAt(size_t n) const override
{
2014-06-08 22:17:44 +00:00
return StringRef(reinterpret_cast<const char *>(&getData()[n]), sizeof(getData()[n]));
}
void insertData(const char * pos, size_t length) override
{
2014-06-08 22:17:44 +00:00
getData().push_back(*reinterpret_cast<const AggregateDataPtr *>(pos));
2014-05-26 16:11:20 +00:00
}
void insertFrom(const IColumn & src, size_t n) override
2014-05-26 16:11:20 +00:00
{
2014-06-08 22:17:44 +00:00
getData().push_back(static_cast<const ColumnAggregateFunction &>(src).getData()[n]);
}
2014-06-05 23:52:28 +00:00
2014-06-08 22:17:44 +00:00
/// Объединить состояние в последней строке с заданным
void insertMergeFrom(const IColumn & src, size_t n)
{
func->merge(getData().back(), static_cast<const ColumnAggregateFunction &>(src).getData()[n]);
2014-06-05 23:52:28 +00:00
}
Arena & createOrGetArena()
{
if (unlikely(arenas.empty()))
arenas.emplace_back(std::make_shared<Arena>());
return *arenas.back().get();
}
void insert(const Field & x) override
2014-06-05 23:52:28 +00:00
{
IAggregateFunction * function = func.get();
2014-06-08 22:17:44 +00:00
Arena & arena = createOrGetArena();
2015-11-10 20:39:11 +00:00
getData().push_back(arena.alloc(function->sizeOfData()));
2014-06-08 22:17:44 +00:00
function->create(getData().back());
ReadBufferFromString read_buffer(x.get<const String &>());
2016-03-12 04:01:03 +00:00
function->deserialize(getData().back(), read_buffer);
2014-06-05 23:52:28 +00:00
}
void insertDefault() override
2014-06-05 23:52:28 +00:00
{
IAggregateFunction * function = func.get();
2016-04-14 05:03:33 +00:00
Arena & arena = createOrGetArena();
getData().push_back(arena.alloc(function->sizeOfData()));
function->create(getData().back());
2014-06-05 23:52:28 +00:00
}
StringRef serializeValueIntoArena(size_t n, Arena & arena, char const *& begin) const override
{
throw Exception("Method serializeValueIntoArena is not supported for " + getName(), ErrorCodes::NOT_IMPLEMENTED);
}
const char * deserializeAndInsertFromArena(const char * pos) override
{
throw Exception("Method deserializeAndInsertFromArena is not supported for " + getName(), ErrorCodes::NOT_IMPLEMENTED);
}
void updateHashWithValue(size_t n, SipHash & hash) const override
{
throw Exception("Method updateHashWithValue is not supported for " + getName(), ErrorCodes::NOT_IMPLEMENTED);
}
size_t byteSize() const override
2014-06-05 23:52:28 +00:00
{
size_t res = getData().size() * sizeof(getData()[0]);
for (const auto & arena : arenas)
res += arena.get()->size();
return res;
}
2014-05-28 14:54:42 +00:00
void insertRangeFrom(const IColumn & from, size_t start, size_t length) override
2013-05-03 05:23:14 +00:00
{
const ColumnAggregateFunction & from_concrete = static_cast<const ColumnAggregateFunction &>(from);
2015-11-29 17:06:30 +00:00
if (start + length > from_concrete.getData().size())
2014-05-28 14:54:42 +00:00
throw Exception("Parameters start = "
+ toString(start) + ", length = "
2015-11-29 17:06:30 +00:00
+ toString(length) + " are out of bound in ColumnAggregateFunction::insertRangeFrom method"
" (data.size() = " + toString(from_concrete.getData().size()) + ").",
2014-05-28 14:54:42 +00:00
ErrorCodes::PARAMETER_OUT_OF_BOUND);
if (src && src.get() != &from_concrete)
{
throw Exception("ColumnAggregateFunction could have only one source that owns aggregation states", ErrorCodes::BAD_ARGUMENTS);
}
else
{
/// Keep shared ownership of aggregation states.
src = from_concrete.shared_from_this();
}
2015-11-29 17:06:30 +00:00
auto & data = getData();
size_t old_size = data.size();
data.resize(old_size + length);
memcpy(&data[old_size], &from_concrete.getData()[start], length * sizeof(data[0]));
2013-05-03 05:23:14 +00:00
}
void popBack(size_t n) override
{
size_t size = data.size();
size_t new_size = size - n;
if (!src)
for (size_t i = new_size; i < size; ++i)
func->destroy(data[i]);
data.resize_assume_reserved(new_size);
}
ColumnPtr filter(const Filter & filter, ssize_t result_size_hint) const override
2013-05-03 05:23:14 +00:00
{
2014-06-08 22:17:44 +00:00
size_t size = getData().size();
if (size != filter.size())
throw Exception("Size of filter doesn't match size of column.", ErrorCodes::SIZES_OF_COLUMNS_DOESNT_MATCH);
std::shared_ptr<ColumnAggregateFunction> res = std::make_shared<ColumnAggregateFunction>(*this);
if (size == 0)
return res;
auto & res_data = res->getData();
if (result_size_hint)
res_data.reserve(result_size_hint > 0 ? result_size_hint : size);
for (size_t i = 0; i < size; ++i)
if (filter[i])
res_data.push_back(getData()[i]);
/// Для экономии оперативки в случае слишком сильной фильтрации.
if (res_data.size() * 2 < res_data.capacity())
res_data = Container_t(res_data.cbegin(), res_data.cend());
return res;
2013-05-03 05:23:14 +00:00
}
ColumnPtr permute(const Permutation & perm, size_t limit) const override
2013-05-03 05:23:14 +00:00
{
2014-06-08 22:17:44 +00:00
size_t size = getData().size();
2014-05-28 14:54:42 +00:00
if (limit == 0)
limit = size;
else
limit = std::min(size, limit);
if (perm.size() < limit)
throw Exception("Size of permutation is less than required.", ErrorCodes::SIZES_OF_COLUMNS_DOESNT_MATCH);
std::shared_ptr<ColumnAggregateFunction> res = std::make_shared<ColumnAggregateFunction>(*this);
2014-06-05 23:52:28 +00:00
res->getData().resize(limit);
2014-05-28 14:54:42 +00:00
for (size_t i = 0; i < limit; ++i)
res->getData()[i] = getData()[perm[i]];
2014-06-05 23:52:28 +00:00
2014-05-28 14:54:42 +00:00
return res;
2013-05-03 05:23:14 +00:00
}
ColumnPtr replicate(const Offsets_t & offsets) const override
{
throw Exception("Method replicate is not supported for ColumnAggregateFunction.", ErrorCodes::NOT_IMPLEMENTED);
}
int compareAt(size_t n, size_t m, const IColumn & rhs_, int nan_direction_hint) const override
2011-09-19 01:42:16 +00:00
{
return 0;
}
2011-09-26 11:05:38 +00:00
void getPermutation(bool reverse, size_t limit, Permutation & res) const override
2011-09-26 11:05:38 +00:00
{
2014-06-08 22:17:44 +00:00
size_t s = getData().size();
res.resize(s);
2011-09-26 11:05:38 +00:00
for (size_t i = 0; i < s; ++i)
res[i] = i;
}
2014-06-05 23:52:28 +00:00
/** Более эффективные методы манипуляции */
Container_t & getData()
{
return data;
2014-06-05 23:52:28 +00:00
}
const Container_t & getData() const
{
return data;
2014-06-05 23:52:28 +00:00
}
private:
void getExtremesImpl(Field & min, Field & max, const NullValuesByteMap * null_map_) const override
{
throw Exception("Method getExtremes is not supported for ColumnAggregateFunction.", ErrorCodes::NOT_IMPLEMENTED);
}
2011-09-19 01:42:16 +00:00
};
}