This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 67779ac54e [python] Introduce 'listagg' aggregator (#8561)
67779ac54e is described below
commit 67779ac54ee1a856c5a684eef16bccfca2500f98
Author: AuroraVoyage <[email protected]>
AuthorDate: Wed Jul 15 08:13:43 2026 +0800
[python] Introduce 'listagg' aggregator (#8561)
---
.../pypaimon/common/options/core_options.py | 23 ++
.../pypaimon/read/reader/aggregate/aggregators.py | 85 ++++++
.../pypaimon/tests/test_field_aggregators.py | 302 ++++++++++++++++++++-
3 files changed, 408 insertions(+), 2 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index aa58e48eb4..03c32f66ed 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -137,6 +137,11 @@ class CoreOptions:
FILE_FORMAT_ROW: str = "row"
FILE_FORMAT_MOSAIC: str = "mosaic"
+ # Field agg constants
+ FIELDS_PREFIX = "fields"
+ DISTINCT = "distinct"
+ LIST_AGG_DELIMITER = "list-agg-delimiter"
+
# Basic options
AUTO_CREATE: ConfigOption[bool] = (
ConfigOptions.key("auto-create")
@@ -1345,3 +1350,21 @@ class CoreOptions:
def dynamic_partition_overwrite(self) -> bool:
return self.options.get(CoreOptions.DYNAMIC_PARTITION_OVERWRITE)
+
+ def field_listagg_delimiter(self, field_name: str) -> str:
+ return self.options.get(
+ ConfigOptions.key(
+
f'{CoreOptions.FIELDS_PREFIX}.{field_name}.{CoreOptions.LIST_AGG_DELIMITER}'
+ )
+ .string_type()
+ .default_value(',')
+ )
+
+ def field_collect_distinct(self, field_name: str) -> bool:
+ return self.options.get(
+ ConfigOptions.key(
+
f'{CoreOptions.FIELDS_PREFIX}.{field_name}.{CoreOptions.DISTINCT}'
+ )
+ .boolean_type()
+ .default_value(False)
+ )
diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
index 961cd10091..13ef86833a 100644
--- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
+++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
@@ -35,6 +35,7 @@ error rather than a silent fallback.
from typing import Any
+from pypaimon.common.options import CoreOptions
from pypaimon.read.reader.aggregate import register_aggregator
from pypaimon.read.reader.aggregate.field_aggregator import FieldAggregator
from pypaimon.schema.data_types import AtomicType, DataType
@@ -52,6 +53,7 @@ NAME_MAX = "max"
NAME_MIN = "min"
NAME_BOOL_OR = "bool_or"
NAME_BOOL_AND = "bool_and"
+NAME_LISTAGG = "listagg"
# Base SQL type names treated as numeric for sum/product-style
@@ -94,6 +96,17 @@ def _check_boolean(name: str, field_type: DataType) -> None:
)
+def is_blank(s: str) -> bool:
+ if s is None:
+ return True
+
+ for ch in s:
+ if not ch.isspace():
+ return False
+
+ return True
+
+
# ---------------------------------------------------------------------------
# Aggregator classes
# ---------------------------------------------------------------------------
@@ -221,6 +234,65 @@ class FieldBoolAndAgg(FieldAggregator):
return bool(accumulator) and bool(input_field)
+class FieldListaggAgg(FieldAggregator):
+ """LISTAGG aggregator for STRING fields.
+
+ Concatenates string values using the configured delimiter.
+
+ If ``distinct`` is True, duplicated tokens are removed while
+ preserving their first appearance order.
+ """
+
+ def __init__(
+ self,
+ name: str,
+ field_type: DataType,
+ field_name: str,
+ options: CoreOptions,
+ ):
+ super().__init__(name, field_type)
+
+ if _atomic_base_name(field_type) != "STRING":
+ raise ValueError(
+ "Data type for '{}' column must be 'STRING' but was '{}'."
+ .format(name, field_type)
+ )
+
+ self.delimiter = options.field_listagg_delimiter(field_name)
+ self.distinct = options.field_collect_distinct(field_name)
+ self._separator = " " if self.delimiter in (None, "") else
self.delimiter
+
+ def agg(self, accumulator: Any, input_field: Any) -> Any:
+ if input_field is None or is_blank(str(input_field)):
+ return accumulator
+
+ if accumulator is None or is_blank(str(accumulator)):
+ return input_field
+
+ accumulator = str(accumulator)
+ input_field = str(input_field)
+
+ if not self.distinct:
+ return accumulator + self.delimiter + input_field
+
+ accumulator_tokens = accumulator.split(self._separator)
+ existing_tokens = set(accumulator_tokens)
+
+ result = [accumulator]
+
+ for token in input_field.split(self._separator):
+ if is_blank(token) or token in existing_tokens:
+ continue
+
+ existing_tokens.add(token)
+ result.append(token)
+
+ if len(result) == 1:
+ return accumulator
+
+ return self.delimiter.join(result)
+
+
# ---------------------------------------------------------------------------
# Registration. Each builder binds an identifier to a factory that
# optionally validates the column DataType before constructing the
@@ -252,6 +324,16 @@ def _build_boolean(cls, identifier: str):
return _factory
+def _build_field_options(cls, identifier: str):
+ """Build a factory that accepts any DataType. Used by
+ ``primary_key`` / ``last_value`` / ``first_value`` variants and by
+ ``max`` / ``min``, all of which work on any orderable DataType.
+ """
+ def _factory(field_type, field_name, options):
+ return cls(identifier, field_type, field_name, options)
+ return _factory
+
+
register_aggregator(
NAME_PRIMARY_KEY,
_build_no_type_check(FieldPrimaryKeyAgg, NAME_PRIMARY_KEY),
@@ -281,3 +363,6 @@ register_aggregator(
register_aggregator(
NAME_BOOL_AND, _build_boolean(FieldBoolAndAgg, NAME_BOOL_AND)
)
+register_aggregator(
+ NAME_LISTAGG, _build_field_options(FieldListaggAgg, NAME_LISTAGG)
+)
diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py
b/paimon-python/pypaimon/tests/test_field_aggregators.py
index f54a67cd95..b995b1c4bc 100644
--- a/paimon-python/pypaimon/tests/test_field_aggregators.py
+++ b/paimon-python/pypaimon/tests/test_field_aggregators.py
@@ -27,7 +27,9 @@ on real PK tables lives in ``test_aggregation_e2e.py``.
import datetime
import unittest
from decimal import Decimal
+from functools import reduce
+from pypaimon.common.options import CoreOptions, Options
from pypaimon.read.reader.aggregate import create_field_aggregator
from pypaimon.read.reader.aggregate.aggregators import (
FieldBoolAndAgg,
@@ -40,16 +42,20 @@ from pypaimon.read.reader.aggregate.aggregators import (
FieldMinAgg,
FieldPrimaryKeyAgg,
FieldSumAgg,
+ FieldListaggAgg,
)
from pypaimon.schema.data_types import AtomicType
-def _make(identifier, sql_type):
+def _make(identifier, sql_type, options: CoreOptions = None):
"""Build an aggregator through the public registry path so we also
exercise the registered factory (including its type validation).
"""
+ if options is None:
+ options = CoreOptions(Options.from_none())
+
return create_field_aggregator(
- AtomicType(sql_type), "field0", identifier, options=None
+ AtomicType(sql_type), "field0", identifier, options=options
)
@@ -247,6 +253,298 @@ class FieldBoolAndAggTest(unittest.TestCase):
self.assertIn("BOOLEAN", str(ctx.exception))
+class FieldListaggAggTest(unittest.TestCase):
+
+ def test_default_delimiter(self):
+ agg = _make("listagg", "STRING")
+ self.assertIsInstance(agg, FieldListaggAgg)
+
+ self.assertEqual(
+ agg.agg("user1", "user2"),
+ "user1,user2",
+ )
+
+ def test_default_delimiter_distinct(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({'fields.field0.distinct': True}))
+ )
+
+ result = reduce(
+ agg.agg,
+ [
+ "user1",
+ "user2",
+ "user1",
+ "user3",
+ ],
+ )
+
+ self.assertEqual(result, "user1,user2,user3")
+
+ def test_whitespace_delimiter_distinct(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({
+ 'fields.field0.list-agg-delimiter': '',
+ 'fields.field0.distinct': True,
+ }))
+ )
+
+ result = reduce(
+ agg.agg,
+ [
+ "AB",
+ "AB C",
+ "D",
+ "EF",
+ "G ",
+ ],
+ )
+ self.assertEqual(result, "ABCDEFG")
+
+ def test_custom_delimiter_empty_strings(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({
+ 'fields.field0.list-agg-delimiter': ';',
+ 'fields.field0.distinct': True
+ })),
+ )
+
+ result = reduce(
+ agg.agg,
+ [
+ "",
+ "",
+ ],
+ )
+
+ self.assertEqual(result, "")
+
+ def test_default_delimiter_distinct_multi_user(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({'fields.field0.distinct': True})),
+ )
+
+ result = reduce(
+ agg.agg,
+ [
+ "user1",
+ "user2",
+ "user1,user3",
+ ],
+ )
+
+ self.assertEqual(result, "user1,user2,user3")
+
+ def test_default_delimiter_distinct_empty_left(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({'fields.field0.distinct': True})),
+ )
+
+ result = reduce(
+ agg.agg,
+ [
+ "",
+ "user2",
+ "user1,user3",
+ ],
+ )
+
+ self.assertEqual(result, "user2,user1,user3")
+
+ def test_custom_delimiter_distinct_multi_kv(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({
+ 'fields.field0.list-agg-delimiter': ';',
+ 'fields.field0.distinct': True
+ }))
+ )
+
+ result = reduce(
+ agg.agg,
+ [
+ "k1=v1;k2=v2",
+ "k1=v1;k3=v3",
+ "",
+ ],
+ )
+
+ self.assertEqual(result, "k1=v1;k2=v2;k3=v3")
+
+ def test_custom_delimiter_whitespace(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({
+ 'fields.field0.list-agg-delimiter': ' ',
+ 'fields.field0.distinct': True
+ })),
+ )
+
+ result = reduce(
+ agg.agg,
+ [
+ "k1=v1 k2=v2",
+ " k1=v1 k3=v3",
+ " ",
+ ],
+ )
+
+ self.assertEqual(result, "k1=v1 k2=v2 k3=v3")
+
+ def test_default_delimiter_distinct_multi_duplicate_kv(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({'fields.field0.distinct': True}))
+ )
+
+ result = reduce(
+ agg.agg,
+ [
+ "k1=v1,k2=v2",
+ "k1=v1,k2=v3",
+ "",
+ ],
+ )
+
+ self.assertEqual(result, "k1=v1,k2=v2,k2=v3")
+
+ def test_custom_delimiter(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({'fields.field0.list-agg-delimiter': '-'})),
+ )
+
+ self.assertEqual(
+ agg.agg("user1", "user2"),
+ "user1-user2",
+ )
+
+ def test_distinct_should_not_match_substring(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({'fields.field0.distinct': True})),
+ )
+
+ result = agg.agg(
+ "abc,def,asd",
+ "ab,xy",
+ )
+
+ self.assertEqual(
+ result,
+ "abc,def,asd,ab,xy",
+ )
+
+ def test_distinct_substring_custom_delimiter(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({
+ 'fields.field0.list-agg-delimiter': ';',
+ 'fields.field0.distinct': True
+ })),
+ )
+
+ result = agg.agg(
+ "abc;def;asd",
+ "ab;xy;def",
+ )
+
+ self.assertEqual(
+ result,
+ "abc;def;asd;ab;xy",
+ )
+
+ def test_ignore_blank_values(self):
+ agg = _make("listagg", "STRING")
+
+ result = reduce(
+ agg.agg,
+ [
+ "user1",
+ "",
+ " ",
+ " ",
+ "\t",
+ "\n",
+ "\r",
+ "\r\n",
+ " \t\n ",
+ " \t\n\r\n \u3000 ",
+ "user2",
+ "\u3000",
+ "\u2000",
+ ],
+ )
+
+ self.assertEqual(
+ result,
+ "user1,user2",
+ )
+
+ def test_distinct_ignore_blank_values(self):
+ agg = _make(
+ "listagg",
+ "STRING",
+ CoreOptions(Options({'fields.field0.distinct': True})),
+ )
+
+ result = reduce(
+ agg.agg,
+ [
+ "user1",
+ "user2",
+ "user1",
+ "user3",
+ "",
+ " ",
+ " ",
+ "\t",
+ "\n",
+ "\r",
+ "\r\n",
+ " \t\n ",
+ " \t\n\r\n \u3000 ",
+ "user2",
+ "user3",
+ "\u3000",
+ "\u2000",
+ ],
+ )
+
+ self.assertEqual(
+ result,
+ "user1,user2,user3",
+ )
+
+ def test_first_non_blank_value_without_leading_delimiter(self):
+ agg = _make("listagg", "STRING")
+
+ acc = None
+ acc = agg.agg(acc, " ")
+ acc = agg.agg(acc, "first line")
+
+ self.assertEqual(
+ acc,
+ "first line",
+ )
+
+
class RegistrationTest(unittest.TestCase):
"""Sanity check that all 10 expected aggregators (the primary-key
placeholder plus 9 value aggregators) are registered when the
