be/src/exprs/aggregate/aggregate_function_group_concat.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 <string.h>

#include <memory>
#include <string>

#include "core/assert_cast.h"
#include "core/column/column_string.h"
#include "core/data_type/data_type_string.h"
#include "core/string_ref.h"
#include "core/types.h"
#include "exprs/aggregate/aggregate_function.h"
#include "exprs/aggregate/aggregate_function_simple_factory.h"

namespace doris {
class Arena;
class BufferReadable;
class BufferWritable;
class IColumn;
} // namespace doris

namespace doris {

struct AggregateFunctionGroupConcatData {
    ColumnString::Chars data;
    std::string separator;
    bool inited = false;

    void add(StringRef ref, StringRef sep) {
        auto delta_size = ref.size;
        if (!inited) {
            separator.assign(sep.data, sep.data + sep.size);
        } else {
            delta_size += separator.size();
        }
        auto offset = data.size();
        data.resize(data.size() + delta_size);

        if (!inited) {
            inited = true;
        } else {
            memcpy(data.data() + offset, separator.data(), separator.size());
            offset += separator.size();
        }
        memcpy(data.data() + offset, ref.data, ref.size);
    }

    void merge(const AggregateFunctionGroupConcatData& rhs) {
        if (!rhs.inited) {
            return;
        }

        if (!inited) {
            inited = true;
            separator = rhs.separator;
            data.assign(rhs.data);
        } else {
            auto offset = data.size();

            auto delta_size = separator.size() + rhs.data.size();
            data.resize(data.size() + delta_size);

            memcpy(data.data() + offset, separator.data(), separator.size());
            offset += separator.size();
            memcpy(data.data() + offset, rhs.data.data(), rhs.data.size());
        }
    }

    StringRef get() const { return StringRef {data.data(), data.size()}; }

    void write(BufferWritable& buf) const {
        buf.write_binary(data);
        buf.write_binary(separator);
        buf.write_binary(inited);
    }

    void read(BufferReadable& buf) {
        buf.read_binary(data);
        buf.read_binary(separator);
        buf.read_binary(inited);
    }

    void reset() {
        data.clear();
        separator = "";
        inited = false;
    }
};

struct AggregateFunctionGroupConcatImplStr {
    static const std::string separator;
    static void add(AggregateFunctionGroupConcatData& __restrict place, const IColumn** columns,
                    size_t row_num) {
        place.add(assert_cast<const ColumnString&, TypeCheckOnRelease::DISABLE>(*columns[0])
                          .get_data_at(row_num),
                  StringRef(separator.data(), separator.length()));
    }
};

struct AggregateFunctionGroupConcatImplStrStr {
    static void add(AggregateFunctionGroupConcatData& __restrict place, const IColumn** columns,
                    size_t row_num) {
        place.add(assert_cast<const ColumnString&, TypeCheckOnRelease::DISABLE>(*columns[0])
                          .get_data_at(row_num),
                  assert_cast<const ColumnString&, TypeCheckOnRelease::DISABLE>(*columns[1])
                          .get_data_at(row_num));
    }
};

template <typename Impl>
class AggregateFunctionGroupConcat final
        : public IAggregateFunctionDataHelper<AggregateFunctionGroupConcatData,
                                              AggregateFunctionGroupConcat<Impl>>,
          VarargsExpression,
          NullableAggregateFunction {
public:
    AggregateFunctionGroupConcat(const DataTypes& argument_types_)
            : IAggregateFunctionDataHelper<AggregateFunctionGroupConcatData,
                                           AggregateFunctionGroupConcat<Impl>>(argument_types_) {}

    String get_name() const override { return "group_concat"; }

    DataTypePtr get_return_type() const override { return std::make_shared<DataTypeString>(); }

    void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
             Arena&) const override {
        Impl::add(this->data(place), columns, row_num);
    }

    void reset(AggregateDataPtr place) const override { this->data(place).reset(); }

    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
               Arena&) const override {
        this->data(place).merge(this->data(rhs));
    }

    void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
        this->data(place).write(buf);
    }

    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
                     Arena&) const override {
        this->data(place).read(buf);
    }

    void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
        const auto result = this->data(place).get();
        assert_cast<ColumnString&>(to).insert_data(result.data, result.size);
    }
};

} // namespace doris

Line	Count	Source
1		// Licensed to the Apache Software Foundation (ASF) under one
2		// or more contributor license agreements. See the NOTICE file
3		// distributed with this work for additional information
4		// regarding copyright ownership. The ASF licenses this file
5		// to you under the Apache License, Version 2.0 (the
6		// "License"); you may not use this file except in compliance
7		// with the License. You may obtain a copy of the License at
8		//
9		// http://www.apache.org/licenses/LICENSE-2.0
10		//
11		// Unless required by applicable law or agreed to in writing,
12		// software distributed under the License is distributed on an
13		// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14		// KIND, either express or implied. See the License for the
15		// specific language governing permissions and limitations
16		// under the License.
17
18		#pragma once
19
20		#include <string.h>
21
22		#include <memory>
23		#include <string>
24
25		#include "core/assert_cast.h"
26		#include "core/column/column_string.h"
27		#include "core/data_type/data_type_string.h"
28		#include "core/string_ref.h"
29		#include "core/types.h"
30		#include "exprs/aggregate/aggregate_function.h"
31		#include "exprs/aggregate/aggregate_function_simple_factory.h"
32
33		namespace doris {
34		class Arena;
35		class BufferReadable;
36		class BufferWritable;
37		class IColumn;
38		} // namespace doris
39
40		namespace doris {
41
42		struct AggregateFunctionGroupConcatData {
43		ColumnString::Chars data;
44		std::string separator;
45		bool inited = false;
46
47	0	void add(StringRef ref, StringRef sep) {
48	0	auto delta_size = ref.size;
49	0	if (!inited) {
50	0	separator.assign(sep.data, sep.data + sep.size);
51	0	} else {
52	0	delta_size += separator.size();
53	0	}
54	0	auto offset = data.size();
55	0	data.resize(data.size() + delta_size);
56
57	0	if (!inited) {
58	0	inited = true;
59	0	} else {
60	0	memcpy(data.data() + offset, separator.data(), separator.size());
61	0	offset += separator.size();
62	0	}
63	0	memcpy(data.data() + offset, ref.data, ref.size);
64	0	}
65
66	0	void merge(const AggregateFunctionGroupConcatData& rhs) {
67	0	if (!rhs.inited) {
68	0	return;
69	0	}
70
71	0	if (!inited) {
72	0	inited = true;
73	0	separator = rhs.separator;
74	0	data.assign(rhs.data);
75	0	} else {
76	0	auto offset = data.size();
77
78	0	auto delta_size = separator.size() + rhs.data.size();
79	0	data.resize(data.size() + delta_size);
80
81	0	memcpy(data.data() + offset, separator.data(), separator.size());
82	0	offset += separator.size();
83	0	memcpy(data.data() + offset, rhs.data.data(), rhs.data.size());
84	0	}
85	0	}
86
87	0	StringRef get() const { return StringRef {data.data(), data.size()}; }
88
89	0	void write(BufferWritable& buf) const {
90	0	buf.write_binary(data);
91	0	buf.write_binary(separator);
92	0	buf.write_binary(inited);
93	0	}
94
95	0	void read(BufferReadable& buf) {
96	0	buf.read_binary(data);
97	0	buf.read_binary(separator);
98	0	buf.read_binary(inited);
99	0	}
100
101	0	void reset() {
102	0	data.clear();
103	0	separator = "";
104	0	inited = false;
105	0	}
106		};
107
108		struct AggregateFunctionGroupConcatImplStr {
109		static const std::string separator;
110		static void add(AggregateFunctionGroupConcatData& __restrict place, const IColumn** columns,
111	0	size_t row_num) {
112	0	place.add(assert_cast<const ColumnString&, TypeCheckOnRelease::DISABLE>(*columns[0])
113	0	.get_data_at(row_num),
114	0	StringRef(separator.data(), separator.length()));
115	0	}
116		};
117
118		struct AggregateFunctionGroupConcatImplStrStr {
119		static void add(AggregateFunctionGroupConcatData& __restrict place, const IColumn** columns,
120	0	size_t row_num) {
121	0	place.add(assert_cast<const ColumnString&, TypeCheckOnRelease::DISABLE>(*columns[0])
122	0	.get_data_at(row_num),
123	0	assert_cast<const ColumnString&, TypeCheckOnRelease::DISABLE>(*columns[1])
124	0	.get_data_at(row_num));
125	0	}
126		};
127
128		template <typename Impl>
129		class AggregateFunctionGroupConcat final
130		: public IAggregateFunctionDataHelper<AggregateFunctionGroupConcatData,
131		AggregateFunctionGroupConcat<Impl>>,
132		VarargsExpression,
133		NullableAggregateFunction {
134		public:
135		AggregateFunctionGroupConcat(const DataTypes& argument_types_)
136	0	: IAggregateFunctionDataHelper<AggregateFunctionGroupConcatData,
137	0	AggregateFunctionGroupConcat<Impl>>(argument_types_) {} Unexecuted instantiation: _ZN5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEEC2ERKSt6vectorISt10shared_ptrIKNS_9IDataTypeEESaIS7_EE Unexecuted instantiation: _ZN5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEEC2ERKSt6vectorISt10shared_ptrIKNS_9IDataTypeEESaIS7_EE
138
139	0	String get_name() const override { return "group_concat"; } Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE8get_nameB5cxx11Ev Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE8get_nameB5cxx11Ev
140
141	0	DataTypePtr get_return_type() const override { return std::make_shared<DataTypeString>(); } Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE15get_return_typeEv Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE15get_return_typeEv
142
143		void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
144	0	Arena&) const override {
145	0	Impl::add(this->data(place), columns, row_num);
146	0	} Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE3addEPcPPKNS_7IColumnElRNS_5ArenaE Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE3addEPcPPKNS_7IColumnElRNS_5ArenaE
147
148	0	void reset(AggregateDataPtr place) const override { this->data(place).reset(); } Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE5resetEPc Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE5resetEPc
149
150		void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
151	0	Arena&) const override {
152	0	this->data(place).merge(this->data(rhs));
153	0	} Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE5mergeEPcPKcRNS_5ArenaE Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE5mergeEPcPKcRNS_5ArenaE
154
155	0	void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
156	0	this->data(place).write(buf);
157	0	} Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE9serializeEPKcRNS_14BufferWritableE Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE9serializeEPKcRNS_14BufferWritableE
158
159		void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
160	0	Arena&) const override {
161	0	this->data(place).read(buf);
162	0	} Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE11deserializeEPcRNS_14BufferReadableERNS_5ArenaE Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE11deserializeEPcRNS_14BufferReadableERNS_5ArenaE
163
164	0	void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
165	0	const auto result = this->data(place).get();
166	0	assert_cast<ColumnString&>(to).insert_data(result.data, result.size);
167	0	} Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE18insert_result_intoEPKcRNS_7IColumnE Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE18insert_result_intoEPKcRNS_7IColumnE
168		};
169
170		} // namespace doris

Coverage Report

Created: 2026-04-20 15:59