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 f0143c98b1 [python] Introduce 'nested_partial_update' aggregator
function (#8578)
f0143c98b1 is described below
commit f0143c98b1ab7efa1a243c51cfbc6a4f05ad60dc
Author: AuroraVoyage <[email protected]>
AuthorDate: Thu Jul 23 21:15:04 2026 +0800
[python] Introduce 'nested_partial_update' aggregator function (#8578)
---
.../pypaimon/read/merge_engine_support.py | 2 +-
.../pypaimon/read/reader/aggregate/aggregators.py | 124 +++++++++++++++++++
.../pypaimon/tests/test_field_aggregators.py | 135 +++++++++++++++++++++
3 files changed, 260 insertions(+), 1 deletion(-)
diff --git a/paimon-python/pypaimon/read/merge_engine_support.py
b/paimon-python/pypaimon/read/merge_engine_support.py
index aab3cf3dd6..321421b892 100644
--- a/paimon-python/pypaimon/read/merge_engine_support.py
+++ b/paimon-python/pypaimon/read/merge_engine_support.py
@@ -63,7 +63,7 @@ _AGGREGATION_SUPPORTED_AGG_FUNCS = frozenset([
"sum", "max", "min",
"bool_or", "bool_and",
"listagg",
- "nested_update",
+ "nested_update", "nested_partial_update",
"collect",
"merge_map_with_keytime",
"merge_map",
diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
index d973acb8f4..5c325d99eb 100644
--- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
+++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
@@ -40,6 +40,7 @@ from pypaimon.common.options.core_options import
NestedKeyNullStrategy
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, ArrayType,
RowType, MapType
+from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.row.internal_row import InternalRow
# aggregator input type hints variables
@@ -60,6 +61,7 @@ NAME_BOOL_OR = "bool_or"
NAME_BOOL_AND = "bool_and"
NAME_LISTAGG = "listagg"
NAME_NESTED_UPDATE = "nested_update"
+NAME_NESTED_PARTIAL_UPDATE = "nested_partial_update"
NAME_COLLECT = "collect"
NAME_MERGE_MAP_WITH_KEYTIME = "merge_map_with_keytime"
NAME_MERGE_MAP = "merge_map"
@@ -732,6 +734,125 @@ class FieldNestedUpdateAgg(FieldAggregator):
)
+class FieldNestedPartialUpdateAgg(FieldAggregator):
+ """
+ Used to partial update a field which representing a nested table.
+ The data type of nested table field is ARRAY<ROW>
+ """
+ def __init__(
+ self,
+ name: str,
+ field_type: ArrayType,
+ field_name: str,
+ options: CoreOptions,
+ ):
+ field_type = _check_array_row(field_name, field_type)
+ super().__init__(name, field_type)
+
+ nested_type: RowType = field_type.element
+ self.nested_fields = len(nested_type.fields)
+
+ self.nested_key =
options.field_nested_update_agg_nested_key(field_name)
+ if not self.nested_key:
+ raise ValueError("nested_update_partial requires 'nested-key' to
be configured.")
+
+ self.key_projection = FieldProjection.from_fields(
+ [nested_type.get_field_index(name) for name in self.nested_key],
+ self.nested_key
+ )
+
+ self.nested_key_null_strategy = (
+
options.field_nested_update_agg_nested_key_null_strategy(field_name)
+ )
+
+ def agg(self, accumulator: Any, input_field: Any) -> Any:
+ if input_field is None:
+ return accumulator
+
+ rows: List[Record] = []
+ if accumulator is not None:
+ self._add_non_null_rows(accumulator, rows)
+ self._add_non_null_rows(input_field, rows)
+
+ row_map: Dict[Tuple[Any, ...], Record] = {}
+ for row in rows:
+ key = self.key_projection.apply(row)
+ if not self._apply_nested_key_null_strategy(key):
+ continue
+
+ to_update = row_map.get(key)
+ if to_update is None:
+ if isinstance(row, InternalRow):
+ to_update = GenericRow([None] * self.nested_fields,
row.fields)
+ elif isinstance(row, dict):
+ to_update = {}.fromkeys(row.keys())
+ else:
+ raise TypeError(
+ "Unsupported row type '{}'. Expected InternalRow or
dict.".format(
+ type(row).__name__
+ )
+ )
+ self._partial_update(to_update, row)
+ row_map[key] = to_update
+
+ return list(row_map.values())
+
+ def _partial_update(self, to_update: Record, input_row: Record) -> None:
+ if isinstance(to_update, InternalRow) and isinstance(input_row,
InternalRow):
+ for i in range(self.nested_fields):
+ value = input_row.get_field(i)
+ if value is not None:
+ to_update.values[i] = value
+ elif isinstance(to_update, dict) and isinstance(input_row, dict):
+ for k, v in input_row.items():
+ if v is not None:
+ to_update[k] = v
+ else:
+ raise TypeError(
+ "Unsupported row types: to_update={}, input_row={}. "
+ "Expected both to be either InternalRow or dict.".format(
+ type(to_update).__name__,
+ type(input_row).__name__,
+ )
+ )
+
+ def _add_non_null_rows(
+ self,
+ array: List[Record],
+ rows: List[Record],
+ ) -> None:
+ """Append non-null rows from array."""
+
+ for row in array:
+ if row is None:
+ continue
+ rows.append(row)
+
+ def _apply_nested_key_null_strategy(self, key: Tuple[Any, ...]) -> bool:
+ """Apply nested-key-null-strategy."""
+
+ if all(v is not None for v in key):
+ return True
+
+ if self.nested_key_null_strategy == NestedKeyNullStrategy.MERGE:
+ return True
+
+ if self.nested_key_null_strategy == NestedKeyNullStrategy.IGNORE:
+ return False
+
+ if self.nested_key_null_strategy == NestedKeyNullStrategy.ERROR:
+ raise ValueError(
+ "Nested key contains null values. "
+ "Primary key fields must not be null."
+ )
+
+ raise ValueError(
+ "Unsupported nested-key-null-strategy '{}'".format(
+ self.nested_key_null_strategy
+ )
+ )
+
+
class FieldMergeMapWithKeyTimeAgg(FieldAggregator):
"""
Aggregator for merging MAP values with key and timestamp.
@@ -1018,6 +1139,9 @@ register_aggregator(
register_aggregator(
NAME_NESTED_UPDATE, _build_field_options(FieldNestedUpdateAgg,
NAME_NESTED_UPDATE)
)
+register_aggregator(
+ NAME_NESTED_PARTIAL_UPDATE,
_build_field_options(FieldNestedPartialUpdateAgg, NAME_NESTED_PARTIAL_UPDATE)
+)
register_aggregator(
NAME_COLLECT, _build_field_options(FieldCollectAgg, NAME_COLLECT)
)
diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py
b/paimon-python/pypaimon/tests/test_field_aggregators.py
index 72b8767ec8..107b160df1 100644
--- a/paimon-python/pypaimon/tests/test_field_aggregators.py
+++ b/paimon-python/pypaimon/tests/test_field_aggregators.py
@@ -46,6 +46,7 @@ from pypaimon.read.reader.aggregate.aggregators import (
FieldSumAgg,
FieldListaggAgg,
FieldNestedUpdateAgg,
+ FieldNestedPartialUpdateAgg,
FieldCollectAgg,
FieldMergeMapWithKeyTimeAgg,
FieldMergeMapAgg,
@@ -1767,6 +1768,140 @@ class FieldNestedUpdateAggTest(unittest.TestCase):
self.assertCountEqual(accumulator, [self.row(0, 0, "A", 1), ])
+class FieldNestedPartialUpdateAggTest(unittest.TestCase):
+ IDENTIFIER = "nested_partial_update"
+
+ DEFAULT_FIELDS = [
+ DataField(10, "k", AtomicType("INT")),
+ DataField(20, "v1", AtomicType("INT")),
+ DataField(30, "v2", AtomicType("STRING")),
+ ]
+
+ def _make_data_type(self, fields: List[DataField] = None):
+ if fields is None:
+ fields = self.DEFAULT_FIELDS
+ return ArrayType(
+ True,
+ RowType(True, fields)
+ )
+
+ def _make(self, data_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(
+ data_type, "field0", self.IDENTIFIER, options=options
+ )
+
+ def row(self, *values, fields: List[DataField] = None):
+ if fields is None:
+ fields = self.DEFAULT_FIELDS
+ return GenericRow(list(values), fields)
+
+ def test_field_nested_partial_update_agg(self):
+ agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k",
+ }
+ )
+ ),
+ )
+ self.assertIsInstance(agg, FieldNestedPartialUpdateAgg)
+
+ accumulator = None
+
+ current = self.row(0, 0, None)
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [current])
+
+ current = self.row(0, None, "A")
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [self.row(0, 0, "A")])
+
+ current = self.row(0, 1, "B")
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [self.row(0, 1, "B")])
+
+ current = self.row(1, 2, "C")
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B"),
+ self.row(1, 2, "C"),
+ ])
+
+ current = self.row(None, 0, "D")
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B"),
+ self.row(1, 2, "C"),
+ self.row(None, 0, "D"),
+ ])
+
+ def
test_field_nested_partial_update_agg_with_nested_key_null_use_ignore_strategy(self):
+ agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k",
+ "fields.field0.nested-key-null-strategy": "IGNORE",
+ }
+ )
+ ),
+ )
+
+ accumulator = None
+
+ current = self.row(0, 0, None)
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [current])
+
+ current = self.row(None, None, "A_ignore")
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [self.row(0, 0, None)])
+
+ current = self.row(None, None, "FirstInput")
+ accumulator = agg.agg(None, [current])
+ self.assertCountEqual(accumulator, [])
+
+ def
test_field_nested_partial_update_agg_with_nested_key_null_use_throw_error_strategy(self):
+ agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k",
+ "fields.field0.nested-key-null-strategy": "ERROR",
+ }
+ )
+ ),
+ )
+
+ accumulator = None
+
+ current = self.row(0, 0, None)
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [current])
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ agg.agg(accumulator, [self.row(None, 0, "A")])
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ agg.agg(None, [self.row(None, None, "FirstInput")])
+
+
class FieldMergeMapWithKeyTimeAggTest(unittest.TestCase):
DEFAULT_FIELDS = [
