be/src/exprs/aggregate/aggregate_function_collect.h

Source
// 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 <cctz/time_zone.h>
#include <glog/logging.h>

#include <cstddef>
#include <cstring>
#include <limits>
#include <memory>
#include <new>
#include <string>
#include <type_traits>

#include "agent/be_exec_version_manager.h"
#include "common/exception.h"
#include "core/assert_cast.h"
#include "core/column/column.h"
#include "core/column/column_array.h"
#include "core/column/column_decimal.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/data_type/data_type.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_nullable.h"
#include "core/pod_array.h"
#include "core/string_buffer.hpp"
#include "core/string_ref.h"
#include "core/types.h"
#include "exprs/aggregate/aggregate_function.h"
#include "util/var_int.h"

namespace doris {
template <PrimitiveType T, bool HasLimit>
struct AggregateFunctionCollectSetData {
    static constexpr bool use_versioned_serde = false;
    static constexpr PrimitiveType PType = T;
    using ElementType = typename PrimitiveTypeTraits<T>::CppType;
    using ColVecType = typename PrimitiveTypeTraits<T>::ColumnType;
    using SelfType = AggregateFunctionCollectSetData;
    using Set = doris::flat_hash_set<ElementType>;
    Set data_set;
    Int64 max_size = -1;

    AggregateFunctionCollectSetData(const DataTypes& argument_types) {}

    size_t size() const { return data_set.size(); }

    void add(const IColumn& column, size_t row_num) {
        data_set.insert(assert_cast<const ColVecType&, TypeCheckOnRelease::DISABLE>(column)
                                .get_data()[row_num]);
    }

    void merge(const SelfType& rhs) {
        if constexpr (HasLimit) {
            if (max_size == -1) {
                max_size = rhs.max_size;
            }

            for (auto& rhs_elem : rhs.data_set) {
                if (size() >= max_size) {
                    return;
                }
                data_set.insert(rhs_elem);
            }
        } else {
            for (const auto& elem : rhs.data_set) {
                data_set.insert(elem);
            }
        }
    }

    void write(BufferWritable& buf) const {
        buf.write_var_uint(data_set.size());
        for (const auto& value : data_set) {
            buf.write_binary(value);
        }
        write_var_int(max_size, buf);
    }

    void read(BufferReadable& buf) {
        uint64_t new_size = 0;
        buf.read_var_uint(new_size);
        ElementType x;
        for (size_t i = 0; i < new_size; ++i) {
            buf.read_binary(x);
            data_set.insert(x);
        }
        read_var_int(max_size, buf);
    }

    void insert_result_into(IColumn& to) const {
        auto& vec = assert_cast<ColVecType&, TypeCheckOnRelease::DISABLE>(to).get_data();
        vec.reserve(size());
        for (const auto& item : data_set) {
            vec.push_back(item);
        }
    }

    void reset() {
        data_set.clear();
        max_size = -1;
    }
};

template <PrimitiveType T, bool HasLimit>
    requires(is_string_type(T))
struct AggregateFunctionCollectSetData<T, HasLimit> {
    static constexpr bool use_versioned_serde = false;
    static constexpr PrimitiveType PType = T;
    using ElementType = StringRef;
    using ColVecType = ColumnString;
    using SelfType = AggregateFunctionCollectSetData<T, HasLimit>;
    using Set = doris::flat_hash_set<ElementType>;
    Set data_set;
    Int64 max_size = -1;

    AggregateFunctionCollectSetData(const DataTypes& argument_types) {}

    size_t size() const { return data_set.size(); }

    void add(const IColumn& column, size_t row_num, Arena& arena) {
        auto key = column.get_data_at(row_num);
        key.data = arena.insert(key.data, key.size);
        data_set.insert(key);
    }

    void merge(const SelfType& rhs, Arena& arena) {
        if (max_size == -1) {
            max_size = rhs.max_size;
        }

        for (const auto& rhs_elem : rhs.data_set) {
            if constexpr (HasLimit) {
                if (size() >= max_size) {
                    return;
                }
            }
            StringRef key = rhs_elem;
            key.data = arena.insert(key.data, key.size);
            data_set.insert(key);
        }
    }

    void write(BufferWritable& buf) const {
        buf.write_var_uint(size());
        for (const auto& elem : data_set) {
            buf.write_binary(elem);
        }
        write_var_int(max_size, buf);
    }

    void read(BufferReadable& buf) {
        UInt64 size;
        buf.read_var_uint(size);
        StringRef ref;
        for (size_t i = 0; i < size; ++i) {
            buf.read_binary(ref);
            data_set.insert(ref);
        }
        read_var_int(max_size, buf);
    }

    void insert_result_into(IColumn& to) const {
        auto& vec = assert_cast<ColVecType&, TypeCheckOnRelease::DISABLE>(to);
        vec.reserve(size());
        for (const auto& item : data_set) {
            vec.insert_data(item.data, item.size);
        }
    }

    void reset() {
        data_set.clear();
        max_size = -1;
    }
};

template <PrimitiveType T, bool HasLimit>
struct AggregateFunctionCollectListData {
    static constexpr bool use_versioned_serde = false;
    static constexpr PrimitiveType PType = T;
    using ElementType = typename PrimitiveTypeTraits<T>::CppType;
    using ColVecType = typename PrimitiveTypeTraits<T>::ColumnType;
    using SelfType = AggregateFunctionCollectListData<T, HasLimit>;
    PaddedPODArray<ElementType> data;
    Int64 max_size = -1;

    AggregateFunctionCollectListData(const DataTypes& argument_types) {}

    size_t size() const { return data.size(); }

    void add(const IColumn& column, size_t row_num) {
        const auto& vec =
                assert_cast<const ColVecType&, TypeCheckOnRelease::DISABLE>(column).get_data();
        data.push_back(vec[row_num]);
    }

    void merge(const SelfType& rhs) {
        if constexpr (HasLimit) {
            if (max_size == -1) {
                max_size = rhs.max_size;
            }
            for (auto& rhs_elem : rhs.data) {
                if (size() >= max_size) {
                    return;
                }
                data.push_back(rhs_elem);
            }
        } else {
            data.insert(rhs.data.begin(), rhs.data.end());
        }
    }

    void write(BufferWritable& buf) const {
        buf.write_var_uint(size());
        buf.write(data.raw_data(), size() * sizeof(ElementType));
        write_var_int(max_size, buf);
    }

    void read(BufferReadable& buf) {
        UInt64 rows = 0;
        buf.read_var_uint(rows);
        data.resize(rows);
        buf.read(reinterpret_cast<char*>(data.data()), rows * sizeof(ElementType));
        read_var_int(max_size, buf);
    }

    void reset() {
        data.clear();
        max_size = -1;
    }

    void insert_result_into(IColumn& to) const {
        auto& vec = assert_cast<ColVecType&, TypeCheckOnRelease::DISABLE>(to).get_data();
        size_t old_size = vec.size();
        vec.resize(old_size + size());
        std::memcpy(vec.data() + old_size, data.data(), size() * sizeof(ElementType));
    }
};

template <PrimitiveType T, bool HasLimit>
    requires(is_string_type(T))
struct AggregateFunctionCollectListData<T, HasLimit> {
    static constexpr bool use_versioned_serde = false;
    static constexpr PrimitiveType PType = T;
    using ElementType = StringRef;
    using ColVecType = ColumnString;
    MutableColumnPtr data;
    Int64 max_size = -1;

    AggregateFunctionCollectListData(const DataTypes& argument_types) {
        data = ColVecType::create();
    }

    size_t size() const { return data->size(); }

    void add(const IColumn& column, size_t row_num) { data->insert_from(column, row_num); }

    void merge(const AggregateFunctionCollectListData& rhs) {
        if constexpr (HasLimit) {
            if (max_size == -1) {
                max_size = rhs.max_size;
            }

            data->insert_range_from(*rhs.data, 0,
                                    std::min(static_cast<size_t>(max_size - size()), rhs.size()));
        } else {
            data->insert_range_from(*rhs.data, 0, rhs.size());
        }
    }

    void write(BufferWritable& buf) const {
        auto& col = assert_cast<ColVecType&>(*data);

        buf.write_var_uint(col.size());
        buf.write(col.get_offsets().raw_data(), col.size() * sizeof(IColumn::Offset));

        buf.write_var_uint(col.get_chars().size());
        buf.write(col.get_chars().raw_data(), col.get_chars().size());
        write_var_int(max_size, buf);
    }

    void read(BufferReadable& buf) {
        auto& col = assert_cast<ColVecType&>(*data);
        UInt64 offs_size = 0;
        buf.read_var_uint(offs_size);
        col.get_offsets().resize(offs_size);
        buf.read(reinterpret_cast<char*>(col.get_offsets().data()),
                 offs_size * sizeof(IColumn::Offset));

        UInt64 chars_size = 0;
        buf.read_var_uint(chars_size);
        col.get_chars().resize(chars_size);
        buf.read(reinterpret_cast<char*>(col.get_chars().data()), chars_size);
        read_var_int(max_size, buf);
    }

    void reset() {
        data->clear();
        max_size = -1;
    }

    void insert_result_into(IColumn& to) const {
        auto& to_str = assert_cast<ColVecType&, TypeCheckOnRelease::DISABLE>(to);
        to_str.insert_range_from(*data, 0, size());
    }
};

template <PrimitiveType T, bool HasLimit>
    requires(!is_string_type(T) && !is_int_or_bool(T) && !is_float_or_double(T) && !is_decimal(T) &&
             !is_date_type(T) && !is_timestamp_ns_type(T) && !is_ip(T) && !is_timestamptz_type(T))
struct AggregateFunctionCollectListData<T, HasLimit> {
    static constexpr bool use_versioned_serde = true;
    static constexpr PrimitiveType PType = T;
    using ElementType = StringRef;
    using Self = AggregateFunctionCollectListData<T, HasLimit>;
    DataTypeSerDeSPtr serde; // for complex serialize && deserialize from multi BE
    MutableColumnPtr column_data;
    Int64 max_size = -1;

    AggregateFunctionCollectListData(const DataTypes& argument_types) {
        DataTypePtr column_type = argument_types[0];
        column_data = column_type->create_column();
        serde = column_type->get_serde();
    }

    size_t size() const { return column_data->size(); }

    void add(const IColumn& column, size_t row_num) { column_data->insert_from(column, row_num); }

    void merge(const AggregateFunctionCollectListData& rhs) {
        if constexpr (HasLimit) {
            if (max_size == -1) {
                max_size = rhs.max_size;
            }

            column_data->insert_range_from(
                    *rhs.column_data, 0,
                    std::min(static_cast<size_t>(max_size - size()), rhs.size()));
        } else {
            column_data->insert_range_from(*rhs.column_data, 0, rhs.size());
        }
    }

    void write(BufferWritable& buf, const IDataType& column_type, int query_version) const {
        // A mixed cluster still uses the old query version: old BEs can only read JSON states.
        if (query_version < SUPPORT_COLLECT_LIST_NATIVE_SERIALIZATION_VERSION) {
            write(buf);
            return;
        }

        // JSON can split the single array string a","b into two elements. Serialize the
        // column in place and frame only the actual native bytes, as nullable array_agg does.
        const auto capacity = cast_set<size_t>(
                column_type.get_uncompressed_serialized_bytes(*column_data, query_version));
        buf.resize(sizeof(UInt64) + capacity);
        auto* begin = buf.data() + sizeof(UInt64);
        const char* end = nullptr;
        try {
            end = column_type.serialize(*column_data, begin, query_version);
        } catch (...) {
            buf.resize(0);
            throw;
        }
        DORIS_CHECK_LE(end, begin + capacity);
        const auto serialized_bytes = static_cast<UInt64>(end - begin);
        memcpy(buf.data(), &serialized_bytes, sizeof(serialized_bytes));
        const auto frame_bytes = sizeof(serialized_bytes) + cast_set<size_t>(serialized_bytes);
        buf.resize(frame_bytes);
        buf.add_offset(frame_bytes);
        write_var_int(max_size, buf);
    }

    void read(BufferReadable& buf, const IDataType& column_type, int query_version) {
        // Match the sender's query version, including when this BE supports a newer version.
        if (query_version < SUPPORT_COLLECT_LIST_NATIVE_SERIALIZATION_VERSION) {
            read(buf);
            return;
        }

        UInt64 serialized_bytes = 0;
        buf.read_binary(serialized_bytes);
        const auto* begin = buf.data();
        column_data->clear();
        const auto* end = column_type.deserialize(begin, &column_data, query_version);
        DORIS_CHECK_EQ(end, begin + serialized_bytes);
        buf.add_offset(serialized_bytes);
        read_var_int(max_size, buf);
    }

    void write(BufferWritable& buf) const {
        const size_t size = column_data->size();
        buf.write_binary(size);

        DataTypeSerDe::FormatOptions opt;
        auto timezone = cctz::utc_time_zone();
        opt.timezone = &timezone;
        // Preserve this text format exactly for queries involving older BEs.
        auto tmp_str = ColumnString::create();
        VectorBufferWriter tmp_buf(*tmp_str.get());

        for (size_t i = 0; i < size; i++) {
            tmp_str->clear();
            if (Status st = serde->serialize_one_cell_to_json(*column_data, i, tmp_buf, opt); !st) {
                throw doris::Exception(ErrorCode::INTERNAL_ERROR,
                                       "Failed to serialize data for " + column_data->get_name() +
                                               " error: " + st.to_string());
            }
            tmp_buf.commit();
            buf.write_binary(tmp_str->get_data_at(0));
        }

        write_var_int(max_size, buf);
    }

    void read(BufferReadable& buf) {
        size_t size = 0;
        buf.read_binary(size);
        column_data->clear();
        column_data->reserve(size);

        StringRef s;
        DataTypeSerDe::FormatOptions opt;
        auto timezone = cctz::utc_time_zone();
        opt.timezone = &timezone;
        for (size_t i = 0; i < size; i++) {
            buf.read_binary(s);
            Slice slice(s.data, s.size);
            if (Status st = serde->deserialize_one_cell_from_json(*column_data, slice, opt); !st) {
                throw doris::Exception(ErrorCode::INTERNAL_ERROR,
                                       "Failed to deserialize data for " + column_data->get_name() +
                                               " error: " + st.to_string());
            }
        }
        read_var_int(max_size, buf);
    }

    void reset() {
        column_data->clear();
        max_size = -1;
    }

    void insert_result_into(IColumn& to) const { to.insert_range_from(*column_data, 0, size()); }
};

template <typename Data, bool HasLimit>
class AggregateFunctionCollect final
        : public IAggregateFunctionDataHelper<Data, AggregateFunctionCollect<Data, HasLimit>, true>,
          VarargsExpression,
          NotNullableAggregateFunction {
    static constexpr bool ENABLE_ARENA =
            std::is_same_v<Data, AggregateFunctionCollectSetData<TYPE_STRING, HasLimit>> ||
            std::is_same_v<Data, AggregateFunctionCollectSetData<TYPE_CHAR, HasLimit>> ||
            std::is_same_v<Data, AggregateFunctionCollectSetData<TYPE_VARCHAR, HasLimit>>;

public:
    AggregateFunctionCollect(const DataTypes& argument_types_)
            : IAggregateFunctionDataHelper<Data, AggregateFunctionCollect<Data, HasLimit>, true>(
                      {argument_types_}),
              return_type(std::make_shared<DataTypeArray>(make_nullable(argument_types_[0]))) {}

    std::string get_name() const override {
        if constexpr (std::is_same_v<AggregateFunctionCollectListData<Data::PType, HasLimit>,
                                     Data>) {
            return "collect_list";
        } else {
            return "collect_set";
        }
    }

    DataTypePtr get_return_type() const override { return return_type; }

    void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
             Arena& arena) const override {
        auto& data = this->data(place);
        if constexpr (HasLimit) {
            if (data.max_size == -1) {
                data.max_size =
                        assert_cast<const ColumnInt32*, TypeCheckOnRelease::DISABLE>(columns[1])
                                ->get_element(row_num);
            }
            if (data.size() >= data.max_size) {
                return;
            }
        }
        if constexpr (ENABLE_ARENA) {
            data.add(*columns[0], row_num, arena);
        } else {
            data.add(*columns[0], row_num);
        }
    }

    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
               Arena& arena) const override {
        auto& data = this->data(place);
        const auto& rhs_data = this->data(rhs);
        if constexpr (HasLimit) {
            if (rhs_data.size() == 0) {
                return;
            }
            if (data.size() == 0) {
                data.max_size = rhs_data.max_size;
            } else if (UNLIKELY(data.max_size != rhs_data.max_size)) {
                throw Exception(ErrorCode::INVALID_ARGUMENT,
                                "{} aggregate states have incompatible limits: {} vs {}",
                                get_name(), data.max_size, rhs_data.max_size);
            }
        }
        if constexpr (ENABLE_ARENA) {
            data.merge(rhs_data, arena);
        } else {
            data.merge(rhs_data);
        }
    }

    void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
        if constexpr (Data::use_versioned_serde) {
            this->data(place).write(buf, *this->argument_types[0], this->version);
        } else {
            this->data(place).write(buf);
        }
    }

    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
                     Arena&) const override {
        if constexpr (Data::use_versioned_serde) {
            this->data(place).read(buf, *this->argument_types[0], this->version);
        } else {
            this->data(place).read(buf);
        }
    }

    void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
        auto& to_arr = assert_cast<ColumnArray&, TypeCheckOnRelease::DISABLE>(to);
        auto& to_nested_col = to_arr.get_data();
        auto* col_null = assert_cast<ColumnNullable*, TypeCheckOnRelease::DISABLE>(&to_nested_col);
        this->data(place).insert_result_into(col_null->get_nested_column());
        col_null->get_null_map_data().resize_fill(col_null->get_nested_column().size(), 0);
        to_arr.get_offsets().push_back(to_nested_col.size());
    }

    void check_input_columns_type(const IColumn** columns) const override {
        IAggregateFunction::check_input_columns_type(columns);
        if constexpr (is_string_type(Data::PType) &&
                      std::is_same_v<AggregateFunctionCollectListData<Data::PType, HasLimit>,
                                     Data>) {
            this->template check_argument_column_type<ColumnString>(columns[0]);
        }
    }

    void check_result_column_type(const IColumn& to) const override {
        IAggregateFunction::check_result_column_type(to);
        if constexpr (is_string_type(Data::PType)) {
            const auto* array_column = check_and_get_column<ColumnArray>(to);
            if (UNLIKELY(array_column == nullptr)) {
                throw doris::Exception(Status::InternalError(
                        "Aggregate function {} result type check failed: Column type {} is not "
                        "ColumnArray",
                        get_name(), to.get_name()));
            }

            const auto& nested_column = array_column->get_data();
            const auto* nullable_column = check_and_get_column<ColumnNullable>(nested_column);
            if (UNLIKELY(nullable_column == nullptr)) {
                throw doris::Exception(Status::InternalError(
                        "Aggregate function {} result type check failed: Column type {} is not "
                        "ColumnNullable",
                        get_name(), nested_column.get_name()));
            }
            this->template check_result_column_type_as<ColumnString>(
                    nullable_column->get_nested_column());
        }
    }

private:
    DataTypePtr return_type;
};

} // namespace doris

Coverage Report

Created: 2026-11-04 19:32