#pragma once #include #include #include #include #include #include #include #include #include namespace DB { template struct AggregateFunctionSparkbarData { using Points = HashMap; Points points; Y min_y = std::numeric_limits::max(); Y max_y = std::numeric_limits::lowest(); void insert(const X & x, const Y & y) { auto result = points.insert({x, y}); if (!result.second) result.first->getMapped() += y; } void add(X x, Y y) { insert(x, y); min_y = std::min(y, min_y); max_y = std::max(y, max_y); } void merge(const AggregateFunctionSparkbarData & other) { if (other.points.empty()) return; for (auto & point : other.points) insert(point.getKey(), point.getMapped()); min_y = std::min(other.min_y, min_y); max_y = std::max(other.max_y, max_y); } void serialize(WriteBuffer & buf) const { writeBinary(min_y, buf); writeBinary(max_y, buf); writeVarUInt(points.size(), buf); for (const auto & elem : points) { writeBinary(elem.getKey(), buf); writeBinary(elem.getMapped(), buf); } } void deserialize(ReadBuffer & buf) { readBinary(min_y, buf); readBinary(max_y, buf); size_t size; readVarUInt(size, buf); /// TODO Protection against huge size X x; Y y; for (size_t i = 0; i < size; ++i) { readBinary(x, buf); readBinary(y, buf); insert(x, y); } } }; template class AggregateFunctionSparkbar final : public IAggregateFunctionDataHelper, AggregateFunctionSparkbar> { private: size_t width; mutable X min_x; mutable X max_x; bool specified_min_max_x; String getBar(const UInt8 value) const { // ▁▂▃▄▅▆▇█ switch (value) { case 1: return "▁"; case 2: return "▂"; case 3: return "▃"; case 4: return "▄"; case 5: return "▅"; case 6: return "▆"; case 7: return "▇"; case 8: return "█"; } return " "; } /** * The minimum value of y is rendered as the lowest height "▁", * the maximum value of y is rendered as the highest height "█", and the middle value will be rendered proportionally. * If a bucket has no y value, it will be rendered as " ". * If the actual number of buckets is greater than the specified bucket, it will be compressed by width. * For example, there are actually 11 buckets, specify 10 buckets, and divide the 11 buckets as follows (11/10): * 0.0-1.1, 1.1-2.2, 2.2-3.3, 3.3-4.4, 4.4-5.5, 5.5-6.6, 6.6-7.7, 7.7-8.8, 8.8-9.9, 9.9-11. * The y value of the first bucket will be calculated as follows: * the actual y value of the first position + the actual second position y*0.1, and the remaining y*0.9 is reserved for the next bucket. * The next bucket will use the last y*0.9 + the actual third position y*0.2, and the remaining y*0.8 will be reserved for the next bucket. And so on. */ String render(const AggregateFunctionSparkbarData & data) const { String value; if (data.points.empty() || !width) return value; size_t diff_x = max_x - min_x; if ((diff_x + 1) <= width) { Y min_y = data.min_y; Y max_y = data.max_y; Float64 diff_y = max_y - min_y; if (diff_y) { for (size_t i = 0; i <= diff_x; ++i) { auto it = data.points.find(min_x + i); bool found = it != data.points.end(); value += getBar(found ? static_cast(std::round(((it->getMapped() - min_y) / diff_y) * 7) + 1) : 0); } } else { for (size_t i = 0; i <= diff_x; ++i) value += getBar(data.points.has(min_x + i) ? 1 : 0); } } else { // begin reshapes to width buckets Float64 multiple_d = (diff_x + 1) / static_cast(width); std::optional min_y; std::optional max_y; std::optional new_y; std::vector> newPoints; newPoints.reserve(width); std::pair bound{0, 0.0}; size_t cur_bucket_num = 0; // upper bound for bucket auto upperBound = [&](size_t bucket_num) { bound.second = (bucket_num + 1) * multiple_d; bound.first = std::floor(bound.second); }; upperBound(cur_bucket_num); for (size_t i = 0; i <= (diff_x + 1); ++i) { if (i == bound.first) // is bound { Float64 proportion = bound.second - bound.first; auto it = data.points.find(min_x + i); bool found = (it != data.points.end()); if (found && proportion > 0) new_y = new_y.value_or(0) + it->getMapped() * proportion; if (new_y) { Float64 avg_y = new_y.value() / multiple_d; newPoints.emplace_back(avg_y); // If min_y has no value, or if the avg_y of the current bucket is less than min_y, update it. if (!min_y || avg_y < min_y) min_y = avg_y; if (!max_y || avg_y > max_y) max_y = avg_y; } else { newPoints.emplace_back(); } // next bucket new_y = found ? ((1 - proportion) * it->getMapped()) : std::optional(); upperBound(++cur_bucket_num); } else { auto it = data.points.find(min_x + i); if (it != data.points.end()) new_y = new_y.value_or(0) + it->getMapped(); } } if (!min_y || !max_y) // No value is set return {}; Float64 diff_y = max_y.value() - min_y.value(); auto getBars = [&] (const std::optional & point_y) { value += getBar(point_y ? static_cast(std::round(((point_y.value() - min_y.value()) / diff_y) * 7) + 1) : 0); }; auto getBarsForConstant = [&] (const std::optional & point_y) { value += getBar(point_y ? 1 : 0); }; if (diff_y) std::for_each(newPoints.begin(), newPoints.end(), getBars); else std::for_each(newPoints.begin(), newPoints.end(), getBarsForConstant); } return value; } public: AggregateFunctionSparkbar(const DataTypes & arguments, const Array & params) : IAggregateFunctionDataHelper, AggregateFunctionSparkbar>( arguments, params) { width = params.at(0).safeGet(); if (params.size() == 3) { specified_min_max_x = true; min_x = params.at(1).safeGet(); max_x = params.at(2).safeGet(); } else { specified_min_max_x = false; min_x = std::numeric_limits::max(); max_x = std::numeric_limits::min(); } } String getName() const override { return "sparkbar"; } DataTypePtr getReturnType() const override { return std::make_shared(); } void add(AggregateDataPtr __restrict place, const IColumn ** columns, size_t row_num, Arena * /*arena*/) const override { X x = assert_cast *>(columns[0])->getData()[row_num]; if (specified_min_max_x && min_x <= x && x <= max_x) { Y y = assert_cast *>(columns[1])->getData()[row_num]; this->data(place).add(x, y); } else if (!specified_min_max_x) { min_x = std::min(x, min_x); max_x = std::max(x, max_x); Y y = assert_cast *>(columns[1])->getData()[row_num]; this->data(place).add(x, y); } } void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena * /*arena*/) const override { this->data(place).merge(this->data(rhs)); } void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf) const override { this->data(place).serialize(buf); } void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, Arena *) const override { this->data(place).deserialize(buf); } bool allocatesMemoryInArena() const override { return false; } void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena * /*arena*/) const override { auto & to_column = assert_cast(to); const auto & data = this->data(place); const String & value = render(data); to_column.insertData(value.data(), value.size()); } }; }