Skip to content

Commit 5ddf57c

Browse files
authored
Merge branch 'main' into main
2 parents 08f8df1 + 1d6b843 commit 5ddf57c

9 files changed

Lines changed: 318 additions & 0 deletions

File tree

include/paimon/defs.h

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,12 @@ struct PAIMON_EXPORT Options {
7979
static const char DEFAULT_AGG_FUNCTION[];
8080
/// IGNORE_RETRACT is "ignore-retract"
8181
static const char IGNORE_RETRACT[];
82+
/// "distinct" - Distinct option for aggregate functions like listagg. Default value is false.
83+
/// Example: fields.f.distinct=true to deduplicate values during aggregation.
84+
static const char DISTINCT[];
85+
/// "list-agg-delimiter" - Delimiter for listagg aggregate function. Default value is ",".
86+
/// Example: fields.f.list-agg-delimiter="-" to concatenate values with "-".
87+
static const char LIST_AGG_DELIMITER[];
8288
/// SEQUENCE_GROUP is "sequence-group"
8389
static const char SEQUENCE_GROUP[];
8490
/// @}

src/paimon/CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -585,6 +585,7 @@ if(PAIMON_BUILD_TESTS)
585585
core/mergetree/compact/aggregate/field_ignore_retract_agg_test.cpp
586586
core/mergetree/compact/aggregate/field_last_non_null_value_agg_test.cpp
587587
core/mergetree/compact/aggregate/field_last_value_agg_test.cpp
588+
core/mergetree/compact/aggregate/field_listagg_agg_test.cpp
588589
core/mergetree/compact/aggregate/field_min_max_agg_test.cpp
589590
core/mergetree/compact/aggregate/field_primary_key_agg_test.cpp
590591
core/mergetree/compact/aggregate/field_sum_agg_test.cpp

src/paimon/common/defs.cpp

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@ const char Options::FIELDS_PREFIX[] = "fields";
2323
const char Options::AGG_FUNCTION[] = "aggregate-function";
2424
const char Options::DEFAULT_AGG_FUNCTION[] = "default-aggregate-function";
2525
const char Options::IGNORE_RETRACT[] = "ignore-retract";
26+
const char Options::DISTINCT[] = "distinct";
27+
const char Options::LIST_AGG_DELIMITER[] = "list-agg-delimiter";
2628
const char Options::SEQUENCE_GROUP[] = "sequence-group";
2729

2830
const char Options::BUCKET[] = "bucket";

src/paimon/core/core_options.cpp

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1114,6 +1114,24 @@ Result<bool> CoreOptions::FieldAggIgnoreRetract(const std::string& field_name) c
11141114
return field_agg_ignore_retract;
11151115
}
11161116

1117+
Result<std::string> CoreOptions::FieldListAggDelimiter(const std::string& field_name) const {
1118+
ConfigParser parser(impl_->raw_options);
1119+
std::string delimiter = ",";
1120+
std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
1121+
std::string(Options::LIST_AGG_DELIMITER);
1122+
PAIMON_RETURN_NOT_OK(parser.ParseString(key, &delimiter));
1123+
return delimiter;
1124+
}
1125+
1126+
Result<bool> CoreOptions::FieldCollectAggDistinct(const std::string& field_name) const {
1127+
ConfigParser parser(impl_->raw_options);
1128+
bool distinct = false;
1129+
std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
1130+
std::string(Options::DISTINCT);
1131+
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(key, &distinct));
1132+
return distinct;
1133+
}
1134+
11171135
bool CoreOptions::DeletionVectorsEnabled() const {
11181136
return impl_->deletion_vectors_enabled;
11191137
}

src/paimon/core/core_options.h

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,8 @@ class PAIMON_EXPORT CoreOptions {
109109
std::optional<std::string> GetFieldsDefaultFunc() const;
110110
Result<std::optional<std::string>> GetFieldAggFunc(const std::string& field_name) const;
111111
Result<bool> FieldAggIgnoreRetract(const std::string& field_name) const;
112+
Result<std::string> FieldListAggDelimiter(const std::string& field_name) const;
113+
Result<bool> FieldCollectAggDistinct(const std::string& field_name) const;
112114
bool DeletionVectorsEnabled() const;
113115
bool DeletionVectorsBitmap64() const;
114116
int64_t DeletionVectorTargetFileSize() const;

src/paimon/core/core_options_test.cpp

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,8 @@ TEST(CoreOptionsTest, TestDefaultValue) {
8383
ASSERT_EQ(std::nullopt, core_options.GetFieldsDefaultFunc());
8484
ASSERT_EQ(std::nullopt, core_options.GetFieldAggFunc("f0").value());
8585
ASSERT_FALSE(core_options.FieldAggIgnoreRetract("f1").value());
86+
ASSERT_EQ(",", core_options.FieldListAggDelimiter("f1").value());
87+
ASSERT_FALSE(core_options.FieldCollectAggDistinct("f1").value());
8688
ASSERT_FALSE(core_options.DeletionVectorsEnabled());
8789
ASSERT_FALSE(core_options.DeletionVectorsBitmap64());
8890
ASSERT_EQ(2 * 1024 * 1024, core_options.DeletionVectorTargetFileSize());
@@ -171,6 +173,8 @@ TEST(CoreOptionsTest, TestFromMap) {
171173
{Options::FIELDS_DEFAULT_AGG_FUNC, "sum"},
172174
{"fields.f0.aggregate-function", "min"},
173175
{"fields.f1.ignore-retract", "true"},
176+
{"fields.f2.list-agg-delimiter", " | "},
177+
{"fields.f2.distinct", "true"},
174178
{Options::DELETION_VECTORS_ENABLED, "true"},
175179
{Options::DELETION_VECTOR_BITMAP64, "true"},
176180
{Options::DELETION_VECTOR_INDEX_FILE_TARGET_SIZE, "4MB"},
@@ -272,6 +276,8 @@ TEST(CoreOptionsTest, TestFromMap) {
272276
ASSERT_EQ("min", core_options.GetFieldAggFunc("f0").value().value());
273277
ASSERT_TRUE(core_options.FieldAggIgnoreRetract("f1").value());
274278
ASSERT_TRUE(core_options.FieldAggIgnoreRetract("f1").value());
279+
ASSERT_EQ(" | ", core_options.FieldListAggDelimiter("f2").value());
280+
ASSERT_TRUE(core_options.FieldCollectAggDistinct("f2").value());
275281
ASSERT_TRUE(core_options.DeletionVectorsEnabled());
276282
ASSERT_TRUE(core_options.DeletionVectorsBitmap64());
277283
ASSERT_EQ(4 * 1024 * 1024, core_options.DeletionVectorTargetFileSize());

src/paimon/core/mergetree/compact/aggregate/field_aggregator_factory.h

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
#include "paimon/core/mergetree/compact/aggregate/field_ignore_retract_agg.h"
3030
#include "paimon/core/mergetree/compact/aggregate/field_last_non_null_value_agg.h"
3131
#include "paimon/core/mergetree/compact/aggregate/field_last_value_agg.h"
32+
#include "paimon/core/mergetree/compact/aggregate/field_listagg_agg.h"
3233
#include "paimon/core/mergetree/compact/aggregate/field_max_agg.h"
3334
#include "paimon/core/mergetree/compact/aggregate/field_min_agg.h"
3435
#include "paimon/core/mergetree/compact/aggregate/field_primary_key_agg.h"
@@ -71,6 +72,9 @@ class FieldAggregatorFactory {
7172
PAIMON_ASSIGN_OR_RAISE(field_aggregator, FieldBoolOrAgg::Create(field_type));
7273
} else if (str_agg == FieldBoolAndAgg::NAME) {
7374
PAIMON_ASSIGN_OR_RAISE(field_aggregator, FieldBoolAndAgg::Create(field_type));
75+
} else if (str_agg == FieldListaggAgg::NAME) {
76+
PAIMON_ASSIGN_OR_RAISE(field_aggregator,
77+
FieldListaggAgg::Create(field_type, options, field_name));
7478
} else {
7579
return Status::Invalid(fmt::format(
7680
"Use unsupported aggregation {} or spell aggregate function incorrectly!",
Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#pragma once
18+
19+
#include <memory>
20+
#include <string>
21+
#include <unordered_set>
22+
23+
#include "paimon/common/data/data_define.h"
24+
#include "paimon/core/core_options.h"
25+
#include "paimon/core/mergetree/compact/aggregate/field_aggregator.h"
26+
27+
namespace arrow {
28+
class DataType;
29+
} // namespace arrow
30+
31+
namespace paimon {
32+
/// listagg aggregate a field of a row.
33+
/// Concatenates string values with a delimiter.
34+
class FieldListaggAgg : public FieldAggregator {
35+
public:
36+
static constexpr char NAME[] = "listagg";
37+
38+
static Result<std::unique_ptr<FieldListaggAgg>> Create(
39+
const std::shared_ptr<arrow::DataType>& field_type, const CoreOptions& options,
40+
const std::string& field_name) {
41+
if (field_type->id() != arrow::Type::type::STRING) {
42+
return Status::Invalid(
43+
fmt::format("invalid field type {} for field '{}' of {}, supposed to be string",
44+
field_type->ToString(), field_name, NAME));
45+
}
46+
PAIMON_ASSIGN_OR_RAISE(std::string delimiter, options.FieldListAggDelimiter(field_name));
47+
PAIMON_ASSIGN_OR_RAISE(bool distinct, options.FieldCollectAggDistinct(field_name));
48+
// When delimiter is empty and distinct is true, fall back to whitespace split.
49+
if (distinct && delimiter.empty()) {
50+
delimiter = " ";
51+
}
52+
return std::unique_ptr<FieldListaggAgg>(
53+
new FieldListaggAgg(field_type, std::move(delimiter), distinct));
54+
}
55+
56+
VariantType Agg(const VariantType& accumulator, const VariantType& input_field) override {
57+
bool accumulator_null = DataDefine::IsVariantNull(accumulator);
58+
bool input_null = DataDefine::IsVariantNull(input_field);
59+
if (accumulator_null || input_null) {
60+
return accumulator_null ? input_field : accumulator;
61+
}
62+
std::string_view acc_str = DataDefine::GetStringView(accumulator);
63+
std::string_view in_str = DataDefine::GetStringView(input_field);
64+
if (in_str.empty()) {
65+
return accumulator;
66+
}
67+
if (acc_str.empty()) {
68+
return input_field;
69+
}
70+
71+
if (distinct_) {
72+
result_ = AggDistinctImpl(acc_str, in_str);
73+
} else {
74+
// Build into a local string to avoid aliasing when acc_str points into result_
75+
std::string new_result;
76+
new_result.reserve(acc_str.size() + delimiter_.size() + in_str.size());
77+
new_result.append(acc_str);
78+
new_result.append(delimiter_);
79+
new_result.append(in_str);
80+
result_ = std::move(new_result);
81+
}
82+
return std::string_view(result_);
83+
}
84+
85+
private:
86+
std::string AggDistinctImpl(std::string_view acc_str, std::string_view in_str) const {
87+
// Split accumulator tokens into a set for dedup
88+
std::unordered_set<std::string_view> seen;
89+
std::string_view remaining = acc_str;
90+
while (true) {
91+
size_t pos = remaining.find(delimiter_);
92+
std::string_view token =
93+
(pos == std::string_view::npos) ? remaining : remaining.substr(0, pos);
94+
if (!token.empty()) {
95+
seen.insert(token);
96+
}
97+
if (pos == std::string_view::npos) {
98+
break;
99+
}
100+
remaining = remaining.substr(pos + delimiter_.size());
101+
}
102+
103+
// Start with the full accumulator, then append delimiter + new distinct tokens from input
104+
std::string result;
105+
result.reserve(acc_str.size() + in_str.size());
106+
result.append(acc_str);
107+
remaining = in_str;
108+
while (true) {
109+
size_t pos = remaining.find(delimiter_);
110+
std::string_view token =
111+
(pos == std::string_view::npos) ? remaining : remaining.substr(0, pos);
112+
if (!token.empty() && seen.insert(token).second) {
113+
result.append(delimiter_);
114+
result.append(token);
115+
}
116+
if (pos == std::string_view::npos) {
117+
break;
118+
}
119+
remaining = remaining.substr(pos + delimiter_.size());
120+
}
121+
return result;
122+
}
123+
124+
explicit FieldListaggAgg(const std::shared_ptr<arrow::DataType>& field_type,
125+
std::string delimiter, bool distinct)
126+
: FieldAggregator(std::string(NAME), field_type),
127+
delimiter_(std::move(delimiter)),
128+
distinct_(distinct) {}
129+
130+
std::string delimiter_;
131+
bool distinct_;
132+
std::string result_;
133+
};
134+
} // namespace paimon
Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "paimon/core/mergetree/compact/aggregate/field_listagg_agg.h"
18+
19+
#include <map>
20+
#include <string>
21+
#include <string_view>
22+
23+
#include "arrow/type_fwd.h"
24+
#include "gtest/gtest.h"
25+
#include "paimon/core/core_options.h"
26+
#include "paimon/status.h"
27+
#include "paimon/testing/utils/testharness.h"
28+
29+
namespace paimon::test {
30+
31+
class FieldListaggAggTest : public testing::Test {
32+
protected:
33+
static Result<std::unique_ptr<FieldListaggAgg>> MakeAgg(const std::string& delimiter = ",",
34+
bool distinct = false) {
35+
std::map<std::string, std::string> opts;
36+
opts["fields.f.list-agg-delimiter"] = delimiter;
37+
opts["fields.f.distinct"] = distinct ? "true" : "false";
38+
PAIMON_ASSIGN_OR_RAISE(auto options, CoreOptions::FromMap(opts));
39+
return FieldListaggAgg::Create(arrow::utf8(), std::move(options), "f");
40+
}
41+
};
42+
43+
TEST_F(FieldListaggAggTest, TestSimple) {
44+
ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg());
45+
auto ret = agg->Agg(std::string_view("hello"), std::string_view(" world"));
46+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "hello, world");
47+
}
48+
49+
TEST_F(FieldListaggAggTest, TestDelimiter) {
50+
ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg("-"));
51+
auto ret = agg->Agg(std::string_view("user1"), std::string_view("user2"));
52+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "user1-user2");
53+
}
54+
55+
TEST_F(FieldListaggAggTest, TestNull) {
56+
ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg());
57+
58+
// input null -> return accumulator
59+
{
60+
auto ret = agg->Agg(std::string_view("hello"), NullType());
61+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "hello");
62+
}
63+
// accumulator null -> return input
64+
{
65+
auto ret = agg->Agg(NullType(), std::string_view("world"));
66+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "world");
67+
}
68+
// both null -> return null
69+
{
70+
auto ret = agg->Agg(NullType(), NullType());
71+
ASSERT_TRUE(DataDefine::IsVariantNull(ret));
72+
}
73+
}
74+
75+
TEST_F(FieldListaggAggTest, TestEmptyString) {
76+
ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg());
77+
78+
// empty input -> return accumulator
79+
{
80+
auto ret = agg->Agg(std::string_view("hello"), std::string_view(""));
81+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "hello");
82+
}
83+
// empty accumulator -> return input
84+
{
85+
auto ret = agg->Agg(std::string_view(""), std::string_view("world"));
86+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "world");
87+
}
88+
// both empty -> return input (which is empty)
89+
{
90+
auto ret = agg->Agg(std::string_view(""), std::string_view(""));
91+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "");
92+
}
93+
}
94+
95+
TEST_F(FieldListaggAggTest, TestMultipleAccumulation) {
96+
ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg());
97+
98+
// "a" + "," + "b" = "a,b", then "a,b" + "," + "c" = "a,b,c"
99+
auto ret = agg->Agg(std::string_view("a"), std::string_view("b"));
100+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "a,b");
101+
ret = agg->Agg(std::move(ret), std::string_view("c"));
102+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "a,b,c");
103+
}
104+
105+
TEST_F(FieldListaggAggTest, TestDistinct) {
106+
ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg(";", true));
107+
108+
// "a;b" + "b;c" -> "a;b;c" (deduplicate "b")
109+
auto ret = agg->Agg(std::string_view("a;b"), std::string_view("b;c"));
110+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "a;b;c");
111+
}
112+
113+
TEST_F(FieldListaggAggTest, TestDistinctNoDuplicates) {
114+
ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg(" ", true));
115+
116+
// "a b" + "c d" -> "a b c d" (no dups to remove)
117+
auto ret = agg->Agg(std::string_view("a b"), std::string_view("c d"));
118+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "a b c d");
119+
}
120+
121+
TEST_F(FieldListaggAggTest, TestDistinctEmptyInput) {
122+
ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg(";", true));
123+
124+
// empty input -> return accumulator
125+
auto ret = agg->Agg(std::string_view("a;b"), std::string_view(""));
126+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "a;b");
127+
}
128+
129+
TEST_F(FieldListaggAggTest, TestDistinctFalse) {
130+
ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg(";", false));
131+
132+
// "a;b" + "b;c" -> "a;b;b;c" (no dedup)
133+
auto ret = agg->Agg(std::string_view("a;b"), std::string_view("b;c"));
134+
ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "a;b;b;c");
135+
}
136+
137+
TEST_F(FieldListaggAggTest, TestInvalidType) {
138+
EXPECT_OK_AND_ASSIGN(auto options, CoreOptions::FromMap({}));
139+
auto result = FieldListaggAgg::Create(arrow::int32(), options, "f");
140+
ASSERT_FALSE(result.ok());
141+
ASSERT_TRUE(result.status().ToString().find("supposed to be string") != std::string::npos)
142+
<< result.status().ToString();
143+
}
144+
145+
} // namespace paimon::test

0 commit comments

Comments
 (0)