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 {
#include "common/compile_check_begin.h"
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

#include "common/compile_check_end.h"

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		#include "common/compile_check_begin.h"
35		class Arena;
36		class BufferReadable;
37		class BufferWritable;
38		class IColumn;
39		} // namespace doris
40
41		namespace doris {
42
43		struct AggregateFunctionGroupConcatData {
44		ColumnString::Chars data;
45		std::string separator;
46		bool inited = false;
47
48	0	void add(StringRef ref, StringRef sep) {
49	0	auto delta_size = ref.size;
50	0	if (!inited) {
51	0	separator.assign(sep.data, sep.data + sep.size);
52	0	} else {
53	0	delta_size += separator.size();
54	0	}
55	0	auto offset = data.size();
56	0	data.resize(data.size() + delta_size);
57
58	0	if (!inited) {
59	0	inited = true;
60	0	} else {
61	0	memcpy(data.data() + offset, separator.data(), separator.size());
62	0	offset += separator.size();
63	0	}
64	0	memcpy(data.data() + offset, ref.data, ref.size);
65	0	}
66
67	0	void merge(const AggregateFunctionGroupConcatData& rhs) {
68	0	if (!rhs.inited) {
69	0	return;
70	0	}
71
72	0	if (!inited) {
73	0	inited = true;
74	0	separator = rhs.separator;
75	0	data.assign(rhs.data);
76	0	} else {
77	0	auto offset = data.size();
78
79	0	auto delta_size = separator.size() + rhs.data.size();
80	0	data.resize(data.size() + delta_size);
81
82	0	memcpy(data.data() + offset, separator.data(), separator.size());
83	0	offset += separator.size();
84	0	memcpy(data.data() + offset, rhs.data.data(), rhs.data.size());
85	0	}
86	0	}
87
88	0	StringRef get() const { return StringRef {data.data(), data.size()}; }
89
90	0	void write(BufferWritable& buf) const {
91	0	buf.write_binary(data);
92	0	buf.write_binary(separator);
93	0	buf.write_binary(inited);
94	0	}
95
96	0	void read(BufferReadable& buf) {
97	0	buf.read_binary(data);
98	0	buf.read_binary(separator);
99	0	buf.read_binary(inited);
100	0	}
101
102	0	void reset() {
103	0	data.clear();
104	0	separator = "";
105	0	inited = false;
106	0	}
107		};
108
109		struct AggregateFunctionGroupConcatImplStr {
110		static const std::string separator;
111		static void add(AggregateFunctionGroupConcatData& __restrict place, const IColumn** columns,
112	0	size_t row_num) {
113	0	place.add(assert_cast<const ColumnString&, TypeCheckOnRelease::DISABLE>(*columns[0])
114	0	.get_data_at(row_num),
115	0	StringRef(separator.data(), separator.length()));
116	0	}
117		};
118
119		struct AggregateFunctionGroupConcatImplStrStr {
120		static void add(AggregateFunctionGroupConcatData& __restrict place, const IColumn** columns,
121	0	size_t row_num) {
122	0	place.add(assert_cast<const ColumnString&, TypeCheckOnRelease::DISABLE>(*columns[0])
123	0	.get_data_at(row_num),
124	0	assert_cast<const ColumnString&, TypeCheckOnRelease::DISABLE>(*columns[1])
125	0	.get_data_at(row_num));
126	0	}
127		};
128
129		template <typename Impl>
130		class AggregateFunctionGroupConcat final
131		: public IAggregateFunctionDataHelper<AggregateFunctionGroupConcatData,
132		AggregateFunctionGroupConcat<Impl>>,
133		VarargsExpression,
134		NullableAggregateFunction {
135		public:
136		AggregateFunctionGroupConcat(const DataTypes& argument_types_)
137	0	: IAggregateFunctionDataHelper<AggregateFunctionGroupConcatData,
138	0	AggregateFunctionGroupConcat<Impl>>(argument_types_) {} Unexecuted instantiation: _ZN5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEEC2ERKSt6vectorISt10shared_ptrIKNS_9IDataTypeEESaIS7_EE Unexecuted instantiation: _ZN5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEEC2ERKSt6vectorISt10shared_ptrIKNS_9IDataTypeEESaIS7_EE
139
140	0	String get_name() const override { return "group_concat"; } Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE8get_nameB5cxx11Ev Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE8get_nameB5cxx11Ev
141
142	0	DataTypePtr get_return_type() const override { return std::make_shared<DataTypeString>(); } Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE15get_return_typeEv Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE15get_return_typeEv
143
144		void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
145	0	Arena&) const override {
146	0	Impl::add(this->data(place), columns, row_num);
147	0	} Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE3addEPcPPKNS_7IColumnElRNS_5ArenaE Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE3addEPcPPKNS_7IColumnElRNS_5ArenaE
148
149	0	void reset(AggregateDataPtr place) const override { this->data(place).reset(); } Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE5resetEPc Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE5resetEPc
150
151		void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
152	0	Arena&) const override {
153	0	this->data(place).merge(this->data(rhs));
154	0	} Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE5mergeEPcPKcRNS_5ArenaE Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE5mergeEPcPKcRNS_5ArenaE
155
156	0	void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
157	0	this->data(place).write(buf);
158	0	} Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE9serializeEPKcRNS_14BufferWritableE Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE9serializeEPKcRNS_14BufferWritableE
159
160		void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
161	0	Arena&) const override {
162	0	this->data(place).read(buf);
163	0	} Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE11deserializeEPcRNS_14BufferReadableERNS_5ArenaE Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE11deserializeEPcRNS_14BufferReadableERNS_5ArenaE
164
165	0	void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
166	0	const auto result = this->data(place).get();
167	0	assert_cast<ColumnString&>(to).insert_data(result.data, result.size);
168	0	} Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_35AggregateFunctionGroupConcatImplStrEE18insert_result_intoEPKcRNS_7IColumnE Unexecuted instantiation: _ZNK5doris28AggregateFunctionGroupConcatINS_38AggregateFunctionGroupConcatImplStrStrEE18insert_result_intoEPKcRNS_7IColumnE
169		};
170
171		} // namespace doris
172
173		#include "common/compile_check_end.h"

Coverage Report

Created: 2026-03-17 00:16