This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f2d8e65c6c7 [fix](be) Own values in single-value aggregate states
(#68210)
f2d8e65c6c7 is described below
commit f2d8e65c6c7d1be19bf6d3f7179cc88f6547ccc4
Author: Mryange <[email protected]>
AuthorDate: Thu Sep 24 14:19:58 2026 +0800
[fix](be) Own values in single-value aggregate states (#68210)
Single-value aggregate state implementations were duplicated across
aggregate functions, and complex values were not handled consistently.
In particular, window functions such as `first_value`, `last_value`,
`nth_value`, `lead`, and `lag` retained a raw input column pointer and
row offset. The input column can be cleared while the window state is
still alive, leaving the state with an invalid reference.
Root cause: the shared state abstraction did not own complex values, so
aggregate functions either maintained separate implementations or
referenced input columns directly.
This change introduces reusable fixed-width, string, and owned-column
single-value states. The column-backed state materializes exactly one
selected row and is reused by complex `min`, `max`, `min_by`, `max_by`,
and window value functions. Unsupported complex comparison types use
non-incremental window evaluation. Existing fixed-width and string state
serialization formats are preserved.
Tests cover value ownership after source-column reclamation, comparison
and serialization, and verify that retained column allocation does not
scale with the number of source rows.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../exprs/aggregate/aggregate_function_min_max.h | 734 ++-------------------
.../aggregate/aggregate_function_min_max_by.cpp | 11 +-
.../aggregate/aggregate_function_min_max_by.h | 146 ++--
.../aggregate/aggregate_function_min_max_impl.h | 12 +-
.../aggregate_function_reader_first_last.h | 128 +---
be/src/exprs/aggregate/aggregate_function_window.h | 36 +-
be/src/exprs/aggregate/single_value_data.h | 417 ++++++++++++
be/test/exprs/aggregate/agg_replace_test.cpp | 120 ----
.../exprs/aggregate/agg_window_nth_value_test.cpp | 47 ++
.../aggregate/aggregate_function_min_max_test.cpp | 197 ++++--
10 files changed, 776 insertions(+), 1072 deletions(-)
diff --git a/be/src/exprs/aggregate/aggregate_function_min_max.h
b/be/src/exprs/aggregate/aggregate_function_min_max.h
index 9a766ddb96a..416bf226196 100644
--- a/be/src/exprs/aggregate/aggregate_function_min_max.h
+++ b/be/src/exprs/aggregate/aggregate_function_min_max.h
@@ -48,6 +48,7 @@
#include "core/type_limit.h"
#include "core/types.h"
#include "exprs/aggregate/aggregate_function.h"
+#include "exprs/aggregate/single_value_data.h"
namespace doris {
class Arena;
@@ -59,650 +60,23 @@ class ColumnVector;
namespace doris {
-/// For numeric values.
-template <PrimitiveType T>
-struct SingleValueDataFixed {
-private:
- using Self = SingleValueDataFixed;
-
- bool has_value =
- false; /// We need to remember if at least one value has been
passed. This is necessary for AggregateFunctionIf.
- typename PrimitiveTypeTraits<T>::CppType value;
-
-public:
- using ColVecType = typename PrimitiveTypeTraits<T>::ColumnType;
- static constexpr bool NeedCheckColumnType = true;
-
- SingleValueDataFixed() = default;
- SingleValueDataFixed(bool has_value_, typename
PrimitiveTypeTraits<T>::CppType value_)
- : has_value(has_value_), value(value_) {}
- bool has() const { return has_value; }
-
- constexpr static bool IsFixedLength = true;
-
- void set_to_min_max(bool max) {
- value = max ? Compare::max_value<typename
PrimitiveTypeTraits<T>::CppType>()
- : Compare::min_value<typename
PrimitiveTypeTraits<T>::CppType>();
- }
-
- void change_if(const IColumn& column, size_t row_num, bool less) {
- has_value = true;
- value = less ? Compare::min(assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
-
TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num],
- value)
- : Compare::max(assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
-
TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num],
- value);
- }
-
- void insert_result_into(IColumn& to) const {
- if (has()) {
- assert_cast<typename PrimitiveTypeTraits<T>::ColumnType&,
TypeCheckOnRelease::DISABLE>(
- to)
- .get_data()
- .push_back(value);
- } else {
- assert_cast<typename PrimitiveTypeTraits<T>::ColumnType&,
TypeCheckOnRelease::DISABLE>(
- to)
- .insert_default();
- }
- }
-
- void reset() {
- if (has()) {
- has_value = false;
- }
- }
-
- void write(BufferWritable& buf) const {
- buf.write_binary(has());
- if (has()) {
- buf.write_binary(value);
- }
- }
-
- void read(BufferReadable& buf, Arena&) {
- buf.read_binary(has_value);
- if (has()) {
- buf.read_binary(value);
- }
- }
-
- void change(const IColumn& column, size_t row_num, Arena&) {
- has_value = true;
- value = assert_cast<const typename PrimitiveTypeTraits<T>::ColumnType&,
- TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num];
- }
-
- /// Assuming to.has()
- void change(const Self& to, Arena&) {
- has_value = true;
- value = to.value;
- }
-
- bool change_if_less(const IColumn& column, size_t row_num, Arena& arena) {
- if (!has() || Compare::less(assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
-
TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num],
- value)) {
- change(column, row_num, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_less(const Self& to, Arena& arena) {
- if (to.has() && (!has() || Compare::less(to.value, value))) {
- change(to, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_greater(const IColumn& column, size_t row_num, Arena&
arena) {
- if (!has() ||
- Compare::greater(assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
- TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num],
- value)) {
- change(column, row_num, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool check_if_equal(const IColumn& column, size_t row_num) const {
- if (!has()) {
- return false;
- }
- return Compare::equal(assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
- TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num],
- value);
- }
-
- bool change_if_greater(const Self& to, Arena& arena) {
- if (to.has() && (!has() || Compare::greater(to.value, value))) {
- change(to, arena);
- return true;
- } else {
- return false;
- }
- }
-
- void change_first_time(const IColumn& column, size_t row_num, Arena&
arena) {
- if (UNLIKELY(!has())) {
- change(column, row_num, arena);
- }
- }
-
- void change_first_time(const Self& to, Arena& arena) {
- if (UNLIKELY(!has() && to.has())) {
- change(to, arena);
- }
- }
-};
-
-/// For decimal values.
-template <PrimitiveType T>
-struct SingleValueDataDecimal {
-private:
- using Self = SingleValueDataDecimal;
-
- bool has_value =
- false; /// We need to remember if at least one value has been
passed. This is necessary for AggregateFunctionIf.
- typename PrimitiveTypeTraits<T>::CppType value;
-
-public:
- using ColVecType = typename PrimitiveTypeTraits<T>::ColumnType;
- static constexpr bool NeedCheckColumnType = true;
-
- SingleValueDataDecimal() = default;
- SingleValueDataDecimal(bool has_value_, typename
PrimitiveTypeTraits<T>::CppType value_)
- : has_value(has_value_), value(value_) {}
- bool has() const { return has_value; }
-
- constexpr static bool IsFixedLength = true;
-
- void set_to_min_max(bool max) {
- value = max ? Compare::max_value<typename
PrimitiveTypeTraits<T>::CppType>()
- : Compare::min_value<typename
PrimitiveTypeTraits<T>::CppType>();
- }
-
- void change_if(const IColumn& column, size_t row_num, bool less) {
- has_value = true;
- value = less ? Compare::min(assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
-
TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num],
- value)
- : Compare::max(assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
-
TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num],
- value);
- }
-
- void insert_result_into(IColumn& to) const {
- if (has()) {
- assert_cast<typename PrimitiveTypeTraits<T>::ColumnType&,
TypeCheckOnRelease::DISABLE>(
- to)
- .insert_data((const char*)&value, 0);
- } else {
- assert_cast<typename PrimitiveTypeTraits<T>::ColumnType&,
TypeCheckOnRelease::DISABLE>(
- to)
- .insert_default();
- }
- }
-
- void reset() {
- if (has()) {
- has_value = false;
- }
- }
-
- void write(BufferWritable& buf) const {
- buf.write_binary(has());
- if (has()) {
- buf.write_binary(value);
- }
- }
-
- void read(BufferReadable& buf, Arena&) {
- buf.read_binary(has_value);
- if (has()) {
- buf.read_binary(value);
- }
- }
-
- void change(const IColumn& column, size_t row_num, Arena&) {
- has_value = true;
- value = assert_cast<const typename PrimitiveTypeTraits<T>::ColumnType&,
- TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num];
- }
-
- /// Assuming to.has()
- void change(const Self& to, Arena&) {
- has_value = true;
- value = to.value;
- }
-
- bool change_if_less(const IColumn& column, size_t row_num, Arena& arena) {
- if (!has() || Compare::less(assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
-
TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num],
- value)) {
- change(column, row_num, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_less(const Self& to, Arena& arena) {
- if (to.has() && (!has() || Compare::less(to.value, value))) {
- change(to, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_greater(const IColumn& column, size_t row_num, Arena&
arena) {
- if (!has() ||
- Compare::greater(assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
- TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num],
- value)) {
- change(column, row_num, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_greater(const Self& to, Arena& arena) {
- if (to.has() && (!has() || Compare::greater(to.value, value))) {
- change(to, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool check_if_equal(const IColumn& column, size_t row_num) const {
- if (!has()) {
- return false;
- }
- return Compare::equal(assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
- TypeCheckOnRelease::DISABLE>(column)
- .get_data()[row_num],
- value);
- }
-
- void change_first_time(const IColumn& column, size_t row_num, Arena&
arena) {
- if (UNLIKELY(!has())) {
- change(column, row_num, arena);
- }
- }
-
- void change_first_time(const Self& to, Arena& arena) {
- if (UNLIKELY(!has() && to.has())) {
- change(to, arena);
- }
- }
-};
-
-/** For strings. Short strings are stored in the object itself, and long
strings are allocated separately.
- * NOTE It could also be suitable for arrays of numbers.
- */
-struct SingleValueDataString {
-private:
- using Self = SingleValueDataString;
- // This function uses int32 for storage, which triggers a 64-bit to 32-bit
conversion warning.
- // However, considering compatibility with future upgrades, no changes
will be made here.
- Int32 size = -1; /// -1 indicates that there is no value.
- Int32 capacity = 0; /// power of two or zero
- DorisUniqueBufferPtr<char> large_data;
-
-public:
- static constexpr Int32 AUTOMATIC_STORAGE_SIZE = 64;
- static constexpr Int32 MAX_SMALL_STRING_SIZE =
- AUTOMATIC_STORAGE_SIZE - sizeof(size) - sizeof(capacity) -
sizeof(large_data);
-
-private:
- char small_data[MAX_SMALL_STRING_SIZE]; /// Including the terminating zero.
-
-public:
- using ColVecType = ColumnString;
- static constexpr bool NeedCheckColumnType = true;
-
- ~SingleValueDataString() = default;
-
- constexpr static bool IsFixedLength = false;
-
- bool has() const { return size >= 0; }
-
- const char* get_data() const {
- return size <= MAX_SMALL_STRING_SIZE ? small_data : large_data.get();
- }
-
- void insert_result_into(IColumn& to) const {
- if (has()) {
- assert_cast<ColumnString&,
TypeCheckOnRelease::DISABLE>(to).insert_data(get_data(),
-
size);
- } else {
- assert_cast<ColumnString&,
TypeCheckOnRelease::DISABLE>(to).insert_default();
- }
- }
-
- void reset() {
- if (size != -1) {
- size = -1;
- capacity = 0;
- large_data.reset();
- }
- }
-
- void write(BufferWritable& buf) const {
- buf.write_binary(size);
- if (has()) {
- buf.write(get_data(), size);
- }
- }
-
- void read(BufferReadable& buf, Arena&) {
- Int32 rhs_size;
- buf.read_binary(rhs_size);
-
- if (rhs_size >= 0) {
- if (rhs_size <= MAX_SMALL_STRING_SIZE) {
- /// Don't free large_data here.
-
- size = rhs_size;
-
- if (size > 0) {
- buf.read(small_data, size);
- }
- } else {
- if (capacity < rhs_size) {
- capacity =
(Int32)round_up_to_power_of_two_or_zero(rhs_size);
- large_data = DorisUniqueBufferPtr<char>(capacity);
- }
-
- size = rhs_size;
- buf.read(large_data.get(), size);
- }
- } else {
- /// Don't free large_data here.
- size = rhs_size;
- }
- }
-
- StringRef get_string_ref() const { return StringRef(get_data(), size); }
-
- /// Assuming to.has()
- void change_impl(StringRef value, Arena&) {
- Int32 value_size = cast_set<Int32>(value.size);
- if (value_size <= MAX_SMALL_STRING_SIZE) {
- /// Don't free large_data here.
- size = value_size;
-
- if (size > 0) {
- memcpy(small_data, value.data, size);
- }
- } else {
- if (capacity < value_size) {
- /// Don't free large_data here.
- capacity = (Int32)round_up_to_power_of_two_or_zero(value_size);
- large_data = DorisUniqueBufferPtr<char>(capacity);
- }
-
- size = value_size;
- memcpy(large_data.get(), value.data, size);
- }
- }
-
- void change(const IColumn& column, size_t row_num, Arena& arena) {
- change_impl(
- assert_cast<const ColumnString&,
TypeCheckOnRelease::DISABLE>(column).get_data_at(
- row_num),
- arena);
- }
-
- void change(const Self& to, Arena& arena) {
change_impl(to.get_string_ref(), arena); }
-
- bool change_if_less(const IColumn& column, size_t row_num, Arena& arena) {
- if (!has() ||
- assert_cast<const ColumnString&,
TypeCheckOnRelease::DISABLE>(column).get_data_at(
- row_num) < get_string_ref()) {
- change(column, row_num, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_greater(const IColumn& column, size_t row_num, Arena&
arena) {
- if (!has() ||
- assert_cast<const ColumnString&,
TypeCheckOnRelease::DISABLE>(column).get_data_at(
- row_num) > get_string_ref()) {
- change(column, row_num, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_less(const Self& to, Arena& arena) {
- if (to.has() && (!has() || to.get_string_ref() < get_string_ref())) {
- change(to, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_greater(const Self& to, Arena& arena) {
- if (to.has() && (!has() || to.get_string_ref() > get_string_ref())) {
- change(to, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool check_if_equal(const IColumn& column, size_t row_num) const {
- if (!has()) {
- return false;
- }
- return assert_cast<const ColumnString&,
TypeCheckOnRelease::DISABLE>(column).get_data_at(
- row_num) == get_string_ref();
- }
-
- void change_first_time(const IColumn& column, size_t row_num, Arena&
arena) {
- if (UNLIKELY(!has())) {
- change(column, row_num, arena);
- }
- }
-
- void change_first_time(const Self& to, Arena& arena) {
- if (UNLIKELY(!has() && to.has())) {
- change(to, arena);
- }
- }
-};
-
-static_assert(sizeof(SingleValueDataString) ==
SingleValueDataString::AUTOMATIC_STORAGE_SIZE);
-
-struct SingleValueDataComplexType {
-private:
- using Self = SingleValueDataComplexType;
-
- DataTypePtr column_type;
- bool has_value = false;
- MutableColumnPtr column_data; // a column ptr only save a single value
- int be_exec_version = -1;
-
-public:
- static constexpr bool NeedCheckColumnType = false;
-
- SingleValueDataComplexType() = default;
- SingleValueDataComplexType(const DataTypes& argument_types, int
be_version) {
- column_type = argument_types[0];
- column_data = column_type->create_column();
- be_exec_version = be_version;
- }
-
- bool has() const { return has_value; }
-
- constexpr static bool IsFixedLength = false;
-
- void insert_result_into(IColumn& to) const {
- if (has()) {
- to.insert_from(*column_data, 0);
- } else {
- to.insert_default();
- }
- }
-
- void reset() {
- has_value = false;
- column_data->clear();
- }
-
- void write(BufferWritable& buf) const {
- buf.write_binary(has_value);
- if (!has()) {
- return;
- }
- auto size_bytes =
- column_type->get_uncompressed_serialized_bytes(*column_data,
be_exec_version);
- buf.write_binary(size_bytes);
- buf.resize(size_bytes);
- auto* p = column_type->serialize(*column_data, buf.data(),
be_exec_version);
- DCHECK_EQ(p, buf.data() + size_bytes);
- buf.add_offset(size_bytes);
- }
-
- void read(BufferReadable& buf, Arena& arena) {
- buf.read_binary(has_value);
- if (!has()) {
- return;
- }
- int64_t size;
- buf.read_binary(size);
- const auto* p = column_type->deserialize(buf.data(), &column_data,
be_exec_version);
- DCHECK_EQ(p, buf.data() + size);
- buf.add_offset(size);
- }
-
- void change(const IColumn& column, size_t row_num, Arena&) {
- has_value = true;
- column_data->clear();
- column_data->insert_from(column, row_num);
- }
-
- /// Assuming to.has()
- void change(const Self& to, Arena&) {
- has_value = true;
- column_data->clear();
- column_data->insert_from(*to.column_data, 0);
- }
-
- bool change_if_less(const IColumn& column, size_t row_num, Arena& arena) {
- if (!has() || column_data->compare_at(0, row_num, column, 1) == 1) {
- change(column, row_num, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_less(const Self& to, Arena& arena) {
- if (to.has() && (!has() || column_data->compare_at(0, 0,
*to.column_data, 1) == 1)) {
- change(to, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_greater(const IColumn& column, size_t row_num, Arena&
arena) {
- if (!has() || column_data->compare_at(0, row_num, column, 1) == -1) {
- change(column, row_num, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool change_if_greater(const Self& to, Arena& arena) {
- if (to.has() && (!has() || column_data->compare_at(0, 0,
*to.column_data, 1) == -1)) {
- change(to, arena);
- return true;
- } else {
- return false;
- }
- }
-
- bool check_if_equal(const IColumn& column, size_t row_num) const {
- if (!has()) {
- return false;
- }
- auto type = column_type->get_primitive_type();
- if (type == TYPE_BITMAP || type == TYPE_HLL || type ==
TYPE_QUANTILE_STATE ||
- type == TYPE_AGG_STATE) {
- return false;
- } else {
- return !column_data->compare_at(0, row_num, column, 1);
- }
- }
-
- void change_first_time(const IColumn& column, size_t row_num, Arena&
arena) {
- if (UNLIKELY(!has())) {
- change(column, row_num, arena);
- }
- }
-
- void change_first_time(const Self& to, Arena& arena) {
- if (UNLIKELY(!has() && to.has())) {
- change(to, arena);
- }
- }
-};
-
template <typename Data>
struct AggregateFunctionMaxData : public Data {
using Self = AggregateFunctionMaxData;
- using Data::IsFixedLength;
+ using Data::UsesFixedLengthStateSerialization;
constexpr static bool IS_ANY = false;
- AggregateFunctionMaxData(const DataTypes& argument_types, int be_version)
- : Data(argument_types, be_version) {
- this->reset();
- }
-
AggregateFunctionMaxData() { reset(); }
void change_if_better(const IColumn& column, size_t row_num, Arena& arena)
{
- if constexpr (Data::IsFixedLength) {
- this->change_if(column, row_num, false);
- } else {
- this->change_if_greater(column, row_num, arena);
- }
+ this->set_if_greater(column, row_num, arena);
}
- void change_if_better(const Self& to, Arena& arena) {
this->change_if_greater(to, arena); }
+ void change_if_better(const Self& to, Arena& arena) {
this->set_if_greater(to, arena); }
void reset() {
- if constexpr (Data::IsFixedLength) {
- this->set_to_min_max(false);
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
+ this->set_value_to_min();
}
Data::reset();
}
@@ -713,28 +87,19 @@ struct AggregateFunctionMaxData : public Data {
template <typename Data>
struct AggregateFunctionMinData : Data {
using Self = AggregateFunctionMinData;
- using Data::IsFixedLength;
+ using Data::UsesFixedLengthStateSerialization;
constexpr static bool IS_ANY = false;
- AggregateFunctionMinData(const DataTypes& argument_types, int be_version)
- : Data(argument_types, be_version) {
- this->reset();
- }
-
AggregateFunctionMinData() { reset(); }
void change_if_better(const IColumn& column, size_t row_num, Arena& arena)
{
- if constexpr (Data::IsFixedLength) {
- this->change_if(column, row_num, true);
- } else {
- this->change_if_less(column, row_num, arena);
- }
+ this->set_if_smaller(column, row_num, arena);
}
- void change_if_better(const Self& to, Arena& arena) {
this->change_if_less(to, arena); }
+ void change_if_better(const Self& to, Arena& arena) {
this->set_if_smaller(to, arena); }
void reset() {
- if constexpr (Data::IsFixedLength) {
- this->set_to_min_max(true);
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
+ this->set_value_to_max();
}
Data::reset();
}
@@ -746,53 +111,47 @@ struct AggregateFunctionMinData : Data {
template <typename Data>
struct AggregateFunctionAnyData : Data {
using Self = AggregateFunctionAnyData;
- using Data::IsFixedLength;
+ using Data::UsesFixedLengthStateSerialization;
static const char* name() { return "any"; }
constexpr static bool IS_ANY = true;
- AggregateFunctionAnyData(const DataTypes& argument_types, int be_version)
- : Data(argument_types, be_version) {};
-
AggregateFunctionAnyData() {};
void change_if_better(const IColumn& column, size_t row_num, Arena& arena)
{
- this->change_first_time(column, row_num, arena);
+ if (UNLIKELY(!this->has())) {
+ this->set(column, row_num, arena);
+ }
}
- void change_if_better(const Self& to, Arena& arena) {
this->change_first_time(to, arena); }
+ void change_if_better(const Self& to, Arena& arena) {
+ if (UNLIKELY(!this->has() && to.has())) {
+ this->set(to, arena);
+ }
+ }
};
template <typename Data>
class AggregateFunctionsSingleValue final
: public IAggregateFunctionDataHelper<Data,
AggregateFunctionsSingleValue<Data>> {
private:
- DataTypePtr& type;
+ const DataTypePtr& _data_type;
using Base = IAggregateFunctionDataHelper<Data,
AggregateFunctionsSingleValue<Data>>;
using IAggregateFunction::argument_types;
public:
AggregateFunctionsSingleValue(const DataTypes& arguments)
: IAggregateFunctionDataHelper<Data,
AggregateFunctionsSingleValue<Data>>(arguments),
- type(this->argument_types[0]) {}
+ _data_type(this->argument_types[0]) {}
- void create(AggregateDataPtr __restrict place) const override {
- if constexpr (std::is_same_v<Data,
AggregateFunctionMaxData<SingleValueDataComplexType>> ||
- std::is_same_v<Data,
AggregateFunctionMinData<SingleValueDataComplexType>> ||
- std::is_same_v<Data,
AggregateFunctionAnyData<SingleValueDataComplexType>>) {
- new (place) Data(argument_types, IAggregateFunction::version);
- } else {
- new (place) Data;
- }
- }
+ void create(AggregateDataPtr __restrict place) const override { new
(place) Data; }
String get_name() const override { return Data::name(); }
- DataTypePtr get_return_type() const override { return type; }
+ DataTypePtr get_return_type() const override { return _data_type; }
- // min/max require sentinel-initialized state (MAX_VALUE for min,
MIN_VALUE for max) via
- // create(), so they cannot use zero-init and must return false. any_value
is safe with
- // zero-init because it checks has_value before comparing
(change_first_time).
- bool is_trivial() const override { return Data::IsFixedLength &&
Data::IS_ANY; }
+ bool is_trivial() const override {
+ return Data::UsesFixedLengthStateSerialization && Data::IS_ANY;
+ }
void add(AggregateDataPtr __restrict place, const IColumn** columns,
ssize_t row_num,
Arena& arena) const override {
@@ -817,12 +176,12 @@ public:
}
void serialize(ConstAggregateDataPtr __restrict place, BufferWritable&
buf) const override {
- this->data(place).write(buf);
+ this->data(place).write(buf, _data_type, IAggregateFunction::version);
}
void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
Arena& arena) const override {
- this->data(place).read(buf, arena);
+ this->data(place).read(buf, _data_type, IAggregateFunction::version,
arena);
}
void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn&
to) const override {
@@ -845,7 +204,7 @@ public:
void serialize_to_column(const std::vector<AggregateDataPtr>& places,
size_t offset,
MutableColumnPtr& dst, const size_t num_rows)
const override {
- if constexpr (Data::IsFixedLength) {
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
auto& dst_column = assert_cast<ColumnFixedLengthObject&>(*dst);
dst_column.resize(num_rows);
auto* dst_data =
reinterpret_cast<Data*>(dst_column.get_data().data());
@@ -859,12 +218,12 @@ public:
void streaming_agg_serialize_to_column(const IColumn** columns,
MutableColumnPtr& dst,
const size_t num_rows, Arena&
arena) const override {
- if constexpr (Data::IsFixedLength) {
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
auto& dst_column = assert_cast<ColumnFixedLengthObject&>(*dst);
dst_column.resize(num_rows);
auto* dst_data =
reinterpret_cast<Data*>(dst_column.get_data().data());
for (size_t i = 0; i != num_rows; ++i) {
- dst_data[i].change(*columns[0], i, arena);
+ dst_data[i].set(*columns[0], i, arena);
}
} else {
Base::streaming_agg_serialize_to_column(columns, dst, num_rows,
arena);
@@ -874,7 +233,7 @@ public:
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena& arena) const override {
- if constexpr (Data::IsFixedLength) {
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
DCHECK(end <= column.size() && begin <= end) << ", begin:" <<
begin << ", end:" << end
<< ", column.size():"
<< column.size();
auto& col = assert_cast<const ColumnFixedLengthObject&>(column);
@@ -890,7 +249,7 @@ public:
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena& arena,
const size_t num_rows) const override {
- if constexpr (Data::IsFixedLength) {
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
const auto& col = assert_cast<const
ColumnFixedLengthObject&>(*column);
const auto* data = col.get_data().data();
this->merge_vec(places, offset, AggregateDataPtr(data), arena,
num_rows);
@@ -904,7 +263,7 @@ public:
void deserialize_and_merge_vec_selected(const AggregateDataPtr* places,
size_t offset,
AggregateDataPtr rhs, const
IColumn* column,
Arena& arena, const size_t
num_rows) const override {
- if constexpr (Data::IsFixedLength) {
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
const auto& col = assert_cast<const
ColumnFixedLengthObject&>(*column);
const auto* data = col.get_data().data();
this->merge_vec_selected(places, offset, AggregateDataPtr(data),
arena, num_rows);
@@ -917,7 +276,7 @@ public:
void serialize_without_key_to_column(ConstAggregateDataPtr __restrict
place,
IColumn& to) const override {
- if constexpr (Data::IsFixedLength) {
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
auto& col = assert_cast<ColumnFixedLengthObject&>(to);
size_t old_size = col.size();
col.resize(old_size + 1);
@@ -928,7 +287,7 @@ public:
}
MutableColumnPtr create_serialize_column() const override {
- if constexpr (Data::IsFixedLength) {
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
return ColumnFixedLengthObject::create(sizeof(Data));
} else {
return ColumnString::create();
@@ -936,14 +295,27 @@ public:
}
DataTypePtr get_serialized_type() const override {
- if constexpr (Data::IsFixedLength) {
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
return std::make_shared<DataTypeFixedLengthObject>();
} else {
return std::make_shared<DataTypeString>();
}
}
- bool supported_incremental_mode() const override { return !(Data::IS_ANY);
}
+ bool supported_incremental_mode() const override {
+ if constexpr (Data::IS_ANY) {
+ return false;
+ }
+ switch (_data_type->get_primitive_type()) {
+ case TYPE_BITMAP:
+ case TYPE_HLL:
+ case TYPE_QUANTILE_STATE:
+ case TYPE_AGG_STATE:
+ return false;
+ default:
+ return true;
+ }
+ }
void execute_function_with_incremental(int64_t partition_start, int64_t
partition_end,
int64_t frame_start, int64_t
frame_end,
@@ -962,7 +334,7 @@ public:
auto incoming_pos = frame_end - 1;
if (!previous_is_nul && outcoming_pos >= partition_start &&
outcoming_pos < partition_end) {
- if (this->data(place).check_if_equal(*columns[0],
outcoming_pos)) {
+ if (this->data(place).is_equal_to(*columns[0], outcoming_pos))
{
this->data(place).reset();
if (has_null) {
const auto& null_map_data =
diff --git a/be/src/exprs/aggregate/aggregate_function_min_max_by.cpp
b/be/src/exprs/aggregate/aggregate_function_min_max_by.cpp
index 70e5508069e..f0136d78701 100644
--- a/be/src/exprs/aggregate/aggregate_function_min_max_by.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_min_max_by.cpp
@@ -22,16 +22,12 @@
#include "exprs/aggregate/aggregate_function_simple_factory.h"
namespace doris {
-std::unique_ptr<MaxMinValueBase> create_max_min_value(const DataTypePtr& type,
int be_version) {
+std::unique_ptr<MaxMinValueBase> create_max_min_value(const DataTypePtr& type)
{
std::unique_ptr<MaxMinValueBase> result;
auto call = [&](const auto& dispatch_type) -> bool {
using DispatchType = std::decay_t<decltype(dispatch_type)>;
constexpr auto PT = DispatchType::PType;
- if constexpr (is_decimal(PT)) {
- result =
std::make_unique<MaxMinValue<SingleValueDataDecimal<PT>>>();
- } else {
- result = std::make_unique<MaxMinValue<SingleValueDataFixed<PT>>>();
- }
+ result = std::make_unique<MaxMinValue<SingleValueDataFixed<PT>>>();
return true;
};
if (type->get_primitive_type() == TYPE_TIMESTAMP_NS) {
@@ -51,8 +47,7 @@ std::unique_ptr<MaxMinValueBase> create_max_min_value(const
DataTypePtr& type, i
case PrimitiveType::TYPE_ARRAY:
case PrimitiveType::TYPE_MAP:
case PrimitiveType::TYPE_STRUCT:
- return
std::make_unique<MaxMinValue<SingleValueDataComplexType>>(DataTypes {type},
-
be_version);
+ return std::make_unique<MaxMinValue<SingleValueDataColumn>>();
default:
throw doris::Exception(ErrorCode::INTERNAL_ERROR,
"Illegal type {} of argument of aggregate
function min/max_by",
diff --git a/be/src/exprs/aggregate/aggregate_function_min_max_by.h
b/be/src/exprs/aggregate/aggregate_function_min_max_by.h
index dfba3fbcd25..a692aca65c0 100644
--- a/be/src/exprs/aggregate/aggregate_function_min_max_by.h
+++ b/be/src/exprs/aggregate/aggregate_function_min_max_by.h
@@ -34,12 +34,14 @@ namespace doris {
struct MaxMinValueBase {
virtual ~MaxMinValueBase() = default;
- virtual void write(BufferWritable& buf) const = 0;
- virtual void read(BufferReadable& buf, Arena& arena) = 0;
+ virtual void write(BufferWritable& buf, const DataTypePtr& data_type,
+ int be_exec_version) const = 0;
+ virtual void read(BufferReadable& buf, const DataTypePtr& data_type, int
be_exec_version,
+ Arena& arena) = 0;
virtual void insert_result_into(IColumn& to) const = 0;
virtual void reset() = 0;
- virtual void change(const IColumn& column, size_t row_num, Arena& arena) =
0;
- virtual void change(const MaxMinValueBase& to, Arena& arena) = 0;
+ virtual void set(const IColumn& column, size_t row_num, Arena& arena) = 0;
+ virtual void set(const MaxMinValueBase& to, Arena& arena) = 0;
};
template <typename VT>
@@ -48,30 +50,33 @@ struct MaxMinValue : public MaxMinValueBase {
MaxMinValue() = default;
- MaxMinValue(const DataTypes& argument_types, int be_version)
- : value(argument_types, be_version) {}
-
~MaxMinValue() override = default;
- void write(BufferWritable& buf) const override { value.write(buf); }
+ void write(BufferWritable& buf, const DataTypePtr& data_type,
+ int be_exec_version) const override {
+ value.write(buf, data_type, be_exec_version);
+ }
- void read(BufferReadable& buf, Arena& arena) override { value.read(buf,
arena); }
+ void read(BufferReadable& buf, const DataTypePtr& data_type, int
be_exec_version,
+ Arena& arena) override {
+ value.read(buf, data_type, be_exec_version, arena);
+ }
void insert_result_into(IColumn& to) const override {
value.insert_result_into(to); }
void reset() override { value.reset(); }
- void change(const IColumn& column, size_t row_num, Arena& arena) override {
- value.change(column, row_num, arena);
+ void set(const IColumn& column, size_t row_num, Arena& arena) override {
+ value.set(column, row_num, arena);
}
- void change(const MaxMinValueBase& to, Arena& arena) override {
+ void set(const MaxMinValueBase& to, Arena& arena) override {
const auto& derived = assert_cast<const MaxMinValue&>(to);
- value.change(derived.value, arena);
+ value.set(derived.value, arena);
}
};
-std::unique_ptr<MaxMinValueBase> create_max_min_value(const DataTypePtr& type,
int be_version);
+std::unique_ptr<MaxMinValueBase> create_max_min_value(const DataTypePtr& type);
/// For bitmap value
struct BitmapValueData {
@@ -98,27 +103,28 @@ public:
}
}
- void write(BufferWritable& buf) const {
+ void write(BufferWritable& buf, const DataTypePtr&, int) const {
buf.write_binary(has());
if (has()) {
DataTypeBitMap::serialize_as_stream(value, buf);
}
}
- void read(BufferReadable& buf, Arena&) {
+ void read(BufferReadable& buf, const DataTypePtr&, int, Arena&) {
buf.read_binary(has_value);
if (has()) {
DataTypeBitMap::deserialize_as_stream(value, buf);
}
}
- void change(const IColumn& column, size_t row_num, Arena&) {
+ void set(const IColumn& column, size_t row_num, Arena&) {
has_value = true;
value = assert_cast<const ColumnBitmap&,
TypeCheckOnRelease::DISABLE>(column)
.get_data()[row_num];
}
- void change(const Self& to, Arena&) {
+ void set(const Self& to, Arena&) {
+ DORIS_CHECK(to.has());
has_value = true;
value = to.value;
}
@@ -139,7 +145,7 @@ public:
* In contrast, the value type VT is intentionally not made a template
parameter.
* On one hand, templating both key and value types would lead to an n × n
* explosion in template instantiations, increasing compile time and code size.
- * On the other hand, value objects typically only invoke the change method;
for
+ * On the other hand, value objects typically only invoke the set method; for
* random data, this method is called approximately log(x) times (where x is
the
* data size), making the overhead acceptable.
*/
@@ -152,16 +158,8 @@ protected:
public:
AggregateFunctionMinMaxByBaseData() {}
- AggregateFunctionMinMaxByBaseData(const DataTypes argument_types, int
be_version)
- requires(std::is_same_v<KT, SingleValueDataComplexType>)
- : key(SingleValueDataComplexType(DataTypes {argument_types[1]},
be_version)) {
- value = create_max_min_value(argument_types[0], be_version);
- }
-
- AggregateFunctionMinMaxByBaseData(const DataTypes argument_types, int
be_version)
- requires(!std::is_same_v<KT, SingleValueDataComplexType>)
- {
- value = create_max_min_value(argument_types[0], be_version);
+ explicit AggregateFunctionMinMaxByBaseData(const DataTypes&
argument_types) {
+ value = create_max_min_value(argument_types[0]);
}
void insert_result_into(IColumn& to) const {
value->insert_result_into(to); }
@@ -170,14 +168,16 @@ public:
value->reset();
key.reset();
}
- void write(BufferWritable& buf) const {
- value->write(buf);
- key.write(buf);
+ void write(BufferWritable& buf, const DataTypePtr& value_type, const
DataTypePtr& key_type,
+ int be_exec_version) const {
+ value->write(buf, value_type, be_exec_version);
+ key.write(buf, key_type, be_exec_version);
}
- void read(BufferReadable& buf, Arena& arena) {
- value->read(buf, arena);
- key.read(buf, arena);
+ void read(BufferReadable& buf, const DataTypePtr& value_type, const
DataTypePtr& key_type,
+ int be_exec_version, Arena& arena) {
+ value->read(buf, value_type, be_exec_version, arena);
+ key.read(buf, key_type, be_exec_version, arena);
}
};
@@ -187,32 +187,32 @@ struct AggregateFunctionMaxByData : public
AggregateFunctionMinMaxByBaseData<KT>
AggregateFunctionMaxByData() {}
- AggregateFunctionMaxByData(const DataTypes argument_types, int be_version)
- : AggregateFunctionMinMaxByBaseData<KT>(argument_types,
be_version) {}
+ explicit AggregateFunctionMaxByData(const DataTypes& argument_types)
+ : AggregateFunctionMinMaxByBaseData<KT>(argument_types) {}
void change_if_better(const IColumn& value_column, const IColumn&
key_column, size_t row_num,
Arena& arena) {
- if (this->key.change_if_greater(key_column, row_num, arena)) {
- this->value->change(value_column, row_num, arena);
+ if (this->key.set_if_greater(key_column, row_num, arena)) {
+ this->value->set(value_column, row_num, arena);
}
}
void change_if_better_batch(const IColumn& value_column, const IColumn&
key_column,
size_t batch_size, Arena& arena) {
- size_t max_pos = -1;
+ size_t best_pos = batch_size;
for (size_t i = 0; i < batch_size; ++i) {
- if (this->key.change_if_greater(key_column, i, arena)) {
- max_pos = i;
+ if (this->key.set_if_greater(key_column, i, arena)) {
+ best_pos = i;
}
}
- if (max_pos != static_cast<size_t>(-1)) {
- this->value->change(value_column, max_pos, arena);
+ if (best_pos < batch_size) {
+ this->value->set(value_column, best_pos, arena);
}
}
void change_if_better(const Self& to, Arena& arena) {
- if (this->key.change_if_greater(to.key, arena)) {
- this->value->change(*to.value, arena);
+ if (this->key.set_if_greater(to.key, arena)) {
+ this->value->set(*to.value, arena);
}
}
@@ -225,32 +225,32 @@ struct AggregateFunctionMinByData : public
AggregateFunctionMinMaxByBaseData<KT>
AggregateFunctionMinByData() {}
- AggregateFunctionMinByData(const DataTypes argument_types, int be_version)
- : AggregateFunctionMinMaxByBaseData<KT>(argument_types,
be_version) {}
+ explicit AggregateFunctionMinByData(const DataTypes& argument_types)
+ : AggregateFunctionMinMaxByBaseData<KT>(argument_types) {}
void change_if_better(const IColumn& value_column, const IColumn&
key_column, size_t row_num,
Arena& arena) {
- if (this->key.change_if_less(key_column, row_num, arena)) {
- this->value->change(value_column, row_num, arena);
+ if (this->key.set_if_smaller(key_column, row_num, arena)) {
+ this->value->set(value_column, row_num, arena);
}
}
void change_if_better_batch(const IColumn& value_column, const IColumn&
key_column,
size_t batch_size, Arena& arena) {
- size_t min_pos = -1;
+ size_t best_pos = batch_size;
for (size_t i = 0; i < batch_size; ++i) {
- if (this->key.change_if_less(key_column, i, arena)) {
- min_pos = i;
+ if (this->key.set_if_smaller(key_column, i, arena)) {
+ best_pos = i;
}
}
- if (min_pos != static_cast<size_t>(-1)) {
- this->value->change(value_column, min_pos, arena);
+ if (best_pos < batch_size) {
+ this->value->set(value_column, best_pos, arena);
}
}
void change_if_better(const Self& to, Arena& arena) {
- if (this->key.change_if_less(to.key, arena)) {
- this->value->change(*to.value, arena);
+ if (this->key.set_if_smaller(to.key, arena)) {
+ this->value->set(*to.value, arena);
}
}
@@ -263,23 +263,23 @@ class AggregateFunctionsMinMaxBy final
MultiExpression,
NullableAggregateFunction {
private:
- DataTypePtr& value_type;
- DataTypePtr& key_type;
+ const DataTypePtr& _value_type;
+ const DataTypePtr& _key_type;
public:
AggregateFunctionsMinMaxBy(const DataTypes& arguments)
: IAggregateFunctionDataHelper<Data,
AggregateFunctionsMinMaxBy<Data>>(
{arguments[0], arguments[1]}),
- value_type(this->argument_types[0]),
- key_type(this->argument_types[1]) {}
+ _value_type(this->argument_types[0]),
+ _key_type(this->argument_types[1]) {}
void create(AggregateDataPtr __restrict place) const override {
- new (place) Data(IAggregateFunction::argument_types,
IAggregateFunction::version);
+ new (place) Data(IAggregateFunction::argument_types);
}
String get_name() const override { return Data::name(); }
- DataTypePtr get_return_type() const override { return value_type; }
+ DataTypePtr get_return_type() const override { return _value_type; }
void add(AggregateDataPtr __restrict place, const IColumn** columns,
ssize_t row_num,
Arena& arena) const override {
@@ -299,12 +299,12 @@ public:
}
void serialize(ConstAggregateDataPtr __restrict place, BufferWritable&
buf) const override {
- this->data(place).write(buf);
+ this->data(place).write(buf, _value_type, _key_type,
IAggregateFunction::version);
}
void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
Arena& arena) const override {
- this->data(place).read(buf, arena);
+ this->data(place).read(buf, _value_type, _key_type,
IAggregateFunction::version, arena);
}
void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn&
to) const override {
@@ -326,15 +326,9 @@ AggregateFunctionPtr
create_aggregate_function_min_max_by(const String& name,
auto call = [&](const auto& dispatch_type) -> bool {
using DispatchType = std::decay_t<decltype(dispatch_type)>;
constexpr auto PT = DispatchType::PType;
- if constexpr (is_decimal(PT)) {
- result = creator_without_type::create_multi_arguments<
-
AggregateFunctionsMinMaxBy<Data<SingleValueDataDecimal<PT>>>>(
- argument_types, result_is_nullable, attr);
- } else {
- result = creator_without_type::create_multi_arguments<
-
AggregateFunctionsMinMaxBy<Data<SingleValueDataFixed<PT>>>>(
- argument_types, result_is_nullable, attr);
- }
+ result = creator_without_type::create_multi_arguments<
+ AggregateFunctionsMinMaxBy<Data<SingleValueDataFixed<PT>>>>(
+ argument_types, result_is_nullable, attr);
return true;
};
// Keep nano dispatch local because the shared scalar dispatcher is also
used by templates
@@ -356,8 +350,8 @@ AggregateFunctionPtr
create_aggregate_function_min_max_by(const String& name,
result_is_nullable, attr);
case PrimitiveType::TYPE_ARRAY:
return creator_without_type::create_multi_arguments<
- AggregateFunctionsMinMaxBy<Data<SingleValueDataComplexType>>>(
- argument_types, result_is_nullable, attr);
+
AggregateFunctionsMinMaxBy<Data<SingleValueDataColumn>>>(argument_types,
+
result_is_nullable, attr);
default:
return nullptr;
}
diff --git a/be/src/exprs/aggregate/aggregate_function_min_max_impl.h
b/be/src/exprs/aggregate/aggregate_function_min_max_impl.h
index 0c49c79aa8a..61753683ac4 100644
--- a/be/src/exprs/aggregate/aggregate_function_min_max_impl.h
+++ b/be/src/exprs/aggregate/aggregate_function_min_max_impl.h
@@ -113,23 +113,23 @@ AggregateFunctionPtr
create_aggregate_function_single_value(const String& name,
argument_types, result_is_nullable, attr);
case PrimitiveType::TYPE_DECIMAL32:
return creator_without_type::create_unary_arguments<
-
AggregateFunctionsSingleValue<Data<SingleValueDataDecimal<TYPE_DECIMAL32>>>>(
+
AggregateFunctionsSingleValue<Data<SingleValueDataFixed<TYPE_DECIMAL32>>>>(
argument_types, result_is_nullable, attr);
case PrimitiveType::TYPE_DECIMAL64:
return creator_without_type::create_unary_arguments<
-
AggregateFunctionsSingleValue<Data<SingleValueDataDecimal<TYPE_DECIMAL64>>>>(
+
AggregateFunctionsSingleValue<Data<SingleValueDataFixed<TYPE_DECIMAL64>>>>(
argument_types, result_is_nullable, attr);
case PrimitiveType::TYPE_DECIMALV2:
return creator_without_type::create_unary_arguments<
-
AggregateFunctionsSingleValue<Data<SingleValueDataDecimal<TYPE_DECIMALV2>>>>(
+
AggregateFunctionsSingleValue<Data<SingleValueDataFixed<TYPE_DECIMALV2>>>>(
argument_types, result_is_nullable, attr);
case PrimitiveType::TYPE_DECIMAL128I:
return creator_without_type::create_unary_arguments<
-
AggregateFunctionsSingleValue<Data<SingleValueDataDecimal<TYPE_DECIMAL128I>>>>(
+
AggregateFunctionsSingleValue<Data<SingleValueDataFixed<TYPE_DECIMAL128I>>>>(
argument_types, result_is_nullable, attr);
case PrimitiveType::TYPE_DECIMAL256:
return creator_without_type::create_unary_arguments<
-
AggregateFunctionsSingleValue<Data<SingleValueDataDecimal<TYPE_DECIMAL256>>>>(
+
AggregateFunctionsSingleValue<Data<SingleValueDataFixed<TYPE_DECIMAL256>>>>(
argument_types, result_is_nullable, attr);
case PrimitiveType::TYPE_ARRAY:
case PrimitiveType::TYPE_MAP:
@@ -139,7 +139,7 @@ AggregateFunctionPtr
create_aggregate_function_single_value(const String& name,
case PrimitiveType::TYPE_HLL:
case PrimitiveType::TYPE_QUANTILE_STATE:
return creator_without_type::create_unary_arguments<
-
AggregateFunctionsSingleValue<Data<SingleValueDataComplexType>>>(
+ AggregateFunctionsSingleValue<Data<SingleValueDataColumn>>>(
argument_types, result_is_nullable, attr);
case PrimitiveType::TYPE_VARBINARY:
// Owning binary values for IO must not implicitly enable single-value
aggregates.
diff --git a/be/src/exprs/aggregate/aggregate_function_reader_first_last.h
b/be/src/exprs/aggregate/aggregate_function_reader_first_last.h
index ff693eec98e..b2ba4225ff0 100644
--- a/be/src/exprs/aggregate/aggregate_function_reader_first_last.h
+++ b/be/src/exprs/aggregate/aggregate_function_reader_first_last.h
@@ -17,104 +17,14 @@
#pragma once
-#include <type_traits>
-
#include "core/column/column_nullable.h"
+#include "exprs/aggregate/single_value_data.h"
namespace doris {
-template <bool arg_is_nullable>
-struct Value {
-public:
- bool is_null() const {
- if (_ptr == nullptr) {
- return true;
- }
- if constexpr (arg_is_nullable) {
- return assert_cast<const ColumnNullable*,
TypeCheckOnRelease::DISABLE>(_ptr)
- ->is_null_at(_offset);
- }
- return false;
- }
-
- template <typename ColVecType>
- void insert_into(IColumn& to) const {
- if constexpr (arg_is_nullable) {
- const auto* col = assert_cast<const ColumnNullable*,
TypeCheckOnRelease::DISABLE>(_ptr);
- assert_cast<ColVecType&,
TypeCheckOnRelease::DISABLE>(to).insert_from(
- col->get_nested_column(), _offset);
- } else {
- assert_cast<ColVecType&,
TypeCheckOnRelease::DISABLE>(to).insert_from(*_ptr, _offset);
- }
- }
-
- // Non-template version: virtual dispatch on IColumn::insert_from.
- // Used by window path where devirtualization is not beneficial.
- void insert_into(IColumn& to) const {
- if constexpr (arg_is_nullable) {
- const auto* col = assert_cast<const ColumnNullable*,
TypeCheckOnRelease::DISABLE>(_ptr);
- to.insert_from(col->get_nested_column(), _offset);
- } else {
- to.insert_from(*_ptr, _offset);
- }
- }
-
- void set_value(const IColumn* column, size_t row) {
- _ptr = column;
- _offset = row;
- }
-
- void reset() {
- _ptr = nullptr;
- _offset = 0;
- }
-
-protected:
- const IColumn* _ptr = nullptr;
- size_t _offset = 0;
-};
-
-template <bool arg_is_nullable>
-struct CopiedValue : public Value<arg_is_nullable> {
-public:
- template <typename ColVecType>
- void insert_into(IColumn& to) const {
- assert_cast<ColVecType&,
TypeCheckOnRelease::DISABLE>(to).insert(_copied_value);
- }
-
- bool is_null() const { return this->_ptr == nullptr; }
-
- template <typename ColVecType>
- void set_value(const IColumn* column, size_t row) {
- // here _ptr, maybe null at row, so call reset to set nullptr
- // But we will use is_null() check first, others have set _ptr column
to a meaningless address
- // because the address have meaningless, only need it to check is
nullptr
- this->_ptr = (IColumn*)0x00000001;
- if constexpr (arg_is_nullable) {
- const auto* col =
- assert_cast<const ColumnNullable*,
TypeCheckOnRelease::DISABLE>(column);
- if (col->is_null_at(row)) {
- this->reset();
- return;
- } else {
- auto& nested_col = assert_cast<const ColVecType&,
TypeCheckOnRelease::DISABLE>(
- col->get_nested_column());
- nested_col.get(row, _copied_value);
- }
- } else {
- column->get(row, _copied_value);
- }
- }
-
-private:
- Field _copied_value;
-};
-
-template <typename ColVecType, bool result_is_nullable, bool arg_is_nullable,
bool is_copy>
+template <bool result_is_nullable, bool arg_is_nullable>
struct ReaderFirstAndLastData {
public:
- using StoreType =
- std::conditional_t<is_copy, CopiedValue<arg_is_nullable>,
Value<arg_is_nullable>>;
static constexpr bool nullable = arg_is_nullable;
static constexpr bool result_nullable = result_is_nullable;
@@ -125,44 +35,40 @@ public:
void insert_result_into(IColumn& to) const {
if constexpr (result_is_nullable) {
- if (_data_value.is_null()) { //_ptr == nullptr || null data at row
+ if (!_data_value.has()) {
auto& col = assert_cast<ColumnNullable&,
TypeCheckOnRelease::DISABLE>(to);
col.insert_default();
} else {
auto& col = assert_cast<ColumnNullable&,
TypeCheckOnRelease::DISABLE>(to);
col.get_null_map_data().push_back(0);
- if constexpr (!std::is_same_v<ColVecType, void>) {
- _data_value.template
insert_into<ColVecType>(col.get_nested_column());
- } else {
- _data_value.insert_into(col.get_nested_column());
- }
+ _data_value.insert_result_into(col.get_nested_column());
}
} else {
- if constexpr (!std::is_same_v<ColVecType, void>) {
- _data_value.template insert_into<ColVecType>(to);
- } else {
- _data_value.insert_into(to);
- }
+ _data_value.insert_result_into(to);
}
}
- // here not check the columns[0] is null at the row,
- // but it is need to check in other
void set_value(const IColumn** columns, size_t pos) {
- if constexpr (is_copy) {
- _data_value.template set_value<ColVecType>(columns[0], pos);
+ if constexpr (arg_is_nullable) {
+ const auto& nullable_column =
+ assert_cast<const ColumnNullable&,
TypeCheckOnRelease::DISABLE>(*columns[0]);
+ if (nullable_column.is_null_at(pos)) {
+ _data_value.reset();
+ } else {
+ _data_value.set(nullable_column.get_nested_column(), pos);
+ }
} else {
- _data_value.set_value(columns[0], pos);
+ _data_value.set(*columns[0], pos);
}
_has_value = true;
}
- bool has_set_value() { return _has_value; }
+ bool has_set_value() const { return _has_value; }
- bool is_null() { return _data_value.is_null(); }
+ bool is_null() const { return !_data_value.has(); }
protected:
- StoreType _data_value;
+ SingleValueDataColumn _data_value;
bool _has_value = false;
};
diff --git a/be/src/exprs/aggregate/aggregate_function_window.h
b/be/src/exprs/aggregate/aggregate_function_window.h
index 56eb337f97c..1a491ef96ca 100644
--- a/be/src/exprs/aggregate/aggregate_function_window.h
+++ b/be/src/exprs/aggregate/aggregate_function_window.h
@@ -397,8 +397,7 @@ public:
};
template <bool result_is_nullable, bool arg_is_nullable>
-struct FirstLastData
- : public ReaderFirstAndLastData<void, result_is_nullable,
arg_is_nullable, false> {
+struct FirstLastData : public ReaderFirstAndLastData<result_is_nullable,
arg_is_nullable> {
public:
void set_is_null() { this->_data_value.reset(); }
};
@@ -417,19 +416,11 @@ public:
int64_t _frame_total_rows = 0;
};
-template <bool arg_is_nullable>
-struct BaseValue : public Value<arg_is_nullable> {
-public:
- bool is_null() const { return this->_ptr == nullptr; }
- // because _ptr pointer to first_argument or third argument, so it's
difficult to cast ptr
- // so here will call virtual function
- StringRef get_value() const { return
this->_ptr->get_data_at(this->_offset); }
-};
-
template <bool result_is_nullable, bool arg_is_nullable>
struct LeadLagData {
public:
static constexpr bool result_nullable = result_is_nullable;
+
void reset() {
_data_value.reset();
_is_inited = false;
@@ -438,17 +429,16 @@ public:
void insert_result_into(IColumn& to) const {
if constexpr (result_is_nullable) {
- if (_data_value.is_null()) {
+ if (!_data_value.has()) {
auto& col = assert_cast<ColumnNullable&,
TypeCheckOnRelease::DISABLE>(to);
col.insert_default();
} else {
auto& col = assert_cast<ColumnNullable&,
TypeCheckOnRelease::DISABLE>(to);
- StringRef value = _data_value.get_value();
- col.insert_data(value.data, value.size);
+ col.get_null_map_data().push_back(0);
+ _data_value.insert_result_into(col.get_nested_column());
}
} else {
- StringRef value = _data_value.get_value();
- to.insert_data(value.data, value.size);
+ _data_value.insert_result_into(to);
}
}
@@ -456,13 +446,15 @@ public:
if constexpr (arg_is_nullable) {
if (assert_cast<const ColumnNullable*,
TypeCheckOnRelease::DISABLE>(columns[0])
->is_null_at(pos)) {
- // ptr == nullptr means nullable
_data_value.reset();
return;
}
+ const auto& nullable =
+ assert_cast<const ColumnNullable&,
TypeCheckOnRelease::DISABLE>(*columns[0]);
+ _data_value.set(nullable.get_nested_column(), pos);
+ } else {
+ _data_value.set(*columns[0], pos);
}
- // here ptr is pointer to nullable column or not null column from first
- _data_value.set_value(columns[0], pos);
}
void set_value_from_default(const IColumn* column, size_t pos) {
@@ -473,10 +465,10 @@ public:
if (nullable_column->is_null_at(pos)) {
this->_data_value.reset();
} else {
-
this->_data_value.set_value(nullable_column->get_nested_column_ptr().get(),
pos);
+ this->_data_value.set(nullable_column->get_nested_column(),
pos);
}
} else {
- this->_data_value.set_value(column, pos);
+ this->_data_value.set(*column, pos);
}
}
@@ -492,7 +484,7 @@ public:
int64_t get_offset_value() const { return _offset_value; }
private:
- BaseValue<arg_is_nullable> _data_value;
+ SingleValueDataColumn _data_value;
bool _is_inited = false;
int64_t _offset_value = 0;
};
diff --git a/be/src/exprs/aggregate/single_value_data.h
b/be/src/exprs/aggregate/single_value_data.h
new file mode 100644
index 00000000000..ff947577d5d
--- /dev/null
+++ b/be/src/exprs/aggregate/single_value_data.h
@@ -0,0 +1,417 @@
+// 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.
+
+#pragma once
+
+#include <cstring>
+
+#include "common/cast_set.h"
+#include "common/compare.h"
+#include "common/logging.h"
+#include "core/assert_cast.h"
+#include "core/column/column.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_string.h"
+#include "core/custom_allocator.h"
+#include "core/data_type/data_type.h"
+#include "core/data_type/primitive_type.h"
+#include "core/string_buffer.hpp"
+#include "core/string_ref.h"
+#include "core/types.h"
+
+namespace doris {
+
+class Arena;
+
+/// Stores one fixed-size scalar value directly in the aggregate state.
+template <PrimitiveType T>
+struct SingleValueDataFixed {
+private:
+ using Self = SingleValueDataFixed;
+ using ValueType = typename PrimitiveTypeTraits<T>::CppType;
+
+ static const ValueType& _value_at(const IColumn& column, size_t row_num) {
+ return assert_cast<const typename PrimitiveTypeTraits<T>::ColumnType&,
+ TypeCheckOnRelease::DISABLE>(column)
+ .get_data()[row_num];
+ }
+
+ // AggregateFunctionIf and state merging must distinguish an empty state
from a default value.
+ bool has_value = false;
+ ValueType value {};
+
+public:
+ using ColVecType = typename PrimitiveTypeTraits<T>::ColumnType;
+ static constexpr bool NeedCheckColumnType = true;
+
+ SingleValueDataFixed() = default;
+ bool has() const { return has_value; }
+
+ static constexpr bool UsesFixedLengthStateSerialization = true;
+
+ void set_value_to_min() { value = Compare::min_value<ValueType>(); }
+
+ void set_value_to_max() { value = Compare::max_value<ValueType>(); }
+
+ void insert_result_into(IColumn& to) const {
+ if (has()) {
+ assert_cast<typename PrimitiveTypeTraits<T>::ColumnType&,
TypeCheckOnRelease::DISABLE>(
+ to)
+ .insert_value(value);
+ } else {
+ assert_cast<typename PrimitiveTypeTraits<T>::ColumnType&,
TypeCheckOnRelease::DISABLE>(
+ to)
+ .insert_default();
+ }
+ }
+
+ void reset() { has_value = false; }
+
+ void write(BufferWritable& buf, const DataTypePtr&, int) const {
+ buf.write_binary(has());
+ if (has()) {
+ buf.write_binary(value);
+ }
+ }
+
+ void read(BufferReadable& buf, const DataTypePtr&, int, Arena&) {
+ buf.read_binary(has_value);
+ if (has()) {
+ buf.read_binary(value);
+ }
+ }
+
+ void set(const IColumn& column, size_t row_num, Arena&) {
+ has_value = true;
+ value = _value_at(column, row_num);
+ }
+
+ void set(const Self& to, Arena&) {
+ DORIS_CHECK(to.has());
+ has_value = true;
+ value = to.value;
+ }
+
+ bool set_if_smaller(const IColumn& column, size_t row_num, Arena& arena) {
+ if (!has() || Compare::less(_value_at(column, row_num), value)) {
+ set(column, row_num, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool set_if_smaller(const Self& to, Arena& arena) {
+ if (to.has() && (!has() || Compare::less(to.value, value))) {
+ set(to, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool set_if_greater(const IColumn& column, size_t row_num, Arena& arena) {
+ if (!has() || Compare::greater(_value_at(column, row_num), value)) {
+ set(column, row_num, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool is_equal_to(const IColumn& column, size_t row_num) const {
+ if (!has()) {
+ return false;
+ }
+ return Compare::equal(_value_at(column, row_num), value);
+ }
+
+ bool set_if_greater(const Self& to, Arena& arena) {
+ if (to.has() && (!has() || Compare::greater(to.value, value))) {
+ set(to, arena);
+ return true;
+ }
+ return false;
+ }
+};
+
+/// Stores short strings inline and allocates a separate buffer for long
strings.
+struct SingleValueDataString {
+private:
+ using Self = SingleValueDataString;
+ // Keep the signed 32-bit size because -1 represents an empty state in the
serialized format.
+ Int32 size = -1; /// -1 indicates that there is no value.
+ Int32 capacity = 0; /// power of two or zero
+ DorisUniqueBufferPtr<char> large_data;
+
+public:
+ static constexpr Int32 AUTOMATIC_STORAGE_SIZE = 64;
+ static constexpr Int32 MAX_SMALL_STRING_SIZE =
+ AUTOMATIC_STORAGE_SIZE - sizeof(size) - sizeof(capacity) -
sizeof(large_data);
+
+private:
+ char small_data[MAX_SMALL_STRING_SIZE];
+
+public:
+ using ColVecType = ColumnString;
+ static constexpr bool NeedCheckColumnType = true;
+
+ ~SingleValueDataString() = default;
+
+ static constexpr bool UsesFixedLengthStateSerialization = false;
+
+ bool has() const { return size >= 0; }
+
+private:
+ static StringRef _value_at(const IColumn& column, size_t row_num) {
+ return assert_cast<const ColumnString&,
TypeCheckOnRelease::DISABLE>(column).get_data_at(
+ row_num);
+ }
+
+ const char* _data() const {
+ return size <= MAX_SMALL_STRING_SIZE ? small_data : large_data.get();
+ }
+
+ StringRef _value() const { return StringRef(_data(), size); }
+
+ void _set(StringRef source) {
+ Int32 value_size = cast_set<Int32>(source.size);
+ if (value_size <= MAX_SMALL_STRING_SIZE) {
+ /// Don't free large_data here.
+ size = value_size;
+
+ if (size > 0) {
+ memcpy(small_data, source.data, size);
+ }
+ } else {
+ if (capacity < value_size) {
+ /// Don't free large_data here.
+ capacity = (Int32)round_up_to_power_of_two_or_zero(value_size);
+ large_data = DorisUniqueBufferPtr<char>(capacity);
+ }
+
+ size = value_size;
+ memcpy(large_data.get(), source.data, size);
+ }
+ }
+
+public:
+ void insert_result_into(IColumn& to) const {
+ if (has()) {
+ assert_cast<ColumnString&,
TypeCheckOnRelease::DISABLE>(to).insert_data(_data(), size);
+ } else {
+ assert_cast<ColumnString&,
TypeCheckOnRelease::DISABLE>(to).insert_default();
+ }
+ }
+
+ void reset() {
+ size = -1;
+ capacity = 0;
+ large_data.reset();
+ }
+
+ void write(BufferWritable& buf, const DataTypePtr&, int) const {
+ buf.write_binary(size);
+ if (has()) {
+ buf.write(_data(), size);
+ }
+ }
+
+ void read(BufferReadable& buf, const DataTypePtr&, int, Arena&) {
+ Int32 rhs_size;
+ buf.read_binary(rhs_size);
+
+ if (rhs_size >= 0) {
+ if (rhs_size <= MAX_SMALL_STRING_SIZE) {
+ /// Don't free large_data here.
+
+ size = rhs_size;
+
+ if (size > 0) {
+ buf.read(small_data, size);
+ }
+ } else {
+ if (capacity < rhs_size) {
+ capacity =
(Int32)round_up_to_power_of_two_or_zero(rhs_size);
+ large_data = DorisUniqueBufferPtr<char>(capacity);
+ }
+
+ size = rhs_size;
+ buf.read(large_data.get(), size);
+ }
+ } else {
+ /// Don't free large_data here.
+ size = rhs_size;
+ }
+ }
+
+ void set(const IColumn& column, size_t row_num, Arena&) {
_set(_value_at(column, row_num)); }
+
+ void set(const Self& to, Arena&) {
+ DORIS_CHECK(to.has());
+ _set(to._value());
+ }
+
+ bool set_if_smaller(const IColumn& column, size_t row_num, Arena& arena) {
+ if (!has() || _value_at(column, row_num) < _value()) {
+ set(column, row_num, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool set_if_greater(const IColumn& column, size_t row_num, Arena& arena) {
+ if (!has() || _value_at(column, row_num) > _value()) {
+ set(column, row_num, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool set_if_smaller(const Self& to, Arena& arena) {
+ if (to.has() && (!has() || to._value() < _value())) {
+ set(to, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool set_if_greater(const Self& to, Arena& arena) {
+ if (to.has() && (!has() || to._value() > _value())) {
+ set(to, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool is_equal_to(const IColumn& column, size_t row_num) const {
+ if (!has()) {
+ return false;
+ }
+ return _value_at(column, row_num) == _value();
+ }
+};
+
+static_assert(sizeof(SingleValueDataString) ==
SingleValueDataString::AUTOMATIC_STORAGE_SIZE);
+
+/// Owns a materialized one-row column for values without a dedicated
representation above.
+class SingleValueDataColumn {
+private:
+ using Self = SingleValueDataColumn;
+
+ ColumnPtr _value;
+
+public:
+ static constexpr bool NeedCheckColumnType = false;
+ static constexpr bool UsesFixedLengthStateSerialization = false;
+
+ bool has() const { return _value.get() != nullptr; }
+
+ size_t allocated_bytes() const { return has() ? _value->allocated_bytes()
: 0; }
+
+ void set(const IColumn& column, size_t row_num) {
+ auto value = column.clone_empty();
+ DCHECK(value->empty());
+ value->reserve(1);
+ value->insert_from(column, row_num);
+ DCHECK_EQ(value->size(), 1);
+ _value = std::move(value);
+ }
+
+ void insert_result_into(IColumn& to) const {
+ if (has()) {
+ to.insert_from(*_value, 0);
+ } else {
+ to.insert_default();
+ }
+ }
+
+ void reset() { _value.reset(); }
+
+ void write(BufferWritable& buf, const DataTypePtr& data_type, int
be_exec_version) const {
+ buf.write_binary(has());
+ if (!has()) {
+ return;
+ }
+ auto size_bytes =
data_type->get_uncompressed_serialized_bytes(*_value, be_exec_version);
+ buf.write_binary(size_bytes);
+ buf.resize(size_bytes);
+ auto* p = data_type->serialize(*_value, buf.data(), be_exec_version);
+ DCHECK_EQ(p, buf.data() + size_bytes);
+ buf.add_offset(size_bytes);
+ }
+
+ void read(BufferReadable& buf, const DataTypePtr& data_type, int
be_exec_version, Arena&) {
+ bool has_value = false;
+ buf.read_binary(has_value);
+ if (!has_value) {
+ reset();
+ return;
+ }
+ int64_t size = 0;
+ buf.read_binary(size);
+ auto value = data_type->create_column();
+ value->reserve(1);
+ const auto* p = data_type->deserialize(buf.data(), &value,
be_exec_version);
+ DCHECK_EQ(p, buf.data() + size);
+ buf.add_offset(size);
+ _value = std::move(value);
+ }
+
+ void set(const IColumn& column, size_t row_num, Arena&) { set(column,
row_num); }
+
+ void set(const Self& to, Arena&) {
+ DORIS_CHECK(to.has());
+ // Stored one-row columns are immutable, so merged states can share
ownership.
+ _value = to._value;
+ }
+
+ bool set_if_smaller(const IColumn& column, size_t row_num, Arena& arena) {
+ if (!has() || column.compare_at(row_num, 0, *_value, 1) < 0) {
+ set(column, row_num, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool set_if_smaller(const Self& to, Arena& arena) {
+ if (to.has() && (!has() || to._value->compare_at(0, 0, *_value, 1) <
0)) {
+ set(to, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool set_if_greater(const IColumn& column, size_t row_num, Arena& arena) {
+ if (!has() || column.compare_at(row_num, 0, *_value, 1) > 0) {
+ set(column, row_num, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool set_if_greater(const Self& to, Arena& arena) {
+ if (to.has() && (!has() || to._value->compare_at(0, 0, *_value, 1) >
0)) {
+ set(to, arena);
+ return true;
+ }
+ return false;
+ }
+
+ bool is_equal_to(const IColumn& column, size_t row_num) const {
+ return has() && column.compare_at(row_num, 0, *_value, 1) == 0;
+ }
+};
+
+} // namespace doris
diff --git a/be/test/exprs/aggregate/agg_replace_test.cpp
b/be/test/exprs/aggregate/agg_replace_test.cpp
index bb00051cbee..af57cb84c86 100644
--- a/be/test/exprs/aggregate/agg_replace_test.cpp
+++ b/be/test/exprs/aggregate/agg_replace_test.cpp
@@ -51,7 +51,6 @@
#include "core/value/hll.h"
#include "exprs/aggregate/aggregate_function.h"
#include "exprs/aggregate/aggregate_function_reader.h"
-#include "exprs/aggregate/aggregate_function_reader_first_last.h"
#include "exprs/aggregate/aggregate_function_simple_factory.h"
#include "gtest/gtest_pred_impl.h"
@@ -307,129 +306,10 @@ public:
agg_function->destroy(place);
}
- template <typename DataType, typename ColumnType, bool nullable>
- void test_basic_data(int8_t input_nums) {
- DataTypePtr data_type = get_data_type<DataType, nullable, false>();
-
- auto data_column = data_type->create_column();
- add_elements<DataType, nullable>(data_column, input_nums);
-
- EXPECT_EQ(input_nums, data_column->size());
- //test Value
- {
- Value<nullable> value;
- EXPECT_TRUE(value.is_null());
- for (int64_t i = 0; i < input_nums; ++i) {
- value.set_value(data_column.get(), i);
- EXPECT_FALSE(value.is_null());
- auto to_column = data_type->create_column();
- if constexpr (nullable) {
- auto& nullable_col =
assert_cast<ColumnNullable&>(*to_column);
- value.insert_into(nullable_col.get_nested_column());
- } else {
- value.insert_into(*to_column);
- }
-
- EXPECT_EQ(1, to_column->size());
- check_column_basic<DataType, nullable>(to_column.get(), i);
- }
- }
- //test CopiedValue
- {
- CopiedValue<nullable> value;
- EXPECT_TRUE(value.is_null());
- for (int64_t i = 0; i < input_nums; ++i) {
- value.template set_value<ColumnType>(data_column.get(), i);
- EXPECT_FALSE(value.is_null());
- auto to_column = data_type->create_column();
- if constexpr (nullable) {
- auto& nullable_col =
assert_cast<ColumnNullable&>(*to_column);
- value.template
insert_into<ColumnType>(nullable_col.get_nested_column());
- } else {
- value.template insert_into<ColumnType>(*to_column);
- }
- EXPECT_EQ(1, to_column->size());
- check_column_basic<DataType, nullable>(to_column.get(), i);
- }
- }
- }
-
- template <typename DataType, typename ColumnType, bool nullable>
- void test_array_data(int8_t input_nums) {
- DataTypePtr data_type = get_data_type<DataType, nullable, true>();
-
- auto data_column = data_type->create_column();
- array_add_elements<DataType, nullable>(data_column, input_nums);
-
- EXPECT_EQ(input_nums + 1, data_column->size());
- //test Value
- {
- Value<nullable> value;
- EXPECT_TRUE(value.is_null());
- for (int64_t i = 0; i <= input_nums; ++i) {
- value.set_value(data_column.get(), i);
- EXPECT_FALSE(value.is_null());
- auto to_column = data_type->create_column();
- if constexpr (nullable) {
- auto& nullable_col =
assert_cast<ColumnNullable&>(*to_column);
- value.insert_into(nullable_col.get_nested_column());
- } else {
- value.insert_into(*to_column);
- }
- EXPECT_EQ(1, to_column->size());
- check_column_array<DataType, nullable>(to_column.get(), i);
- }
- }
- //test CopiedValue
- {
- CopiedValue<nullable> value;
- EXPECT_TRUE(value.is_null());
- for (int64_t i = 0; i <= input_nums; ++i) {
- value.template set_value<ColumnArray>(data_column.get(), i);
- EXPECT_FALSE(value.is_null());
- auto to_column = data_type->create_column();
- if constexpr (nullable) {
- auto& nullable_col =
assert_cast<ColumnNullable&>(*to_column);
- value.template
insert_into<ColumnArray>(nullable_col.get_nested_column());
- } else {
- value.template insert_into<ColumnArray>(*to_column);
- }
- EXPECT_EQ(1, to_column->size());
- check_column_array<DataType, nullable>(to_column.get(), i);
- }
- }
- }
-
private:
Arena _agg_arena_pool;
};
-TEST_F(VAggReplaceTest, test_basic_data) {
- test_basic_data<DataTypeInt8, ColumnInt8, false>(11);
- test_basic_data<DataTypeInt16, ColumnInt16, false>(11);
- test_basic_data<DataTypeInt32, ColumnInt32, false>(11);
- test_basic_data<DataTypeInt64, ColumnInt64, false>(11);
- test_basic_data<DataTypeInt128, ColumnInt128, false>(11);
- test_basic_data<DataTypeDecimalV2, ColumnDecimal128V2, false>(11);
- test_basic_data<DataTypeString, ColumnString, false>(11);
- test_basic_data<DataTypeInt128, ColumnInt128, false>(11);
- test_basic_data<DataTypeDate, ColumnDate, false>(11);
- test_basic_data<DataTypeDateTime, ColumnDateTime, false>(11);
-}
-
-TEST_F(VAggReplaceTest, test_array_data) {
- test_array_data<DataTypeInt8, ColumnArray, false>(11);
- test_array_data<DataTypeInt16, ColumnArray, false>(11);
- test_array_data<DataTypeInt32, ColumnArray, false>(11);
- test_array_data<DataTypeInt64, ColumnArray, false>(11);
- test_array_data<DataTypeInt128, ColumnArray, false>(11);
- test_array_data<DataTypeDecimalV2, ColumnArray, false>(11);
- test_array_data<DataTypeString, ColumnArray, false>(11);
- test_array_data<DataTypeInt128, ColumnArray, false>(11);
- test_array_data<DataTypeDate, ColumnArray, false>(11);
- test_array_data<DataTypeDateTime, ColumnArray, false>(11);
-}
-
TEST_F(VAggReplaceTest, test_basic_replace_reader) {
test_agg_replace<DataTypeInt8, false>("replace_reader", 10, 0);
test_agg_replace<DataTypeInt16, false>("replace_reader", 10, 0);
diff --git a/be/test/exprs/aggregate/agg_window_nth_value_test.cpp
b/be/test/exprs/aggregate/agg_window_nth_value_test.cpp
index 61ac798ba56..dce7a0729e7 100644
--- a/be/test/exprs/aggregate/agg_window_nth_value_test.cpp
+++ b/be/test/exprs/aggregate/agg_window_nth_value_test.cpp
@@ -78,4 +78,51 @@ TEST(AggregateWindowNthValueTest,
UpperBoundedLowerUnboundedFrame) {
function->destroy(place);
}
+TEST(AggregateWindowValueTest, OwnsValueAfterSourceColumnIsCleared) {
+ AggregateFunctionSimpleFactory factory;
+ register_aggregate_function_window_lead_lag_first_last(factory);
+
+ for (const std::string function_name :
+ {"first_value", "last_value", "nth_value", "lead", "lag"}) {
+ SCOPED_TRACE(function_name);
+ DataTypes argument_types = {std::make_shared<DataTypeString>()};
+ if (function_name == "nth_value") {
+ argument_types.push_back(std::make_shared<DataTypeInt64>());
+ } else if (function_name == "lead" || function_name == "lag") {
+ argument_types.push_back(std::make_shared<DataTypeInt64>());
+ argument_types.push_back(std::make_shared<DataTypeString>());
+ }
+ auto function = factory.get(function_name, argument_types, nullptr,
true, -1,
+ {.is_window_function = true, .column_names
= {}});
+ ASSERT_NE(function, nullptr);
+
+ auto value_column = ColumnString::create();
+ value_column->insert_data("A", 1);
+ value_column->insert_data("B", 1);
+ value_column->insert_data("C", 1);
+ auto offset_column = ColumnInt64::create();
+ offset_column->insert_value(1);
+ auto default_column = ColumnString::create();
+ default_column->insert_data("default", 7);
+ const IColumn* columns[] = {value_column.get(), offset_column.get(),
default_column.get()};
+
+ Arena arena;
+ auto* place =
reinterpret_cast<AggregateDataPtr>(arena.alloc(function->size_of_data()));
+ function->create(place);
+ UInt8 use_null_result = false;
+ UInt8 could_use_previous_result = false;
+ function->add_range_single_place(0, 3, 1, 2, place, columns, arena,
&use_null_result,
+ &could_use_previous_result);
+
+ value_column->clear();
+ auto result_column = ColumnNullable::create(ColumnString::create(),
ColumnUInt8::create());
+ function->insert_result_into(place, *result_column);
+
+ ASSERT_EQ(result_column->size(), 1);
+ EXPECT_FALSE(result_column->is_null_at(0));
+ EXPECT_EQ(result_column->get_data_at(0).to_string(), "B");
+ function->destroy(place);
+ }
+}
+
} // namespace doris
diff --git a/be/test/exprs/aggregate/aggregate_function_min_max_test.cpp
b/be/test/exprs/aggregate/aggregate_function_min_max_test.cpp
index 57f8941ced4..411de8f5500 100644
--- a/be/test/exprs/aggregate/aggregate_function_min_max_test.cpp
+++ b/be/test/exprs/aggregate/aggregate_function_min_max_test.cpp
@@ -21,14 +21,30 @@
#include <string>
+#include "agent/be_exec_version_manager.h"
#include "core/arena.h"
#include "core/column/column_string.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
#include "core/string_buffer.hpp"
namespace doris {
class SingleValueDataStringTest : public testing::Test {
protected:
+ void set_value(SingleValueDataString& data, const std::string& value) {
+ auto column = ColumnString::create();
+ column->insert_data(value.data(), value.size());
+ data.set(*column, 0, arena);
+ }
+
+ std::string get_value(const SingleValueDataString& data) {
+ auto column = ColumnString::create();
+ data.insert_result_into(*column);
+ return column->get_data_at(0).to_string();
+ }
+
Arena arena;
};
@@ -44,126 +60,123 @@ TEST_F(SingleValueDataStringTest, ResetWhenNoValue) {
ASSERT_FALSE(data.has());
}
-TEST_F(SingleValueDataStringTest, SmallStringChangeImpl) {
+TEST_F(SingleValueDataStringTest, SetSmallString) {
SingleValueDataString data;
std::string small = "hello";
- data.change_impl(StringRef(small.data(), small.size()), arena);
+ set_value(data, small);
ASSERT_TRUE(data.has());
- auto ref = data.get_string_ref();
- ASSERT_EQ(ref.size, small.size());
- ASSERT_EQ(std::string(ref.data, ref.size), small);
+ ASSERT_EQ(get_value(data), small);
}
-TEST_F(SingleValueDataStringTest, LargeStringChangeImpl) {
+TEST_F(SingleValueDataStringTest, SetLargeString) {
SingleValueDataString data;
// Create a string larger than MAX_SMALL_STRING_SIZE
std::string large(SingleValueDataString::MAX_SMALL_STRING_SIZE + 10, 'x');
- data.change_impl(StringRef(large.data(), large.size()), arena);
+ set_value(data, large);
ASSERT_TRUE(data.has());
- auto ref = data.get_string_ref();
- ASSERT_EQ(ref.size, large.size());
- ASSERT_EQ(std::string(ref.data, ref.size), large);
+ ASSERT_EQ(get_value(data), large);
}
-TEST_F(SingleValueDataStringTest, ResetAfterChange) {
+TEST_F(SingleValueDataStringTest, ResetAfterSet) {
SingleValueDataString data;
std::string s = "test";
- data.change_impl(StringRef(s.data(), s.size()), arena);
+ set_value(data, s);
ASSERT_TRUE(data.has());
data.reset();
ASSERT_FALSE(data.has());
}
-TEST_F(SingleValueDataStringTest, ChangeIfLess) {
+TEST_F(SingleValueDataStringTest, SetIfSmaller) {
SingleValueDataString data;
std::string a = "banana";
std::string b = "apple";
- data.change_impl(StringRef(a.data(), a.size()), arena);
- ASSERT_EQ(std::string(data.get_string_ref().data,
data.get_string_ref().size), a);
+ set_value(data, a);
+ ASSERT_EQ(get_value(data), a);
SingleValueDataString other;
- other.change_impl(StringRef(b.data(), b.size()), arena);
+ set_value(other, b);
- ASSERT_TRUE(data.change_if_less(other, arena));
- ASSERT_EQ(std::string(data.get_string_ref().data,
data.get_string_ref().size), b);
+ ASSERT_TRUE(data.set_if_smaller(other, arena));
+ ASSERT_EQ(get_value(data), b);
// "apple" is not less than "apple"
- ASSERT_FALSE(data.change_if_less(other, arena));
+ ASSERT_FALSE(data.set_if_smaller(other, arena));
}
-TEST_F(SingleValueDataStringTest, ChangeIfGreater) {
+TEST_F(SingleValueDataStringTest, SetIfGreater) {
SingleValueDataString data;
std::string a = "apple";
std::string b = "banana";
- data.change_impl(StringRef(a.data(), a.size()), arena);
+ set_value(data, a);
SingleValueDataString other;
- other.change_impl(StringRef(b.data(), b.size()), arena);
+ set_value(other, b);
- ASSERT_TRUE(data.change_if_greater(other, arena));
- ASSERT_EQ(std::string(data.get_string_ref().data,
data.get_string_ref().size), b);
+ ASSERT_TRUE(data.set_if_greater(other, arena));
+ ASSERT_EQ(get_value(data), b);
- ASSERT_FALSE(data.change_if_greater(other, arena));
+ ASSERT_FALSE(data.set_if_greater(other, arena));
}
-TEST_F(SingleValueDataStringTest, ChangeFirstTime) {
+TEST_F(SingleValueDataStringTest, SetFromState) {
SingleValueDataString data;
SingleValueDataString src;
std::string s = "first";
- src.change_impl(StringRef(s.data(), s.size()), arena);
+ set_value(src, s);
- data.change_first_time(src, arena);
+ data.set(src, arena);
ASSERT_TRUE(data.has());
- ASSERT_EQ(std::string(data.get_string_ref().data,
data.get_string_ref().size), s);
+ ASSERT_EQ(get_value(data), s);
- // Second call should not change
SingleValueDataString other;
std::string s2 = "second";
- other.change_impl(StringRef(s2.data(), s2.size()), arena);
- data.change_first_time(other, arena);
- ASSERT_EQ(std::string(data.get_string_ref().data,
data.get_string_ref().size), s);
+ set_value(other, s2);
+ data.set(other, arena);
+ ASSERT_EQ(get_value(data), s2);
}
TEST_F(SingleValueDataStringTest, WriteReadSmallString) {
SingleValueDataString data;
std::string s = "serialize_me";
- data.change_impl(StringRef(s.data(), s.size()), arena);
+ set_value(data, s);
// Write
auto col_write = ColumnString::create();
BufferWritable writer(*col_write);
- data.write(writer);
+ auto data_type = std::make_shared<DataTypeString>();
+ data.write(writer, data_type, -1);
writer.commit();
// Read
auto ref = col_write->get_data_at(0);
BufferReadable reader(ref);
SingleValueDataString data2;
- data2.read(reader, arena);
+ data2.read(reader, data_type, -1, arena);
ASSERT_TRUE(data2.has());
- ASSERT_EQ(std::string(data2.get_string_ref().data,
data2.get_string_ref().size), s);
+ ASSERT_EQ(get_value(data2), s);
}
TEST_F(SingleValueDataStringTest, WriteReadLargeString) {
SingleValueDataString data;
std::string s(SingleValueDataString::MAX_SMALL_STRING_SIZE + 20, 'L');
- data.change_impl(StringRef(s.data(), s.size()), arena);
+ set_value(data, s);
auto col_write = ColumnString::create();
BufferWritable writer(*col_write);
- data.write(writer);
+ auto data_type = std::make_shared<DataTypeString>();
+ data.write(writer, data_type, -1);
writer.commit();
auto ref = col_write->get_data_at(0);
BufferReadable reader(ref);
SingleValueDataString data2;
- data2.read(reader, arena);
+ data2.read(reader, data_type, -1, arena);
ASSERT_TRUE(data2.has());
- ASSERT_EQ(std::string(data2.get_string_ref().data,
data2.get_string_ref().size), s);
+ ASSERT_EQ(get_value(data2), s);
}
TEST_F(SingleValueDataStringTest, WriteReadNoValue) {
@@ -171,13 +184,14 @@ TEST_F(SingleValueDataStringTest, WriteReadNoValue) {
auto col_write = ColumnString::create();
BufferWritable writer(*col_write);
- data.write(writer);
+ auto data_type = std::make_shared<DataTypeString>();
+ data.write(writer, data_type, -1);
writer.commit();
auto ref = col_write->get_data_at(0);
BufferReadable reader(ref);
SingleValueDataString data2;
- data2.read(reader, arena);
+ data2.read(reader, data_type, -1, arena);
ASSERT_FALSE(data2.has());
}
@@ -185,7 +199,7 @@ TEST_F(SingleValueDataStringTest, WriteReadNoValue) {
TEST_F(SingleValueDataStringTest, InsertResultIntoWithValue) {
SingleValueDataString data;
std::string s = "result";
- data.change_impl(StringRef(s.data(), s.size()), arena);
+ set_value(data, s);
auto col = ColumnString::create();
data.insert_result_into(*col);
@@ -208,17 +222,104 @@ TEST_F(SingleValueDataStringTest, LargeStringRealloc) {
SingleValueDataString data;
// First large allocation
std::string s1(SingleValueDataString::MAX_SMALL_STRING_SIZE + 10, 'A');
- data.change_impl(StringRef(s1.data(), s1.size()), arena);
- ASSERT_EQ(std::string(data.get_string_ref().data,
data.get_string_ref().size), s1);
+ set_value(data, s1);
+ ASSERT_EQ(get_value(data), s1);
// Second larger allocation triggers realloc
std::string s2(SingleValueDataString::MAX_SMALL_STRING_SIZE + 200, 'B');
- data.change_impl(StringRef(s2.data(), s2.size()), arena);
- ASSERT_EQ(std::string(data.get_string_ref().data,
data.get_string_ref().size), s2);
+ set_value(data, s2);
+ ASSERT_EQ(get_value(data), s2);
}
TEST_F(SingleValueDataStringTest, SizeStaticAssert) {
static_assert(sizeof(SingleValueDataString) ==
SingleValueDataString::AUTOMATIC_STORAGE_SIZE);
}
+TEST(SingleValueDataColumnTest, OwnsValueAndCreatesColumnLazily) {
+ auto array_type =
std::make_shared<DataTypeArray>(std::make_shared<DataTypeInt32>());
+ SingleValueDataColumn data;
+ EXPECT_FALSE(data.has());
+ EXPECT_EQ(data.allocated_bytes(), 0);
+
+ auto source = array_type->create_column();
+ source->insert(Field::create_field<TYPE_ARRAY>(
+ Array {Field::create_field<TYPE_INT>(1),
Field::create_field<TYPE_INT>(2)}));
+ Arena arena;
+ data.set(*source, 0, arena);
+ ASSERT_TRUE(data.has());
+
+ source->clear();
+ auto result = array_type->create_column();
+ data.insert_result_into(*result);
+ ASSERT_EQ(result->size(), 1);
+ EXPECT_EQ((*result)[0],
+ Field::create_field<TYPE_ARRAY>(
+ Array {Field::create_field<TYPE_INT>(1),
Field::create_field<TYPE_INT>(2)}));
+}
+
+TEST(SingleValueDataColumnTest, AllocationDoesNotScaleWithSourceRows) {
+ constexpr size_t large_source_rows = 4096;
+ auto array_type =
std::make_shared<DataTypeArray>(std::make_shared<DataTypeInt32>());
+ auto create_source = [&](size_t rows) {
+ auto source = array_type->create_column();
+ const auto value = Field::create_field<TYPE_ARRAY>(
+ Array {Field::create_field<TYPE_INT>(1),
Field::create_field<TYPE_INT>(2)});
+ for (size_t i = 0; i < rows; ++i) {
+ source->insert(value);
+ }
+ return source;
+ };
+
+ auto single_row_source = create_source(1);
+ auto large_source = create_source(large_source_rows);
+ ASSERT_GT(large_source->allocated_bytes(),
single_row_source->allocated_bytes());
+
+ Arena arena;
+ SingleValueDataColumn single_row_state;
+ single_row_state.set(*single_row_source, 0, arena);
+ SingleValueDataColumn large_source_state;
+ large_source_state.set(*large_source, large_source_rows - 1, arena);
+
+ EXPECT_EQ(large_source_state.allocated_bytes(),
single_row_state.allocated_bytes());
+ EXPECT_LT(large_source_state.allocated_bytes(),
large_source->allocated_bytes());
+
+ auto result = array_type->create_column();
+ large_source_state.insert_result_into(*result);
+ ASSERT_EQ(result->size(), 1);
+ EXPECT_EQ((*result)[0], (*large_source)[large_source_rows - 1]);
+}
+
+TEST(SingleValueDataColumnTest, CompareCopyAndSerialize) {
+ auto array_type =
std::make_shared<DataTypeArray>(std::make_shared<DataTypeInt32>());
+ const int be_exec_version = BeExecVersionManager::get_newest_version();
+ auto source = array_type->create_column();
+ source->insert(Field::create_field<TYPE_ARRAY>(Array
{Field::create_field<TYPE_INT>(2)}));
+ source->insert(Field::create_field<TYPE_ARRAY>(Array
{Field::create_field<TYPE_INT>(1)}));
+
+ Arena arena;
+ SingleValueDataColumn data;
+ EXPECT_TRUE(data.set_if_smaller(*source, 0, arena));
+ EXPECT_TRUE(data.set_if_smaller(*source, 1, arena));
+ EXPECT_TRUE(data.is_equal_to(*source, 1));
+
+ SingleValueDataColumn copied;
+ copied.set(data, arena);
+ data.reset();
+ ASSERT_TRUE(copied.has());
+
+ auto serialized = ColumnString::create();
+ BufferWritable writer(*serialized);
+ copied.write(writer, array_type, be_exec_version);
+ writer.commit();
+
+ SingleValueDataColumn restored;
+ BufferReadable reader(serialized->get_data_at(0));
+ restored.read(reader, array_type, be_exec_version, arena);
+ auto result = array_type->create_column();
+ restored.insert_result_into(*result);
+ ASSERT_EQ(result->size(), 1);
+ EXPECT_EQ((*result)[0],
+ Field::create_field<TYPE_ARRAY>(Array
{Field::create_field<TYPE_INT>(1)}));
+}
+
} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
