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 87aeb13552 [python] Introduce merge_map_with_keytime aggregator
function (#8734)
87aeb13552 is described below
commit 87aeb135526ebcc92dc0b36f6215670943fc95c5
Author: AuroraVoyage <[email protected]>
AuthorDate: Tue Jul 21 14:06:42 2026 +0800
[python] Introduce merge_map_with_keytime aggregator function (#8734)
---
.../pypaimon/common/options/core_options.py | 10 ++
.../pypaimon/read/merge_engine_support.py | 5 +-
.../pypaimon/read/reader/aggregate/aggregators.py | 126 ++++++++++++++++-
.../pypaimon/tests/test_field_aggregators.py | 155 +++++++++++++++++++++
4 files changed, 293 insertions(+), 3 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index a6aa5f96b5..36fd975f6d 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -153,6 +153,7 @@ class CoreOptions:
NESTED_KEY_NULL_STRATEGY = "nested-key-null-strategy"
NESTED_SEQUENCE_FIELD = "nested-sequence-field"
COUNT_LIMIT = "count-limit"
+ MERGE_MAP_TS_FIELD = "ts-field"
# Basic options
AUTO_CREATE: ConfigOption[bool] = (
@@ -1597,6 +1598,15 @@ class CoreOptions:
.default_value(2147483647) # Integer.MAX_VALUE
)
+ def field_merge_map_ts_field(self, field_name: str) -> str:
+ return self.options.get(
+ ConfigOptions.key(
+
f'{CoreOptions.FIELDS_PREFIX}.{field_name}.{CoreOptions.MERGE_MAP_TS_FIELD}'
+ )
+ .string_type()
+ .no_default_value()
+ )
+
@property
def query_auth_enabled(self) -> bool:
return self.options.get(CoreOptions.QUERY_AUTH_ENABLED)
diff --git a/paimon-python/pypaimon/read/merge_engine_support.py
b/paimon-python/pypaimon/read/merge_engine_support.py
index 89680141e3..ff551f9ab5 100644
--- a/paimon-python/pypaimon/read/merge_engine_support.py
+++ b/paimon-python/pypaimon/read/merge_engine_support.py
@@ -64,7 +64,8 @@ _AGGREGATION_SUPPORTED_AGG_FUNCS = frozenset([
"bool_or", "bool_and",
"listagg",
"nested_update",
- "collect"
+ "collect",
+ "merge_map_with_keytime",
])
_FIELDS_PREFIX = "fields."
_FIELD_SEQUENCE_GROUP_SUFFIX = ".sequence-group"
@@ -211,7 +212,7 @@ def check_supported(table) -> None:
"(aggregation.remove-record-on-delete, "
"fields.<f>.ignore-retract) "
"and other aggregators (product / listagg / collect / "
- "merge_map* / nested_update* / theta_sketch / "
+ "nested_update* / theta_sketch / "
"hll_sketch / roaring_bitmap_*) are not yet supported. "
"Open an issue to track support.".format(
", ".join(sorted(unsupported)),
diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
index 060082c1cc..321a70daca 100644
--- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
+++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
@@ -39,7 +39,7 @@ from pypaimon.common.options import CoreOptions
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
+from pypaimon.schema.data_types import AtomicType, DataType, ArrayType,
RowType, MapType
from pypaimon.table.row.internal_row import InternalRow
# aggregator input type hints variables
@@ -61,6 +61,7 @@ NAME_BOOL_AND = "bool_and"
NAME_LISTAGG = "listagg"
NAME_NESTED_UPDATE = "nested_update"
NAME_COLLECT = "collect"
+NAME_MERGE_MAP_WITH_KEYTIME = "merge_map_with_keytime"
# Base SQL type names treated as numeric for sum/product-style
@@ -730,6 +731,126 @@ class FieldNestedUpdateAgg(FieldAggregator):
)
+class FieldMergeMapWithKeyTimeAgg(FieldAggregator):
+ """
+ Aggregator for merging MAP values with key and timestamp.
+
+ The input field type must be MAP with ROW values. Each ROW value must
+ contain a timestamp field, which is used to resolve conflicts when the
+ same key appears multiple times.
+
+ For the same key:
+ - A value with a newer timestamp replaces the existing value.
+ - A value with a null timestamp is ignored.
+ - A null value removes the corresponding key from the map.
+ """
+ def __init__(
+ self,
+ name: str,
+ field_type: DataType,
+ field_name: str,
+ options: CoreOptions,
+ ):
+ super().__init__(name, field_type)
+ if not isinstance(field_type, MapType):
+ raise ValueError(
+ "Data type for field '{}' must be 'MAP' but was
'{}'".format(field_name, field_type)
+ )
+ if not isinstance(field_type.value, RowType):
+ raise ValueError(
+ "Value type of MAP for field '{}' must be 'ROW' but was
'{}'".format(field_name, field_type.value)
+ )
+ if len(field_type.value.fields) < 2:
+ raise ValueError(
+ "ROW type for field '{}' must have at least 2 fields, but
found {}".format(
+ field_name, len(field_type.value.fields)
+ )
+ )
+
+ self._resolve_ts_field_index(field_type.value, options, field_name)
+
+ def agg(self, accumulator: Any, input_field: Any) -> Any:
+ if accumulator is None or input_field is None:
+ return input_field if accumulator is None else accumulator
+
+ result_map = {}
+ self._put_to_map(result_map, accumulator)
+ self._merge_input_map(result_map, input_field)
+
+ return result_map
+
+ def _resolve_ts_field_index(self, row_type: RowType, options: CoreOptions,
field_name: str) -> None:
+ ts_field_name = options.field_merge_map_ts_field(field_name)
+ if ts_field_name is None:
+ # default to the last field
+ ts_field_index = len(row_type.fields) - 1
+ ts_field_name = row_type.fields[ts_field_index].name
+ else:
+ ts_field_index = row_type.get_field_index(ts_field_name)
+ if ts_field_index < 0:
+ raise ValueError(
+ "Timestamp field '{}' not found in ROW type for field
'{}'. Available fields: {}".format(
+ ts_field_name, field_name, [field.name for field in
row_type.fields]
+ )
+ )
+
+ self.ts_field_name = ts_field_name
+ self.ts_field_index = ts_field_index
+
+ def _put_to_map(self, maps: Dict[Any, Any], input_field: Any):
+ if isinstance(input_field, dict):
+ maps.update(input_field)
+ elif isinstance(input_field, list):
+ tmp_map = {}
+ for item in input_field:
+ if not isinstance(item, dict):
+ raise TypeError(
+ "list element must be dict, got {}".format(type(item))
+ )
+ tmp_map[item['key']] = item['value']
+
+ maps.update(tmp_map)
+ else:
+ raise TypeError(
+ "input_field must be dict or list[dict], got
{}".format(type(input_field))
+ )
+
+ def _merge_input_map(self, result_map: Dict[Any, Any], input_field: Any):
+ input_map = {}
+ self._put_to_map(input_map, input_field)
+
+ for key, new_row in input_map.items():
+ if new_row is None:
+ result_map.pop(key, None)
+ continue
+
+ new_ts = self._get_ts_field(new_row)
+ if new_ts is None:
+ continue
+
+ existing_row = result_map.get(key)
+ if existing_row is None:
+ result_map[key] = new_row
+ continue
+
+ existing_ts = self._get_ts_field(existing_row)
+ if existing_ts is None or new_ts > existing_ts:
+ result_map[key] = new_row
+
+ def _get_ts_field(self, input_row: Any) -> Any:
+ if isinstance(input_row, dict):
+ return input_row.get(self.ts_field_name)
+
+ if isinstance(input_row, InternalRow):
+ return input_row.get_field(self.ts_field_index)
+
+ raise TypeError(
+ "Unsupported row type '{}', expected InternalRow or dict.".format(
+ type(input_row).__name__
+ )
+ )
+
+
# ---------------------------------------------------------------------------
# Registration. Each builder binds an identifier to a factory that
# optionally validates the column DataType before constructing the
@@ -809,3 +930,6 @@ register_aggregator(
register_aggregator(
NAME_COLLECT, _build_field_options(FieldCollectAgg, NAME_COLLECT)
)
+register_aggregator(
+ NAME_MERGE_MAP_WITH_KEYTIME,
_build_field_options(FieldMergeMapWithKeyTimeAgg, NAME_MERGE_MAP_WITH_KEYTIME)
+)
diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py
b/paimon-python/pypaimon/tests/test_field_aggregators.py
index 2f4d07c39d..c6a188d90a 100644
--- a/paimon-python/pypaimon/tests/test_field_aggregators.py
+++ b/paimon-python/pypaimon/tests/test_field_aggregators.py
@@ -47,6 +47,7 @@ from pypaimon.read.reader.aggregate.aggregators import (
FieldListaggAgg,
FieldNestedUpdateAgg,
FieldCollectAgg,
+ FieldMergeMapWithKeyTimeAgg,
)
from pypaimon.schema.data_types import AtomicType, DataField, RowType,
ArrayType, MapType
from pypaimon.table.row.generic_row import GenericRow
@@ -1765,6 +1766,160 @@ class FieldNestedUpdateAggTest(unittest.TestCase):
self.assertCountEqual(accumulator, [self.row(0, 0, "A", 1), ])
+class FieldMergeMapWithKeyTimeAggTest(unittest.TestCase):
+
+ DEFAULT_FIELDS = [
+ DataField(0, "actual_value", AtomicType("STRING")),
+ DataField(1, "dbsync_ts", AtomicType("STRING")),
+ ]
+
+ def _make(self, row_type: RowType = None, options: CoreOptions = None):
+ if options is None:
+ options = CoreOptions(Options.from_none())
+
+ if not row_type:
+ row_type = RowType(True, self.DEFAULT_FIELDS)
+
+ return create_field_aggregator(
+ MapType(True, AtomicType("STRING"), row_type),
+ "field0",
+ "merge_map_with_keytime",
+ 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_merge_map_with_key_time_dict_row_default_ts(self):
+ agg = self._make()
+
+ result = agg.agg(
+ None,
+ {"key1": {"actual_value": "A", "dbsync_ts": "100"}},
+ )
+
+ self.assertEqual(
+ result,
+ {"key1": {"actual_value": "A", "dbsync_ts": "100"}},
+ )
+
+ result = agg.agg(
+ result,
+ {
+ "key1": {"actual_value": "A+", "dbsync_ts": "110"},
+ "key2": {"actual_value": "B", "dbsync_ts": "100"}
+ },
+ )
+
+ self.assertEqual(
+ result,
+ {
+ "key1": {"actual_value": "A+", "dbsync_ts": "110"},
+ "key2": {"actual_value": "B", "dbsync_ts": "100"}
+ },
+ )
+
+ def test_field_merge_map_with_key_time_agg_using_default_ts(self):
+ agg = self._make()
+ self.assertIsInstance(agg, FieldMergeMapWithKeyTimeAgg)
+
+ self.assertIsNone(agg.agg(None, None))
+
+ acc = agg.agg(
+ None,
+ {
+ "key1": self.row("A", "17682882903686900100"),
+ "key2": self.row("B", "17682882903686900100"),
+ },
+ )
+ self.assertEqual(
+ acc,
+ {
+ "key1": self.row("A", "17682882903686900100"),
+ "key2": self.row("B", "17682882903686900100"),
+ },
+ )
+
+ # newer timestamp replaces old value
+ acc = agg.agg(
+ acc,
+ {
+ "key1": self.row("A1", "17682882903686900200"),
+ "key3": self.row("C", "17682882903686900200"),
+ },
+ )
+ self.assertEqual(
+ acc,
+ {
+ "key1": self.row("A1", "17682882903686900200"),
+ "key2": self.row("B", "17682882903686900100"),
+ "key3": self.row("C", "17682882903686900200"),
+ },
+ )
+
+ # older timestamp keeps existing value
+ acc = agg.agg(
+ acc,
+ {
+ "key2": self.row("B2", "17682882903686900050"),
+ },
+ )
+ self.assertEqual(
+ acc["key2"],
+ self.row("B", "17682882903686900100"),
+ )
+
+ def test_field_merge_map_with_key_time_agg_using_event_time(self):
+ fields = [
+ DataField(0, "actual_value", AtomicType("STRING")),
+ DataField(1, "other_field", AtomicType("STRING")),
+ DataField(2, "event_time", AtomicType("STRING")),
+ DataField(3, "dbsync_ts", AtomicType("STRING")),
+ ]
+ agg = self._make(
+ row_type=RowType(True, fields),
+ options=CoreOptions(Options(
+ {"fields.field0.ts-field": "event_time"}
+ ))
+ )
+
+ acc = agg.agg(None, {
+ "key1": self.row("A", "other_a", "2026-07-19 10:00:00",
"2026-07-19 12:00:00")
+ })
+
+ acc = agg.agg(acc, {
+ "key1": self.row("A1", "other_a1", "2026-07-19 11:00:00",
"2026-07-19 10:00:00")
+ })
+ self.assertEqual(
+ acc["key1"],
+ self.row("A1", "other_a1", "2026-07-19 11:00:00", "2026-07-19
10:00:00"),
+ )
+
+ acc = agg.agg(acc, {
+ "key1": self.row("A2", "other_a2", "2026-07-19 09:00:00",
"2026-07-19 13:00:00")
+ })
+ self.assertEqual(
+ acc["key1"],
+ self.row("A1", "other_a1", "2026-07-19 11:00:00", "2026-07-19
10:00:00"),
+ )
+
+ def test_field_merge_map_with_key_time_agg_retract(self):
+ agg = self._make()
+
+ with self.assertRaises(NotImplementedError):
+ agg.retract(
+ {
+ "key1": self.row("A", "17682882903686900100"),
+ "key2": self.row("B", "17682882903686900100"),
+ },
+ {
+ "key1": self.row("A", "17682882903686900100"),
+ },
+ )
+
+
class RegistrationTest(unittest.TestCase):
"""Sanity check that all 10 expected aggregators (the primary-key
placeholder plus 9 value aggregators) are registered when the
