[Feature](func) Support AggFn percentile_approx_array (#65847)

### Release note

doc: https://github.com/apache/doris-website/pull/4016

Support function `PERCENTILE_APPROX_ARRAY`

performance:

```text
Doris> select percentile_approx(FUniqID, 0.01), percentile_approx(FUniqID, 0.99) from hits_100m;
+----------------------------------+----------------------------------+
| percentile_approx(FUniqID, 0.01) | percentile_approx(FUniqID, 0.99) |
+----------------------------------+----------------------------------+
|                                0 |            9.170476181709914e+18 |
+----------------------------------+----------------------------------+
1 row in set (13.962 sec)

Doris> select percentile_approx_array(FUniqID, [0.01, 0.99]) from hits_100m;
+------------------------------------------------+
| percentile_approx_array(FUniqID, [0.01, 0.99]) |
+------------------------------------------------+
| [0, 9.170476181709914e+18]                     |
+------------------------------------------------+
1 row in set (7.869 sec)
```
This commit is contained in:
linrrarity authored and GitHub committed 2026-08-03 09:53:04 +08:00
1 parent a27bf6f179
commit dc4fc151b9
18 files changed
+1297 -110

No files matched your search

@@ -59,6 +59,24 @@ AggregateFunctionPtr create_aggregate_function_percentile_approx_weighted(
return nullptr;
}
AggregateFunctionPtr create_aggregate_function_percentile_approx_array(
const std::string& name, const DataTypes& argument_types, const DataTypePtr& result_type,
const bool result_is_nullable, const AggregateFunctionAttr& attr) {
const DataTypePtr& argument_type = remove_nullable(argument_types[0]);
if (argument_type->get_primitive_type() != PrimitiveType::TYPE_DOUBLE) {
return nullptr;
}
if (argument_types.size() == 2) {
return creator_without_type::create<AggregateFunctionPercentileApproxArray<false>>(
argument_types, result_is_nullable, attr);
}
if (argument_types.size() == 3) {
return creator_without_type::create<AggregateFunctionPercentileApproxArray<true>>(
argument_types, result_is_nullable, attr);
}
return nullptr;
}
void register_aggregate_function_percentile(AggregateFunctionSimpleFactory& factory) {
using creator = creator_with_type_list<TYPE_TINYINT, TYPE_SMALLINT, TYPE_INT, TYPE_BIGINT,
TYPE_LARGEINT, TYPE_FLOAT, TYPE_DOUBLE>;
@@ -85,6 +103,8 @@ void register_aggregate_function_percentile_old(AggregateFunctionSimpleFactory&
void register_aggregate_function_percentile_approx(AggregateFunctionSimpleFactory& factory) {
factory.register_function_both("percentile_approx",
create_aggregate_function_percentile_approx);
factory.register_function_both("percentile_approx_array",
create_aggregate_function_percentile_approx_array);
factory.register_function_both("percentile_approx_weighted",
create_aggregate_function_percentile_approx_weighted);
@@ -86,7 +86,7 @@ struct PercentileApproxState {
//The compression parameter setting range is [2048, 10000].
//If the value of compression parameter is not specified set, or is outside the range of [2048, 10000],
//will use the default value of 10000
if (compression < 2048 || compression > 10000) {
if (!std::isfinite(compression) || compression < 2048 || compression > 10000) {
compression = 10000;
}
digest = TDigest::create_unique(compression);
@@ -329,6 +329,238 @@ public:
}
};
struct PercentileApproxArrayState {
void init(const PaddedPODArray<Float64>& quantiles, const NullMap& null_map, size_t start,
size_t size, float compression = 10000) {
if (init_flag) {
return;
}
if (!std::isfinite(compression) || compression < 2048 || compression > 10000) {
compression = 10000;
}
compressions = compression;
levels.quantiles.resize(size);
levels.permutation.resize(size);
for (size_t i = 0; i < size; ++i) {
if (null_map[start + i]) {
throw Exception(ErrorCode::INVALID_ARGUMENT,
"percentile_approx_array quantile should not be null");
}
check_quantile(quantiles[start + i]);
levels.quantiles[i] = quantiles[start + i];
levels.permutation[i] = i;
}
if (!levels.empty()) {
digest = TDigest::create_unique(compressions);
}
init_flag = true;
}
void add_range(const Float64* values, size_t size) {
if (levels.empty()) {
return;
}
for (size_t i = 0; i < size; ++i) {
digest->add(static_cast<float>(values[i]));
}
}
void write(BufferWritable& buf) const {
buf.write_binary(init_flag);
if (!init_flag) {
return;
}
levels.write(buf);
buf.write_binary(compressions);
if (levels.empty()) {
return;
}
uint32_t serialize_size = digest->serialized_size();
std::string result(serialize_size, '0');
digest->serialize(reinterpret_cast<uint8_t*>(result.data()));
buf.write_binary(result);
}
void read(BufferReadable& buf) {
reset();
buf.read_binary(init_flag);
if (!init_flag) {
return;
}
levels.read(buf);
buf.read_binary(compressions);
if (levels.empty()) {
return;
}
std::string str;
buf.read_binary(str);
digest = TDigest::create_unique(compressions);
digest->unserialize(reinterpret_cast<const uint8_t*>(str.data()));
}
void merge(const PercentileApproxArrayState& rhs) {
if (!rhs.init_flag) {
return;
}
if (!init_flag) {
levels.merge(rhs.levels);
compressions = rhs.compressions;
if (!levels.empty()) {
digest = TDigest::create_unique(compressions);
}
init_flag = true;
} else {
if (compressions != rhs.compressions || levels.quantiles != rhs.levels.quantiles) {
throw Exception(
ErrorCode::INVALID_ARGUMENT,
"percentile_approx_array aggregate states have incompatible quantiles "
"or compression");
}
}
if (!levels.empty()) {
digest->merge(rhs.digest.get());
}
}
void reset() {
init_flag = false;
levels.clear();
digest.reset();
compressions = 10000;
}
void insert_result_into(IColumn& to) const {
auto& column_data = assert_cast<ColumnFloat64&, TypeCheckOnRelease::DISABLE>(to).get_data();
if (levels.empty()) {
return;
}
const size_t old_size = column_data.size();
const size_t size = levels.quantiles.size();
column_data.resize(old_size + size);
digest->quantiles(levels.quantiles.data(), levels.get_permutation().data(), size,
column_data.data() + old_size);
}
bool init_flag = false;
PercentileLevels levels;
std::unique_ptr<TDigest> digest;
float compressions = 10000;
};
template <bool has_compression>
class AggregateFunctionPercentileApproxArray final
: public IAggregateFunctionDataHelper<
PercentileApproxArrayState,
AggregateFunctionPercentileApproxArray<has_compression>>,
MultiExpression,
NotNullableAggregateFunction {
public:
using Base =
IAggregateFunctionDataHelper<PercentileApproxArrayState,
AggregateFunctionPercentileApproxArray<has_compression>>;
AggregateFunctionPercentileApproxArray(const DataTypes& argument_types_)
: Base(argument_types_) {}
String get_name() const override { return "percentile_approx_array"; }
DataTypePtr get_return_type() const override {
return std::make_shared<DataTypeArray>(make_nullable(std::make_shared<DataTypeFloat64>()));
}
void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
Arena&) const override {
const auto& sources =
assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(*columns[0]);
_add_values(this->data(place), &sources.get_data()[row_num], 1, columns,
cast_set<size_t>(row_num));
}
void add_batch_single_place(size_t batch_size, AggregateDataPtr place, const IColumn** columns,
Arena&) const override {
const auto& sources =
assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(*columns[0]);
DCHECK_EQ(sources.get_data().size(), batch_size);
_add_values(this->data(place), sources.get_data().data(), batch_size, columns, 0);
}
void add_batch_range(size_t batch_begin, size_t batch_end, AggregateDataPtr place,
const IColumn** columns, Arena&, bool has_null) override {
DCHECK(!has_null);
const auto& sources =
assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(*columns[0]);
_add_values(this->data(place), sources.get_data().data() + batch_begin,
batch_end - batch_begin + 1, columns, batch_begin);
}
void check_input_columns_type(const IColumn** columns) const override {
this->template check_argument_column_type<ColumnFloat64>(columns[0]);
check_percentile_array_column_type(*this, *columns[1], 1);
if constexpr (has_compression) {
this->template check_argument_column_type<ColumnFloat64>(columns[2]);
}
}
void reset(AggregateDataPtr __restrict place) const override { this->data(place).reset(); }
void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
Arena&) const override {
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
this->data(place).write(buf);
}
void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
Arena&) const override {
this->data(place).read(buf);
}
void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
auto& to_arr = assert_cast<ColumnArray&, TypeCheckOnRelease::DISABLE>(to);
auto& nullable_data =
assert_cast<ColumnNullable&, TypeCheckOnRelease::DISABLE>(to_arr.get_data());
auto& nested_data = nullable_data.get_nested_column();
this->data(place).insert_result_into(nested_data);
nullable_data.get_null_map_data().resize_fill(nested_data.size(), 0);
to_arr.get_offsets().push_back(nested_data.size());
}
private:
void _add_values(PercentileApproxArrayState& state, const Float64* values, size_t value_size,
const IColumn** columns, size_t quantile_row) const {
if (!state.init_flag) {
const auto& quantile_array =
assert_cast<const ColumnArray&, TypeCheckOnRelease::DISABLE>(*columns[1]);
const auto& offsets = quantile_array.get_offsets();
const auto& nullable_quantiles =
assert_cast<const ColumnNullable&, TypeCheckOnRelease::DISABLE>(
quantile_array.get_data());
const auto& quantiles = assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(
nullable_quantiles.get_nested_column())
.get_data();
const auto& null_map = nullable_quantiles.get_null_map_data();
const size_t start = quantile_row == 0 ? 0 : offsets[quantile_row - 1];
const size_t quantile_size = offsets[quantile_row] - start;
float compression = 10000;
if constexpr (has_compression) {
const auto& compression_column =
assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(*columns[2]);
compression = static_cast<float>(compression_column.get_element(0));
}
state.init(quantiles, null_map, start, quantile_size, compression);
}
state.add_range(values, value_size);
}
};
template <PrimitiveType T>
struct PercentileState {
mutable std::vector<Counts<typename PrimitiveTypeTraits<T>::CppType>> vec_counts;
+1 -1
View File
@@ -35,7 +35,7 @@
namespace doris {
inline void check_quantile(double quantile) {
if (quantile < 0 || quantile > 1) {
if (!std::isfinite(quantile) || quantile < 0 || quantile > 1) {
throw Exception(ErrorCode::INVALID_ARGUMENT,
"quantile in func percentile should in [0, 1], but real data is:" +
std::to_string(quantile));
+163 -104
View File
@@ -62,7 +62,7 @@ using Value = float;
using Weight = float;
using Index = size_t;
const size_t kHighWater = 40000;
constexpr size_t K_HIGH_WATER = 40000;
class Centroid {
public:
@@ -126,7 +126,7 @@ class TDigest {
TDigestComparator() = default;
bool operator()(const TDigest* left, const TDigest* right) const {
return left->totalSize() > right->totalSize();
return left->total_size() > right->total_size();
}
};
using TDigestQueue =
@@ -137,19 +137,19 @@ public:
explicit TDigest(Value compression) : TDigest(compression, 0) {}
TDigest(Value compression, Index bufferSize) : TDigest(compression, bufferSize, 0) {}
TDigest(Value compression, Index buffer_size) : TDigest(compression, buffer_size, 0) {}
TDigest(Value compression, Index unmergedSize, Index mergedSize)
TDigest(Value compression, Index unmerged_size, Index merged_size)
: _compression(compression),
_max_processed(processedSize(mergedSize, compression)),
_max_unprocessed(unprocessedSize(unmergedSize, compression)) {
_max_processed(processed_size(merged_size, compression)),
_max_unprocessed(unprocessed_size(unmerged_size, compression)) {
_processed.reserve(_max_processed);
_unprocessed.reserve(_max_unprocessed + 1);
}
TDigest(std::vector<Centroid>&& processed, std::vector<Centroid>&& unprocessed,
Value compression, Index unmergedSize, Index mergedSize)
: TDigest(compression, unmergedSize, mergedSize) {
Value compression, Index unmerged_size, Index merged_size)
: TDigest(compression, unmerged_size, merged_size) {
_processed = std::move(processed);
_unprocessed = std::move(unprocessed);
@@ -159,7 +159,7 @@ public:
_min = std::min(_min, _processed[0].mean());
_max = std::max(_max, (_processed.cend() - 1)->mean());
}
updateCumulative();
_update_cumulative();
}
static Weight weight(std::vector<Centroid>& centroids) noexcept {
@@ -188,11 +188,11 @@ public:
: TDigest(std::move(o._processed), std::move(o._unprocessed), o._compression,
o._max_unprocessed, o._max_processed) {}
static inline Index processedSize(Index size, Value compression) noexcept {
static inline Index processed_size(Index size, Value compression) noexcept {
return (size == 0) ? static_cast<Index>(2 * std::ceil(compression)) : size;
}
static inline Index unprocessedSize(Index size, Value compression) noexcept {
static inline Index unprocessed_size(Index size, Value compression) noexcept {
return (size == 0) ? static_cast<Index>(8 * std::ceil(compression)) : size;
}
@@ -206,15 +206,15 @@ public:
const std::vector<Centroid>& unprocessed() const { return _unprocessed; }
Index maxUnprocessed() const { return _max_unprocessed; }
Index max_unprocessed() const { return _max_unprocessed; }
Index maxProcessed() const { return _max_processed; }
Index max_processed() const { return _max_processed; }
void add(std::vector<const TDigest*> digests) { add(digests.cbegin(), digests.cend()); }
// merge in a vector of tdigests in the most efficient manner possible
// in constant space
// works for any value of kHighWater
// works for any value of K_HIGH_WATER
void add(std::vector<const TDigest*>::const_iterator iter,
std::vector<const TDigest*>::const_iterator end) {
if (iter != end) {
@@ -226,48 +226,48 @@ public:
std::vector<const TDigest*> batch;
batch.reserve(size);
size_t totalSize = 0;
size_t total_size = 0;
while (!pq.empty()) {
const auto* td = pq.top();
batch.push_back(td);
pq.pop();
totalSize += td->totalSize();
if (totalSize >= kHighWater || pq.empty()) {
mergeProcessed(batch);
mergeUnprocessed(batch);
processIfNecessary();
total_size += td->total_size();
if (total_size >= K_HIGH_WATER || pq.empty()) {
_merge_processed(batch);
_merge_unprocessed(batch);
_process_if_necessary();
batch.clear();
totalSize = 0;
total_size = 0;
}
}
updateCumulative();
_update_cumulative();
}
}
Weight processedWeight() const { return _processed_weight; }
Weight processed_weight() const { return _processed_weight; }
Weight unprocessedWeight() const { return _unprocessed_weight; }
Weight unprocessed_weight() const { return _unprocessed_weight; }
bool haveUnprocessed() const { return _unprocessed.size() > 0; }
bool have_unprocessed() const { return _unprocessed.size() > 0; }
size_t totalSize() const { return _processed.size() + _unprocessed.size(); }
size_t total_size() const { return _processed.size() + _unprocessed.size(); }
long totalWeight() const { return static_cast<long>(_processed_weight + _unprocessed_weight); }
long total_weight() const { return static_cast<long>(_processed_weight + _unprocessed_weight); }
// return the cdf on the t-digest
Value cdf(Value x) {
if (haveUnprocessed() || isDirty()) {
process();
if (have_unprocessed() || is_dirty()) {
_process();
}
return cdfProcessed(x);
return cdf_processed(x);
}
bool isDirty() {
bool is_dirty() {
return _processed.size() > _max_processed || _unprocessed.size() > _max_unprocessed;
}
// return the cdf on the processed values
Value cdfProcessed(Value x) const {
Value cdf_processed(Value x) const {
VLOG_CRITICAL << "cdf value " << x;
VLOG_CRITICAL << "processed size " << _processed.size();
if (_processed.size() == 0) {
@@ -306,13 +306,13 @@ public:
}
// check for the left tail
if (x <= mean(0)) {
if (x <= _mean(0)) {
VLOG_CRITICAL << "left tail "
<< " _min " << _min << " mean(0) " << mean(0) << " x " << x;
<< " _min " << _min << " mean(0) " << _mean(0) << " x " << x;
// note that this is different than mean(0) > _min ... this guarantees interpolation works
if (mean(0) - _min > 0) {
return static_cast<Value>((x - _min) / (mean(0) - _min) * weight(0) /
if (_mean(0) - _min > 0) {
return static_cast<Value>((x - _min) / (_mean(0) - _min) * _weight(0) /
_processed_weight / 2.0);
} else {
return 0;
@@ -320,13 +320,13 @@ public:
}
// and the right tail
if (x >= mean(n - 1)) {
if (x >= _mean(n - 1)) {
VLOG_CRITICAL << "right tail"
<< " _max " << _max << " mean(n - 1) " << mean(n - 1) << " x " << x;
<< " _max " << _max << " mean(n - 1) " << _mean(n - 1) << " x " << x;
if (_max - mean(n - 1) > 0) {
return static_cast<Value>(1.0 - (_max - x) / (_max - mean(n - 1)) *
weight(n - 1) / _processed_weight /
if (_max - _mean(n - 1) > 0) {
return static_cast<Value>(1.0 - (_max - x) / (_max - _mean(n - 1)) *
_weight(n - 1) / _processed_weight /
2.0);
} else {
return 1;
@@ -345,21 +345,78 @@ public:
VLOG_CRITICAL << "middle "
<< " z1 " << z1 << " z2 " << z2 << " x " << x;
return weightedAverage(_cumulative[i - 1], z2, _cumulative[i], z1) / _processed_weight;
return _weighted_average(_cumulative[i - 1], z2, _cumulative[i], z1) /
_processed_weight;
}
}
// this returns a quantile on the t-digest
Value quantile(Value q) {
if (haveUnprocessed() || isDirty()) {
process();
if (have_unprocessed() || is_dirty()) {
_process();
}
return quantile_processed(q);
}
void quantiles(const double* quantile_levels, const size_t* permutation, size_t size,
double* result) {
if (size == 0) {
return;
}
if (have_unprocessed() || is_dirty()) {
_process();
}
if (_processed.empty()) {
std::fill(result, result + size, NAN);
return;
}
if (_processed.size() == 1) {
std::fill(result, result + size, static_cast<double>(_mean(0)));
return;
}
const auto n = _processed.size();
size_t cumulative_index = 0;
for (size_t result_index = 0; result_index < size; ++result_index) {
const size_t level_index = permutation[result_index];
const auto q = static_cast<Value>(quantile_levels[level_index]);
DCHECK_GE(q, 0);
DCHECK_LE(q, 1);
const auto index = q * _processed_weight;
if (index <= _weight(0) / 2.0) {
DCHECK_GT(_weight(0), 0);
result[level_index] =
static_cast<Value>(_min + 2.0 * index / _weight(0) * (_mean(0) - _min));
continue;
}
while (cumulative_index < _cumulative.size() && _cumulative[cumulative_index] < index) {
++cumulative_index;
}
if (cumulative_index > 0 && cumulative_index + 1 < _cumulative.size()) {
auto z1 = index - _cumulative[cumulative_index - 1];
auto z2 = _cumulative[cumulative_index] - index;
result[level_index] = static_cast<double>(_weighted_average(
_mean(cumulative_index - 1), z2, _mean(cumulative_index), z1));
continue;
}
DCHECK_LE(index, _processed_weight);
DCHECK_GE(index, _processed_weight - _weight(n - 1) / 2.0);
auto z1 = static_cast<Value>(index - _processed_weight - _weight(n - 1) / 2.0);
auto z2 = static_cast<Value>(_weight(n - 1) / 2 - z1);
result[level_index] =
static_cast<double>(_weighted_average(_mean(n - 1), z1, _max, z2));
}
return quantileProcessed(q);
}
// this returns a quantile on the currently processed values without changing the t-digest
// the value will not represent the unprocessed values
Value quantileProcessed(Value q) const {
Value quantile_processed(Value q) const {
if (q < 0 || q > 1) {
VLOG_CRITICAL << "q should be in [0,1], got " << q;
return NAN;
@@ -371,7 +428,7 @@ public:
} else if (_processed.size() == 1) {
// with one data point, all quantiles lead to Rome
return mean(0);
return _mean(0);
}
// we know that there are at least two sorted now
@@ -381,9 +438,9 @@ public:
const auto index = q * _processed_weight;
// at the boundaries, we return _min or _max
if (index <= weight(0) / 2.0) {
DCHECK_GT(weight(0), 0);
return static_cast<Value>(_min + 2.0 * index / weight(0) * (mean(0) - _min));
if (index <= _weight(0) / 2.0) {
DCHECK_GT(_weight(0), 0);
return static_cast<Value>(_min + 2.0 * index / _weight(0) * (_mean(0) - _min));
}
auto iter = std::lower_bound(_cumulative.cbegin(), _cumulative.cend(), index);
@@ -394,22 +451,22 @@ public:
auto z1 = index - *(iter - 1);
auto z2 = *(iter)-index;
// VLOG_CRITICAL << "z2 " << z2 << " index " << index << " z1 " << z1;
return weightedAverage(mean(i - 1), z2, mean(i), z1);
return _weighted_average(_mean(i - 1), z2, _mean(i), z1);
}
DCHECK_LE(index, _processed_weight);
DCHECK_GE(index, _processed_weight - weight(n - 1) / 2.0);
DCHECK_GE(index, _processed_weight - _weight(n - 1) / 2.0);
auto z1 = static_cast<Value>(index - _processed_weight - weight(n - 1) / 2.0);
auto z2 = static_cast<Value>(weight(n - 1) / 2 - z1);
return weightedAverage(mean(n - 1), z1, _max, z2);
auto z1 = static_cast<Value>(index - _processed_weight - _weight(n - 1) / 2.0);
auto z2 = static_cast<Value>(_weight(n - 1) / 2 - z1);
return _weighted_average(_mean(n - 1), z1, _max, z2);
}
Value compression() const { return _compression; }
void add(Value x) { add(x, 1); }
void compress() { process(); }
void compress() { _process(); }
// add a single centroid to the unprocessed vector, processing previously unprocessed sorted if our limit has
// been reached.
@@ -419,7 +476,7 @@ public:
}
_unprocessed.emplace_back(x, w);
_unprocessed_weight += w;
processIfNecessary();
_process_if_necessary();
return true;
}
@@ -433,7 +490,7 @@ public:
_unprocessed.push_back(*(iter++));
}
if (_unprocessed.size() >= _max_unprocessed) {
process();
_process();
}
}
}
@@ -541,7 +598,9 @@ private:
Value _min = std::numeric_limits<Value>::max();
Value _max = std::numeric_limits<Value>::min();
// min() is the smallest positive value, so use lowest() for all-negative input,
// e.g. {-3, -2, -1} must set _max to -1.
Value _max = std::numeric_limits<Value>::lowest();
Index _max_processed;
@@ -558,13 +617,13 @@ private:
std::vector<Weight> _cumulative;
// return mean of i-th centroid
Value mean(int64_t i) const noexcept { return _processed[i].mean(); }
Value _mean(int64_t i) const noexcept { return _processed[i].mean(); }
// return weight of i-th centroid
Weight weight(int64_t i) const noexcept { return _processed[i].weight(); }
Weight _weight(int64_t i) const noexcept { return _processed[i].weight(); }
// append all unprocessed centroids into current unprocessed vector
void mergeUnprocessed(const std::vector<const TDigest*>& tdigests) {
void _merge_unprocessed(const std::vector<const TDigest*>& tdigests) {
if (tdigests.size() == 0) {
return;
}
@@ -583,7 +642,7 @@ private:
}
// merge all processed centroids together into a single sorted vector
void mergeProcessed(const std::vector<const TDigest*>& tdigests) {
void _merge_processed(const std::vector<const TDigest*>& tdigests) {
if (tdigests.size() == 0) {
return;
}
@@ -627,21 +686,21 @@ private:
}
}
void processIfNecessary() {
if (isDirty()) {
process();
void _process_if_necessary() {
if (is_dirty()) {
_process();
}
}
void updateCumulative() {
void _update_cumulative() {
const auto n = _processed.size();
_cumulative.clear();
_cumulative.reserve(n + 1);
Weight previous = 0.0;
for (Index i = 0; i < n; i++) {
Weight current = weight(i);
auto halfCurrent = static_cast<Weight>(current / 2.0);
_cumulative.push_back(previous + halfCurrent);
Weight current = _weight(i);
auto half_current = static_cast<Weight>(current / 2.0);
_cumulative.push_back(previous + half_current);
previous = previous + current;
}
_cumulative.push_back(previous);
@@ -649,7 +708,7 @@ private:
// merges _unprocessed centroids and _processed centroids together and processes them
// when complete, _unprocessed will be empty and _processed will have at most _max_processed centroids
void process() {
void _process() {
CentroidComparator cc;
// select percentile_approx(lo_orderkey,0.5) from lineorder;
// have test pdqsort and RadixSort, find here pdqsort performance is better when data is struct Centroid
@@ -665,20 +724,20 @@ private:
_processed.clear();
_processed.push_back(_unprocessed[0]);
Weight wSoFar = _unprocessed[0].weight();
Weight wLimit = _processed_weight * integratedQ(1.0);
Weight w_so_far = _unprocessed[0].weight();
Weight w_limit = _processed_weight * _integrated_q(1.0);
auto end = _unprocessed.end();
for (auto iter = _unprocessed.cbegin() + 1; iter < end; iter++) {
const auto& centroid = *iter;
Weight projectedW = wSoFar + centroid.weight();
if (projectedW <= wLimit) {
wSoFar = projectedW;
Weight projected_w = w_so_far + centroid.weight();
if (projected_w <= w_limit) {
w_so_far = projected_w;
(_processed.end() - 1)->add(centroid);
} else {
auto k1 = integratedLocation(wSoFar / _processed_weight);
wLimit = _processed_weight * integratedQ(static_cast<Value>(k1 + 1.0));
wSoFar += centroid.weight();
auto k1 = _integrated_location(w_so_far / _processed_weight);
w_limit = _processed_weight * _integrated_q(static_cast<Value>(k1 + 1.0));
w_so_far += centroid.weight();
_processed.emplace_back(centroid);
}
}
@@ -687,34 +746,34 @@ private:
VLOG_CRITICAL << "new _min " << _min;
_max = std::max(_max, (_processed.cend() - 1)->mean());
VLOG_CRITICAL << "new _max " << _max;
updateCumulative();
_update_cumulative();
}
size_t checkWeights(const std::vector<Centroid>& sorted, Value total) {
size_t badWeight = 0;
size_t _check_weights(const std::vector<Centroid>& sorted, Value total) {
size_t bad_weight = 0;
auto k1 = 0.0;
auto q = 0.0;
for (auto iter = sorted.cbegin(); iter != sorted.cend(); iter++) {
auto w = iter->weight();
auto dq = w / total;
auto k2 = integratedLocation(static_cast<Value>(q + dq));
auto k2 = _integrated_location(static_cast<Value>(q + dq));
if (k2 - k1 > 1 && w != 1) {
VLOG_CRITICAL << "Oversize centroid at " << std::distance(sorted.cbegin(), iter)
<< " k1 " << k1 << " k2 " << k2 << " dk " << (k2 - k1) << " w " << w
<< " q " << q;
badWeight++;
bad_weight++;
}
if (k2 - k1 > 1.5 && w != 1) {
VLOG_CRITICAL << "Egregiously Oversize centroid at "
<< std::distance(sorted.cbegin(), iter) << " k1 " << k1 << " k2 "
<< k2 << " dk " << (k2 - k1) << " w " << w << " q " << q;
badWeight++;
bad_weight++;
}
q += dq;
k1 = k2;
}
return badWeight;
return bad_weight;
}
/**
@@ -734,23 +793,23 @@ private:
* @param q The quantile scale value to be mapped.
* @return The centroid scale value corresponding to q.
*/
Value integratedLocation(Value q) const {
Value _integrated_location(Value q) const {
return static_cast<Value>(_compression * (std::asin(2.0 * q - 1.0) + M_PI / 2) / M_PI);
}
Value integratedQ(Value k) const {
Value _integrated_q(Value k) const {
return static_cast<Value>(
(std::sin(std::min(k, _compression) * M_PI / _compression - M_PI / 2) + 1) / 2);
}
/**
* Same as {@link #weightedAverageSorted(Value, Value, Value, Value)} but flips
* Same as {@link #_weighted_average_sorted(Value, Value, Value, Value)} but flips
* the order of the variables if <code>x2</code> is greater than
* <code>x1</code>.
*/
static Value weightedAverage(Value x1, Value w1, Value x2, Value w2) {
return (x1 <= x2) ? weightedAverageSorted(x1, w1, x2, w2)
: weightedAverageSorted(x2, w2, x1, w1);
static Value _weighted_average(Value x1, Value w1, Value x2, Value w2) {
return (x1 <= x2) ? _weighted_average_sorted(x1, w1, x2, w2)
: _weighted_average_sorted(x2, w2, x1, w1);
}
/**
@@ -760,13 +819,13 @@ private:
* and is guaranteed to return a number between <code>x1</code> and
* <code>x2</code>.
*/
static Value weightedAverageSorted(Value x1, Value w1, Value x2, Value w2) {
static Value _weighted_average_sorted(Value x1, Value w1, Value x2, Value w2) {
DCHECK_LE(x1, x2);
const Value x = (x1 * w1 + x2 * w2) / (w1 + w2);
return std::max(x1, std::min(x, x2));
}
static Value interpolate(Value x, Value x0, Value x1) { return (x - x0) / (x1 - x0); }
static Value _interpolate(Value x, Value x0, Value x1) { return (x - x0) / (x1 - x0); }
/**
* Computes an interpolated value of a quantile that is between two sorted.
@@ -774,18 +833,18 @@ private:
* Index is the quantile desired multiplied by the total number of samples - 1.
*
* @param index Denormalized quantile desired
* @param previousIndex The denormalized quantile corresponding to the center of the previous centroid.
* @param nextIndex The denormalized quantile corresponding to the center of the following centroid.
* @param previousMean The mean of the previous centroid.
* @param nextMean The mean of the following centroid.
* @param previous_index The denormalized quantile corresponding to the center of the previous centroid.
* @param next_index The denormalized quantile corresponding to the center of the following centroid.
* @param previous_mean The mean of the previous centroid.
* @param next_mean The mean of the following centroid.
* @return The interpolated mean.
*/
static Value quantile(Value index, Value previousIndex, Value nextIndex, Value previousMean,
Value nextMean) {
const auto delta = nextIndex - previousIndex;
const auto previousWeight = (nextIndex - index) / delta;
const auto nextWeight = (index - previousIndex) / delta;
return previousMean * previousWeight + nextMean * nextWeight;
static Value _quantile(Value index, Value previous_index, Value next_index, Value previous_mean,
Value next_mean) {
const auto delta = next_index - previous_index;
const auto previous_weight = (next_index - index) / delta;
const auto next_weight = (index - previous_index) / delta;
return previous_mean * previous_weight + next_mean * next_weight;
}
};
} // namespace doris
@@ -0,0 +1,353 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#include <gtest/gtest.h>
#include <limits>
#include <memory>
#include <optional>
#include <vector>
#include "core/column/column_array.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "core/string_buffer.hpp"
#include "exprs/aggregate/aggregate_function_percentile.h"
#include "exprs/aggregate/aggregate_function_simple_factory.h"
#include "util/tdigest.h"
namespace doris {
namespace {
AggregateFunctionPtr create_percentile_approx_array_function(bool has_compression) {
DataTypes argument_types {
std::make_shared<DataTypeFloat64>(),
std::make_shared<DataTypeArray>(make_nullable(std::make_shared<DataTypeFloat64>()))};
if (has_compression) {
argument_types.push_back(std::make_shared<DataTypeFloat64>());
}
auto result_type =
std::make_shared<DataTypeArray>(make_nullable(std::make_shared<DataTypeFloat64>()));
return AggregateFunctionSimpleFactory::instance().get(
"percentile_approx_array", argument_types, result_type, false,
BeExecVersionManager::get_newest_version(), {.column_names = {}});
}
MutableColumnPtr create_values(const std::vector<double>& values) {
auto column = ColumnFloat64::create();
for (double value : values) {
column->insert_value(value);
}
return column;
}
MutableColumnPtr create_quantiles(const std::vector<double>& quantiles, size_t rows,
std::optional<size_t> null_index = std::nullopt) {
auto nested_column = ColumnFloat64::create();
auto null_map = ColumnUInt8::create();
auto offsets = ColumnArray::ColumnOffsets::create();
for (size_t row = 0; row < rows; ++row) {
for (size_t i = 0; i < quantiles.size(); ++i) {
nested_column->insert_value(quantiles[i]);
null_map->insert_value(null_index.has_value() && *null_index == i);
}
offsets->insert_value((row + 1) * quantiles.size());
}
return ColumnArray::create(
ColumnNullable::create(std::move(nested_column), std::move(null_map)),
std::move(offsets));
}
std::vector<double> read_result(const AggregateFunctionPtr& function, AggregateDataPtr place) {
auto result_column = ColumnArray::create(
ColumnNullable::create(ColumnFloat64::create(), ColumnUInt8::create()),
ColumnArray::ColumnOffsets::create());
function->insert_result_into(place, *result_column);
const auto& nullable_data = assert_cast<const ColumnNullable&>(result_column->get_data());
const auto& data = assert_cast<const ColumnFloat64&>(nullable_data.get_nested_column());
return {data.get_data().begin(), data.get_data().end()};
}
std::vector<double> expected_quantiles(const std::vector<double>& values,
const std::vector<double>& quantiles, float compression) {
TDigest digest(compression);
for (double value : values) {
digest.add(value);
}
std::vector<double> result;
result.reserve(quantiles.size());
for (double quantile : quantiles) {
result.push_back(digest.quantile(quantile));
}
return result;
}
void expect_results_equal(const std::vector<double>& actual, const std::vector<double>& expected) {
ASSERT_EQ(actual.size(), expected.size());
for (size_t i = 0; i < actual.size(); ++i) {
EXPECT_DOUBLE_EQ(actual[i], expected[i]);
}
}
} // namespace
TEST(AggregateFunctionPercentileApproxArrayTest, AddAndBatchPaths) {
const std::vector<double> values {1, 2, 3, 4, 5, 100};
const std::vector<double> quantiles {0.9, 0.0, 0.5, 0.5, 1.0};
auto function = create_percentile_approx_array_function(false);
ASSERT_NE(function, nullptr);
auto value_column = create_values(values);
auto quantile_column = create_quantiles(quantiles, values.size());
const IColumn* columns[] = {value_column.get(), quantile_column.get()};
Arena arena;
std::unique_ptr<char[]> row_memory(new char[function->size_of_data()]);
AggregateDataPtr row_place = row_memory.get();
function->create(row_place);
for (size_t i = 0; i < values.size(); ++i) {
function->add(row_place, columns, i, arena);
}
std::unique_ptr<char[]> batch_memory(new char[function->size_of_data()]);
AggregateDataPtr batch_place = batch_memory.get();
function->create(batch_place);
function->add_batch_single_place(values.size(), batch_place, columns, arena);
std::unique_ptr<char[]> range_memory(new char[function->size_of_data()]);
AggregateDataPtr range_place = range_memory.get();
function->create(range_place);
function->add_batch_range(1, 4, range_place, columns, arena, false);
const auto expected = expected_quantiles(values, quantiles, 10000);
expect_results_equal(read_result(function, row_place), expected);
expect_results_equal(read_result(function, batch_place), expected);
expect_results_equal(read_result(function, range_place),
expected_quantiles({2, 3, 4, 5}, quantiles, 10000));
function->destroy(row_place);
function->destroy(batch_place);
function->destroy(range_place);
}
TEST(AggregateFunctionPercentileApproxArrayTest, CompressionSerializationAndMerge) {
const std::vector<double> values {5, 10, 15, 20, 25, 30};
const std::vector<double> quantiles {0.25, 0.5, 0.75};
auto function = create_percentile_approx_array_function(true);
ASSERT_NE(function, nullptr);
auto value_column = create_values(values);
auto quantile_column = create_quantiles(quantiles, values.size());
auto compression_column = ColumnFloat64::create();
compression_column->insert_value(2048);
const IColumn* columns[] = {value_column.get(), quantile_column.get(),
compression_column.get()};
Arena arena;
std::unique_ptr<char[]> source_memory(new char[function->size_of_data()]);
AggregateDataPtr source_place = source_memory.get();
function->create(source_place);
function->add_batch_single_place(values.size(), source_place, columns, arena);
ColumnString serialized_column;
VectorBufferWriter writer(serialized_column);
function->serialize(source_place, writer);
writer.commit();
std::unique_ptr<char[]> restored_memory(new char[function->size_of_data()]);
AggregateDataPtr restored_place = restored_memory.get();
function->create(restored_place);
VectorBufferReader reader(serialized_column.get_data_at(0));
function->deserialize(restored_place, reader, arena);
std::unique_ptr<char[]> merged_memory(new char[function->size_of_data()]);
AggregateDataPtr merged_place = merged_memory.get();
function->create(merged_place);
function->merge(merged_place, restored_place, arena);
const auto expected = expected_quantiles(values, quantiles, 2048);
expect_results_equal(read_result(function, restored_place), expected);
expect_results_equal(read_result(function, merged_place), expected);
function->destroy(source_place);
function->destroy(restored_place);
function->destroy(merged_place);
}
TEST(AggregateFunctionPercentileApproxArrayTest, MergeCompatibleInitializedStates) {
const std::vector<double> left_values {5, 10, 15};
const std::vector<double> right_values {20, 25, 30};
const std::vector<double> quantiles {0.25, 0.5, 0.75};
auto function = create_percentile_approx_array_function(true);
ASSERT_NE(function, nullptr);
Arena arena;
auto create_state = [&](const std::vector<double>& values) {
auto value_column = create_values(values);
auto quantile_column = create_quantiles(quantiles, values.size());
auto compression_column = ColumnFloat64::create();
compression_column->insert_value(2048);
const IColumn* columns[] = {value_column.get(), quantile_column.get(),
compression_column.get()};
std::unique_ptr<char[]> memory(new char[function->size_of_data()]);
function->create(memory.get());
function->add_batch_single_place(values.size(), memory.get(), columns, arena);
return memory;
};
auto left = create_state(left_values);
auto right = create_state(right_values);
function->merge(left.get(), right.get(), arena);
std::vector<double> all_values = left_values;
all_values.insert(all_values.end(), right_values.begin(), right_values.end());
expect_results_equal(read_result(function, left.get()),
expected_quantiles(all_values, quantiles, 2048));
function->destroy(left.get());
function->destroy(right.get());
}
TEST(AggregateFunctionPercentileApproxArrayTest, EmptyQuantilesAndInvalidQuantile) {
auto function = create_percentile_approx_array_function(false);
ASSERT_NE(function, nullptr);
Arena arena;
auto value_column = create_values({1, 2, 3});
auto empty_quantile_column = create_quantiles({}, 3);
const IColumn* empty_columns[] = {value_column.get(), empty_quantile_column.get()};
std::unique_ptr<char[]> empty_memory(new char[function->size_of_data()]);
AggregateDataPtr empty_place = empty_memory.get();
function->create(empty_place);
function->add_batch_single_place(3, empty_place, empty_columns, arena);
EXPECT_TRUE(read_result(function, empty_place).empty());
const auto& empty_state = *reinterpret_cast<const PercentileApproxArrayState*>(empty_place);
EXPECT_TRUE(empty_state.init_flag);
EXPECT_EQ(empty_state.digest.get(), nullptr);
ColumnString serialized_column;
VectorBufferWriter writer(serialized_column);
function->serialize(empty_place, writer);
writer.commit();
std::unique_ptr<char[]> restored_memory(new char[function->size_of_data()]);
AggregateDataPtr restored_place = restored_memory.get();
function->create(restored_place);
VectorBufferReader reader(serialized_column.get_data_at(0));
function->deserialize(restored_place, reader, arena);
const auto& restored_state =
*reinterpret_cast<const PercentileApproxArrayState*>(restored_place);
EXPECT_TRUE(restored_state.init_flag);
EXPECT_EQ(restored_state.digest.get(), nullptr);
EXPECT_TRUE(read_result(function, restored_place).empty());
std::unique_ptr<char[]> merged_memory(new char[function->size_of_data()]);
AggregateDataPtr merged_place = merged_memory.get();
function->create(merged_place);
function->merge(merged_place, restored_place, arena);
const auto& merged_state = *reinterpret_cast<const PercentileApproxArrayState*>(merged_place);
EXPECT_TRUE(merged_state.init_flag);
EXPECT_EQ(merged_state.digest.get(), nullptr);
EXPECT_TRUE(read_result(function, merged_place).empty());
auto invalid_quantile_column = create_quantiles({0.5, 0.9}, 3, 1);
const IColumn* invalid_columns[] = {value_column.get(), invalid_quantile_column.get()};
std::unique_ptr<char[]> invalid_memory(new char[function->size_of_data()]);
AggregateDataPtr invalid_place = invalid_memory.get();
function->create(invalid_place);
EXPECT_THROW(function->add_batch_single_place(3, invalid_place, invalid_columns, arena),
Exception);
function->destroy(empty_place);
function->destroy(restored_place);
function->destroy(merged_place);
function->destroy(invalid_place);
}
TEST(AggregateFunctionPercentileApproxArrayTest, RejectsIncompatibleStates) {
auto function = create_percentile_approx_array_function(true);
ASSERT_NE(function, nullptr);
Arena arena;
auto create_state = [&](const std::vector<double>& quantiles, double compression) {
auto value_column = create_values({1, 2, 3});
auto quantile_column = create_quantiles(quantiles, 3);
auto compression_column = ColumnFloat64::create();
compression_column->insert_value(compression);
const IColumn* columns[] = {value_column.get(), quantile_column.get(),
compression_column.get()};
std::unique_ptr<char[]> memory(new char[function->size_of_data()]);
function->create(memory.get());
function->add_batch_single_place(3, memory.get(), columns, arena);
return memory;
};
auto destination = create_state({0.5}, 2048);
auto different_quantiles = create_state({0.9}, 2048);
auto different_compression = create_state({0.5}, 10000);
EXPECT_THROW(function->merge(destination.get(), different_quantiles.get(), arena), Exception);
EXPECT_THROW(function->merge(destination.get(), different_compression.get(), arena), Exception);
function->destroy(destination.get());
function->destroy(different_quantiles.get());
function->destroy(different_compression.get());
}
TEST(AggregateFunctionPercentileApproxArrayTest, HandlesNonFiniteParameters) {
const std::vector<double> non_finite_values {std::numeric_limits<double>::quiet_NaN(),
std::numeric_limits<double>::infinity(),
-std::numeric_limits<double>::infinity()};
auto value_column = create_values({1, 2, 3});
Arena arena;
auto function_without_compression = create_percentile_approx_array_function(false);
ASSERT_NE(function_without_compression, nullptr);
for (double quantile : non_finite_values) {
auto quantile_column = create_quantiles({quantile}, 3);
const IColumn* columns[] = {value_column.get(), quantile_column.get()};
std::unique_ptr<char[]> memory(new char[function_without_compression->size_of_data()]);
function_without_compression->create(memory.get());
EXPECT_THROW(function_without_compression->add_batch_single_place(3, memory.get(), columns,
arena),
Exception);
function_without_compression->destroy(memory.get());
}
auto function_with_compression = create_percentile_approx_array_function(true);
ASSERT_NE(function_with_compression, nullptr);
auto quantile_column = create_quantiles({0.5}, 3);
const auto expected = expected_quantiles({1, 2, 3}, {0.5}, 10000);
for (double compression : non_finite_values) {
auto compression_column = ColumnFloat64::create();
compression_column->insert_value(compression);
const IColumn* columns[] = {value_column.get(), quantile_column.get(),
compression_column.get()};
std::unique_ptr<char[]> memory(new char[function_with_compression->size_of_data()]);
function_with_compression->create(memory.get());
function_with_compression->add_batch_single_place(3, memory.get(), columns, arena);
const auto& state = *reinterpret_cast<const PercentileApproxArrayState*>(memory.get());
EXPECT_FLOAT_EQ(state.compressions, 10000);
expect_results_equal(read_result(function_with_compression, memory.get()), expected);
function_with_compression->destroy(memory.get());
}
}
} // namespace doris
+79 -2
View File
@@ -21,6 +21,7 @@
#include <gtest/gtest-test-part.h>
#include <memory>
#include <numeric>
#include <random>
#include "gtest/gtest_pred_impl.h"
@@ -158,8 +159,8 @@ TEST_F(TDigestTest, MoreThan2BValues) {
const auto count = 1L << 28;
digest.add(next, count);
}
EXPECT_EQ(static_cast<long>(1000 + float(10L * (1 << 28))), digest.totalWeight());
EXPECT_GT(digest.totalWeight(), std::numeric_limits<int32_t>::max());
EXPECT_EQ(static_cast<long>(1000 + float(10L * (1 << 28))), digest.total_weight());
EXPECT_GT(digest.total_weight(), std::numeric_limits<int32_t>::max());
std::vector<double> quantiles {0, 0.1, 0.5, 0.9, 1, reals(gen)};
std::sort(quantiles.begin(), quantiles.end());
auto prev = std::numeric_limits<double>::min();
@@ -216,6 +217,82 @@ TEST_F(TDigestTest, ExtremeQuantiles) {
}
}
TEST_F(TDigestTest, BatchQuantilesMatchSingleQuantiles) {
TDigest digest(1000);
for (int i = 0; i < 10000; ++i) {
digest.add(static_cast<double>((i * 37) % 1000));
}
std::vector<double> levels {0.9, 0.0, 0.5, 0.1, 1.0, 0.5, 0.99};
std::vector<size_t> permutation(levels.size());
std::iota(permutation.begin(), permutation.end(), 0);
std::sort(permutation.begin(), permutation.end(),
[&levels](size_t lhs, size_t rhs) { return levels[lhs] < levels[rhs]; });
std::vector<double> results(levels.size());
digest.quantiles(levels.data(), permutation.data(), levels.size(), results.data());
for (size_t i = 0; i < levels.size(); ++i) {
EXPECT_DOUBLE_EQ(results[i], static_cast<double>(digest.quantile(levels[i])));
}
}
TEST_F(TDigestTest, BatchQuantilesMatchSingleQuantileInLeftTail) {
TDigest digest(2);
digest.add(0.0F);
digest.compress();
digest.add(1.5F, 2.0F);
digest.add(3.0F, 3.0F);
digest.compress();
ASSERT_EQ(digest.processed().size(), 2);
ASSERT_FLOAT_EQ(digest.processed()[0].mean(), 1.0F);
ASSERT_FLOAT_EQ(digest.processed()[0].weight(), 3.0F);
ASSERT_EQ(digest.total_weight(), 6);
std::vector<double> levels {0.1};
std::vector<size_t> permutation {0};
std::vector<double> results(levels.size());
digest.quantiles(levels.data(), permutation.data(), levels.size(), results.data());
EXPECT_DOUBLE_EQ(results[0], static_cast<double>(digest.quantile(levels[0])));
}
TEST_F(TDigestTest, AllNegativeValuesHaveCorrectMaximum) {
TDigest digest(1000);
digest.add(-3.0);
digest.add(-2.0);
digest.add(-1.0);
std::vector<double> levels {1.0};
std::vector<size_t> permutation {0};
std::vector<double> results(levels.size());
digest.quantiles(levels.data(), permutation.data(), levels.size(), results.data());
EXPECT_FLOAT_EQ(digest.quantile(levels[0]), -1.0F);
EXPECT_DOUBLE_EQ(results[0], -1.0);
EXPECT_DOUBLE_EQ(results[0], static_cast<double>(digest.quantile(levels[0])));
}
TEST_F(TDigestTest, BatchQuantilesHandleEmptyAndSingleValueDigests) {
std::vector<double> levels {0.0, 0.5, 1.0};
std::vector<size_t> permutation {0, 1, 2};
std::vector<double> results(levels.size());
TDigest empty_digest(1000);
empty_digest.quantiles(levels.data(), permutation.data(), levels.size(), results.data());
for (double result : results) {
EXPECT_TRUE(std::isnan(result));
}
TDigest single_value_digest(1000);
single_value_digest.add(42.0);
single_value_digest.quantiles(levels.data(), permutation.data(), levels.size(), results.data());
for (double result : results) {
EXPECT_DOUBLE_EQ(result, 42.0);
}
}
TEST_F(TDigestTest, Montonicity) {
TDigest digest(1000);
std::uniform_real_distribution<> reals(0.0, 1.0);
@@ -77,6 +77,7 @@ import org.apache.doris.nereids.trees.expressions.functions.agg.OrthogonalBitmap
import org.apache.doris.nereids.trees.expressions.functions.agg.OrthogonalBitmapUnionCount;
import org.apache.doris.nereids.trees.expressions.functions.agg.Percentile;
import org.apache.doris.nereids.trees.expressions.functions.agg.PercentileApprox;
import org.apache.doris.nereids.trees.expressions.functions.agg.PercentileApproxArray;
import org.apache.doris.nereids.trees.expressions.functions.agg.PercentileApproxWeighted;
import org.apache.doris.nereids.trees.expressions.functions.agg.PercentileArray;
import org.apache.doris.nereids.trees.expressions.functions.agg.PercentileReservoir;
@@ -190,6 +191,7 @@ public class BuiltinAggregateFunctions implements FunctionHelper {
agg(Percentile.class, "percentile", "percentile_cont"),
agg(PercentileReservoir.class, "percentile_reservoir"),
agg(PercentileApprox.class, "percentile_approx"),
agg(PercentileApproxArray.class, "percentile_approx_array"),
agg(PercentileApproxWeighted.class, "percentile_approx_weighted"),
agg(PercentileArray.class, "percentile_array"),
agg(QuantileUnion.class, "quantile_union"),
@@ -0,0 +1,127 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
package org.apache.doris.nereids.trees.expressions.functions.agg;
import org.apache.doris.catalog.FunctionSignature;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral;
import org.apache.doris.nereids.trees.expressions.literal.Literal;
import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
import org.apache.doris.nereids.types.ArrayType;
import org.apache.doris.nereids.types.DoubleType;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
import java.util.List;
/**
* AggregateFunction 'percentile_approx_array'.
*/
public class PercentileApproxArray extends NotNullableAggregateFunction
implements ExplicitlyCastableSignature {
public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
FunctionSignature.ret(ArrayType.of(DoubleType.INSTANCE))
.args(DoubleType.INSTANCE, ArrayType.of(DoubleType.INSTANCE)),
FunctionSignature.ret(ArrayType.of(DoubleType.INSTANCE))
.args(DoubleType.INSTANCE, ArrayType.of(DoubleType.INSTANCE), DoubleType.INSTANCE)
);
public PercentileApproxArray(Expression arg0, Expression arg1) {
this(false, arg0, arg1);
}
public PercentileApproxArray(boolean distinct, Expression arg0, Expression arg1) {
super("percentile_approx_array", distinct, arg0, arg1);
}
public PercentileApproxArray(Expression arg0, Expression arg1, Expression arg2) {
this(false, arg0, arg1, arg2);
}
public PercentileApproxArray(boolean distinct, Expression arg0, Expression arg1,
Expression arg2) {
super("percentile_approx_array", distinct, arg0, arg1, arg2);
}
/** constructor for withChildren and reuse signature */
private PercentileApproxArray(AggregateFunctionParams functionParams) {
super(functionParams);
}
@Override
public void checkLegalityBeforeTypeCoercion() {
if (!getArgument(1).isConstant()) {
throw new AnalysisException(
"percentile_approx_array requires second parameter must be a constant : " + this.toSql());
}
if (arity() == 3 && !getArgument(2).isConstant()) {
throw new AnalysisException(
"percentile_approx_array requires the third parameter must be a constant : " + this.toSql());
}
}
@Override
public void checkLegalityAfterRewrite() {
checkLegalityBeforeTypeCoercion();
Expression quantiles = getArgument(1);
if (!(quantiles instanceof ArrayLiteral)) {
return;
}
for (Expression item : ((ArrayLiteral) quantiles).getValue()) {
if (item instanceof NullLiteral) {
throw new AnalysisException(
"percentile_approx_array quantile should not be null : " + this.toSql());
}
if (!(item instanceof Literal) || !item.getDataType().isNumericType()) {
continue;
}
double value = ((Literal) item).getDouble();
if (!Double.isFinite(value) || value < 0.0 || value > 1.0) {
throw new AnalysisException("percentile_approx_array quantile must be in [0, 1], but got "
+ value + ": " + this.toSql());
}
}
}
@Override
public PercentileApproxArray withDistinctAndChildren(boolean distinct, List<Expression> children) {
Preconditions.checkArgument(children.size() == 2 || children.size() == 3);
return new PercentileApproxArray(getFunctionParams(distinct, children));
}
@Override
public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
return visitor.visitPercentileApproxArray(this, context);
}
@Override
public List<FunctionSignature> getSignatures() {
return SIGNATURES;
}
@Override
public Expression resultForEmptyInput() {
return new ArrayLiteral(new ArrayList<>(), this.getDataType());
}
}
@@ -48,6 +48,7 @@ public class ForEachCombinator extends NullableAggregateFunction
add("percentile");
add("percentile_array");
add("percentile_approx");
add("percentile_approx_array");
add("percentile_approx_weighted");
}
});
@@ -117,6 +117,12 @@ public class MergeCombinator extends AggregateFunction
@Override
public void checkLegalityBeforeTypeCoercion() {
nested.checkLegalityBeforeTypeCoercion();
// A directly nested state retains the original value expressions, so their legality
// checks are still meaningful. After a subquery or stored-column boundary, the nested
// function is rebuilt with mocked slots from AggStateType; replaying value-expression
// checks there would reject valid states whose constants are already serialized.
if (getArgument(0) instanceof StateCombinator) {
nested.checkLegalityBeforeTypeCoercion();
}
}
}
@@ -144,4 +144,9 @@ public class StateCombinator extends ScalarFunction
public void checkLegalityBeforeTypeCoercion() {
nested.checkLegalityBeforeTypeCoercion();
}
@Override
public void checkLegalityAfterRewrite() {
nested.withChildren(children()).checkLegalityAfterRewrite();
}
}
@@ -112,6 +112,12 @@ public class UnionCombinator extends AggregateFunction
@Override
public void checkLegalityBeforeTypeCoercion() {
nested.checkLegalityBeforeTypeCoercion();
// A directly nested state retains the original value expressions, so their legality
// checks are still meaningful. After a subquery or stored-column boundary, the nested
// function is rebuilt with mocked slots from AggStateType; replaying value-expression
// checks there would reject valid states whose constants are already serialized.
if (getArgument(0) instanceof StateCombinator) {
nested.checkLegalityBeforeTypeCoercion();
}
}
}
@@ -71,6 +71,7 @@ import org.apache.doris.nereids.trees.expressions.functions.agg.OrthogonalBitmap
import org.apache.doris.nereids.trees.expressions.functions.agg.OrthogonalBitmapUnionCount;
import org.apache.doris.nereids.trees.expressions.functions.agg.Percentile;
import org.apache.doris.nereids.trees.expressions.functions.agg.PercentileApprox;
import org.apache.doris.nereids.trees.expressions.functions.agg.PercentileApproxArray;
import org.apache.doris.nereids.trees.expressions.functions.agg.PercentileApproxWeighted;
import org.apache.doris.nereids.trees.expressions.functions.agg.PercentileArray;
import org.apache.doris.nereids.trees.expressions.functions.agg.PercentileReservoir;
@@ -331,6 +332,10 @@ public interface AggregateFunctionVisitor<R, C> {
return visitNullableAggregateFunction(percentileApprox, context);
}
default R visitPercentileApproxArray(PercentileApproxArray percentileApproxArray, C context) {
return visitAggregateFunction(percentileApproxArray, context);
}
default R visitPercentileArray(PercentileArray percentileArray, C context) {
return visitAggregateFunction(percentileArray, context);
}
@@ -48,3 +48,8 @@ k2 agg_state<max_by(int not null, int null)> No false \N GENERIC
-- !max_by_null --
\N \N
-- !percentile_approx_merge_after_subquery --
2
-- !percentile_approx_union_after_subquery --
2
@@ -0,0 +1,38 @@
-- This file is automatically generated. You should know what you did if you want to edit this
-- !percentile_approx_array_1 --
[1, 2.25, 10, 27.5, 100]
-- !percentile_approx_array_2 --
[2.25, 10, 27.5]
-- !percentile_approx_array_3 --
[85.99997711181641, 1.200000047683716, 10, 10]
-- !percentile_approx_array_4 --
[2.25, 10, 27.5] [2.25, 10, 27.5]
-- !percentile_approx_array_5 --
[10, 10, 10]
-- !percentile_approx_array_6 --
[]
-- !percentile_approx_array_7 --
1 [1, 2, 3]
2 [10, 25, 100]
3 []
-- !percentile_approx_array_8 --
[]
-- !percentile_approx_array_9 --
[2.25, 10, 27.5] [2.25, 10, 27.5] [2.25, 10, 27.5] [2.25, 10, 27.5]
-- !percentile_approx_array_direct_merge --
[2.25, 10, 27.5]
-- !percentile_approx_array_subquery_merge --
[2.25, 10, 27.5]
-- !percentile_approx_array_subquery_union --
[2.25, 10, 27.5]
@@ -74,4 +74,39 @@ suite("test_agg_state_nereids") {
qt_union """ select max_by_merge(kstate) from (select k1,max_by_union(k2) kstate from a_table group by k1 order by k1) t; """
qt_max_by_null """ select max_by_merge(max_by_state(k1,null)),min_by_merge(min_by_state(null,k3)) from d_table; """
test {
sql """
SELECT percentile_approx_merge(percentile_approx_state(k1, k2 / 10.0))
FROM d_table
"""
exception "percentile_approx requires second parameter must be a constant"
}
test {
sql """
SELECT percentile_approx_union(percentile_approx_state(k1, k2 / 10.0))
FROM d_table
"""
exception "percentile_approx requires second parameter must be a constant"
}
qt_percentile_approx_merge_after_subquery """
SELECT percentile_approx_merge(state)
FROM (
SELECT percentile_approx_state(k1, 0.5) AS state
FROM d_table
) states
"""
qt_percentile_approx_union_after_subquery """
SELECT percentile_approx_merge(state)
FROM (
SELECT percentile_approx_union(state) AS state
FROM (
SELECT percentile_approx_state(k1, 0.5) AS state
FROM d_table
) states
) unions
"""
}
@@ -145,4 +145,9 @@ suite("test_agg_foreach") {
sql """select PERCENTILE_APPROX_foreach(a,a) from foreach_table;"""
exception "Unsupport the func"
}
test {
sql """select PERCENTILE_APPROX_ARRAY_foreach(a,a) from foreach_table;"""
exception "Unsupport the func"
}
}
@@ -0,0 +1,211 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
suite("test_aggregate_percentile_approx_array") {
sql "SET enable_agg_state = true"
sql "DROP TABLE IF EXISTS percentile_approx_array_test"
sql """
CREATE TABLE percentile_approx_array_test (
group_id INT,
value DOUBLE NULL
)
DUPLICATE KEY(group_id)
DISTRIBUTED BY HASH(group_id) BUCKETS 1
PROPERTIES("replication_num" = "1")
"""
sql """
INSERT INTO percentile_approx_array_test VALUES
(1, 1.0), (1, 2.0), (1, 3.0), (1, NULL),
(2, 10.0), (2, 20.0), (2, 30.0), (2, 100.0),
(3, NULL)
"""
qt_percentile_approx_array_1 """
SELECT percentile_approx_array(value, [0.0, 0.25, 0.5, 0.75, 1.0])
FROM percentile_approx_array_test
"""
qt_percentile_approx_array_2 """
SELECT percentile_approx_array(value, [0.25, 0.5, 0.75], 2048)
FROM percentile_approx_array_test
"""
qt_percentile_approx_array_3 """
SELECT percentile_approx_array(value, [0.9, 0.1, 0.5, 0.5])
FROM percentile_approx_array_test
"""
qt_percentile_approx_array_4 """
SELECT
percentile_approx_array(value, [0.25, 0.5, 0.75]),
percentile_approx_array(value, [0.25, 0.5, 0.75], 10001)
FROM percentile_approx_array_test
"""
qt_percentile_approx_array_5 """
SELECT percentile_approx_array(10, [0.0, 0.5, 1.0])
FROM percentile_approx_array_test
"""
qt_percentile_approx_array_6 """
SELECT percentile_approx_array(CAST(NULL AS DOUBLE), [0.0, 0.5, 1.0])
FROM percentile_approx_array_test
"""
order_qt_percentile_approx_array_7 """
SELECT group_id, percentile_approx_array(value, [0.0, 0.5, 1.0])
FROM percentile_approx_array_test
GROUP BY group_id
ORDER BY group_id
"""
qt_percentile_approx_array_8 """
SELECT percentile_approx_array(value, [])
FROM percentile_approx_array_test
"""
qt_percentile_approx_array_9 """
SELECT
percentile_approx_array(value, [0.25, 0.5, 0.75]),
percentile_approx_array(value, [0.25, 0.5, 0.75], CAST('NaN' AS DOUBLE)),
percentile_approx_array(value, [0.25, 0.5, 0.75], CAST('Infinity' AS DOUBLE)),
percentile_approx_array(value, [0.25, 0.5, 0.75], CAST('-Infinity' AS DOUBLE))
FROM percentile_approx_array_test
"""
qt_percentile_approx_array_direct_merge """
SELECT percentile_approx_array_merge(
percentile_approx_array_state(value, [0.25, 0.5, 0.75], 2048)
)
FROM percentile_approx_array_test
"""
qt_percentile_approx_array_subquery_merge """
SELECT percentile_approx_array_merge(state)
FROM (
SELECT percentile_approx_array_state(
value, [0.25, 0.5, 0.75], 2048
) AS state
FROM percentile_approx_array_test
) states
"""
qt_percentile_approx_array_subquery_union """
SELECT percentile_approx_array_merge(state)
FROM (
SELECT percentile_approx_array_union(state) AS state
FROM (
SELECT percentile_approx_array_state(
value, [0.25, 0.5, 0.75], 2048
) AS state
FROM percentile_approx_array_test
) states
) unions
"""
test {
sql """
SELECT percentile_approx_array_merge(
percentile_approx_array_state(value, array(value / 100.0, 0.5))
)
FROM percentile_approx_array_test
"""
exception "percentile_approx_array requires second parameter must be a constant"
}
test {
sql """
SELECT percentile_approx_array_merge(
percentile_approx_array_union(
percentile_approx_array_state(value, array(value / 100.0, 0.5))
)
)
FROM percentile_approx_array_test
"""
exception "percentile_approx_array requires second parameter must be a constant"
}
explain {
sql """
SELECT percentile_approx_array(value, [0.25, 0.5, 0.75])
FROM percentile_approx_array_test
"""
notContains "cast(value as DOUBLE)"
}
test {
sql """
SELECT percentile_approx_array(value, array(value / 100.0, 0.5))
FROM percentile_approx_array_test
"""
exception "percentile_approx_array requires second parameter must be a constant"
}
test {
sql """
SELECT percentile_approx_array(value, [0.5, NULL])
FROM percentile_approx_array_test
"""
exception "percentile_approx_array quantile should not be null"
}
test {
sql """
SELECT percentile_approx_array(value, [0.5, 1.1])
FROM percentile_approx_array_test
"""
exception "quantile"
}
test {
sql """
SELECT percentile_approx_array_state(CAST(NULL AS DOUBLE), [1.1])
"""
exception "percentile_approx_array quantile must be in [0, 1]"
}
test {
sql """
SELECT percentile_approx_array(value, array(CAST('NaN' AS DOUBLE)))
FROM percentile_approx_array_test
"""
exception "quantile must be in [0, 1]"
}
test {
sql """
SELECT percentile_approx_array(value, array(CAST('Infinity' AS DOUBLE)))
FROM percentile_approx_array_test
"""
exception "quantile must be in [0, 1]"
}
test {
sql """
SELECT percentile_approx_array(value, array(CAST('-Infinity' AS DOUBLE)))
FROM percentile_approx_array_test
"""
exception "quantile must be in [0, 1]"
}
test {
sql """
SELECT percentile_approx_array(value, [0.5], value)
FROM percentile_approx_array_test
"""
exception "percentile_approx_array requires the third parameter must be a constant"
}
}