You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by GitBox <gi...@apache.org> on 2022/02/24 14:14:26 UTC
[GitHub] [incubator-doris] spaces-X opened a new pull request #8234: [Feature] Support pre-aggregation for quantile type
spaces-X opened a new pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234
# Proposed changes
close #7782
## Problem Summary:
Add a new column-type to speed up the approximation of quantiles.
1. The new column-type is named `quantile_state` with fixed aggregation function `quantile_union` , which
2. support pre-aggregation of new column-type and quantile_state related functions.
I will update this page to add some test results about this column-type later.
## Checklist(Required)
1. Does it affect the original behavior: (Almost No)
4. Has unit tests been added: (Yes)
5. Has document been added or modified: (Yes)
6. Does it need to update dependencies: (No)
7. Are there any changes that cannot be rolled back: (Yes/No)
## Further comments
If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r827742974
##########
File path: be/src/util/quantile_state.h
##########
@@ -0,0 +1,74 @@
+// 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.
+#ifndef DORIS_BE_SRC_OLAP_QUANTILE_STATE_H
+#define DORIS_BE_SRC_OLAP_QUANTILE_STATE_H
+
+
+#include <string>
+#include <vector>
+#include <memory>
+#include "tdigest.h"
+#include "slice.h"
+
+namespace doris {
+
+
+class Slice;
+class TDigest;
+
+const static int QUANTILE_STATE_EXPLICIT_NUM = 2048;
+const static int QUANTILE_STATE_COMPRESSION_MAX = 2048;
+const static int QUANTILE_STATE_COMPRESSION_MIN = 10000;
Review comment:
messed up of min and max
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r820042927
##########
File path: be/src/util/quantile_state.cpp
##########
@@ -0,0 +1,377 @@
+// 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 <string.h>
+#include <cmath>
+#include "util/quantile_state.h"
+#include "util/coding.h"
+#include "common/logging.h"
+
+
+namespace doris{
+
+template<typename T>
+QuantileState<T>::QuantileState():_type(EMPTY),compression(2048){}
+
+template<typename T>
+QuantileState<T>::QuantileState(float compression):_type(EMPTY),compression(compression){}
+
+
+template<typename T>
+QuantileState<T>::QuantileState(const Slice& slice) {
+ if (!deserialize(slice)) {
Review comment:
Add a ```DCHECK``` for deserialize result.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r824717556
##########
File path: fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java
##########
@@ -386,25 +386,18 @@ public void analyze(Analyzer analyzer) throws UserException {
}
}
- if (columnDef.getType().isHllType()) {
- hasHll = true;
- }
-
- if (columnDef.getAggregateType() == AggregateType.BITMAP_UNION) {
- hasBitmap = columnDef.getType().isBitmapType();
+ if (columnDef.getType().isObjectStored()) {
+ hasObjectStored = true;
+ objectStoredColumn = columnDef.getName();
}
if (!columnSet.add(columnDef.getName())) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_DUP_FIELDNAME, columnDef.getName());
}
}
- if (hasHll && keysDesc.getKeysType() != KeysType.AGG_KEYS) {
- throw new AnalysisException("HLL must be used in AGG_KEYS");
- }
-
- if (hasBitmap && keysDesc.getKeysType() != KeysType.AGG_KEYS) {
- throw new AnalysisException("BITMAP_UNION must be used in AGG_KEYS");
+ if (hasObjectStored && keysDesc.getKeysType() != KeysType.AGG_KEYS) {
+ throw new AnalysisException("object stored column must be used in AGG_KEYS table, object stored column:" + objectStoredColumn);
Review comment:
ok i will remove the object stored key words just keep the column name in message
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangshuo128 commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangshuo128 commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r824382538
##########
File path: fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java
##########
@@ -386,25 +386,18 @@ public void analyze(Analyzer analyzer) throws UserException {
}
}
- if (columnDef.getType().isHllType()) {
- hasHll = true;
- }
-
- if (columnDef.getAggregateType() == AggregateType.BITMAP_UNION) {
- hasBitmap = columnDef.getType().isBitmapType();
+ if (columnDef.getType().isObjectStored()) {
+ hasObjectStored = true;
+ objectStoredColumn = columnDef.getName();
}
if (!columnSet.add(columnDef.getName())) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_DUP_FIELDNAME, columnDef.getName());
}
}
- if (hasHll && keysDesc.getKeysType() != KeysType.AGG_KEYS) {
- throw new AnalysisException("HLL must be used in AGG_KEYS");
- }
-
- if (hasBitmap && keysDesc.getKeysType() != KeysType.AGG_KEYS) {
- throw new AnalysisException("BITMAP_UNION must be used in AGG_KEYS");
+ if (hasObjectStored && keysDesc.getKeysType() != KeysType.AGG_KEYS) {
+ throw new AnalysisException("object stored column must be used in AGG_KEYS table, object stored column:" + objectStoredColumn);
Review comment:
`object stored` is the implementation details, which should be transparent to the end-users. Let's just use the specific column type name.
```suggestion
throw new AnalysisException(objectStoredColumn + " must be used in AGG_KEYS");
```
##########
File path: fe/fe-core/src/main/java/org/apache/doris/catalog/PrimitiveType.java
##########
@@ -49,11 +49,12 @@
VARCHAR("VARCHAR", 16, TPrimitiveType.VARCHAR),
DECIMALV2("DECIMALV2", 16, TPrimitiveType.DECIMALV2),
-
- HLL("HLL", 16, TPrimitiveType.HLL),
TIME("TIME", 8, TPrimitiveType.TIME),
- // we use OBJECT type represent BITMAP type in Backend
+ // these following type are stored as object binary in BE.
Review comment:
```suggestion
// these following types are stored as object binary in BE.
```
##########
File path: be/src/exprs/quantile_function.cpp
##########
@@ -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.
+
+#include "exprs/quantile_function.h"
+#include "exprs/anyval_util.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "util/quantile_state.h"
+#include "util/string_parser.hpp"
+#include "util/slice.h"
+
+namespace doris {
+
+using doris_udf::DoubleVal;
+using doris_udf::StringVal;
+using doris_udf::FloatVal;
+
+void QuantileStateFunctions::init(){}
+
+void QuantileStateFunctions::quantile_state_init(FunctionContext* ctx, StringVal* dst) {
+ dst->is_null = false;
+ dst->len = sizeof(QuantileState<double>);
+ dst->ptr = (uint8_t*) new QuantileState<double>();
+}
+
+static StringVal serialize(FunctionContext* ctx, QuantileState<double>* value) {
+ StringVal result(ctx, value->get_serialized_size());
+ value->serialize(result.ptr);
+ return result;
+}
+
+StringVal QuantileStateFunctions::to_quantile_state(FunctionContext* ctx, const StringVal& src) {
+ QuantileState<double> quantile_state;
+ quantile_state.set_compression(2048);
Review comment:
It's better to use a named constant to represent `2048` and `10000`, etc, rather than these magic numbers.
##########
File path: be/src/olap/row_block.cpp
##########
@@ -89,8 +89,7 @@ void RowBlock::_compute_layout() {
_field_offset_in_memory.push_back(memory_size);
// All field has a nullbyte in memory
- if (column.type() == OLAP_FIELD_TYPE_VARCHAR || column.type() == OLAP_FIELD_TYPE_HLL ||
- column.type() == OLAP_FIELD_TYPE_CHAR || column.type() == OLAP_FIELD_TYPE_OBJECT ||column.type() == OLAP_FIELD_TYPE_STRING) {
+ if (column.is_slice_stored()) {
Review comment:
I prefer `is_length_variable_type` to `is_variable_type`
##########
File path: be/src/olap/aggregate_func.h
##########
@@ -580,6 +581,53 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_
: public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_OBJECT> {
};
+template <>
+struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_QUANTILE_UNION, OLAP_FIELD_TYPE_QUANTILE_STATE> {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool,
+ ObjectPool* agg_pool) {
+ DCHECK_EQ(src_null, false);
+ dst->set_not_null();
+
+ auto* src_slice = reinterpret_cast<const Slice*>(src);
+ auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
+
+ // we use zero size represent this slice is a agg object
+ dst_slice->size = 0;
+ auto* quantile_state = new QuantileState<double>(*src_slice);
+
+ mem_pool->mem_tracker()->Consume(quantile_state->get_serialized_size());
+
+ dst_slice->data = reinterpret_cast<char*>(quantile_state);
+
+ agg_pool->add(quantile_state);
+ }
+
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
+ DCHECK_EQ(src.is_null(), false);
+
+ auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
+ auto* src_slice = reinterpret_cast<const Slice*>(src.cell_ptr());
Review comment:
```suggestion
const auto* src_slice = reinterpret_cast<Slice*>(src.cell_ptr());
```
##########
File path: be/src/util/quantile_state.h
##########
@@ -0,0 +1,72 @@
+// 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.
+#ifndef DORIS_BE_SRC_OLAP_QUANTILE_STATE_H
+#define DORIS_BE_SRC_OLAP_QUANTILE_STATE_H
+
+
+#include <string>
+#include <vector>
+#include <memory>
+#include "tdigest.h"
+#include "slice.h"
+
+namespace doris {
+
+
+class Slice;
+class TDigest;
+
+const static int QUANTILE_STATE_EXPLICIT_NUM = 2048;
+
+enum QuantileStateType {
+ EMPTY = 0,
+ SINGLE = 1, // single element
+ EXPLICIT = 2, // more than one elements,stored in vector
+ TDIGEST = 3 // TDIGEST object
+};
+
+template<typename T>
+class QuantileState
+{
+public:
+ QuantileState();
+ explicit QuantileState(float compression);
+ explicit QuantileState(const Slice& slice);
+ void set_compression(float compression);
+ bool deserialize(const Slice& slice);
+ size_t serialize(uint8_t* dst) const;
+ void merge(QuantileState<T>& other);
+ void add_value(const T& value);
+ void clear();
+ bool is_valid(const Slice& slice);
+ size_t get_serialized_size();
+ T get_value_by_percentile(float percentile);
+ T get_explicit_value_by_percentile(float percentile);
+ ~QuantileState();
+
+private:
+ QuantileStateType _type = EMPTY;
+ TDigest* tdigest_ptr = nullptr;
Review comment:
```suggestion
TDigest* _tdigest_ptr = nullptr;
```
##########
File path: be/src/util/quantile_state.cpp
##########
@@ -0,0 +1,377 @@
+// 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 <string.h>
+#include <cmath>
+#include "util/quantile_state.h"
+#include "util/coding.h"
+#include "common/logging.h"
+
+
+namespace doris{
+
+template<typename T>
+QuantileState<T>::QuantileState():_type(EMPTY),compression(2048){}
+
+template<typename T>
+QuantileState<T>::QuantileState(float compression):_type(EMPTY),compression(compression){}
+
+
+template<typename T>
+QuantileState<T>::QuantileState(const Slice& slice) {
+ if (!deserialize(slice)) {
+ _type = EMPTY;
+ }
+}
+
+template<typename T>
+QuantileState<T>::~QuantileState() {
+ clear();
+}
+
+template<typename T>
+size_t QuantileState<T>::get_serialized_size() {
+ size_t size = 1 + sizeof(float); // type(QuantileStateType) + compression(float)
+ switch(_type) {
+ case EMPTY:
+ break;
+ case SINGLE:
+ size += sizeof(T);
+ break;
+ case EXPLICIT:
+ size += sizeof(uint16_t) + sizeof(T) * _explicit_data.size();
+ break;
+ case TDIGEST:
+ size += tdigest_ptr->serialized_size();
+ break;
+ }
+ return size;
+}
+
+template<typename T>
+void QuantileState<T>::set_compression(float compression) {
+ this->compression = compression;
+}
+
+template<typename T>
+bool QuantileState<T>::is_valid(const Slice& slice) {
+
+ if (slice.size < 1) {
+ return false;
+ }
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ const uint8_t* end = (uint8_t*)slice.data + slice.size;
+ float compress_value = *reinterpret_cast<const float*>(ptr);
+ if (compress_value < 2048 || compress_value > 10000) {
+ return false;
+ }
+ ptr += sizeof(float);
+
+
+ auto type = (QuantileStateType)*ptr++;
+ switch (type) {
+ case EMPTY:
+ break;
+ case SINGLE:{
+ if ((ptr + sizeof(T)) > end) {
+ return false;
+ }
+ ptr += sizeof(T);
+ // _single_data = *reinterpret_cast<T*>(ptr)++;
+ break;
+ }
+ case EXPLICIT: {
+ if ((ptr + sizeof(uint16_t)) > end) {
+ return false;
+ }
+ uint16_t num_explicits = decode_fixed16_le(ptr);
+ ptr += sizeof(uint16_t);
+ ptr += num_explicits * sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ if ((ptr + sizeof(uint32_t)) > end) {
+ return false;
+ }
+ uint32_t tdigest_serialized_length = decode_fixed32_le(ptr);
+ ptr += tdigest_serialized_length;
+ break;
+ }
+ default:
+ return false;
+ }
+ return ptr == end;
+}
+
+template<typename T>
+T QuantileState<T>::get_explicit_value_by_percentile(float percentile) {
+ DCHECK(_type == EXPLICIT);
+ if (percentile < 0 || percentile > 1) {
+ LOG(WARNING) << "get_explicit_value_by_percentile failed caused by percentile:" << percentile <<" is invalid";
+ return NAN;
+ }
+ int n = _explicit_data.size();
+ std::sort(_explicit_data.begin(), _explicit_data.end());
+
+ double index = (n - 1) * percentile;
+ int intIdx = (int) index;
+ if (intIdx == n-1 ){
+ return _explicit_data[intIdx];
+ }
+ return _explicit_data[intIdx+1] * (index - intIdx) + _explicit_data[intIdx] * (intIdx + 1 - index);
+}
+
+template<typename T>
+T QuantileState<T>::get_value_by_percentile(float percentile) {
+ switch(_type) {
+ case EMPTY: {
+ return NAN;
+ }
+ case SINGLE: {
+ return _single_data;
+ }
+ case EXPLICIT: {
+ return get_explicit_value_by_percentile(percentile);
+ }
+ case TDIGEST: {
+ return tdigest_ptr->quantile(percentile);
+ }
+ default:
+ break;
+ }
+ return NAN;
+}
+
+template<typename T>
+bool QuantileState<T>::deserialize(const Slice& slice) {
+ DCHECK(_type == EMPTY);
+
+ // in case of insert error data caused be crashed
+ if (slice.data == nullptr || slice.size <= 0) {
+ return false;
+ }
+ // check input is valid
+ if (!is_valid(slice)) {
+ LOG(WARNING) << "QuantileState deserialize failed: slice is invalid";
+ return false;
+ }
+
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ compression = *reinterpret_cast<const float*>(ptr);
+ ptr += sizeof(float);
+ // first byte : type
+ _type = (QuantileStateType)*ptr++;
+ switch (_type) {
+ case EMPTY:
+ // 1: empty
+ break;
+ case SINGLE: {
+ // 2: single_data value
+ _single_data = *reinterpret_cast<const T*>(ptr);
+ ptr += sizeof(T);
+ break;
+ }
+ case EXPLICIT: {
+ // 3: number of explicit values
+ // make sure that num_explicit is positive
+ uint16_t num_explicits = decode_fixed16_le(ptr);
+ ptr += sizeof(uint16_t);
+ _explicit_data.reserve(std::min(num_explicits*2, QUANTILE_STATE_EXPLICIT_NUM));
+ _explicit_data.resize(num_explicits);
+ memcpy(&_explicit_data[0], ptr, num_explicits*sizeof(T));
+ ptr += num_explicits*sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ // 4: Tdigest object value
+ tdigest_ptr = new TDigest(0);
+ tdigest_ptr->unserialize(ptr);
+ break;
+ }
+ default:
+ // revert type to EMPTY
+ _type = EMPTY;
+ return false;
+ }
+ return true;
+
+}
+
+template<typename T>
+size_t QuantileState<T>::serialize(uint8_t* dst) const{
+ uint8_t* ptr = dst;
+ *reinterpret_cast<float *>(ptr) = compression;
+ ptr += sizeof(float);
+ switch (_type) {
+ case EMPTY: {
+ *ptr++ = EMPTY;
+ break;
+ }
+ case SINGLE: {
+ *ptr++ = SINGLE;
+ *reinterpret_cast<T *>(ptr) = _single_data;
+ ptr += sizeof(T);
+ break;
+ }
+ case EXPLICIT: {
+ *ptr++ = EXPLICIT;
+ uint16_t size = _explicit_data.size();
+ *reinterpret_cast<uint16_t *>(ptr) = size;
+ ptr += sizeof(uint16_t);
+ memcpy(ptr, &_explicit_data[0], size*sizeof(T));
+ ptr += size*sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ *ptr++ = TDIGEST;
+ size_t tdigest_size = tdigest_ptr->serialize(ptr);
+ ptr += tdigest_size;
+ break;
+ }
+ default:
+ break;
+ }
+ return ptr - dst;
+}
+
+template<typename T>
+void QuantileState<T>::merge(QuantileState<T>& other) {
+ switch(other._type) {
+ case EMPTY:
+ break;
+ case SINGLE: {
+ add_value(other._single_data);
+ break;
+ }
+ case EXPLICIT: {
+ switch(_type) {
+ case EMPTY:
+ _type = EXPLICIT;
+ _explicit_data.swap(other._explicit_data);
+ break;
+ case SINGLE:
+ _type = EXPLICIT;
+ _explicit_data.swap(other._explicit_data);
+ add_value(_single_data);
+ break;
+ case EXPLICIT:
+ if (_explicit_data.size() + other._explicit_data.size() > QUANTILE_STATE_EXPLICIT_NUM) {
+ _type = TDIGEST;
+ tdigest_ptr = new TDigest(compression);
+ for (int i = 0; i < _explicit_data.size(); i++) {
+ tdigest_ptr->add(_explicit_data[i]);
+ }
+ for (int i = 0; i < other._explicit_data.size(); i++) {
+ tdigest_ptr->add(other._explicit_data[i]);
+ }
+ } else {
+ _explicit_data.insert(_explicit_data.end(), other._explicit_data.begin(), other._explicit_data.end());
+ }
+ break;
+ case TDIGEST:
+ for (int i = 0; i < other._explicit_data.size(); i++) {
+ tdigest_ptr->add(other._explicit_data[i]);
+ }
+ break;
+ default:
+ break;
+ }
+ break;
+ }
+ case TDIGEST: {
+ switch(_type) {
+ case EMPTY:
+ _type = TDIGEST;
+ tdigest_ptr = other.tdigest_ptr;
+ other.tdigest_ptr = nullptr;
+ break;
+ case SINGLE:
+ _type = TDIGEST;
+ tdigest_ptr = other.tdigest_ptr;
+ other.tdigest_ptr = nullptr;
+ tdigest_ptr->add(_single_data);
+ break;
+ case EXPLICIT:
+ _type = TDIGEST;
+ tdigest_ptr = other.tdigest_ptr;
+ other.tdigest_ptr = nullptr;
+ for (int i = 0; i < _explicit_data.size(); i++) {
+ tdigest_ptr->add(_explicit_data[i]);
+ }
+ break;
+ case TDIGEST:
+ tdigest_ptr->merge(other.tdigest_ptr);
+ break;
+ default:
+ break;
+ }
+ break;
+ }
+ default:
+ return;
Review comment:
The `default` branch should never be reached. What about leaving a warning log?
##########
File path: be/src/exprs/quantile_function.cpp
##########
@@ -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.
+
+#include "exprs/quantile_function.h"
+#include "exprs/anyval_util.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "util/quantile_state.h"
+#include "util/string_parser.hpp"
+#include "util/slice.h"
+
+namespace doris {
+
+using doris_udf::DoubleVal;
+using doris_udf::StringVal;
+using doris_udf::FloatVal;
+
+void QuantileStateFunctions::init(){}
+
+void QuantileStateFunctions::quantile_state_init(FunctionContext* ctx, StringVal* dst) {
+ dst->is_null = false;
+ dst->len = sizeof(QuantileState<double>);
+ dst->ptr = (uint8_t*) new QuantileState<double>();
+}
+
+static StringVal serialize(FunctionContext* ctx, QuantileState<double>* value) {
+ StringVal result(ctx, value->get_serialized_size());
+ value->serialize(result.ptr);
+ return result;
+}
+
+StringVal QuantileStateFunctions::to_quantile_state(FunctionContext* ctx, const StringVal& src) {
+ QuantileState<double> quantile_state;
+ quantile_state.set_compression(2048);
+ const AnyVal* digest_compression = ctx->get_constant_arg(1);
+ if (digest_compression != nullptr) {
+ float compression = reinterpret_cast<const FloatVal*>(digest_compression)->val;
+ if (compression >= 2048 && compression <= 10000) {
+ quantile_state.set_compression(compression);
+ }
+ }
+
+ if(!src.is_null) {
+ StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+ double double_value = StringParser::string_to_float<double>(
+ reinterpret_cast<char*>(src.ptr), src.len, &parse_result);
+ if (UNLIKELY(parse_result != StringParser::PARSE_SUCCESS)) {
+ std::stringstream error_msg;
+ error_msg << "The input: " << std::string(reinterpret_cast<char*>(src.ptr), src.len)
+ << " is not valid, to_bitmap only support bigint value from 0 to "
+ "18446744073709551615 currently";
+ ctx->set_error(error_msg.str().c_str());
+ return StringVal::null();
+ }
+ quantile_state.add_value(double_value);
+ }
+ return serialize(ctx, &quantile_state);
+}
+
+void QuantileStateFunctions::quantile_union(FunctionContext* ctx, const StringVal& src, StringVal* dst) {
+ if(src.is_null) {
+ return;
+ }
+ auto dst_quantile = reinterpret_cast<QuantileState<double>*>(dst->ptr);
+ if(src.len == 0) {
+ dst_quantile->merge(*reinterpret_cast<QuantileState<double>*>(src.ptr));
+ } else {
+ QuantileState<double> state(Slice(src.ptr, src.len));
+ dst_quantile->merge(state);
+ }
+}
+
+DoubleVal QuantileStateFunctions::quantile_percent(FunctionContext* ctx, StringVal& src) {
+ const AnyVal* percentile = ctx->get_constant_arg(1);
Review comment:
I'm afraid we should add a `quantile_percent _prepare` function to do some preparation work, e.g., handle constant arguments checks.
##########
File path: be/src/util/quantile_state.cpp
##########
@@ -0,0 +1,377 @@
+// 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 <string.h>
+#include <cmath>
+#include "util/quantile_state.h"
+#include "util/coding.h"
+#include "common/logging.h"
+
+
+namespace doris{
+
+template<typename T>
+QuantileState<T>::QuantileState():_type(EMPTY),compression(2048){}
+
+template<typename T>
+QuantileState<T>::QuantileState(float compression):_type(EMPTY),compression(compression){}
+
+
+template<typename T>
+QuantileState<T>::QuantileState(const Slice& slice) {
+ if (!deserialize(slice)) {
+ _type = EMPTY;
+ }
+}
+
+template<typename T>
+QuantileState<T>::~QuantileState() {
+ clear();
+}
+
+template<typename T>
+size_t QuantileState<T>::get_serialized_size() {
+ size_t size = 1 + sizeof(float); // type(QuantileStateType) + compression(float)
+ switch(_type) {
+ case EMPTY:
+ break;
+ case SINGLE:
+ size += sizeof(T);
+ break;
+ case EXPLICIT:
+ size += sizeof(uint16_t) + sizeof(T) * _explicit_data.size();
+ break;
+ case TDIGEST:
+ size += tdigest_ptr->serialized_size();
+ break;
+ }
+ return size;
+}
+
+template<typename T>
+void QuantileState<T>::set_compression(float compression) {
+ this->compression = compression;
+}
+
+template<typename T>
+bool QuantileState<T>::is_valid(const Slice& slice) {
+
+ if (slice.size < 1) {
+ return false;
+ }
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ const uint8_t* end = (uint8_t*)slice.data + slice.size;
+ float compress_value = *reinterpret_cast<const float*>(ptr);
+ if (compress_value < 2048 || compress_value > 10000) {
+ return false;
+ }
+ ptr += sizeof(float);
+
+
+ auto type = (QuantileStateType)*ptr++;
+ switch (type) {
+ case EMPTY:
+ break;
+ case SINGLE:{
+ if ((ptr + sizeof(T)) > end) {
+ return false;
+ }
+ ptr += sizeof(T);
+ // _single_data = *reinterpret_cast<T*>(ptr)++;
+ break;
+ }
+ case EXPLICIT: {
+ if ((ptr + sizeof(uint16_t)) > end) {
+ return false;
+ }
+ uint16_t num_explicits = decode_fixed16_le(ptr);
+ ptr += sizeof(uint16_t);
+ ptr += num_explicits * sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ if ((ptr + sizeof(uint32_t)) > end) {
+ return false;
+ }
+ uint32_t tdigest_serialized_length = decode_fixed32_le(ptr);
+ ptr += tdigest_serialized_length;
+ break;
+ }
+ default:
+ return false;
+ }
+ return ptr == end;
+}
+
+template<typename T>
+T QuantileState<T>::get_explicit_value_by_percentile(float percentile) {
+ DCHECK(_type == EXPLICIT);
+ if (percentile < 0 || percentile > 1) {
+ LOG(WARNING) << "get_explicit_value_by_percentile failed caused by percentile:" << percentile <<" is invalid";
+ return NAN;
+ }
+ int n = _explicit_data.size();
+ std::sort(_explicit_data.begin(), _explicit_data.end());
+
+ double index = (n - 1) * percentile;
+ int intIdx = (int) index;
+ if (intIdx == n-1 ){
+ return _explicit_data[intIdx];
+ }
+ return _explicit_data[intIdx+1] * (index - intIdx) + _explicit_data[intIdx] * (intIdx + 1 - index);
+}
+
+template<typename T>
+T QuantileState<T>::get_value_by_percentile(float percentile) {
+ switch(_type) {
+ case EMPTY: {
+ return NAN;
+ }
+ case SINGLE: {
+ return _single_data;
+ }
+ case EXPLICIT: {
+ return get_explicit_value_by_percentile(percentile);
+ }
+ case TDIGEST: {
+ return tdigest_ptr->quantile(percentile);
+ }
+ default:
+ break;
+ }
+ return NAN;
+}
+
+template<typename T>
+bool QuantileState<T>::deserialize(const Slice& slice) {
+ DCHECK(_type == EMPTY);
+
+ // in case of insert error data caused be crashed
+ if (slice.data == nullptr || slice.size <= 0) {
+ return false;
+ }
+ // check input is valid
+ if (!is_valid(slice)) {
+ LOG(WARNING) << "QuantileState deserialize failed: slice is invalid";
+ return false;
+ }
+
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ compression = *reinterpret_cast<const float*>(ptr);
+ ptr += sizeof(float);
+ // first byte : type
+ _type = (QuantileStateType)*ptr++;
+ switch (_type) {
+ case EMPTY:
+ // 1: empty
+ break;
+ case SINGLE: {
+ // 2: single_data value
+ _single_data = *reinterpret_cast<const T*>(ptr);
+ ptr += sizeof(T);
+ break;
+ }
+ case EXPLICIT: {
+ // 3: number of explicit values
+ // make sure that num_explicit is positive
+ uint16_t num_explicits = decode_fixed16_le(ptr);
+ ptr += sizeof(uint16_t);
+ _explicit_data.reserve(std::min(num_explicits*2, QUANTILE_STATE_EXPLICIT_NUM));
+ _explicit_data.resize(num_explicits);
+ memcpy(&_explicit_data[0], ptr, num_explicits*sizeof(T));
+ ptr += num_explicits*sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ // 4: Tdigest object value
+ tdigest_ptr = new TDigest(0);
+ tdigest_ptr->unserialize(ptr);
+ break;
+ }
+ default:
+ // revert type to EMPTY
+ _type = EMPTY;
+ return false;
+ }
+ return true;
+
+}
+
+template<typename T>
+size_t QuantileState<T>::serialize(uint8_t* dst) const{
+ uint8_t* ptr = dst;
+ *reinterpret_cast<float *>(ptr) = compression;
+ ptr += sizeof(float);
+ switch (_type) {
+ case EMPTY: {
+ *ptr++ = EMPTY;
+ break;
+ }
+ case SINGLE: {
+ *ptr++ = SINGLE;
+ *reinterpret_cast<T *>(ptr) = _single_data;
+ ptr += sizeof(T);
+ break;
+ }
+ case EXPLICIT: {
+ *ptr++ = EXPLICIT;
+ uint16_t size = _explicit_data.size();
+ *reinterpret_cast<uint16_t *>(ptr) = size;
+ ptr += sizeof(uint16_t);
+ memcpy(ptr, &_explicit_data[0], size*sizeof(T));
+ ptr += size*sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ *ptr++ = TDIGEST;
+ size_t tdigest_size = tdigest_ptr->serialize(ptr);
+ ptr += tdigest_size;
+ break;
+ }
+ default:
+ break;
+ }
+ return ptr - dst;
+}
+
+template<typename T>
+void QuantileState<T>::merge(QuantileState<T>& other) {
+ switch(other._type) {
+ case EMPTY:
+ break;
+ case SINGLE: {
+ add_value(other._single_data);
+ break;
+ }
+ case EXPLICIT: {
+ switch(_type) {
+ case EMPTY:
+ _type = EXPLICIT;
+ _explicit_data.swap(other._explicit_data);
+ break;
+ case SINGLE:
+ _type = EXPLICIT;
+ _explicit_data.swap(other._explicit_data);
+ add_value(_single_data);
+ break;
+ case EXPLICIT:
+ if (_explicit_data.size() + other._explicit_data.size() > QUANTILE_STATE_EXPLICIT_NUM) {
+ _type = TDIGEST;
+ tdigest_ptr = new TDigest(compression);
+ for (int i = 0; i < _explicit_data.size(); i++) {
+ tdigest_ptr->add(_explicit_data[i]);
+ }
+ for (int i = 0; i < other._explicit_data.size(); i++) {
+ tdigest_ptr->add(other._explicit_data[i]);
+ }
+ } else {
+ _explicit_data.insert(_explicit_data.end(), other._explicit_data.begin(), other._explicit_data.end());
+ }
+ break;
+ case TDIGEST:
+ for (int i = 0; i < other._explicit_data.size(); i++) {
+ tdigest_ptr->add(other._explicit_data[i]);
+ }
+ break;
+ default:
+ break;
+ }
+ break;
+ }
+ case TDIGEST: {
+ switch(_type) {
+ case EMPTY:
+ _type = TDIGEST;
+ tdigest_ptr = other.tdigest_ptr;
+ other.tdigest_ptr = nullptr;
+ break;
+ case SINGLE:
+ _type = TDIGEST;
+ tdigest_ptr = other.tdigest_ptr;
+ other.tdigest_ptr = nullptr;
+ tdigest_ptr->add(_single_data);
+ break;
+ case EXPLICIT:
+ _type = TDIGEST;
+ tdigest_ptr = other.tdigest_ptr;
+ other.tdigest_ptr = nullptr;
+ for (int i = 0; i < _explicit_data.size(); i++) {
+ tdigest_ptr->add(_explicit_data[i]);
+ }
+ break;
+ case TDIGEST:
+ tdigest_ptr->merge(other.tdigest_ptr);
+ break;
+ default:
+ break;
+ }
+ break;
+ }
+ default:
+ return;
+ }
+
+
+}
+
+template<typename T>
+void QuantileState<T>::add_value(const T& value) {
+ switch(_type) {
+ case EMPTY:
+ _single_data = value;
+ _type = SINGLE;
+ break;
+ case SINGLE:
+ _explicit_data.emplace_back(_single_data);
+ _explicit_data.emplace_back(value);
+ _type = EXPLICIT;
+ break;
+ case EXPLICIT:
+ if (_explicit_data.size() == QUANTILE_STATE_EXPLICIT_NUM) {
+ tdigest_ptr = new TDigest(compression);
+ for (int i = 0; i < _explicit_data.size(); i++) {
+ tdigest_ptr->add(_explicit_data[i]);
+ }
+ _explicit_data.clear();
+ _explicit_data.shrink_to_fit();
+ _type = TDIGEST;
+
+ } else {
+ _explicit_data.emplace_back(value);
+ }
+ break;
+ case TDIGEST:
+ tdigest_ptr->add(value);
+ break;
+ }
+
+}
+
+template<typename T>
+void QuantileState<T>::clear() {
+ _type = EMPTY;
+ if (tdigest_ptr != nullptr) {
+ delete tdigest_ptr;
Review comment:
It's safe to delete a `nullptr`, no need to check here.
##########
File path: be/src/util/quantile_state.h
##########
@@ -0,0 +1,72 @@
+// 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.
+#ifndef DORIS_BE_SRC_OLAP_QUANTILE_STATE_H
+#define DORIS_BE_SRC_OLAP_QUANTILE_STATE_H
+
+
+#include <string>
+#include <vector>
+#include <memory>
+#include "tdigest.h"
+#include "slice.h"
+
+namespace doris {
+
+
+class Slice;
+class TDigest;
+
+const static int QUANTILE_STATE_EXPLICIT_NUM = 2048;
+
+enum QuantileStateType {
+ EMPTY = 0,
+ SINGLE = 1, // single element
+ EXPLICIT = 2, // more than one elements,stored in vector
+ TDIGEST = 3 // TDIGEST object
+};
+
+template<typename T>
+class QuantileState
+{
+public:
+ QuantileState();
+ explicit QuantileState(float compression);
+ explicit QuantileState(const Slice& slice);
+ void set_compression(float compression);
+ bool deserialize(const Slice& slice);
+ size_t serialize(uint8_t* dst) const;
+ void merge(QuantileState<T>& other);
+ void add_value(const T& value);
+ void clear();
+ bool is_valid(const Slice& slice);
+ size_t get_serialized_size();
+ T get_value_by_percentile(float percentile);
+ T get_explicit_value_by_percentile(float percentile);
+ ~QuantileState();
+
+private:
+ QuantileStateType _type = EMPTY;
+ TDigest* tdigest_ptr = nullptr;
+ T _single_data;
+ std::vector<T> _explicit_data;
+ float compression;
Review comment:
```suggestion
float _compression;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r820044470
##########
File path: be/src/util/quantile_state.cpp
##########
@@ -0,0 +1,377 @@
+// 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 <string.h>
+#include <cmath>
+#include "util/quantile_state.h"
+#include "util/coding.h"
+#include "common/logging.h"
+
+
+namespace doris{
+
+template<typename T>
+QuantileState<T>::QuantileState():_type(EMPTY),compression(2048){}
+
+template<typename T>
+QuantileState<T>::QuantileState(float compression):_type(EMPTY),compression(compression){}
+
+
+template<typename T>
+QuantileState<T>::QuantileState(const Slice& slice) {
+ if (!deserialize(slice)) {
+ _type = EMPTY;
+ }
+}
+
+template<typename T>
+QuantileState<T>::~QuantileState() {
+ clear();
Review comment:
Maybe just delete ```tdigest_ptr``` is enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r819469487
##########
File path: be/src/olap/aggregate_func.h
##########
@@ -580,6 +581,53 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_
: public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_OBJECT> {
};
+template <>
+struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_QUANTILE_UNION, OLAP_FIELD_TYPE_QUANTILE_STATE> {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool,
+ ObjectPool* agg_pool) {
+ DCHECK_EQ(src_null, false);
+ dst->set_not_null();
+
+ auto* src_slice = reinterpret_cast<const Slice*>(src);
+ auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
+
+ // we use zero size represent this slice is a agg object
+ dst_slice->size = 0;
+ auto* quantile_state = new QuantileState<double>(*src_slice);
Review comment:
```suggestion
auto* dst_quantile_state = new QuantileState<double>(*src_slice);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#issuecomment-1069110038
> Please format the code style. Maybe https://doris.apache.org/branch-0.13/zh-CN/developer-guide/format-code.html#%E4%BB%A3%E7%A0%81%E9%A3%8E%E6%A0%BC%E5%AE%9A%E5%88%B6 would help.
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangshuo128 commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangshuo128 commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r825293491
##########
File path: be/src/exprs/quantile_function.cpp
##########
@@ -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.
+
+#include "exprs/quantile_function.h"
+#include "exprs/anyval_util.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "util/quantile_state.h"
+#include "util/string_parser.hpp"
+#include "util/slice.h"
+
+namespace doris {
+
+using doris_udf::DoubleVal;
+using doris_udf::StringVal;
+using doris_udf::FloatVal;
+
+void QuantileStateFunctions::init(){}
+
+void QuantileStateFunctions::quantile_state_init(FunctionContext* ctx, StringVal* dst) {
+ dst->is_null = false;
+ dst->len = sizeof(QuantileState<double>);
+ dst->ptr = (uint8_t*) new QuantileState<double>();
+}
+
+static StringVal serialize(FunctionContext* ctx, QuantileState<double>* value) {
+ StringVal result(ctx, value->get_serialized_size());
+ value->serialize(result.ptr);
+ return result;
+}
+
+StringVal QuantileStateFunctions::to_quantile_state(FunctionContext* ctx, const StringVal& src) {
+ QuantileState<double> quantile_state;
+ quantile_state.set_compression(2048);
+ const AnyVal* digest_compression = ctx->get_constant_arg(1);
+ if (digest_compression != nullptr) {
+ float compression = reinterpret_cast<const FloatVal*>(digest_compression)->val;
+ if (compression >= 2048 && compression <= 10000) {
+ quantile_state.set_compression(compression);
+ }
+ }
+
+ if(!src.is_null) {
+ StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+ double double_value = StringParser::string_to_float<double>(
+ reinterpret_cast<char*>(src.ptr), src.len, &parse_result);
+ if (UNLIKELY(parse_result != StringParser::PARSE_SUCCESS)) {
+ std::stringstream error_msg;
+ error_msg << "The input: " << std::string(reinterpret_cast<char*>(src.ptr), src.len)
+ << " is not valid, to_bitmap only support bigint value from 0 to "
+ "18446744073709551615 currently";
+ ctx->set_error(error_msg.str().c_str());
+ return StringVal::null();
+ }
+ quantile_state.add_value(double_value);
+ }
+ return serialize(ctx, &quantile_state);
+}
+
+void QuantileStateFunctions::quantile_union(FunctionContext* ctx, const StringVal& src, StringVal* dst) {
+ if(src.is_null) {
+ return;
+ }
+ auto dst_quantile = reinterpret_cast<QuantileState<double>*>(dst->ptr);
+ if(src.len == 0) {
+ dst_quantile->merge(*reinterpret_cast<QuantileState<double>*>(src.ptr));
+ } else {
+ QuantileState<double> state(Slice(src.ptr, src.len));
+ dst_quantile->merge(state);
+ }
+}
+
+DoubleVal QuantileStateFunctions::quantile_percent(FunctionContext* ctx, StringVal& src) {
+ const AnyVal* percentile = ctx->get_constant_arg(1);
Review comment:
Please ignore the comments before.
Hmm... I think we should do arguments' validation check in FE, i.e., `FunctionCall.analyzeImpl`, rather than in BE.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#issuecomment-1069064562
@wangbo @wangshuo128
Already accepted and completed some of your review comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r819444398
##########
File path: gensrc/thrift/PlanNodes.thrift
##########
@@ -157,8 +157,8 @@ struct TBrokerRangeDesc {
}
struct TBrokerScanRangeParams {
- 1: required i8 column_separator;
- 2: required i8 line_delimiter;
+ 1: required byte column_separator;
Review comment:
Why using byte instead of i8 here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r820043972
##########
File path: be/src/util/quantile_state.cpp
##########
@@ -0,0 +1,377 @@
+// 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 <string.h>
+#include <cmath>
+#include "util/quantile_state.h"
+#include "util/coding.h"
+#include "common/logging.h"
+
+
+namespace doris{
+
+template<typename T>
+QuantileState<T>::QuantileState():_type(EMPTY),compression(2048){}
+
+template<typename T>
+QuantileState<T>::QuantileState(float compression):_type(EMPTY),compression(compression){}
+
+
+template<typename T>
+QuantileState<T>::QuantileState(const Slice& slice) {
+ if (!deserialize(slice)) {
+ _type = EMPTY;
+ }
+}
+
+template<typename T>
+QuantileState<T>::~QuantileState() {
+ clear();
+}
+
+template<typename T>
+size_t QuantileState<T>::get_serialized_size() {
+ size_t size = 1 + sizeof(float); // type(QuantileStateType) + compression(float)
+ switch(_type) {
+ case EMPTY:
+ break;
+ case SINGLE:
+ size += sizeof(T);
+ break;
+ case EXPLICIT:
+ size += sizeof(uint16_t) + sizeof(T) * _explicit_data.size();
+ break;
+ case TDIGEST:
+ size += tdigest_ptr->serialized_size();
+ break;
+ }
+ return size;
+}
+
+template<typename T>
+void QuantileState<T>::set_compression(float compression) {
+ this->compression = compression;
+}
+
+template<typename T>
+bool QuantileState<T>::is_valid(const Slice& slice) {
+
+ if (slice.size < 1) {
+ return false;
+ }
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ const uint8_t* end = (uint8_t*)slice.data + slice.size;
+ float compress_value = *reinterpret_cast<const float*>(ptr);
+ if (compress_value < 2048 || compress_value > 10000) {
+ return false;
+ }
+ ptr += sizeof(float);
+
+
+ auto type = (QuantileStateType)*ptr++;
+ switch (type) {
+ case EMPTY:
+ break;
+ case SINGLE:{
+ if ((ptr + sizeof(T)) > end) {
+ return false;
+ }
+ ptr += sizeof(T);
+ // _single_data = *reinterpret_cast<T*>(ptr)++;
+ break;
+ }
+ case EXPLICIT: {
+ if ((ptr + sizeof(uint16_t)) > end) {
+ return false;
+ }
+ uint16_t num_explicits = decode_fixed16_le(ptr);
+ ptr += sizeof(uint16_t);
+ ptr += num_explicits * sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ if ((ptr + sizeof(uint32_t)) > end) {
+ return false;
+ }
+ uint32_t tdigest_serialized_length = decode_fixed32_le(ptr);
+ ptr += tdigest_serialized_length;
+ break;
+ }
+ default:
+ return false;
+ }
+ return ptr == end;
+}
+
+template<typename T>
+T QuantileState<T>::get_explicit_value_by_percentile(float percentile) {
+ DCHECK(_type == EXPLICIT);
+ if (percentile < 0 || percentile > 1) {
+ LOG(WARNING) << "get_explicit_value_by_percentile failed caused by percentile:" << percentile <<" is invalid";
+ return NAN;
+ }
+ int n = _explicit_data.size();
+ std::sort(_explicit_data.begin(), _explicit_data.end());
+
+ double index = (n - 1) * percentile;
+ int intIdx = (int) index;
+ if (intIdx == n-1 ){
+ return _explicit_data[intIdx];
+ }
+ return _explicit_data[intIdx+1] * (index - intIdx) + _explicit_data[intIdx] * (intIdx + 1 - index);
+}
+
+template<typename T>
+T QuantileState<T>::get_value_by_percentile(float percentile) {
+ switch(_type) {
+ case EMPTY: {
+ return NAN;
+ }
+ case SINGLE: {
+ return _single_data;
+ }
+ case EXPLICIT: {
+ return get_explicit_value_by_percentile(percentile);
+ }
+ case TDIGEST: {
+ return tdigest_ptr->quantile(percentile);
+ }
+ default:
+ break;
+ }
+ return NAN;
+}
+
+template<typename T>
+bool QuantileState<T>::deserialize(const Slice& slice) {
+ DCHECK(_type == EMPTY);
+
+ // in case of insert error data caused be crashed
+ if (slice.data == nullptr || slice.size <= 0) {
+ return false;
+ }
+ // check input is valid
+ if (!is_valid(slice)) {
+ LOG(WARNING) << "QuantileState deserialize failed: slice is invalid";
+ return false;
+ }
+
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ compression = *reinterpret_cast<const float*>(ptr);
+ ptr += sizeof(float);
+ // first byte : type
+ _type = (QuantileStateType)*ptr++;
+ switch (_type) {
+ case EMPTY:
+ // 1: empty
+ break;
+ case SINGLE: {
+ // 2: single_data value
+ _single_data = *reinterpret_cast<const T*>(ptr);
+ ptr += sizeof(T);
+ break;
+ }
+ case EXPLICIT: {
+ // 3: number of explicit values
+ // make sure that num_explicit is positive
+ uint16_t num_explicits = decode_fixed16_le(ptr);
+ ptr += sizeof(uint16_t);
+ _explicit_data.reserve(std::min(num_explicits*2, QUANTILE_STATE_EXPLICIT_NUM));
+ _explicit_data.resize(num_explicits);
+ memcpy(&_explicit_data[0], ptr, num_explicits*sizeof(T));
+ ptr += num_explicits*sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ // 4: Tdigest object value
+ tdigest_ptr = new TDigest(0);
+ tdigest_ptr->unserialize(ptr);
+ break;
+ }
+ default:
+ // revert type to EMPTY
+ _type = EMPTY;
+ return false;
+ }
+ return true;
+
+}
+
+template<typename T>
+size_t QuantileState<T>::serialize(uint8_t* dst) const{
+ uint8_t* ptr = dst;
+ *reinterpret_cast<float *>(ptr) = compression;
+ ptr += sizeof(float);
+ switch (_type) {
+ case EMPTY: {
+ *ptr++ = EMPTY;
+ break;
+ }
+ case SINGLE: {
+ *ptr++ = SINGLE;
+ *reinterpret_cast<T *>(ptr) = _single_data;
+ ptr += sizeof(T);
+ break;
+ }
+ case EXPLICIT: {
+ *ptr++ = EXPLICIT;
+ uint16_t size = _explicit_data.size();
+ *reinterpret_cast<uint16_t *>(ptr) = size;
+ ptr += sizeof(uint16_t);
+ memcpy(ptr, &_explicit_data[0], size*sizeof(T));
+ ptr += size*sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ *ptr++ = TDIGEST;
+ size_t tdigest_size = tdigest_ptr->serialize(ptr);
+ ptr += tdigest_size;
+ break;
+ }
+ default:
+ break;
+ }
+ return ptr - dst;
+}
+
+template<typename T>
+void QuantileState<T>::merge(QuantileState<T>& other) {
+ switch(other._type) {
+ case EMPTY:
+ break;
+ case SINGLE: {
+ add_value(other._single_data);
+ break;
+ }
+ case EXPLICIT: {
+ switch(_type) {
+ case EMPTY:
+ _type = EXPLICIT;
+ _explicit_data.swap(other._explicit_data);
+ break;
+ case SINGLE:
+ _type = EXPLICIT;
+ _explicit_data.swap(other._explicit_data);
+ add_value(_single_data);
+ break;
+ case EXPLICIT:
+ if (_explicit_data.size() + other._explicit_data.size() > QUANTILE_STATE_EXPLICIT_NUM) {
+ _type = TDIGEST;
+ tdigest_ptr = new TDigest(compression);
+ for (int i = 0; i < _explicit_data.size(); i++) {
+ tdigest_ptr->add(_explicit_data[i]);
+ }
+ for (int i = 0; i < other._explicit_data.size(); i++) {
+ tdigest_ptr->add(other._explicit_data[i]);
+ }
+ } else {
+ _explicit_data.insert(_explicit_data.end(), other._explicit_data.begin(), other._explicit_data.end());
+ }
+ break;
+ case TDIGEST:
+ for (int i = 0; i < other._explicit_data.size(); i++) {
+ tdigest_ptr->add(other._explicit_data[i]);
+ }
+ break;
+ default:
+ break;
+ }
+ break;
+ }
+ case TDIGEST: {
+ switch(_type) {
+ case EMPTY:
+ _type = TDIGEST;
+ tdigest_ptr = other.tdigest_ptr;
+ other.tdigest_ptr = nullptr;
+ break;
+ case SINGLE:
+ _type = TDIGEST;
+ tdigest_ptr = other.tdigest_ptr;
+ other.tdigest_ptr = nullptr;
+ tdigest_ptr->add(_single_data);
+ break;
+ case EXPLICIT:
+ _type = TDIGEST;
+ tdigest_ptr = other.tdigest_ptr;
+ other.tdigest_ptr = nullptr;
+ for (int i = 0; i < _explicit_data.size(); i++) {
+ tdigest_ptr->add(_explicit_data[i]);
+ }
+ break;
+ case TDIGEST:
+ tdigest_ptr->merge(other.tdigest_ptr);
+ break;
+ default:
+ break;
+ }
+ break;
+ }
+ default:
+ return;
+ }
+
+
+}
+
+template<typename T>
+void QuantileState<T>::add_value(const T& value) {
+ switch(_type) {
+ case EMPTY:
+ _single_data = value;
+ _type = SINGLE;
+ break;
+ case SINGLE:
+ _explicit_data.emplace_back(_single_data);
+ _explicit_data.emplace_back(value);
+ _type = EXPLICIT;
+ break;
+ case EXPLICIT:
+ if (_explicit_data.size() == QUANTILE_STATE_EXPLICIT_NUM) {
+ tdigest_ptr = new TDigest(compression);
+ for (int i = 0; i < _explicit_data.size(); i++) {
+ tdigest_ptr->add(_explicit_data[i]);
+ }
+ _explicit_data.clear();
+ _explicit_data.shrink_to_fit();
+ _type = TDIGEST;
+
+ } else {
+ _explicit_data.emplace_back(value);
+ }
+ break;
+ case TDIGEST:
+ tdigest_ptr->add(value);
+ break;
+ }
+
+}
+
+template<typename T>
+void QuantileState<T>::clear() {
+ _type = EMPTY;
+ if (tdigest_ptr != nullptr) {
+ delete tdigest_ptr;
+ tdigest_ptr = nullptr;
+ }
+ _explicit_data.clear();
+ _explicit_data.shrink_to_fit();
Review comment:
```_explicit_data``` is a field of quantitle_state, is it necessary to clear it manually?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r819470765
##########
File path: be/src/olap/aggregate_func.h
##########
@@ -580,6 +581,53 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_
: public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_OBJECT> {
};
+template <>
+struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_QUANTILE_UNION, OLAP_FIELD_TYPE_QUANTILE_STATE> {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool,
+ ObjectPool* agg_pool) {
+ DCHECK_EQ(src_null, false);
+ dst->set_not_null();
+
+ auto* src_slice = reinterpret_cast<const Slice*>(src);
+ auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
+
+ // we use zero size represent this slice is a agg object
+ dst_slice->size = 0;
+ auto* quantile_state = new QuantileState<double>(*src_slice);
+
+ mem_pool->mem_tracker()->Consume(quantile_state->get_serialized_size());
Review comment:
Why not using ```sizeof(QuantileState)``` instead of ```quantile_state->get_serialized_size()```?
Which one is more precise?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r816756313
##########
File path: be/src/olap/tablet_schema.cpp
##########
@@ -246,6 +256,7 @@ uint32_t TabletColumn::get_field_length_by_type(TPrimitiveType::type type, uint3
return 4;
case TPrimitiveType::DOUBLE:
return 8;
+ case TPrimitiveType::QUANTILE_STATE:
Review comment:
QUANTILE_STATE is variable length type, which is same to BITMAP.
It only store ptr(8 byte) and length(8 byte) in the TupleRow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r820040951
##########
File path: be/src/exprs/quantile_function.cpp
##########
@@ -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.
+
+#include "exprs/quantile_function.h"
+#include "exprs/anyval_util.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "util/quantile_state.h"
+#include "util/string_parser.hpp"
+#include "util/slice.h"
+
+namespace doris {
+
+using doris_udf::DoubleVal;
+using doris_udf::StringVal;
+using doris_udf::FloatVal;
+
+void QuantileStateFunctions::init(){}
+
+void QuantileStateFunctions::quantile_state_init(FunctionContext* ctx, StringVal* dst) {
+ dst->is_null = false;
+ dst->len = sizeof(QuantileState<double>);
+ dst->ptr = (uint8_t*) new QuantileState<double>();
+}
+
+static StringVal serialize(FunctionContext* ctx, QuantileState<double>* value) {
+ StringVal result(ctx, value->get_serialized_size());
+ value->serialize(result.ptr);
+ return result;
+}
+
+StringVal QuantileStateFunctions::to_quantile_state(FunctionContext* ctx, const StringVal& src) {
+ QuantileState<double> quantile_state;
+ quantile_state.set_compression(2048);
+ const AnyVal* digest_compression = ctx->get_constant_arg(1);
+ if (digest_compression != nullptr) {
+ float compression = reinterpret_cast<const FloatVal*>(digest_compression)->val;
+ if (compression >= 2048 && compression <= 10000) {
+ quantile_state.set_compression(compression);
+ }
+ }
+
+ if(!src.is_null) {
+ StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+ double double_value = StringParser::string_to_float<double>(
+ reinterpret_cast<char*>(src.ptr), src.len, &parse_result);
+ if (UNLIKELY(parse_result != StringParser::PARSE_SUCCESS)) {
+ std::stringstream error_msg;
+ error_msg << "The input: " << std::string(reinterpret_cast<char*>(src.ptr), src.len)
+ << " is not valid, to_bitmap only support bigint value from 0 to "
Review comment:
why ```to_bitmap``` appears here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r816704629
##########
File path: be/src/olap/row_block.cpp
##########
@@ -89,8 +89,7 @@ void RowBlock::_compute_layout() {
_field_offset_in_memory.push_back(memory_size);
// All field has a nullbyte in memory
- if (column.type() == OLAP_FIELD_TYPE_VARCHAR || column.type() == OLAP_FIELD_TYPE_HLL ||
- column.type() == OLAP_FIELD_TYPE_CHAR || column.type() == OLAP_FIELD_TYPE_OBJECT ||column.type() == OLAP_FIELD_TYPE_STRING) {
+ if (column.is_slice_stored()) {
Review comment:
Maybe ```is_variable_type``` be better than ```is_slice_stored``` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] HappenLee commented on pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
HappenLee commented on pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#issuecomment-1053850074
before we release the feature, maybe we should support it in vec exec engine both.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r816706217
##########
File path: be/src/olap/tablet_schema.cpp
##########
@@ -246,6 +256,7 @@ uint32_t TabletColumn::get_field_length_by_type(TPrimitiveType::type type, uint3
return 4;
case TPrimitiveType::DOUBLE:
return 8;
+ case TPrimitiveType::QUANTILE_STATE:
Review comment:
Why ```QUANTILE_STATE``` share the same type length with ```Object Type```?
Is QUANTILE variable length type or fixed length?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangshuo128 commented on pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangshuo128 commented on pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#issuecomment-1069073608
Please format the code style. Maybe https://doris.apache.org/branch-0.13/zh-CN/developer-guide/format-code.html#%E4%BB%A3%E7%A0%81%E9%A3%8E%E6%A0%BC%E5%AE%9A%E5%88%B6 would help.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] github-actions[bot] commented on pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#issuecomment-1076095275
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangshuo128 commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangshuo128 commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r824363371
##########
File path: be/src/exprs/anyval_util.cpp
##########
@@ -146,6 +147,10 @@ FunctionContext::TypeDesc AnyValUtil::column_type_to_type_desc(const TypeDescrip
case TYPE_OBJECT:
out.type = FunctionContext::TYPE_OBJECT;
// FIXME(cmy): is this fallthrough meaningful?
+ break; //TODO(weixiang):why no break here?
Review comment:
If not sure why no `break` here, it's better to not change the original behavior here.
Avoid introducing weird and unrelated bug by this feature.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangshuo128 commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangshuo128 commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r824419074
##########
File path: be/src/exprs/quantile_function.cpp
##########
@@ -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.
+
+#include "exprs/quantile_function.h"
+#include "exprs/anyval_util.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "util/quantile_state.h"
+#include "util/string_parser.hpp"
+#include "util/slice.h"
+
+namespace doris {
+
+using doris_udf::DoubleVal;
+using doris_udf::StringVal;
+using doris_udf::FloatVal;
+
+void QuantileStateFunctions::init(){}
+
+void QuantileStateFunctions::quantile_state_init(FunctionContext* ctx, StringVal* dst) {
+ dst->is_null = false;
+ dst->len = sizeof(QuantileState<double>);
+ dst->ptr = (uint8_t*) new QuantileState<double>();
+}
+
+static StringVal serialize(FunctionContext* ctx, QuantileState<double>* value) {
+ StringVal result(ctx, value->get_serialized_size());
+ value->serialize(result.ptr);
+ return result;
+}
+
+StringVal QuantileStateFunctions::to_quantile_state(FunctionContext* ctx, const StringVal& src) {
+ QuantileState<double> quantile_state;
+ quantile_state.set_compression(2048);
+ const AnyVal* digest_compression = ctx->get_constant_arg(1);
+ if (digest_compression != nullptr) {
+ float compression = reinterpret_cast<const FloatVal*>(digest_compression)->val;
+ if (compression >= 2048 && compression <= 10000) {
+ quantile_state.set_compression(compression);
+ }
+ }
+
+ if(!src.is_null) {
+ StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+ double double_value = StringParser::string_to_float<double>(
+ reinterpret_cast<char*>(src.ptr), src.len, &parse_result);
+ if (UNLIKELY(parse_result != StringParser::PARSE_SUCCESS)) {
+ std::stringstream error_msg;
+ error_msg << "The input: " << std::string(reinterpret_cast<char*>(src.ptr), src.len)
+ << " is not valid, to_bitmap only support bigint value from 0 to "
+ "18446744073709551615 currently";
+ ctx->set_error(error_msg.str().c_str());
+ return StringVal::null();
+ }
+ quantile_state.add_value(double_value);
+ }
+ return serialize(ctx, &quantile_state);
+}
+
+void QuantileStateFunctions::quantile_union(FunctionContext* ctx, const StringVal& src, StringVal* dst) {
+ if(src.is_null) {
+ return;
+ }
+ auto dst_quantile = reinterpret_cast<QuantileState<double>*>(dst->ptr);
+ if(src.len == 0) {
+ dst_quantile->merge(*reinterpret_cast<QuantileState<double>*>(src.ptr));
+ } else {
+ QuantileState<double> state(Slice(src.ptr, src.len));
+ dst_quantile->merge(state);
+ }
+}
+
+DoubleVal QuantileStateFunctions::quantile_percent(FunctionContext* ctx, StringVal& src) {
+ const AnyVal* percentile = ctx->get_constant_arg(1);
Review comment:
I'm afraid we should add a `quantile_percent_prepare` function to do some preparation work, e.g., handle constant arguments checks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r822330366
##########
File path: gensrc/thrift/PlanNodes.thrift
##########
@@ -157,8 +157,8 @@ struct TBrokerRangeDesc {
}
struct TBrokerScanRangeParams {
- 1: required i8 column_separator;
- 2: required i8 line_delimiter;
+ 1: required byte column_separator;
Review comment:
i got a compile error before when use i8 type.
i will try again in the next commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r822332679
##########
File path: be/src/olap/aggregate_func.h
##########
@@ -580,6 +581,53 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_
: public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_OBJECT> {
};
+template <>
+struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_QUANTILE_UNION, OLAP_FIELD_TYPE_QUANTILE_STATE> {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool,
+ ObjectPool* agg_pool) {
+ DCHECK_EQ(src_null, false);
+ dst->set_not_null();
+
+ auto* src_slice = reinterpret_cast<const Slice*>(src);
+ auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
+
+ // we use zero size represent this slice is a agg object
+ dst_slice->size = 0;
+ auto* quantile_state = new QuantileState<double>(*src_slice);
+
+ mem_pool->mem_tracker()->Consume(quantile_state->get_serialized_size());
Review comment:
The result of sizeof() is not incorrect, which is much lower than actual memory usage, cause we have vector or obj ptr in the member variables.
The result of get_serialized_size is closer to real memory usage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r820044960
##########
File path: be/src/util/quantile_state.cpp
##########
@@ -0,0 +1,377 @@
+// 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 <string.h>
+#include <cmath>
+#include "util/quantile_state.h"
+#include "util/coding.h"
+#include "common/logging.h"
+
+
+namespace doris{
+
+template<typename T>
+QuantileState<T>::QuantileState():_type(EMPTY),compression(2048){}
+
+template<typename T>
+QuantileState<T>::QuantileState(float compression):_type(EMPTY),compression(compression){}
+
+
+template<typename T>
+QuantileState<T>::QuantileState(const Slice& slice) {
+ if (!deserialize(slice)) {
+ _type = EMPTY;
+ }
+}
+
+template<typename T>
+QuantileState<T>::~QuantileState() {
+ clear();
+}
+
+template<typename T>
+size_t QuantileState<T>::get_serialized_size() {
+ size_t size = 1 + sizeof(float); // type(QuantileStateType) + compression(float)
+ switch(_type) {
+ case EMPTY:
+ break;
+ case SINGLE:
+ size += sizeof(T);
+ break;
+ case EXPLICIT:
+ size += sizeof(uint16_t) + sizeof(T) * _explicit_data.size();
+ break;
+ case TDIGEST:
+ size += tdigest_ptr->serialized_size();
+ break;
+ }
+ return size;
+}
+
+template<typename T>
+void QuantileState<T>::set_compression(float compression) {
+ this->compression = compression;
+}
+
+template<typename T>
+bool QuantileState<T>::is_valid(const Slice& slice) {
+
+ if (slice.size < 1) {
+ return false;
+ }
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ const uint8_t* end = (uint8_t*)slice.data + slice.size;
+ float compress_value = *reinterpret_cast<const float*>(ptr);
+ if (compress_value < 2048 || compress_value > 10000) {
+ return false;
+ }
+ ptr += sizeof(float);
+
+
+ auto type = (QuantileStateType)*ptr++;
+ switch (type) {
+ case EMPTY:
+ break;
+ case SINGLE:{
+ if ((ptr + sizeof(T)) > end) {
+ return false;
+ }
+ ptr += sizeof(T);
+ // _single_data = *reinterpret_cast<T*>(ptr)++;
Review comment:
remove this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r824814803
##########
File path: be/src/olap/aggregate_func.h
##########
@@ -580,6 +581,53 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_
: public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_OBJECT> {
};
+template <>
+struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_QUANTILE_UNION, OLAP_FIELD_TYPE_QUANTILE_STATE> {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool,
+ ObjectPool* agg_pool) {
+ DCHECK_EQ(src_null, false);
+ dst->set_not_null();
+
+ auto* src_slice = reinterpret_cast<const Slice*>(src);
+ auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
+
+ // we use zero size represent this slice is a agg object
+ dst_slice->size = 0;
+ auto* quantile_state = new QuantileState<double>(*src_slice);
+
+ mem_pool->mem_tracker()->Consume(quantile_state->get_serialized_size());
+
+ dst_slice->data = reinterpret_cast<char*>(quantile_state);
+
+ agg_pool->add(quantile_state);
+ }
+
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
+ DCHECK_EQ(src.is_null(), false);
+
+ auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
+ auto* src_slice = reinterpret_cast<const Slice*>(src.cell_ptr());
Review comment:
It will meet compile error cause `srr.cell_ptr` is const.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#issuecomment-1075100669
@HappenLee @wangbo
I've made the modifications as suggested by the reviewer.
Please review it soon otherwise I need to keep resolving conflicts with master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r815555742
##########
File path: be/src/exprs/quantile_function.cpp
##########
@@ -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.
+
+#include "exprs/quantile_function.h"
+#include "exprs/anyval_util.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "util/quantile_state.h"
+#include "util/string_parser.hpp"
+#include "util/slice.h"
+
+namespace doris {
+
+using doris_udf::DoubleVal;
+using doris_udf::StringVal;
+using doris_udf::FloatVal;
+
+void QuantileStateFunctions::init(){}
+
+void QuantileStateFunctions::quantile_state_init(FunctionContext* ctx, StringVal* dst) {
+ dst->is_null = false;
+ dst->len = sizeof(QuantileState<double>);
+ dst->ptr = (uint8_t*) new QuantileState<double>();
+}
+
+static StringVal serialize(FunctionContext* ctx, QuantileState<double>* value) {
+ StringVal result(ctx, value->get_serialized_size());
+ value->serialize(result.ptr);
+ return result;
+}
+
+StringVal QuantileStateFunctions::to_quantile_state(FunctionContext* ctx, const StringVal& src) {
+ QuantileState<double> quantile_state;
+ quantile_state.set_compression(2048);
+ const AnyVal* digest_compression = ctx->get_constant_arg(1);
+ if (digest_compression != nullptr) {
+ float compression = reinterpret_cast<const FloatVal*>(digest_compression)->val;
+ if (compression >= 2048 && compression <= 10000) {
+ quantile_state.set_compression(compression);
+ }
+ }
+
+ if(!src.is_null) {
+ StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+ double double_value = StringParser::string_to_float<double>(
+ reinterpret_cast<char*>(src.ptr), src.len, &parse_result);
+ if (UNLIKELY(parse_result != StringParser::PARSE_SUCCESS)) {
+ std::stringstream error_msg;
+ error_msg << "The input: " << std::string(reinterpret_cast<char*>(src.ptr), src.len)
+ << " is not valid, to_bitmap only support bigint value from 0 to "
+ "18446744073709551615 currently";
+ ctx->set_error(error_msg.str().c_str());
+ return StringVal::null();
+ }
+ quantile_state.add_value(double_value);
+ }
+ return serialize(ctx, &quantile_state);
+}
+
+void QuantileStateFunctions::quantile_union(FunctionContext* ctx, const StringVal& src, StringVal* dst) {
+ if(src.is_null) {
+ return;
+ }
+ auto dst_quantile = reinterpret_cast<QuantileState<double>*>(dst->ptr);
+ if(src.len == 0) {
+ dst_quantile->merge(*reinterpret_cast<QuantileState<double>*>(src.ptr));
+ } else {
+ QuantileState<double> state(Slice(src.ptr, src.len));
+ dst_quantile->merge(state);
+ }
+}
+
+DoubleVal QuantileStateFunctions::quantile_percent(FunctionContext* ctx, StringVal& src) {
+ const AnyVal* percentile = ctx->get_constant_arg(1);
+ if (percentile != nullptr) {
+ float percentile_value = reinterpret_cast<const FloatVal*>(percentile)->val;
+ if (percentile_value > 1 || percentile_value <0)
+ {
+ std::stringstream error_msg;
+ error_msg<< "The percentile must between 0 and 1, but input is:" << std::to_string(percentile_value);
+ ctx->set_error(error_msg.str().c_str());
+ return DoubleVal::null();
+ }
+
+ if (src.len == 0) {
+ auto quantile_state = reinterpret_cast<QuantileState<double>*>(src.ptr);
+ return {static_cast<double>(quantile_state->get_value_by_percentile(percentile_value))};
+ } else {
+ QuantileState<double> quantile_state(Slice(src.ptr, src.len));
+ return {static_cast<double>(quantile_state.get_value_by_percentile(percentile_value))};
+ }
+
+
+ } else {
+ std::stringstream error_msg;
+ error_msg << "quantile_percent function's second argument must be constant. eg: quantile_percent(col, 0.95)";
+ ctx->set_error(error_msg.str().c_str());
+ }
+ return DoubleVal::null();
+}
+
+StringVal QuantileStateFunctions::quantile_state_serialize(FunctionContext* ctx, const StringVal& src) {
+ if(src.is_null) {
+ return src;
+
+ }
+ auto tmp_ptr = reinterpret_cast<QuantileState<double>*>(src.ptr);
+ StringVal result = serialize(ctx, tmp_ptr);
+ delete tmp_ptr;
Review comment:
Why delete src.ptr here? Where is its memory allocated?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r816755862
##########
File path: be/src/olap/tablet_schema.cpp
##########
@@ -246,6 +256,7 @@ uint32_t TabletColumn::get_field_length_by_type(TPrimitiveType::type type, uint3
return 4;
case TPrimitiveType::DOUBLE:
return 8;
+ case TPrimitiveType::QUANTILE_STATE:
Review comment:
QUANTILE_STATE is variable length type, which is same to BITMAP.
It only store `ptr`(8 byte) and `length`(8 byte) in the TupleRow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r822616794
##########
File path: be/src/util/quantile_state.cpp
##########
@@ -0,0 +1,377 @@
+// 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 <string.h>
+#include <cmath>
+#include "util/quantile_state.h"
+#include "util/coding.h"
+#include "common/logging.h"
+
+
+namespace doris{
+
+template<typename T>
+QuantileState<T>::QuantileState():_type(EMPTY),compression(2048){}
+
+template<typename T>
+QuantileState<T>::QuantileState(float compression):_type(EMPTY),compression(compression){}
+
+
+template<typename T>
+QuantileState<T>::QuantileState(const Slice& slice) {
+ if (!deserialize(slice)) {
+ _type = EMPTY;
+ }
+}
+
+template<typename T>
+QuantileState<T>::~QuantileState() {
+ clear();
+}
+
+template<typename T>
+size_t QuantileState<T>::get_serialized_size() {
+ size_t size = 1 + sizeof(float); // type(QuantileStateType) + compression(float)
+ switch(_type) {
+ case EMPTY:
+ break;
+ case SINGLE:
+ size += sizeof(T);
+ break;
+ case EXPLICIT:
+ size += sizeof(uint16_t) + sizeof(T) * _explicit_data.size();
+ break;
+ case TDIGEST:
+ size += tdigest_ptr->serialized_size();
+ break;
+ }
+ return size;
+}
+
+template<typename T>
+void QuantileState<T>::set_compression(float compression) {
+ this->compression = compression;
+}
+
+template<typename T>
+bool QuantileState<T>::is_valid(const Slice& slice) {
+
+ if (slice.size < 1) {
+ return false;
+ }
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ const uint8_t* end = (uint8_t*)slice.data + slice.size;
+ float compress_value = *reinterpret_cast<const float*>(ptr);
+ if (compress_value < 2048 || compress_value > 10000) {
+ return false;
+ }
+ ptr += sizeof(float);
+
+
+ auto type = (QuantileStateType)*ptr++;
+ switch (type) {
+ case EMPTY:
+ break;
+ case SINGLE:{
+ if ((ptr + sizeof(T)) > end) {
+ return false;
+ }
+ ptr += sizeof(T);
+ // _single_data = *reinterpret_cast<T*>(ptr)++;
+ break;
+ }
+ case EXPLICIT: {
+ if ((ptr + sizeof(uint16_t)) > end) {
+ return false;
+ }
+ uint16_t num_explicits = decode_fixed16_le(ptr);
+ ptr += sizeof(uint16_t);
+ ptr += num_explicits * sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ if ((ptr + sizeof(uint32_t)) > end) {
+ return false;
+ }
+ uint32_t tdigest_serialized_length = decode_fixed32_le(ptr);
+ ptr += tdigest_serialized_length;
Review comment:
Here just check whether the total length is correct.
If check the content , we should call `deserialize()` first. It will take lots of time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangshuo128 commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangshuo128 commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r825293002
##########
File path: be/src/olap/aggregate_func.h
##########
@@ -580,6 +581,53 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_
: public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_OBJECT> {
};
+template <>
+struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_QUANTILE_UNION, OLAP_FIELD_TYPE_QUANTILE_STATE> {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool,
+ ObjectPool* agg_pool) {
+ DCHECK_EQ(src_null, false);
+ dst->set_not_null();
+
+ auto* src_slice = reinterpret_cast<const Slice*>(src);
+ auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
+
+ // we use zero size represent this slice is a agg object
+ dst_slice->size = 0;
+ auto* quantile_state = new QuantileState<double>(*src_slice);
+
+ mem_pool->mem_tracker()->Consume(quantile_state->get_serialized_size());
+
+ dst_slice->data = reinterpret_cast<char*>(quantile_state);
+
+ agg_pool->add(quantile_state);
+ }
+
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
+ DCHECK_EQ(src.is_null(), false);
+
+ auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
+ auto* src_slice = reinterpret_cast<const Slice*>(src.cell_ptr());
Review comment:
Ah, I see, my mistake.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r814433675
##########
File path: be/src/util/quantile_state.cpp
##########
@@ -0,0 +1,377 @@
+// 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 <string.h>
+#include <cmath>
+#include "util/quantile_state.h"
+#include "util/coding.h"
+#include "common/logging.h"
+
+
+namespace doris{
+
+template<typename T>
+QuantileState<T>::QuantileState():_type(EMPTY),compression(2048){}
+
+template<typename T>
+QuantileState<T>::QuantileState(float compression):_type(EMPTY),compression(compression){}
+
+
+template<typename T>
+QuantileState<T>::QuantileState(const Slice& slice) {
+ if (!deserialize(slice)) {
+ _type = EMPTY;
+ }
+}
+
+template<typename T>
+QuantileState<T>::~QuantileState() {
+ clear();
+}
+
+template<typename T>
+size_t QuantileState<T>::get_serialized_size() {
+ size_t size = 1 + sizeof(float); // type(QuantileStateType) + compression(float)
+ switch(_type) {
+ case EMPTY:
+ break;
+ case SINGLE:
+ size += sizeof(T);
+ break;
+ case EXPLICIT:
+ size += sizeof(uint16_t) + sizeof(T) * _explicit_data.size();
+ break;
+ case TDIGEST:
+ size += tdigest_ptr->serialized_size();
+ break;
+ }
+ return size;
+}
+
+template<typename T>
+void QuantileState<T>::set_compression(float compression) {
+ this->compression = compression;
+}
+
+template<typename T>
+bool QuantileState<T>::is_valid(const Slice& slice) {
+
+ if (slice.size < 1) {
+ return false;
+ }
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ const uint8_t* end = (uint8_t*)slice.data + slice.size;
+ float compress_value = *reinterpret_cast<const float*>(ptr);
+ if (compress_value < 2048 || compress_value > 10000) {
+ return false;
+ }
+ ptr += sizeof(float);
+
+
+ auto type = (QuantileStateType)*ptr++;
+ switch (type) {
+ case EMPTY:
+ break;
+ case SINGLE:{
+ if ((ptr + sizeof(T)) > end) {
+ return false;
+ }
+ ptr += sizeof(T);
+ // _single_data = *reinterpret_cast<T*>(ptr)++;
+ break;
+ }
+ case EXPLICIT: {
+ if ((ptr + sizeof(uint16_t)) > end) {
+ return false;
+ }
+ uint16_t num_explicits = decode_fixed16_le(ptr);
+ ptr += sizeof(uint16_t);
+ ptr += num_explicits * sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ if ((ptr + sizeof(uint32_t)) > end) {
+ return false;
+ }
+ uint32_t tdigest_serialized_length = decode_fixed32_le(ptr);
+ ptr += tdigest_serialized_length;
+ break;
+ }
+ default:
+ return false;
+ }
+ return ptr == end;
+}
+
+template<typename T>
+T QuantileState<T>::get_explicit_value_by_percentile(float percentile) {
+ DCHECK(_type == EXPLICIT);
+ if (percentile < 0 || percentile > 1) {
+ LOG(WARNING) << "get_explicit_value_by_percentile failed caused by percentile:" << percentile <<" is invalid";
+ return NAN;
+ }
+ int n = _explicit_data.size();
+ std::sort(_explicit_data.begin(), _explicit_data.end());
+
+ double index = (n - 1) * percentile;
+ int intIdx = (int) index;
+ if (intIdx == n-1 ){
+ return _explicit_data[intIdx];
+ }
+ return _explicit_data[intIdx+1] * (index - intIdx) + _explicit_data[intIdx] * (intIdx + 1 - index);
+}
+
+template<typename T>
+T QuantileState<T>::get_value_by_percentile(float percentile) {
+ switch(_type) {
+ case EMPTY: {
+ return NAN;
+ }
+ case SINGLE: {
+ return _single_data;
+ }
+ case EXPLICIT: {
+ return get_explicit_value_by_percentile(percentile);
+ }
+ case TDIGEST: {
+ return tdigest_ptr->quantile(percentile);
+ }
+ default:
+ break;
+ }
+ return NAN;
+}
+
+template<typename T>
+bool QuantileState<T>::deserialize(const Slice& slice) {
+ DCHECK(_type == EMPTY);
+
+ // in case of insert error data caused be crashed
+ if (slice.data == nullptr || slice.size <= 0) {
+ return false;
+ }
+ // check input is valid
+ if (!is_valid(slice)) {
+ LOG(WARNING) << "QuantileState deserialize failed: slice is invalid";
+ return false;
+ }
+
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ compression = *reinterpret_cast<const float*>(ptr);
+ ptr += sizeof(float);
+ // first byte : type
+ _type = (QuantileStateType)*ptr++;
+ switch (_type) {
+ case EMPTY:
+ // 1: empty
+ break;
+ case SINGLE: {
+ // 2: single_data value
+ _single_data = *reinterpret_cast<const T*>(ptr);
+ ptr += sizeof(T);
+ break;
+ }
+ case EXPLICIT: {
+ // 3: number of explicit values
+ // make sure that num_explicit is positive
+ uint16_t num_explicits = decode_fixed16_le(ptr);
+ ptr += sizeof(uint16_t);
+ _explicit_data.reserve(std::min(num_explicits*2, QUANTILE_STATE_EXPLICIT_NUM));
+ _explicit_data.resize(num_explicits);
+ memcpy(&_explicit_data[0], ptr, num_explicits*sizeof(T));
+ ptr += num_explicits*sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ // 4: Tdigest object value
+ tdigest_ptr = new TDigest(0);
+ tdigest_ptr->unserialize(ptr);
+ break;
+ }
+ default:
+ // revert type to EMPTY
+ _type = EMPTY;
+ return false;
+ }
+ return true;
+
+}
+
+template<typename T>
+size_t QuantileState<T>::serialize(uint8_t* dst) const{
+ uint8_t* ptr = dst;
+ *reinterpret_cast<float *>(ptr) = compression;
+ ptr += sizeof(float);
+ switch (_type) {
+ case EMPTY: {
+ *ptr++ = EMPTY;
+ break;
+ }
+ case SINGLE: {
+ *ptr++ = SINGLE;
+ *reinterpret_cast<T *>(ptr) = _single_data;
+ ptr += sizeof(T);
+ break;
+ }
+ case EXPLICIT: {
+ *ptr++ = EXPLICIT;
+ uint16_t size = _explicit_data.size();
+ *reinterpret_cast<uint16_t *>(ptr) = size;
+ ptr += sizeof(uint16_t);
+ memcpy(ptr, &_explicit_data[0], size*sizeof(T));
+ ptr += size*sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ *ptr++ = TDIGEST;
+ size_t tdigest_size = tdigest_ptr->serialize(ptr);
+ ptr += tdigest_size;
+ break;
+ }
+ default:
+ break;
+ }
+ return ptr - dst;
+}
+
+template<typename T>
+void QuantileState<T>::merge(QuantileState<T>& other) {
+ switch(other._type) {
+ case EMPTY:
+ break;
+ case SINGLE: {
+ add_value(other._single_data);
+ break;
+ }
+ case EXPLICIT: {
+ switch(_type) {
+ case EMPTY:
+ _type = EXPLICIT;
+ _explicit_data.swap(other._explicit_data);
+ break;
+ case SINGLE:
+ _type = EXPLICIT;
+ _explicit_data.swap(other._explicit_data);
+ add_value(_single_data);
+ break;
+ case EXPLICIT:
+ if (_explicit_data.size() + other._explicit_data.size() > QUANTILE_STATE_EXPLICIT_NUM) {
+ _type = TDIGEST;
+ tdigest_ptr = new TDigest(compression);
Review comment:
use `new TDigest(compression,compression,compression)` will save lots of memory.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#issuecomment-1054058228
> before we release the feature, maybe we should support it in vec exec engine both.
yeap, i think it is more reasonable to put it in another pr and i will implement it in the near future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r816756606
##########
File path: be/src/olap/row_block.cpp
##########
@@ -89,8 +89,7 @@ void RowBlock::_compute_layout() {
_field_offset_in_memory.push_back(memory_size);
// All field has a nullbyte in memory
- if (column.type() == OLAP_FIELD_TYPE_VARCHAR || column.type() == OLAP_FIELD_TYPE_HLL ||
- column.type() == OLAP_FIELD_TYPE_CHAR || column.type() == OLAP_FIELD_TYPE_OBJECT ||column.type() == OLAP_FIELD_TYPE_STRING) {
+ if (column.is_slice_stored()) {
Review comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r824834713
##########
File path: be/src/exprs/quantile_function.cpp
##########
@@ -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.
+
+#include "exprs/quantile_function.h"
+#include "exprs/anyval_util.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "util/quantile_state.h"
+#include "util/string_parser.hpp"
+#include "util/slice.h"
+
+namespace doris {
+
+using doris_udf::DoubleVal;
+using doris_udf::StringVal;
+using doris_udf::FloatVal;
+
+void QuantileStateFunctions::init(){}
+
+void QuantileStateFunctions::quantile_state_init(FunctionContext* ctx, StringVal* dst) {
+ dst->is_null = false;
+ dst->len = sizeof(QuantileState<double>);
+ dst->ptr = (uint8_t*) new QuantileState<double>();
+}
+
+static StringVal serialize(FunctionContext* ctx, QuantileState<double>* value) {
+ StringVal result(ctx, value->get_serialized_size());
+ value->serialize(result.ptr);
+ return result;
+}
+
+StringVal QuantileStateFunctions::to_quantile_state(FunctionContext* ctx, const StringVal& src) {
+ QuantileState<double> quantile_state;
+ quantile_state.set_compression(2048);
+ const AnyVal* digest_compression = ctx->get_constant_arg(1);
+ if (digest_compression != nullptr) {
+ float compression = reinterpret_cast<const FloatVal*>(digest_compression)->val;
+ if (compression >= 2048 && compression <= 10000) {
+ quantile_state.set_compression(compression);
+ }
+ }
+
+ if(!src.is_null) {
+ StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+ double double_value = StringParser::string_to_float<double>(
+ reinterpret_cast<char*>(src.ptr), src.len, &parse_result);
+ if (UNLIKELY(parse_result != StringParser::PARSE_SUCCESS)) {
+ std::stringstream error_msg;
+ error_msg << "The input: " << std::string(reinterpret_cast<char*>(src.ptr), src.len)
+ << " is not valid, to_bitmap only support bigint value from 0 to "
+ "18446744073709551615 currently";
+ ctx->set_error(error_msg.str().c_str());
+ return StringVal::null();
+ }
+ quantile_state.add_value(double_value);
+ }
+ return serialize(ctx, &quantile_state);
+}
+
+void QuantileStateFunctions::quantile_union(FunctionContext* ctx, const StringVal& src, StringVal* dst) {
+ if(src.is_null) {
+ return;
+ }
+ auto dst_quantile = reinterpret_cast<QuantileState<double>*>(dst->ptr);
+ if(src.len == 0) {
+ dst_quantile->merge(*reinterpret_cast<QuantileState<double>*>(src.ptr));
+ } else {
+ QuantileState<double> state(Slice(src.ptr, src.len));
+ dst_quantile->merge(state);
+ }
+}
+
+DoubleVal QuantileStateFunctions::quantile_percent(FunctionContext* ctx, StringVal& src) {
+ const AnyVal* percentile = ctx->get_constant_arg(1);
Review comment:
I did not get the benfit of adding a quantile_percent_prepare function. May be this can be implemented in support vec-functions of quantile_state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] spaces-X commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
spaces-X commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r815708161
##########
File path: be/src/exprs/quantile_function.cpp
##########
@@ -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.
+
+#include "exprs/quantile_function.h"
+#include "exprs/anyval_util.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "util/quantile_state.h"
+#include "util/string_parser.hpp"
+#include "util/slice.h"
+
+namespace doris {
+
+using doris_udf::DoubleVal;
+using doris_udf::StringVal;
+using doris_udf::FloatVal;
+
+void QuantileStateFunctions::init(){}
+
+void QuantileStateFunctions::quantile_state_init(FunctionContext* ctx, StringVal* dst) {
+ dst->is_null = false;
+ dst->len = sizeof(QuantileState<double>);
+ dst->ptr = (uint8_t*) new QuantileState<double>();
+}
+
+static StringVal serialize(FunctionContext* ctx, QuantileState<double>* value) {
+ StringVal result(ctx, value->get_serialized_size());
+ value->serialize(result.ptr);
+ return result;
+}
+
+StringVal QuantileStateFunctions::to_quantile_state(FunctionContext* ctx, const StringVal& src) {
+ QuantileState<double> quantile_state;
+ quantile_state.set_compression(2048);
+ const AnyVal* digest_compression = ctx->get_constant_arg(1);
+ if (digest_compression != nullptr) {
+ float compression = reinterpret_cast<const FloatVal*>(digest_compression)->val;
+ if (compression >= 2048 && compression <= 10000) {
+ quantile_state.set_compression(compression);
+ }
+ }
+
+ if(!src.is_null) {
+ StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+ double double_value = StringParser::string_to_float<double>(
+ reinterpret_cast<char*>(src.ptr), src.len, &parse_result);
+ if (UNLIKELY(parse_result != StringParser::PARSE_SUCCESS)) {
+ std::stringstream error_msg;
+ error_msg << "The input: " << std::string(reinterpret_cast<char*>(src.ptr), src.len)
+ << " is not valid, to_bitmap only support bigint value from 0 to "
+ "18446744073709551615 currently";
+ ctx->set_error(error_msg.str().c_str());
+ return StringVal::null();
+ }
+ quantile_state.add_value(double_value);
+ }
+ return serialize(ctx, &quantile_state);
+}
+
+void QuantileStateFunctions::quantile_union(FunctionContext* ctx, const StringVal& src, StringVal* dst) {
+ if(src.is_null) {
+ return;
+ }
+ auto dst_quantile = reinterpret_cast<QuantileState<double>*>(dst->ptr);
+ if(src.len == 0) {
+ dst_quantile->merge(*reinterpret_cast<QuantileState<double>*>(src.ptr));
+ } else {
+ QuantileState<double> state(Slice(src.ptr, src.len));
+ dst_quantile->merge(state);
+ }
+}
+
+DoubleVal QuantileStateFunctions::quantile_percent(FunctionContext* ctx, StringVal& src) {
+ const AnyVal* percentile = ctx->get_constant_arg(1);
+ if (percentile != nullptr) {
+ float percentile_value = reinterpret_cast<const FloatVal*>(percentile)->val;
+ if (percentile_value > 1 || percentile_value <0)
+ {
+ std::stringstream error_msg;
+ error_msg<< "The percentile must between 0 and 1, but input is:" << std::to_string(percentile_value);
+ ctx->set_error(error_msg.str().c_str());
+ return DoubleVal::null();
+ }
+
+ if (src.len == 0) {
+ auto quantile_state = reinterpret_cast<QuantileState<double>*>(src.ptr);
+ return {static_cast<double>(quantile_state->get_value_by_percentile(percentile_value))};
+ } else {
+ QuantileState<double> quantile_state(Slice(src.ptr, src.len));
+ return {static_cast<double>(quantile_state.get_value_by_percentile(percentile_value))};
+ }
+
+
+ } else {
+ std::stringstream error_msg;
+ error_msg << "quantile_percent function's second argument must be constant. eg: quantile_percent(col, 0.95)";
+ ctx->set_error(error_msg.str().c_str());
+ }
+ return DoubleVal::null();
+}
+
+StringVal QuantileStateFunctions::quantile_state_serialize(FunctionContext* ctx, const StringVal& src) {
+ if(src.is_null) {
+ return src;
+
+ }
+ auto tmp_ptr = reinterpret_cast<QuantileState<double>*>(src.ptr);
+ StringVal result = serialize(ctx, tmp_ptr);
+ delete tmp_ptr;
Review comment:
This function will be called in the PartitionedAggregationNode.
The memory is allocated from `quantile_state_init` initialize with a `empty quantile_state object`.
The `empty quantile_state` object is for aggregate calculations and it can be deleted after the aggregated results are serialized.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] morningman merged pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
morningman merged pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wangbo commented on a change in pull request #8234: [Feature] Support pre-aggregation for quantile type
Posted by GitBox <gi...@apache.org>.
wangbo commented on a change in pull request #8234:
URL: https://github.com/apache/incubator-doris/pull/8234#discussion_r820046585
##########
File path: be/src/util/quantile_state.cpp
##########
@@ -0,0 +1,377 @@
+// 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 <string.h>
+#include <cmath>
+#include "util/quantile_state.h"
+#include "util/coding.h"
+#include "common/logging.h"
+
+
+namespace doris{
+
+template<typename T>
+QuantileState<T>::QuantileState():_type(EMPTY),compression(2048){}
+
+template<typename T>
+QuantileState<T>::QuantileState(float compression):_type(EMPTY),compression(compression){}
+
+
+template<typename T>
+QuantileState<T>::QuantileState(const Slice& slice) {
+ if (!deserialize(slice)) {
+ _type = EMPTY;
+ }
+}
+
+template<typename T>
+QuantileState<T>::~QuantileState() {
+ clear();
+}
+
+template<typename T>
+size_t QuantileState<T>::get_serialized_size() {
+ size_t size = 1 + sizeof(float); // type(QuantileStateType) + compression(float)
+ switch(_type) {
+ case EMPTY:
+ break;
+ case SINGLE:
+ size += sizeof(T);
+ break;
+ case EXPLICIT:
+ size += sizeof(uint16_t) + sizeof(T) * _explicit_data.size();
+ break;
+ case TDIGEST:
+ size += tdigest_ptr->serialized_size();
+ break;
+ }
+ return size;
+}
+
+template<typename T>
+void QuantileState<T>::set_compression(float compression) {
+ this->compression = compression;
+}
+
+template<typename T>
+bool QuantileState<T>::is_valid(const Slice& slice) {
+
+ if (slice.size < 1) {
+ return false;
+ }
+ const uint8_t* ptr = (uint8_t*)slice.data;
+ const uint8_t* end = (uint8_t*)slice.data + slice.size;
+ float compress_value = *reinterpret_cast<const float*>(ptr);
+ if (compress_value < 2048 || compress_value > 10000) {
+ return false;
+ }
+ ptr += sizeof(float);
+
+
+ auto type = (QuantileStateType)*ptr++;
+ switch (type) {
+ case EMPTY:
+ break;
+ case SINGLE:{
+ if ((ptr + sizeof(T)) > end) {
+ return false;
+ }
+ ptr += sizeof(T);
+ // _single_data = *reinterpret_cast<T*>(ptr)++;
+ break;
+ }
+ case EXPLICIT: {
+ if ((ptr + sizeof(uint16_t)) > end) {
+ return false;
+ }
+ uint16_t num_explicits = decode_fixed16_le(ptr);
+ ptr += sizeof(uint16_t);
+ ptr += num_explicits * sizeof(T);
+ break;
+ }
+ case TDIGEST: {
+ if ((ptr + sizeof(uint32_t)) > end) {
+ return false;
+ }
+ uint32_t tdigest_serialized_length = decode_fixed32_le(ptr);
+ ptr += tdigest_serialized_length;
Review comment:
Why not check content of tdigest? Such as ```isDirty()```.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org