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 35ca2c0425 [python] Introduce collect aggregator function. (#8737)
35ca2c0425 is described below
commit 35ca2c0425de91a949562e83c70c5ea5b41c44be
Author: AuroraVoyage <[email protected]>
AuthorDate: Mon Jul 20 11:28:18 2026 +0800
[python] Introduce collect aggregator function. (#8737)
---
.../pypaimon/read/merge_engine_support.py | 1 +
.../pypaimon/read/reader/aggregate/aggregators.py | 69 +++++++
.../pypaimon/tests/test_aggregation_e2e.py | 14 +-
.../pypaimon/tests/test_field_aggregators.py | 209 ++++++++++++++++++++-
4 files changed, 286 insertions(+), 7 deletions(-)
diff --git a/paimon-python/pypaimon/read/merge_engine_support.py
b/paimon-python/pypaimon/read/merge_engine_support.py
index cefcee2e67..89680141e3 100644
--- a/paimon-python/pypaimon/read/merge_engine_support.py
+++ b/paimon-python/pypaimon/read/merge_engine_support.py
@@ -64,6 +64,7 @@ _AGGREGATION_SUPPORTED_AGG_FUNCS = frozenset([
"bool_or", "bool_and",
"listagg",
"nested_update",
+ "collect"
])
_FIELDS_PREFIX = "fields."
_FIELD_SEQUENCE_GROUP_SUFFIX = ".sequence-group"
diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
index 11b182edaf..060082c1cc 100644
--- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
+++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
@@ -60,6 +60,7 @@ NAME_BOOL_OR = "bool_or"
NAME_BOOL_AND = "bool_and"
NAME_LISTAGG = "listagg"
NAME_NESTED_UPDATE = "nested_update"
+NAME_COLLECT = "collect"
# Base SQL type names treated as numeric for sum/product-style
@@ -455,6 +456,71 @@ class FieldListaggAgg(FieldAggregator):
return self.delimiter.join(result)
+class FieldCollectAgg(FieldAggregator):
+
+ def __init__(
+ self,
+ name: str,
+ field_type: ArrayType,
+ field_name: str,
+ options: CoreOptions,
+ ):
+ super().__init__(name, field_type)
+
+ if not isinstance(field_type, ArrayType):
+ raise ValueError(
+ "Data type for collect column must be 'Array' but was
'{}'.".format(field_type)
+ )
+
+ self.distinct = options.field_collect_distinct(field_name)
+
+ def agg_reversed(self, accumulator: Any, input_field: Any) -> Any:
+ # we don't need to actually do the reverse here for this agg
+ # because accumulator has been distinct, just let accumulator be
accumulator will speed up
+ # distinct process
+ return self.agg(accumulator, input_field)
+
+ def agg(self, accumulator: Any, input_field: Any) -> Any:
+ if accumulator is None and input_field is None:
+ return None
+
+ if (accumulator is None or input_field is None) and not self.distinct:
+ return input_field if accumulator is None else accumulator
+
+ if self.distinct:
+ result = []
+ self._collect(result, accumulator)
+ self._collect(result, input_field)
+
+ return result
+ else:
+ return list(accumulator) + list(input_field)
+
+ def retract(self, accumulator: Any, retract_field: Any) -> Any:
+ if accumulator is None:
+ return None
+
+ if retract_field is None:
+ return accumulator
+
+ if len(retract_field) == 0:
+ return accumulator
+
+ result = list(accumulator)
+ for element in retract_field:
+ if element in result:
+ result.remove(element)
+
+ return result
+
+ def _collect(self, result: List[Any], data: List[Any]):
+ if data is None:
+ return
+ for element in data:
+ if element not in result:
+ result.append(element)
+
+
class FieldNestedUpdateAgg(FieldAggregator):
"""
Used to update a field which representing a nested table.
@@ -740,3 +806,6 @@ register_aggregator(
register_aggregator(
NAME_NESTED_UPDATE, _build_field_options(FieldNestedUpdateAgg,
NAME_NESTED_UPDATE)
)
+register_aggregator(
+ NAME_COLLECT, _build_field_options(FieldCollectAgg, NAME_COLLECT)
+)
diff --git a/paimon-python/pypaimon/tests/test_aggregation_e2e.py
b/paimon-python/pypaimon/tests/test_aggregation_e2e.py
index 0d557799e9..b70fe97d51 100644
--- a/paimon-python/pypaimon/tests/test_aggregation_e2e.py
+++ b/paimon-python/pypaimon/tests/test_aggregation_e2e.py
@@ -294,20 +294,22 @@ class AggregationMergeEngineE2ETest(unittest.TestCase):
)
def test_out_of_scope_field_aggregator_rejected(self):
- # collect is one of the aggregator identifiers this engine
- # doesn't support yet. The guard must reject the config rather
+ # hll_sketch is one of the aggregator identifiers this engine
+ # doesn't support yet. collect is now supported by the Python
+ # aggregation engine, so use hll_sketch instead.
+ # The guard must reject the config rather
# than let the per-field factory build a (silently wrong)
# fallback.
self._create_and_expect_unsupported(
- 'agg_reject_collect',
- {'fields.label.aggregate-function': 'collect'},
+ 'agg_reject_hll_sketch',
+ {'fields.label.aggregate-function': 'hll_sketch'},
'fields.label.aggregate-function',
)
def test_out_of_scope_default_aggregator_rejected(self):
self._create_and_expect_unsupported(
- 'agg_reject_default_collect',
- {'fields.default-aggregate-function': 'product'},
+ 'agg_reject_default_hll_sketch',
+ {'fields.default-aggregate-function': 'hll_sketch'},
'fields.default-aggregate-function',
)
diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py
b/paimon-python/pypaimon/tests/test_field_aggregators.py
index 9c9d5876fe..2f4d07c39d 100644
--- a/paimon-python/pypaimon/tests/test_field_aggregators.py
+++ b/paimon-python/pypaimon/tests/test_field_aggregators.py
@@ -46,8 +46,9 @@ from pypaimon.read.reader.aggregate.aggregators import (
FieldSumAgg,
FieldListaggAgg,
FieldNestedUpdateAgg,
+ FieldCollectAgg,
)
-from pypaimon.schema.data_types import AtomicType, DataField, RowType,
ArrayType
+from pypaimon.schema.data_types import AtomicType, DataField, RowType,
ArrayType, MapType
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.row.internal_row import InternalRow
@@ -550,6 +551,212 @@ class FieldListaggAggTest(unittest.TestCase):
)
+class FieldCollectAggTest(unittest.TestCase):
+
+ def _make(self, distinct, element_type=None):
+ if element_type is None:
+ element_type = AtomicType("INT")
+
+ options = CoreOptions(Options({"fields.field0.distinct": distinct}))
+
+ return create_field_aggregator(
+ ArrayType(True, element_type),
+ "field0",
+ "collect",
+ options=options,
+ )
+
+ def row(self, *values, fields: List[DataField]):
+ return GenericRow(list(values), fields)
+
+ def test_field_collect_agg_with_distinct(self):
+ agg = self._make(distinct=True)
+ self.assertIsInstance(agg, FieldCollectAgg)
+
+ self.assertIsNone(agg.agg(None, None))
+
+ result = agg.agg(None, [1, 1, 2])
+ self.assertEqual(result, [1, 2])
+
+ result = agg.agg([1, 1, 2], [2, 3])
+ self.assertEqual(result, [1, 2, 3])
+
+ def test_field_collect_agg_without_distinct(self):
+ agg = self._make(distinct=False)
+
+ self.assertIsNone(agg.agg(None, None))
+
+ result = agg.agg(None, [1, 1, 2])
+ self.assertEqual(result, [1, 1, 2])
+
+ result = agg.agg([1, 1, 2], [2, 3])
+ self.assertEqual(result, [1, 1, 2, 2, 3])
+
+ def test_field_collect_agg_retract(self):
+ agg = self._make(distinct=True)
+
+ result = agg.retract([1, 2, 3], [1])
+ self.assertEqual(result, [2, 3])
+ self.assertIsNone(agg.retract(None, [1]))
+ self.assertEqual(agg.retract([1, 2], None), [1, 2])
+
+ def test_field_collect_agg_retract_duplicate_elements(self):
+ # primitive type
+ agg = self._make(distinct=True)
+
+ self.assertEqual(
+ agg.retract([1, 1, 2, 2, 3], [1, 2, 3]),
+ [1, 2],
+ )
+
+ # row type
+ fields = [
+ DataField(0, "id", AtomicType("INT")),
+ DataField(1, "name", AtomicType("STRING")),
+ ]
+ agg = self._make(
+ distinct=True,
+ element_type=RowType(True, fields),
+ )
+
+ self.assertEqual(
+ agg.retract(
+ [
+ self.row(1, "A", fields=fields),
+ self.row(1, "A", fields=fields),
+ self.row(1, "B", fields=fields),
+ self.row(2, "B", fields=fields),
+ ],
+ [
+ self.row(1, "A", fields=fields),
+ self.row(2, "B", fields=fields),
+ ],
+ ),
+ [
+ self.row(1, "A", fields=fields),
+ self.row(1, "B", fields=fields),
+ ],
+ )
+
+ # array type
+ agg = self._make(
+ distinct=True,
+ element_type=ArrayType(True, AtomicType("INT")),
+ )
+
+ self.assertEqual(
+ agg.retract(
+ [[1, 1], [1, 1], [1, 2], [2, 1]],
+ [[1, 1], [1, 2]],
+ ),
+ [[1, 1], [2, 1]],
+ )
+
+ # map type
+ agg = self._make(
+ distinct=True,
+ element_type=MapType(True, AtomicType("INT"),
AtomicType("STRING")),
+ )
+
+ self.assertEqual(
+ agg.retract(
+ [{1: "A"}, {1: "A"}, {1: "A", 2: "B"}, {1: "C"}],
+ [{1: "A"}, {2: "B", 1: "A"}],
+ ),
+ [{1: "A"}, {1: "C"}],
+ )
+
+ def test_field_collect_agg_with_row_type(self):
+ fields = [
+ DataField(0, "id", AtomicType("INT")),
+ DataField(1, "name", AtomicType("STRING")),
+ ]
+ agg = self._make(
+ distinct=True,
+ element_type=RowType(True, fields),
+ )
+
+ input1 = [
+ self.row(1, "A", fields=fields),
+ self.row(1, "B", fields=fields),
+ ]
+
+ result = agg.agg(None, input1)
+ self.assertEqual(result, input1)
+
+ input2 = [
+ self.row(1, "A", fields=fields),
+ self.row(2, "A", fields=fields),
+ ]
+
+ result = agg.agg(input1, input2)
+ self.assertEqual(result, [
+ self.row(1, "A", fields=fields),
+ self.row(1, "B", fields=fields),
+ self.row(2, "A", fields=fields),
+ ])
+
+ # retract
+ result = agg.retract(
+ [
+ self.row(1, "A", fields=fields),
+ self.row(1, "B", fields=fields),
+ self.row(2, "B", fields=fields),
+ ],
+ [
+ self.row(1, "A", fields=fields),
+ self.row(2, "B", fields=fields),
+ ],
+ )
+ self.assertEqual(result, [self.row(1, "B", fields=fields)])
+
+ def test_field_collect_agg_with_array_type(self):
+ agg = self._make(
+ distinct=True,
+ element_type=ArrayType(True, AtomicType("INT")),
+ )
+
+ input1 = [[1, 1], [1, 2]]
+ acc = agg.agg(None, input1)
+ self.assertEqual(acc, input1)
+
+ input2 = [[1, 1], [1, 2], [2, 1]]
+ acc = agg.agg(acc, input2)
+ self.assertEqual(acc, [[1, 1], [1, 2], [2, 1]])
+
+ # retract
+ acc = agg.retract(
+ [[1, 1], [1, 2], [2, 1]],
+ [[1, 1], [1, 2]],
+ )
+ self.assertEqual(acc, [[2, 1]])
+
+ def test_field_collect_agg_with_map_type(self):
+ agg = self._make(
+ distinct=True,
+ element_type=MapType(
+ True,
+ AtomicType("INT"),
+ AtomicType("STRING"),
+ ),
+ )
+
+ input1 = [{1: "A"}, {1: "A", 2: "B"}]
+ acc = agg.agg(None, input1)
+ self.assertEqual(acc, input1)
+
+ input2 = [{1: "A"}, {2: "B", 1: "A"}, {1: "C"}]
+ acc = agg.agg(acc, input2)
+ self.assertEqual(acc, [{1: "A"}, {1: "A", 2: "B"}, {1: "C"}])
+
+ # retract
+ acc = agg.retract(
+ [{1: "A"}, {1: "A", 2: "B"}, {1: "C"}],
+ [{1: "A"}, {2: "B", 1: "A"}],
+ )
+ self.assertEqual(acc, [{1: "C"}])
+
+
class FieldNestedUpdateAggTest(unittest.TestCase):
IDENTIFIER = "nested_update"
