#pragma once #include #include #include #include #include #include #include #include namespace DB { /** В качестве состояния используется массив, в который складываются все значения. * NOTE Если различных значений мало, то это не оптимально. * Для 8 и 16-битных значений возможно, было бы лучше использовать lookup-таблицу. */ template struct AggregateFunctionQuantileExactData { /// Сразу будет выделена память на несколько элементов так, чтобы состояние занимало 64 байта. static constexpr size_t bytes_in_arena = 64 - sizeof(PODArray); using Array = PODArray, bytes_in_arena>>; Array array; }; /** Точно вычисляет квантиль. * В качестве типа аргумента может быть только числовой тип (в том числе, дата и дата-с-временем). * Тип результата совпадает с типом аргумента. */ template class AggregateFunctionQuantileExact final : public IUnaryAggregateFunction, AggregateFunctionQuantileExact> { private: double level; DataTypePtr type; public: AggregateFunctionQuantileExact(double level_ = 0.5) : level(level_) {} String getName() const override { return "quantileExact"; } DataTypePtr getReturnType() const override { return type; } void setArgument(const DataTypePtr & argument) { type = argument; } void setParameters(const Array & params) override { if (params.size() != 1) throw Exception("Aggregate function " + getName() + " requires exactly one parameter.", ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH); level = apply_visitor(FieldVisitorConvertToNumber(), params[0]); } void addImpl(AggregateDataPtr place, const IColumn & column, size_t row_num) const { this->data(place).array.push_back(static_cast &>(column).getData()[row_num]); } void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs) const override { this->data(place).array.insert(this->data(rhs).array.begin(), this->data(rhs).array.end()); } void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override { const auto & array = this->data(place).array; size_t size = array.size(); writeVarUInt(size, buf); buf.write(reinterpret_cast(&array[0]), size * sizeof(array[0])); } void deserialize(AggregateDataPtr place, ReadBuffer & buf) const override { auto & array = this->data(place).array; size_t size = 0; readVarUInt(size, buf); array.resize(size); buf.read(reinterpret_cast(&array[0]), size * sizeof(array[0])); } void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override { /// Сортировка массива не будет считаться нарушением константности. auto & array = const_cast::Array &>(this->data(place).array); T quantile = T(); if (!array.empty()) { size_t n = level < 1 ? level * array.size() : (array.size() - 1); std::nth_element(array.begin(), array.begin() + n, array.end()); /// NOTE Можно придумать алгоритм radix-select. quantile = array[n]; } static_cast &>(to).getData().push_back(quantile); } }; /** То же самое, но позволяет вычислить сразу несколько квантилей. * Для этого, принимает в качестве параметров несколько уровней. Пример: quantilesExact(0.5, 0.8, 0.9, 0.95)(ConnectTiming). * Возвращает массив результатов. */ template class AggregateFunctionQuantilesExact final : public IUnaryAggregateFunction, AggregateFunctionQuantilesExact> { private: QuantileLevels levels; DataTypePtr type; public: String getName() const override { return "quantilesExact"; } DataTypePtr getReturnType() const override { return new DataTypeArray(type); } void setArgument(const DataTypePtr & argument) { type = argument; } void setParameters(const Array & params) override { levels.set(params); } void addImpl(AggregateDataPtr place, const IColumn & column, size_t row_num) const { this->data(place).array.push_back(static_cast &>(column).getData()[row_num]); } void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs) const override { this->data(place).array.insert(this->data(rhs).array.begin(), this->data(rhs).array.end()); } void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override { const auto & array = this->data(place).array; size_t size = array.size(); writeVarUInt(size, buf); buf.write(reinterpret_cast(&array[0]), size * sizeof(array[0])); } void deserialize(AggregateDataPtr place, ReadBuffer & buf) const override { auto & array = this->data(place).array; size_t size = 0; readVarUInt(size, buf); array.resize(size); buf.read(reinterpret_cast(&array[0]), size * sizeof(array[0])); } void insertResultInto(ConstAggregateDataPtr place, IColumn & to) const override { /// Сортировка массива не будет считаться нарушением константности. auto & array = const_cast::Array &>(this->data(place).array); ColumnArray & arr_to = static_cast(to); ColumnArray::Offsets_t & offsets_to = arr_to.getOffsets(); size_t num_levels = levels.size(); offsets_to.push_back((offsets_to.size() == 0 ? 0 : offsets_to.back()) + num_levels); typename ColumnVector::Container_t & data_to = static_cast &>(arr_to.getData()).getData(); size_t old_size = data_to.size(); data_to.resize(old_size + num_levels); if (!array.empty()) { size_t prev_n = 0; for (auto level_index : levels.permutation) { auto level = levels.levels[level_index]; size_t n = level < 1 ? level * array.size() : (array.size() - 1); std::nth_element(array.begin() + prev_n, array.begin() + n, array.end()); data_to[old_size + level_index] = array[n]; prev_n = n; } } else { for (size_t i = 0; i < num_levels; ++i) data_to[old_size + i] = T(); } } }; }