This is an automated email from the ASF dual-hosted git repository.
HappenLee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new dc4fc151b99 [Feature](func) Support AggFn percentile_approx_array
(#65847)
dc4fc151b99 is described below
commit dc4fc151b997a6d1909e01ebdc6bcc60eed5b8a2
Author: linrrarity <[email protected]>
AuthorDate: Mon Aug 3 09:53:04 2026 +0800
[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)
```
---
.../aggregate/aggregate_function_percentile.cpp | 20 ++
.../aggregate/aggregate_function_percentile.h | 234 +++++++++++++-
be/src/util/percentile_util.h | 2 +-
be/src/util/tdigest.h | 267 ++++++++++------
be/test/exprs/aggregate/agg_percentile_test.cpp | 353 +++++++++++++++++++++
be/test/util/tdigest_test.cpp | 81 ++++-
.../doris/catalog/BuiltinAggregateFunctions.java | 2 +
.../functions/agg/PercentileApproxArray.java | 127 ++++++++
.../functions/combinator/ForEachCombinator.java | 1 +
.../functions/combinator/MergeCombinator.java | 8 +-
.../functions/combinator/StateCombinator.java | 5 +
.../functions/combinator/UnionCombinator.java | 8 +-
.../visitor/AggregateFunctionVisitor.java | 5 +
.../agg_state/nereids/test_agg_state_nereids.out | 5 +
.../test_aggregate_percentile_approx_array.out | 38 +++
.../nereids/test_agg_state_nereids.groovy | 35 ++
.../suites/function_p0/test_agg_foreach.groovy | 5 +
.../test_aggregate_percentile_approx_array.groovy | 211 ++++++++++++
18 files changed, 1297 insertions(+), 110 deletions(-)
diff --git a/be/src/exprs/aggregate/aggregate_function_percentile.cpp
b/be/src/exprs/aggregate/aggregate_function_percentile.cpp
index b1309ea0508..eb829f1e318 100644
--- a/be/src/exprs/aggregate/aggregate_function_percentile.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_percentile.cpp
@@ -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);
diff --git a/be/src/exprs/aggregate/aggregate_function_percentile.h
b/be/src/exprs/aggregate/aggregate_function_percentile.h
index fdf85b9a994..cfd8e239538 100644
--- a/be/src/exprs/aggregate/aggregate_function_percentile.h
+++ b/be/src/exprs/aggregate/aggregate_function_percentile.h
@@ -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;
diff --git a/be/src/util/percentile_util.h b/be/src/util/percentile_util.h
index 05003e5cb63..030e8df8069 100644
--- a/be/src/util/percentile_util.h
+++ b/be/src/util/percentile_util.h
@@ -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));
diff --git a/be/src/util/tdigest.h b/be/src/util/tdigest.h
index 7209c5f8dab..cd5570edeb6 100644
--- a/be/src/util/tdigest.h
+++ b/be/src/util/tdigest.h
@@ -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
diff --git a/be/test/exprs/aggregate/agg_percentile_test.cpp
b/be/test/exprs/aggregate/agg_percentile_test.cpp
new file mode 100644
index 00000000000..8e42fc2cc5e
--- /dev/null
+++ b/be/test/exprs/aggregate/agg_percentile_test.cpp
@@ -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
diff --git a/be/test/util/tdigest_test.cpp b/be/test/util/tdigest_test.cpp
index efb40774324..b5fec84ba48 100644
--- a/be/test/util/tdigest_test.cpp
+++ b/be/test/util/tdigest_test.cpp
@@ -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);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
index 1f60a228c11..7f3fa4cddec 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
@@ -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"),
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/PercentileApproxArray.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/PercentileApproxArray.java
new file mode 100644
index 00000000000..14010a401fe
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/PercentileApproxArray.java
@@ -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());
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/ForEachCombinator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/ForEachCombinator.java
index 3c5fce06159..0067be76642 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/ForEachCombinator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/ForEachCombinator.java
@@ -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");
}
});
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/MergeCombinator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/MergeCombinator.java
index 0ac2a6a8f3f..071df33f566 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/MergeCombinator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/MergeCombinator.java
@@ -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();
+ }
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/StateCombinator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/StateCombinator.java
index c0a45ddc609..b10db1b3e65 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/StateCombinator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/StateCombinator.java
@@ -144,4 +144,9 @@ public class StateCombinator extends ScalarFunction
public void checkLegalityBeforeTypeCoercion() {
nested.checkLegalityBeforeTypeCoercion();
}
+
+ @Override
+ public void checkLegalityAfterRewrite() {
+ nested.withChildren(children()).checkLegalityAfterRewrite();
+ }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/UnionCombinator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/UnionCombinator.java
index 5e4a701a4b7..4c54762e5a5 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/UnionCombinator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/UnionCombinator.java
@@ -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();
+ }
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
index 81b0830be66..fd76221f962 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
@@ -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);
}
diff --git
a/regression-test/data/datatype_p0/agg_state/nereids/test_agg_state_nereids.out
b/regression-test/data/datatype_p0/agg_state/nereids/test_agg_state_nereids.out
index c06bf577d2e..5ccaee80cf2 100644
---
a/regression-test/data/datatype_p0/agg_state/nereids/test_agg_state_nereids.out
+++
b/regression-test/data/datatype_p0/agg_state/nereids/test_agg_state_nereids.out
@@ -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
diff --git
a/regression-test/data/query_p0/sql_functions/aggregate_functions/test_aggregate_percentile_approx_array.out
b/regression-test/data/query_p0/sql_functions/aggregate_functions/test_aggregate_percentile_approx_array.out
new file mode 100644
index 00000000000..24c27e94901
--- /dev/null
+++
b/regression-test/data/query_p0/sql_functions/aggregate_functions/test_aggregate_percentile_approx_array.out
@@ -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]
diff --git
a/regression-test/suites/datatype_p0/agg_state/nereids/test_agg_state_nereids.groovy
b/regression-test/suites/datatype_p0/agg_state/nereids/test_agg_state_nereids.groovy
index 5c0217e9d47..75a9758a2d9 100644
---
a/regression-test/suites/datatype_p0/agg_state/nereids/test_agg_state_nereids.groovy
+++
b/regression-test/suites/datatype_p0/agg_state/nereids/test_agg_state_nereids.groovy
@@ -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
+ """
}
diff --git a/regression-test/suites/function_p0/test_agg_foreach.groovy
b/regression-test/suites/function_p0/test_agg_foreach.groovy
index 16fa21141c6..d2280e0c9ce 100644
--- a/regression-test/suites/function_p0/test_agg_foreach.groovy
+++ b/regression-test/suites/function_p0/test_agg_foreach.groovy
@@ -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"
+ }
}
diff --git
a/regression-test/suites/query_p0/sql_functions/aggregate_functions/test_aggregate_percentile_approx_array.groovy
b/regression-test/suites/query_p0/sql_functions/aggregate_functions/test_aggregate_percentile_approx_array.groovy
new file mode 100644
index 00000000000..90777b79eeb
--- /dev/null
+++
b/regression-test/suites/query_p0/sql_functions/aggregate_functions/test_aggregate_percentile_approx_array.groovy
@@ -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"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
