2013-07-28 20:53:31 +00:00
|
|
|
#pragma once
|
|
|
|
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <IO/WriteHelpers.h>
|
|
|
|
#include <IO/ReadHelpers.h>
|
2013-07-28 20:53:31 +00:00
|
|
|
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <DataTypes/DataTypeArray.h>
|
|
|
|
#include <DataTypes/DataTypesNumber.h>
|
|
|
|
#include <DataTypes/DataTypeString.h>
|
2013-07-28 20:53:31 +00:00
|
|
|
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <Columns/ColumnArray.h>
|
2013-07-28 20:53:31 +00:00
|
|
|
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <Common/HashTable/HashSet.h>
|
2019-07-31 15:44:03 +00:00
|
|
|
#include <Common/HashTable/HashTableKeyHolder.h>
|
2019-08-21 02:28:04 +00:00
|
|
|
#include <Common/assert_cast.h>
|
2013-07-28 20:53:31 +00:00
|
|
|
|
2017-12-20 07:36:30 +00:00
|
|
|
#include <AggregateFunctions/IAggregateFunction.h>
|
2013-07-28 20:53:31 +00:00
|
|
|
|
|
|
|
#define AGGREGATE_FUNCTION_GROUP_ARRAY_UNIQ_MAX_SIZE 0xFFFFFF
|
|
|
|
|
|
|
|
|
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
|
|
|
|
|
|
|
|
template <typename T>
|
|
|
|
struct AggregateFunctionGroupUniqArrayData
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
/// When creating, the hash table must be small.
|
|
|
|
using Set = HashSet<
|
|
|
|
T,
|
|
|
|
DefaultHash<T>,
|
|
|
|
HashTableGrower<4>,
|
|
|
|
HashTableAllocatorWithStackMemory<sizeof(T) * (1 << 4)>
|
|
|
|
>;
|
|
|
|
|
|
|
|
Set value;
|
2013-07-28 20:53:31 +00:00
|
|
|
};
|
|
|
|
|
|
|
|
|
2017-03-09 04:18:41 +00:00
|
|
|
/// Puts all values to the hash set. Returns an array of unique values. Implemented for numeric types.
|
2019-04-15 06:36:24 +00:00
|
|
|
template <typename T, typename Tlimit_num_elem>
|
2016-07-10 17:19:35 +00:00
|
|
|
class AggregateFunctionGroupUniqArray
|
2019-04-15 06:36:24 +00:00
|
|
|
: public IAggregateFunctionDataHelper<AggregateFunctionGroupUniqArrayData<T>, AggregateFunctionGroupUniqArray<T, Tlimit_num_elem>>
|
2013-07-28 20:53:31 +00:00
|
|
|
{
|
2019-04-15 06:36:24 +00:00
|
|
|
static constexpr bool limit_num_elems = Tlimit_num_elem::value;
|
|
|
|
UInt64 max_elems;
|
|
|
|
|
2013-07-28 20:53:31 +00:00
|
|
|
private:
|
2017-04-01 07:20:54 +00:00
|
|
|
using State = AggregateFunctionGroupUniqArrayData<T>;
|
2015-11-11 02:04:23 +00:00
|
|
|
|
2013-07-28 20:53:31 +00:00
|
|
|
public:
|
2019-04-15 06:36:24 +00:00
|
|
|
AggregateFunctionGroupUniqArray(const DataTypePtr & argument_type, UInt64 max_elems_ = std::numeric_limits<UInt64>::max())
|
|
|
|
: IAggregateFunctionDataHelper<AggregateFunctionGroupUniqArrayData<T>,
|
|
|
|
AggregateFunctionGroupUniqArray<T, Tlimit_num_elem>>({argument_type}, {}),
|
|
|
|
max_elems(max_elems_) {}
|
2019-02-11 19:26:32 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
String getName() const override { return "groupUniqArray"; }
|
|
|
|
|
|
|
|
DataTypePtr getReturnType() const override
|
|
|
|
{
|
|
|
|
return std::make_shared<DataTypeArray>(std::make_shared<DataTypeNumber<T>>());
|
|
|
|
}
|
|
|
|
|
2017-12-20 07:36:30 +00:00
|
|
|
void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena *) const override
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
2019-04-15 06:36:24 +00:00
|
|
|
if (limit_num_elems && this->data(place).value.size() >= max_elems)
|
|
|
|
return;
|
2019-08-21 02:28:04 +00:00
|
|
|
this->data(place).value.insert(assert_cast<const ColumnVector<T> &>(*columns[0]).getData()[row_num]);
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
2017-12-01 21:51:50 +00:00
|
|
|
void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena *) const override
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
2019-04-15 09:09:52 +00:00
|
|
|
if (!limit_num_elems)
|
|
|
|
this->data(place).value.merge(this->data(rhs).value);
|
|
|
|
else
|
|
|
|
{
|
|
|
|
auto & cur_set = this->data(place).value;
|
|
|
|
auto & rhs_set = this->data(rhs).value;
|
|
|
|
|
|
|
|
for (auto & rhs_elem : rhs_set)
|
|
|
|
{
|
|
|
|
if (cur_set.size() >= max_elems)
|
2019-04-23 22:40:16 +00:00
|
|
|
return;
|
2019-04-15 09:09:52 +00:00
|
|
|
cur_set.insert(rhs_elem.getValue());
|
|
|
|
}
|
|
|
|
}
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
|
|
|
|
{
|
|
|
|
auto & set = this->data(place).value;
|
|
|
|
size_t size = set.size();
|
|
|
|
writeVarUInt(size, buf);
|
2019-08-02 16:30:09 +00:00
|
|
|
for (const auto & elem : set)
|
2017-04-01 07:20:54 +00:00
|
|
|
writeIntBinary(elem, buf);
|
|
|
|
}
|
|
|
|
|
|
|
|
void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override
|
|
|
|
{
|
|
|
|
this->data(place).value.read(buf);
|
|
|
|
}
|
|
|
|
|
|
|
|
void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override
|
|
|
|
{
|
2019-08-21 02:28:04 +00:00
|
|
|
ColumnArray & arr_to = assert_cast<ColumnArray &>(to);
|
2017-12-15 21:32:25 +00:00
|
|
|
ColumnArray::Offsets & offsets_to = arr_to.getOffsets();
|
2017-04-01 07:20:54 +00:00
|
|
|
|
|
|
|
const typename State::Set & set = this->data(place).value;
|
|
|
|
size_t size = set.size();
|
|
|
|
|
2018-12-24 14:26:38 +00:00
|
|
|
offsets_to.push_back(offsets_to.back() + size);
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2019-08-21 02:28:04 +00:00
|
|
|
typename ColumnVector<T>::Container & data_to = assert_cast<ColumnVector<T> &>(arr_to.getData()).getData();
|
2017-04-01 07:20:54 +00:00
|
|
|
size_t old_size = data_to.size();
|
|
|
|
data_to.resize(old_size + size);
|
|
|
|
|
|
|
|
size_t i = 0;
|
|
|
|
for (auto it = set.begin(); it != set.end(); ++it, ++i)
|
2019-02-28 09:35:38 +00:00
|
|
|
data_to[old_size + i] = it->getValue();
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
2017-09-17 20:22:39 +00:00
|
|
|
|
|
|
|
const char * getHeaderFilePath() const override { return __FILE__; }
|
2013-07-28 20:53:31 +00:00
|
|
|
};
|
|
|
|
|
2016-09-13 13:24:24 +00:00
|
|
|
|
2016-09-23 23:33:17 +00:00
|
|
|
/// Generic implementation, it uses serialized representation as object descriptor.
|
2019-06-26 12:32:53 +00:00
|
|
|
struct AggregateFunctionGroupUniqArrayGenericData
|
2016-09-21 16:39:44 +00:00
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
static constexpr size_t INIT_ELEMS = 2; /// adjustable
|
|
|
|
static constexpr size_t ELEM_SIZE = sizeof(HashSetCellWithSavedHash<StringRef, StringRefHash>);
|
|
|
|
using Set = HashSetWithSavedHash<StringRef, StringRefHash, HashTableGrower<INIT_ELEMS>, HashTableAllocatorWithStackMemory<INIT_ELEMS * ELEM_SIZE>>;
|
2016-09-26 16:50:13 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
Set value;
|
2016-09-21 16:39:44 +00:00
|
|
|
};
|
|
|
|
|
2019-04-15 11:10:56 +00:00
|
|
|
template <bool is_plain_column>
|
|
|
|
static void deserializeAndInsertImpl(StringRef str, IColumn & data_to);
|
|
|
|
|
2016-09-22 23:26:08 +00:00
|
|
|
/** Template parameter with true value should be used for columns that store their elements in memory continuously.
|
2019-01-22 19:56:53 +00:00
|
|
|
* For such columns groupUniqArray() can be implemented more efficiently (especially for small numeric arrays).
|
2016-09-22 23:26:08 +00:00
|
|
|
*/
|
2019-04-15 06:36:24 +00:00
|
|
|
template <bool is_plain_column = false, typename Tlimit_num_elem = std::false_type>
|
2019-06-26 12:32:53 +00:00
|
|
|
class AggregateFunctionGroupUniqArrayGeneric
|
|
|
|
: public IAggregateFunctionDataHelper<AggregateFunctionGroupUniqArrayGenericData, AggregateFunctionGroupUniqArrayGeneric<is_plain_column, Tlimit_num_elem>>
|
2016-09-21 16:39:44 +00:00
|
|
|
{
|
2019-02-11 19:26:32 +00:00
|
|
|
DataTypePtr & input_data_type;
|
2016-09-21 16:39:44 +00:00
|
|
|
|
2019-04-15 06:36:24 +00:00
|
|
|
static constexpr bool limit_num_elems = Tlimit_num_elem::value;
|
|
|
|
UInt64 max_elems;
|
|
|
|
|
2019-06-26 12:32:53 +00:00
|
|
|
using State = AggregateFunctionGroupUniqArrayGenericData;
|
2016-09-21 16:39:44 +00:00
|
|
|
|
2019-07-31 15:44:03 +00:00
|
|
|
static auto getKeyHolder(const IColumn & column, size_t row_num, Arena & arena)
|
2019-04-17 10:24:02 +00:00
|
|
|
{
|
2019-07-31 15:44:03 +00:00
|
|
|
if constexpr (is_plain_column)
|
|
|
|
{
|
|
|
|
return ArenaKeyHolder{column.getDataAt(row_num), arena};
|
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
|
|
|
const char * begin = nullptr;
|
|
|
|
StringRef serialized = column.serializeValueIntoArena(row_num, arena, begin);
|
|
|
|
return SerializedKeyHolder{serialized, arena};
|
|
|
|
}
|
2019-04-15 11:10:56 +00:00
|
|
|
}
|
2016-09-21 16:39:44 +00:00
|
|
|
|
2019-04-17 10:24:02 +00:00
|
|
|
static void deserializeAndInsert(StringRef str, IColumn & data_to)
|
|
|
|
{
|
2019-04-15 11:10:56 +00:00
|
|
|
return deserializeAndInsertImpl<is_plain_column>(str, data_to);
|
|
|
|
}
|
2016-09-21 16:39:44 +00:00
|
|
|
|
|
|
|
public:
|
2019-08-03 11:02:40 +00:00
|
|
|
AggregateFunctionGroupUniqArrayGeneric(const DataTypePtr & input_data_type_, UInt64 max_elems_ = std::numeric_limits<UInt64>::max())
|
|
|
|
: IAggregateFunctionDataHelper<AggregateFunctionGroupUniqArrayGenericData, AggregateFunctionGroupUniqArrayGeneric<is_plain_column, Tlimit_num_elem>>({input_data_type_}, {})
|
2019-04-15 06:36:24 +00:00
|
|
|
, input_data_type(this->argument_types[0])
|
|
|
|
, max_elems(max_elems_) {}
|
2016-09-21 16:39:44 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
String getName() const override { return "groupUniqArray"; }
|
|
|
|
|
|
|
|
DataTypePtr getReturnType() const override
|
|
|
|
{
|
2017-12-18 01:11:48 +00:00
|
|
|
return std::make_shared<DataTypeArray>(input_data_type);
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
bool allocatesMemoryInArena() const override
|
|
|
|
{
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
|
|
|
|
void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
|
|
|
|
{
|
|
|
|
auto & set = this->data(place).value;
|
|
|
|
writeVarUInt(set.size(), buf);
|
|
|
|
|
|
|
|
for (const auto & elem : set)
|
|
|
|
{
|
2019-02-28 09:35:38 +00:00
|
|
|
writeStringBinary(elem.getValue(), buf);
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena * arena) const override
|
|
|
|
{
|
|
|
|
auto & set = this->data(place).value;
|
|
|
|
size_t size;
|
|
|
|
readVarUInt(size, buf);
|
|
|
|
//TODO: set.reserve(size);
|
|
|
|
|
2017-12-27 21:45:05 +00:00
|
|
|
for (size_t i = 0; i < size; ++i)
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
|
|
|
set.insert(readStringBinaryInto(*arena, buf));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-12-20 07:36:30 +00:00
|
|
|
void add(AggregateDataPtr place, const IColumn ** columns, size_t row_num, Arena * arena) const override
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
|
|
|
auto & set = this->data(place).value;
|
2019-04-15 09:09:52 +00:00
|
|
|
if (limit_num_elems && set.size() >= max_elems)
|
|
|
|
return;
|
|
|
|
|
2019-07-31 15:44:03 +00:00
|
|
|
bool inserted;
|
2019-08-20 09:58:44 +00:00
|
|
|
State::Set::LookupResult it;
|
2019-07-31 15:44:03 +00:00
|
|
|
auto key_holder = getKeyHolder(*columns[0], row_num, *arena);
|
|
|
|
set.emplace(key_holder, it, inserted);
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena * arena) const override
|
|
|
|
{
|
|
|
|
auto & cur_set = this->data(place).value;
|
|
|
|
auto & rhs_set = this->data(rhs).value;
|
|
|
|
|
|
|
|
bool inserted;
|
2019-08-20 09:58:44 +00:00
|
|
|
State::Set::LookupResult it;
|
2017-04-01 07:20:54 +00:00
|
|
|
for (auto & rhs_elem : rhs_set)
|
|
|
|
{
|
2019-04-15 09:09:52 +00:00
|
|
|
if (limit_num_elems && cur_set.size() >= max_elems)
|
2019-07-31 15:44:03 +00:00
|
|
|
return;
|
|
|
|
|
|
|
|
// We have to copy the keys to our arena.
|
|
|
|
assert(arena != nullptr);
|
|
|
|
cur_set.emplace(ArenaKeyHolder{rhs_elem.getValue(), *arena}, it, inserted);
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override
|
|
|
|
{
|
2019-08-21 02:28:04 +00:00
|
|
|
ColumnArray & arr_to = assert_cast<ColumnArray &>(to);
|
2017-12-15 21:32:25 +00:00
|
|
|
ColumnArray::Offsets & offsets_to = arr_to.getOffsets();
|
2017-04-01 07:20:54 +00:00
|
|
|
IColumn & data_to = arr_to.getData();
|
|
|
|
|
|
|
|
auto & set = this->data(place).value;
|
2018-12-24 14:26:38 +00:00
|
|
|
offsets_to.push_back(offsets_to.back() + set.size());
|
2017-04-01 07:20:54 +00:00
|
|
|
|
|
|
|
for (auto & elem : set)
|
|
|
|
{
|
2019-02-28 09:35:38 +00:00
|
|
|
deserializeAndInsert(elem.getValue(), data_to);
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
}
|
2017-09-17 20:22:39 +00:00
|
|
|
|
|
|
|
const char * getHeaderFilePath() const override { return __FILE__; }
|
2016-09-21 16:39:44 +00:00
|
|
|
};
|
|
|
|
|
|
|
|
template <>
|
2019-04-15 11:10:56 +00:00
|
|
|
inline void deserializeAndInsertImpl<false>(StringRef str, IColumn & data_to)
|
2019-04-15 09:09:52 +00:00
|
|
|
{
|
|
|
|
data_to.deserializeAndInsertFromArena(str.data);
|
|
|
|
}
|
|
|
|
|
|
|
|
template <>
|
2019-04-15 11:10:56 +00:00
|
|
|
inline void deserializeAndInsertImpl<true>(StringRef str, IColumn & data_to)
|
2019-04-15 09:09:52 +00:00
|
|
|
{
|
|
|
|
data_to.insertData(str.data, str.size);
|
|
|
|
}
|
2016-09-21 16:39:44 +00:00
|
|
|
|
2013-07-28 20:53:31 +00:00
|
|
|
#undef AGGREGATE_FUNCTION_GROUP_ARRAY_UNIQ_MAX_SIZE
|
|
|
|
|
|
|
|
}
|