From 30c7406198fe6cc3cc0e95f6461901c404b1061d Mon Sep 17 00:00:00 2001 From: Mryange Date: Fri, 18 Sep 2026 23:49:02 +0800 Subject: [PATCH] [fix](be) Own values in single-value aggregate states ### What problem does this PR solve? Issue Number: N/A Related PR: #67439 Problem Summary: Single-value aggregate states duplicated storage logic, while first_value, last_value, nth_value, lead, and lag retained raw input column references that could outlive reclaimed input columns. Introduce reusable fixed-width, string, and owned-column state implementations, reuse them across min/max, min_by/max_by, any, and window value functions, and keep complex window types on non-incremental evaluation when equality comparison is unsupported. ### Release note Fix single-value window functions retaining references to reclaimed input columns. ### Check List (For Author) - Test: Unit Test - ./run-be-ut.sh --run --filter=SingleValueDataStringTest.*:SingleValueDataColumnTest.* -j 48 (18 tests passed under ASAN) - Behavior changed: Yes. Window states now own selected values instead of retaining raw input column references. - Does this need documentation: No --- .../aggregate/aggregate_function_min_max.h | 734 ++---------------- .../aggregate_function_min_max_by.cpp | 11 +- .../aggregate/aggregate_function_min_max_by.h | 146 ++-- .../aggregate_function_min_max_impl.h | 12 +- .../aggregate_function_reader_first_last.h | 128 +-- .../aggregate/aggregate_function_window.h | 36 +- be/src/exprs/aggregate/single_value_data.h | 417 ++++++++++ be/test/exprs/aggregate/agg_replace_test.cpp | 120 --- .../aggregate/agg_window_nth_value_test.cpp | 47 ++ .../aggregate_function_min_max_test.cpp | 197 +++-- 10 files changed, 776 insertions(+), 1072 deletions(-) create mode 100644 be/src/exprs/aggregate/single_value_data.h diff --git a/be/src/exprs/aggregate/aggregate_function_min_max.h b/be/src/exprs/aggregate/aggregate_function_min_max.h index 9a766ddb96a917..416bf226196bb1 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 -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::CppType value; - -public: - using ColVecType = typename PrimitiveTypeTraits::ColumnType; - static constexpr bool NeedCheckColumnType = true; - - SingleValueDataFixed() = default; - SingleValueDataFixed(bool has_value_, typename PrimitiveTypeTraits::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::CppType>() - : Compare::min_value::CppType>(); - } - - void change_if(const IColumn& column, size_t row_num, bool less) { - has_value = true; - value = less ? Compare::min(assert_cast::ColumnType&, - TypeCheckOnRelease::DISABLE>(column) - .get_data()[row_num], - value) - : Compare::max(assert_cast::ColumnType&, - TypeCheckOnRelease::DISABLE>(column) - .get_data()[row_num], - value); - } - - void insert_result_into(IColumn& to) const { - if (has()) { - assert_cast::ColumnType&, TypeCheckOnRelease::DISABLE>( - to) - .get_data() - .push_back(value); - } else { - assert_cast::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::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::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::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::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 -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::CppType value; - -public: - using ColVecType = typename PrimitiveTypeTraits::ColumnType; - static constexpr bool NeedCheckColumnType = true; - - SingleValueDataDecimal() = default; - SingleValueDataDecimal(bool has_value_, typename PrimitiveTypeTraits::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::CppType>() - : Compare::min_value::CppType>(); - } - - void change_if(const IColumn& column, size_t row_num, bool less) { - has_value = true; - value = less ? Compare::min(assert_cast::ColumnType&, - TypeCheckOnRelease::DISABLE>(column) - .get_data()[row_num], - value) - : Compare::max(assert_cast::ColumnType&, - TypeCheckOnRelease::DISABLE>(column) - .get_data()[row_num], - value); - } - - void insert_result_into(IColumn& to) const { - if (has()) { - assert_cast::ColumnType&, TypeCheckOnRelease::DISABLE>( - to) - .insert_data((const char*)&value, 0); - } else { - assert_cast::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::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::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::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::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 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(to).insert_data(get_data(), - size); - } else { - assert_cast(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(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(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(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(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(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(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(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 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 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 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 class AggregateFunctionsSingleValue final : public IAggregateFunctionDataHelper> { private: - DataTypePtr& type; + const DataTypePtr& _data_type; using Base = IAggregateFunctionDataHelper>; using IAggregateFunction::argument_types; public: AggregateFunctionsSingleValue(const DataTypes& arguments) : IAggregateFunctionDataHelper>(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> || - std::is_same_v> || - std::is_same_v>) { - 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 @@ class AggregateFunctionsSingleValue final } 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 @@ class AggregateFunctionsSingleValue final void serialize_to_column(const std::vector& 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(*dst); dst_column.resize(num_rows); auto* dst_data = reinterpret_cast(dst_column.get_data().data()); @@ -859,12 +218,12 @@ class AggregateFunctionsSingleValue final 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(*dst); dst_column.resize(num_rows); auto* dst_data = reinterpret_cast(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 @@ class AggregateFunctionsSingleValue final 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(column); @@ -890,7 +249,7 @@ class AggregateFunctionsSingleValue final 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(*column); const auto* data = col.get_data().data(); this->merge_vec(places, offset, AggregateDataPtr(data), arena, num_rows); @@ -904,7 +263,7 @@ class AggregateFunctionsSingleValue final 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(*column); const auto* data = col.get_data().data(); this->merge_vec_selected(places, offset, AggregateDataPtr(data), arena, num_rows); @@ -917,7 +276,7 @@ class AggregateFunctionsSingleValue final 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(to); size_t old_size = col.size(); col.resize(old_size + 1); @@ -928,7 +287,7 @@ class AggregateFunctionsSingleValue final } 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 @@ class AggregateFunctionsSingleValue final } DataTypePtr get_serialized_type() const override { - if constexpr (Data::IsFixedLength) { + if constexpr (Data::UsesFixedLengthStateSerialization) { return std::make_shared(); } else { return std::make_shared(); } } - 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 @@ class AggregateFunctionsSingleValue final 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 70e5508069e9ea..f0136d78701662 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 create_max_min_value(const DataTypePtr& type, int be_version) { +std::unique_ptr create_max_min_value(const DataTypePtr& type) { std::unique_ptr result; auto call = [&](const auto& dispatch_type) -> bool { using DispatchType = std::decay_t; constexpr auto PT = DispatchType::PType; - if constexpr (is_decimal(PT)) { - result = std::make_unique>>(); - } else { - result = std::make_unique>>(); - } + result = std::make_unique>>(); return true; }; if (type->get_primitive_type() == TYPE_TIMESTAMP_NS) { @@ -51,8 +47,7 @@ std::unique_ptr create_max_min_value(const DataTypePtr& type, i case PrimitiveType::TYPE_ARRAY: case PrimitiveType::TYPE_MAP: case PrimitiveType::TYPE_STRUCT: - return std::make_unique>(DataTypes {type}, - be_version); + return std::make_unique>(); 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 dfba3fbcd25038..a692aca65c0615 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 @@ -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(to); - value.change(derived.value, arena); + value.set(derived.value, arena); } }; -std::unique_ptr create_max_min_value(const DataTypePtr& type, int be_version); +std::unique_ptr create_max_min_value(const DataTypePtr& type); /// For bitmap value struct BitmapValueData { @@ -98,27 +103,28 @@ struct BitmapValueData { } } - 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(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 @@ struct BitmapValueData { * 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 @@ struct AggregateFunctionMinMaxByBaseData { public: AggregateFunctionMinMaxByBaseData() {} - AggregateFunctionMinMaxByBaseData(const DataTypes argument_types, int be_version) - requires(std::is_same_v) - : 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) - { - 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 @@ struct AggregateFunctionMinMaxByBaseData { 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 AggregateFunctionMaxByData() {} - AggregateFunctionMaxByData(const DataTypes argument_types, int be_version) - : AggregateFunctionMinMaxByBaseData(argument_types, be_version) {} + explicit AggregateFunctionMaxByData(const DataTypes& argument_types) + : AggregateFunctionMinMaxByBaseData(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(-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 AggregateFunctionMinByData() {} - AggregateFunctionMinByData(const DataTypes argument_types, int be_version) - : AggregateFunctionMinMaxByBaseData(argument_types, be_version) {} + explicit AggregateFunctionMinByData(const DataTypes& argument_types) + : AggregateFunctionMinMaxByBaseData(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(-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>( {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 @@ class AggregateFunctionsMinMaxBy final } 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; constexpr auto PT = DispatchType::PType; - if constexpr (is_decimal(PT)) { - result = creator_without_type::create_multi_arguments< - AggregateFunctionsMinMaxBy>>>( - argument_types, result_is_nullable, attr); - } else { - result = creator_without_type::create_multi_arguments< - AggregateFunctionsMinMaxBy>>>( - argument_types, result_is_nullable, attr); - } + result = creator_without_type::create_multi_arguments< + AggregateFunctionsMinMaxBy>>>( + 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>>( - argument_types, result_is_nullable, attr); + AggregateFunctionsMinMaxBy>>(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 9717cc0461c7c1..c5e217a1a0b838 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>>>( + AggregateFunctionsSingleValue>>>( argument_types, result_is_nullable, attr); case PrimitiveType::TYPE_DECIMAL64: return creator_without_type::create_unary_arguments< - AggregateFunctionsSingleValue>>>( + AggregateFunctionsSingleValue>>>( argument_types, result_is_nullable, attr); case PrimitiveType::TYPE_DECIMALV2: return creator_without_type::create_unary_arguments< - AggregateFunctionsSingleValue>>>( + AggregateFunctionsSingleValue>>>( argument_types, result_is_nullable, attr); case PrimitiveType::TYPE_DECIMAL128I: return creator_without_type::create_unary_arguments< - AggregateFunctionsSingleValue>>>( + AggregateFunctionsSingleValue>>>( argument_types, result_is_nullable, attr); case PrimitiveType::TYPE_DECIMAL256: return creator_without_type::create_unary_arguments< - AggregateFunctionsSingleValue>>>( + AggregateFunctionsSingleValue>>>( 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>>( + AggregateFunctionsSingleValue>>( argument_types, result_is_nullable, attr); default: return nullptr; 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 ff693eec98e598..b2ba4225ff0f32 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 - #include "core/column/column_nullable.h" +#include "exprs/aggregate/single_value_data.h" namespace doris { -template -struct Value { -public: - bool is_null() const { - if (_ptr == nullptr) { - return true; - } - if constexpr (arg_is_nullable) { - return assert_cast(_ptr) - ->is_null_at(_offset); - } - return false; - } - - template - void insert_into(IColumn& to) const { - if constexpr (arg_is_nullable) { - const auto* col = assert_cast(_ptr); - assert_cast(to).insert_from( - col->get_nested_column(), _offset); - } else { - assert_cast(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(_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 -struct CopiedValue : public Value { -public: - template - void insert_into(IColumn& to) const { - assert_cast(to).insert(_copied_value); - } - - bool is_null() const { return this->_ptr == nullptr; } - - template - 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(column); - if (col->is_null_at(row)) { - this->reset(); - return; - } else { - auto& nested_col = assert_cast( - col->get_nested_column()); - nested_col.get(row, _copied_value); - } - } else { - column->get(row, _copied_value); - } - } - -private: - Field _copied_value; -}; - -template +template struct ReaderFirstAndLastData { public: - using StoreType = - std::conditional_t, Value>; static constexpr bool nullable = arg_is_nullable; static constexpr bool result_nullable = result_is_nullable; @@ -125,44 +35,40 @@ struct ReaderFirstAndLastData { 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(to); col.insert_default(); } else { auto& col = assert_cast(to); col.get_null_map_data().push_back(0); - if constexpr (!std::is_same_v) { - _data_value.template insert_into(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) { - _data_value.template insert_into(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(columns[0], pos); + if constexpr (arg_is_nullable) { + const auto& nullable_column = + assert_cast(*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 56eb337f97c4a1..1a491ef96ca9ee 100644 --- a/be/src/exprs/aggregate/aggregate_function_window.h +++ b/be/src/exprs/aggregate/aggregate_function_window.h @@ -397,8 +397,7 @@ class WindowFunctionNTile final }; template -struct FirstLastData - : public ReaderFirstAndLastData { +struct FirstLastData : public ReaderFirstAndLastData { public: void set_is_null() { this->_data_value.reset(); } }; @@ -417,19 +416,11 @@ struct NthValueData : public FirstLastData int64_t _frame_total_rows = 0; }; -template -struct BaseValue : public Value { -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 struct LeadLagData { public: static constexpr bool result_nullable = result_is_nullable; + void reset() { _data_value.reset(); _is_inited = false; @@ -438,17 +429,16 @@ struct LeadLagData { 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(to); col.insert_default(); } else { auto& col = assert_cast(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 @@ struct LeadLagData { if constexpr (arg_is_nullable) { if (assert_cast(columns[0]) ->is_null_at(pos)) { - // ptr == nullptr means nullable _data_value.reset(); return; } + const auto& nullable = + assert_cast(*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 @@ struct LeadLagData { 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 @@ struct LeadLagData { int64_t get_offset_value() const { return _offset_value; } private: - BaseValue _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 00000000000000..ff947577d5d7ba --- /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 + +#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 +struct SingleValueDataFixed { +private: + using Self = SingleValueDataFixed; + using ValueType = typename PrimitiveTypeTraits::CppType; + + static const ValueType& _value_at(const IColumn& column, size_t row_num) { + return assert_cast::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::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(); } + + void set_value_to_max() { value = Compare::max_value(); } + + void insert_result_into(IColumn& to) const { + if (has()) { + assert_cast::ColumnType&, TypeCheckOnRelease::DISABLE>( + to) + .insert_value(value); + } else { + assert_cast::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 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(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(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(capacity); + } + + size = value_size; + memcpy(large_data.get(), source.data, size); + } + } + +public: + void insert_result_into(IColumn& to) const { + if (has()) { + assert_cast(to).insert_data(_data(), size); + } else { + assert_cast(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(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 bb00051cbeeb5e..af57cb84c861c9 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 @@ class VAggReplaceTest : public testing::Test { agg_function->destroy(place); } - template - void test_basic_data(int8_t input_nums) { - DataTypePtr data_type = get_data_type(); - - auto data_column = data_type->create_column(); - add_elements(data_column, input_nums); - - EXPECT_EQ(input_nums, data_column->size()); - //test Value - { - Value 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(*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(to_column.get(), i); - } - } - //test CopiedValue - { - CopiedValue value; - EXPECT_TRUE(value.is_null()); - for (int64_t i = 0; i < input_nums; ++i) { - value.template 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(*to_column); - value.template insert_into(nullable_col.get_nested_column()); - } else { - value.template insert_into(*to_column); - } - EXPECT_EQ(1, to_column->size()); - check_column_basic(to_column.get(), i); - } - } - } - - template - void test_array_data(int8_t input_nums) { - DataTypePtr data_type = get_data_type(); - - auto data_column = data_type->create_column(); - array_add_elements(data_column, input_nums); - - EXPECT_EQ(input_nums + 1, data_column->size()); - //test Value - { - Value 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(*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(to_column.get(), i); - } - } - //test CopiedValue - { - CopiedValue value; - EXPECT_TRUE(value.is_null()); - for (int64_t i = 0; i <= input_nums; ++i) { - value.template 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(*to_column); - value.template insert_into(nullable_col.get_nested_column()); - } else { - value.template insert_into(*to_column); - } - EXPECT_EQ(1, to_column->size()); - check_column_array(to_column.get(), i); - } - } - } - private: Arena _agg_arena_pool; }; -TEST_F(VAggReplaceTest, test_basic_data) { - test_basic_data(11); - test_basic_data(11); - test_basic_data(11); - test_basic_data(11); - test_basic_data(11); - test_basic_data(11); - test_basic_data(11); - test_basic_data(11); - test_basic_data(11); - test_basic_data(11); -} - -TEST_F(VAggReplaceTest, test_array_data) { - test_array_data(11); - test_array_data(11); - test_array_data(11); - test_array_data(11); - test_array_data(11); - test_array_data(11); - test_array_data(11); - test_array_data(11); - test_array_data(11); - test_array_data(11); -} - TEST_F(VAggReplaceTest, test_basic_replace_reader) { test_agg_replace("replace_reader", 10, 0); test_agg_replace("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 61ac798ba56662..dce7a0729e71fb 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()}; + if (function_name == "nth_value") { + argument_types.push_back(std::make_shared()); + } else if (function_name == "lead" || function_name == "lag") { + argument_types.push_back(std::make_shared()); + argument_types.push_back(std::make_shared()); + } + 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(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 57f8941ced4ac2..411de8f5500f7d 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 +#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(); + 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(); + 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(); + 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(std::make_shared()); + SingleValueDataColumn data; + EXPECT_FALSE(data.has()); + EXPECT_EQ(data.allocated_bytes(), 0); + + auto source = array_type->create_column(); + source->insert(Field::create_field( + Array {Field::create_field(1), Field::create_field(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( + Array {Field::create_field(1), Field::create_field(2)})); +} + +TEST(SingleValueDataColumnTest, AllocationDoesNotScaleWithSourceRows) { + constexpr size_t large_source_rows = 4096; + auto array_type = std::make_shared(std::make_shared()); + auto create_source = [&](size_t rows) { + auto source = array_type->create_column(); + const auto value = Field::create_field( + Array {Field::create_field(1), Field::create_field(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(std::make_shared()); + const int be_exec_version = BeExecVersionManager::get_newest_version(); + auto source = array_type->create_column(); + source->insert(Field::create_field(Array {Field::create_field(2)})); + source->insert(Field::create_field(Array {Field::create_field(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(Array {Field::create_field(1)})); +} + } // namespace doris