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 7234e4c34e [python] Introduce 'nested_update' aggregator function
(#8577)
7234e4c34e is described below
commit 7234e4c34e9a705af66a816f5c859afd55b8f7f4
Author: AuroraVoyage <[email protected]>
AuthorDate: Sat Jul 18 13:35:37 2026 +0800
[python] Introduce 'nested_update' aggregator function (#8577)
---
.../pypaimon/common/options/core_options.py | 55 ++
.../pypaimon/read/merge_engine_support.py | 28 +-
.../pypaimon/read/reader/aggregate/aggregators.py | 378 +++++++-
paimon-python/pypaimon/schema/data_types.py | 6 +
paimon-python/pypaimon/table/row/projected_row.py | 12 +-
.../pypaimon/tests/test_field_aggregators.py | 1015 +++++++++++++++++++-
.../pypaimon/tests/test_sequence_field_read.py | 6 +-
7 files changed, 1467 insertions(+), 33 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 471a080ddf..4d2e6a7bb8 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -98,6 +98,13 @@ class GlobalIndexSearchMode(str, Enum):
DETAIL = "detail"
+class NestedKeyNullStrategy(str, Enum):
+ """Strategy for handling rows whose nested-key contains null values."""
+ MERGE = "merge"
+ IGNORE = "ignore"
+ ERROR = "error"
+
+
class CoreOptions:
"""Core options for Paimon tables."""
@@ -142,6 +149,10 @@ class CoreOptions:
FIELDS_PREFIX = "fields"
DISTINCT = "distinct"
LIST_AGG_DELIMITER = "list-agg-delimiter"
+ NESTED_KEY = "nested-key"
+ NESTED_KEY_NULL_STRATEGY = "nested-key-null-strategy"
+ NESTED_SEQUENCE_FIELD = "nested-sequence-field"
+ COUNT_LIMIT = "count-limit"
# Basic options
AUTO_CREATE: ConfigOption[bool] = (
@@ -1513,6 +1524,50 @@ class CoreOptions:
.default_value(False)
)
+ def field_nested_update_agg_nested_key(self, field_name: str) -> List[str]:
+ key_string = self.options.get(
+ ConfigOptions.key(
+
f'{CoreOptions.FIELDS_PREFIX}.{field_name}.{CoreOptions.NESTED_KEY}'
+ )
+ .string_type()
+ .no_default_value()
+ )
+
+ if not key_string:
+ return []
+ return list(map(str.strip, key_string.split(",")))
+
+ def field_nested_update_agg_nested_sequence_field(self, field_name: str)
-> List[str]:
+ key_string = self.options.get(
+ ConfigOptions.key(
+
f'{CoreOptions.FIELDS_PREFIX}.{field_name}.{CoreOptions.NESTED_SEQUENCE_FIELD}'
+ )
+ .string_type()
+ .no_default_value()
+ )
+
+ if not key_string:
+ return []
+ return list(map(str.strip, key_string.split(",")))
+
+ def field_nested_update_agg_nested_key_null_strategy(self, field_name:
str) -> NestedKeyNullStrategy:
+ return self.options.get(
+ ConfigOptions.key(
+
f'{CoreOptions.FIELDS_PREFIX}.{field_name}.{CoreOptions.NESTED_KEY_NULL_STRATEGY}'
+ )
+ .enum_type(NestedKeyNullStrategy)
+ .default_value(NestedKeyNullStrategy.MERGE)
+ )
+
+ def field_nested_update_agg_count_limit(self, field_name: str) -> int:
+ return self.options.get(
+ ConfigOptions.key(
+
f'{CoreOptions.FIELDS_PREFIX}.{field_name}.{CoreOptions.COUNT_LIMIT}'
+ )
+ .int_type()
+ .default_value(2147483647) # Integer.MAX_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 998dd72e80..cefcee2e67 100644
--- a/paimon-python/pypaimon/read/merge_engine_support.py
+++ b/paimon-python/pypaimon/read/merge_engine_support.py
@@ -63,31 +63,15 @@ _AGGREGATION_SUPPORTED_AGG_FUNCS = frozenset([
"sum", "max", "min",
"bool_or", "bool_and",
"listagg",
+ "nested_update",
])
_FIELDS_PREFIX = "fields."
_FIELD_SEQUENCE_GROUP_SUFFIX = ".sequence-group"
_FIELD_AGGREGATE_FUNCTION_SUFFIX = ".aggregate-function"
_FIELD_IGNORE_RETRACT_SUFFIX = ".ignore-retract"
-_FIELD_NESTED_SEQUENCE_SUFFIX = ".nested-sequence-field"
_DEFAULT_AGGREGATE_FUNCTION_KEY = "fields.default-aggregate-function"
-def _nested_sequence_field_options(table) -> Set[str]:
- """Option keys configuring ``nested-sequence-field`` (a per-field
- nested sequence ordering distinct from the top-level
- ``sequence.field``). pypaimon implements top-level ``sequence.field``
- but not nested sequence fields, so reject them on every PK engine
- rather than silently ignoring them.
- """
- flagged: Set[str] = set()
- raw = table.options.options.to_map()
- for key in raw:
- if key.startswith(_FIELDS_PREFIX) and key.endswith(
- _FIELD_NESTED_SEQUENCE_SUFFIX):
- flagged.add(key)
- return flagged
-
-
def _unsupported_sequence_fields(table) -> Set[str]:
"""Configured ``sequence.field`` names whose type pypaimon cannot order.
Java's ``UserDefinedSeqComparator`` delegates to ``RecordComparator``
@@ -175,16 +159,6 @@ def check_supported(table) -> None:
"""
if not table.is_primary_key_table:
return
- # ``nested-sequence-field`` is unimplemented on every engine; reject it
- # before per-engine dispatch so it can't be silently ignored by the
- # top-level ``sequence.field`` comparator.
- nested_seq = _nested_sequence_field_options(table)
- if nested_seq:
- raise NotImplementedError(
- "nested-sequence-field is not implemented in pypaimon yet: {}. "
- "Top-level 'sequence.field' is supported; open an issue to track "
- "nested sequence field support.".format(",
".join(sorted(nested_seq)))
- )
# ``sequence.field`` validity is engine-independent in Java
# (SchemaValidation.validateSequenceField). pypaimon has no
# schema-creation validation, so enforce the same invariants here on
diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
index 13ef86833a..11b182edaf 100644
--- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
+++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
@@ -33,12 +33,17 @@ the registry will report them as unsupported so users see a
clear
error rather than a silent fallback.
"""
-from typing import Any
+from typing import Any, List, Dict, Optional, Tuple, Union
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
+from pypaimon.schema.data_types import AtomicType, DataType, ArrayType, RowType
+from pypaimon.table.row.internal_row import InternalRow
+
+# aggregator input type hints variables
+Record = Union[InternalRow, Dict[str, Any]]
# Aggregator identifiers exposed via ``fields.<name>.aggregate-function``
@@ -54,6 +59,7 @@ NAME_MIN = "min"
NAME_BOOL_OR = "bool_or"
NAME_BOOL_AND = "bool_and"
NAME_LISTAGG = "listagg"
+NAME_NESTED_UPDATE = "nested_update"
# Base SQL type names treated as numeric for sum/product-style
@@ -96,6 +102,24 @@ def _check_boolean(name: str, field_type: DataType) -> None:
)
+def _check_array_row(name: str, field_type: DataType) -> ArrayType:
+ """Check field_type is ARRAY<ROW> and return the ArrayType."""
+
+ if not isinstance(field_type, ArrayType):
+ raise ValueError(
+ "Data type for '{}' column must be 'ARRAY<ROW>' but was '{}'."
+ .format(name, field_type)
+ )
+
+ if not isinstance(field_type.element, RowType):
+ raise ValueError(
+ "Data type for '{}' column must be 'ARRAY<ROW>' but was '{}'."
+ .format(name, field_type)
+ )
+
+ return field_type
+
+
def is_blank(s: str) -> bool:
if s is None:
return True
@@ -107,6 +131,144 @@ def is_blank(s: str) -> bool:
return True
+def _compare_objects(left: Any, right: Any) -> int:
+ """
+ Compare two comparable Python objects using Paimon's ordering.
+
+ Nulls are ordered before non-null values (Nulls First).
+ """
+
+ if left is None:
+ return 0 if right is None else -1
+
+ if right is None:
+ return 1
+
+ return (left > right) - (left < right)
+
+
+def _compare_tuple(left: Tuple[Any, ...], right: Tuple[Any, ...]) -> int:
+ """
+ Lexicographical comparison with Nulls First.
+ """
+
+ for l, r in zip(left, right):
+ cmp = _compare_objects(l, r)
+ if cmp != 0:
+ return cmp
+
+ if len(left) == len(right):
+ return 0
+
+ return -1 if len(left) < len(right) else 1
+
+
+def _row_equals(left: Record, right: Record) -> bool:
+ """
+ Compare two records for equality.
+
+ Supports both ``InternalRow`` and ``dict`` representations.
+ """
+ if isinstance(left, dict) and isinstance(right, dict):
+ return left == right
+
+ if isinstance(left, InternalRow) and isinstance(right, InternalRow):
+ if len(left) != len(right):
+ return False
+
+ for i in range(len(left)):
+ if left.get_field(i) != right.get_field(i):
+ return False
+
+ return True
+
+ raise TypeError(
+ "Cannot compare records of different or unsupported types: "
+ f"{type(left).__name__} and {type(right).__name__}. "
+ "Expected both records to be either InternalRow or dict."
+ )
+
+
+class FieldProjection:
+ """
+ Extracts selected fields from a row.
+
+ This helper is primarily used by nested aggregators (e.g.
+ ``nested_update`` and ``nested_partial_update``) to retrieve
+ configured fields from nested rows.
+
+ It supports both row representations currently used by pypaimon:
+
+ * :class:`InternalRow` - fields are accessed by ordinal position.
+ * ``dict`` - fields are accessed by field name (used by the
+ PyArrow -> Polars read path).
+
+ The extracted values are returned as a tuple so they can be used
+ directly as comparison keys, dictionary keys, or sequence values.
+ """
+
+ def __init__(
+ self,
+ index_mapping: List[int],
+ field_names: List[str],
+ ):
+ """
+ Create a FieldProjection.
+
+ Args:
+ index_mapping: Ordinal positions of the selected fields.
+ field_names: Corresponding field names. Used when the input
+ row is represented as a dict.
+ """
+ if len(index_mapping) != len(field_names):
+ raise ValueError(
+ "index_mapping and field_names must have the same length."
+ )
+
+ self.index_mapping = index_mapping
+ self.field_names = field_names
+
+ @staticmethod
+ def from_fields(
+ index_mapping: List[int],
+ field_names: List[str],
+ ) -> "FieldProjection":
+ """Create a FieldProjection from field indexes and names."""
+ return FieldProjection(index_mapping, field_names)
+
+ def apply(self, element: Record) -> Tuple[Any, ...]:
+ """
+ Return the projected fields as a tuple.
+
+ Args:
+ element: Either an ``InternalRow`` or a ``dict``.
+
+ Returns:
+ Tuple containing projected field values.
+
+ Raises:
+ TypeError: If the row type is unsupported.
+ """
+
+ if isinstance(element, InternalRow):
+ return tuple(
+ element.get_field(index) if index >= 0 else None
+ for index in self.index_mapping
+ )
+
+ if isinstance(element, dict):
+ return tuple(
+ element.get(name)
+ for name in self.field_names
+ )
+
+ raise TypeError(
+ "Unsupported row type '{}', expected InternalRow or dict.".format(
+ type(element).__name__
+ )
+ )
+
+
# ---------------------------------------------------------------------------
# Aggregator classes
# ---------------------------------------------------------------------------
@@ -293,6 +455,215 @@ class FieldListaggAgg(FieldAggregator):
return self.delimiter.join(result)
+class FieldNestedUpdateAgg(FieldAggregator):
+ """
+ Used to 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)
+ self._check_option_dependencies(options, field_name)
+
+ super().__init__(name, field_type)
+
+ nested_type: RowType = field_type.element
+
+ self.nested_key =
options.field_nested_update_agg_nested_key(field_name)
+ self.nested_key_null_strategy =
options.field_nested_update_agg_nested_key_null_strategy(field_name)
+ self.nested_sequence_field =
options.field_nested_update_agg_nested_sequence_field(field_name)
+ self.count_limit =
options.field_nested_update_agg_count_limit(field_name)
+
+ if self.nested_key:
+ self.key_projection = FieldProjection.from_fields(
+ [nested_type.get_field_index(name) for name in
self.nested_key],
+ self.nested_key
+ )
+ else:
+ self.key_projection = None
+
+ if self.nested_sequence_field:
+ self.sequence_projection = FieldProjection.from_fields(
+ [nested_type.get_field_index(name) for name in
self.nested_sequence_field],
+ self.nested_sequence_field
+ )
+ self.has_sequence_field = True
+ else:
+ self.sequence_projection = None
+ self.has_sequence_field = False
+
+ def agg(self, accumulator: Any, input_field: Any) -> Any:
+ if input_field is None:
+ return accumulator
+
+ if self.key_projection is None:
+ if accumulator is None:
+ rows: List[Record] = []
+ self._add_non_null_rows(input_field, rows, self.count_limit)
+ return rows
+
+ if len(accumulator) >= self.count_limit:
+ return accumulator
+
+ remain_count = self.count_limit - len(accumulator)
+ rows: List[Record] = []
+ self._add_non_null_rows(accumulator, rows)
+ self._add_non_null_rows(input_field, rows, remain_count)
+ return rows
+ else:
+ row_map: Dict[Tuple[Any, ...], Record] = {}
+ if accumulator is not None:
+ self._add_nested_rows(accumulator, row_map, False)
+ self._add_nested_rows(input_field, row_map, True)
+ return list(row_map.values())
+
+ def retract(self, accumulator: Any, retract_field: Any) -> Any:
+ if accumulator is None or retract_field is None:
+ return accumulator
+
+ if self.key_projection is None:
+ rows: List[Record] = []
+ self._add_non_null_rows(accumulator, rows)
+ for retract_row in retract_field:
+ if retract_row is None:
+ continue
+ rows = [row for row in rows if not _row_equals(row,
retract_row)]
+ return rows
+ else:
+ row_map: Dict[Tuple[Any, ...], Record] = {}
+ for row in accumulator:
+ if row is None:
+ continue
+ key = self.key_projection.apply(row)
+ if not self._apply_nested_key_null_strategy(key):
+ continue
+ row_map[key] = row
+
+ for row in retract_field:
+ if row is None:
+ continue
+ key = self.key_projection.apply(row)
+ if not self._apply_nested_key_null_strategy(key):
+ continue
+ row_map.pop(key, None)
+ return list(row_map.values())
+
+ @staticmethod
+ def _check_option_dependencies(
+ options: CoreOptions,
+ field: str,
+ ) -> None:
+ nested_key = options.field_nested_update_agg_nested_key(field)
+ strategy_configured = options.options.contains_key(
+
f"{CoreOptions.FIELDS_PREFIX}.{field}.{CoreOptions.NESTED_KEY_NULL_STRATEGY}")
+
+ if strategy_configured and not nested_key:
+ raise ValueError(
+ "Option 'fields.<field-name>.nested-key-null-strategy' "
+ "requires 'fields.<field-name>.nested-key' to be configured."
+ )
+
+ if (
+ options.field_nested_update_agg_nested_sequence_field(field)
+ and not nested_key
+ ):
+ raise ValueError(
+ "Option 'fields.<field-name>.nested-sequence-field' "
+ "requires 'fields.<field-name>.nested-key' to be configured."
+ )
+
+ def _compare_sequence(self, new_row: Record, old_row: Record) -> int:
+ if not self.has_sequence_field:
+ raise ValueError(
+ "compare_sequence() called but no nested_sequence_field
configured."
+ )
+
+ new_seq_key = self.sequence_projection.apply(new_row)
+ old_seq_key = self.sequence_projection.apply(old_row)
+
+ return _compare_tuple(new_seq_key, old_seq_key)
+
+ def _add_non_null_rows(
+ self,
+ array: List[Record],
+ rows: List[Record],
+ remain_size: Optional[int] = None,
+ ) -> None:
+ """Append non-null rows from array.
+
+ If remain_size is specified, append at most remain_size rows.
+ """
+
+ count = 0
+
+ for row in array:
+ if row is None:
+ continue
+
+ if remain_size is not None and count >= remain_size:
+ break
+
+ rows.append(row)
+ count += 1
+
+ def _add_nested_rows(
+ self,
+ array: List[Record],
+ row_map: Dict[Tuple[Any, ...], Record],
+ limit_new_keys: bool,
+ ) -> None:
+ """Merge rows from ``array`` into ``rows`` using nested keys."""
+
+ if self.key_projection is None:
+ raise ValueError(
+ "key_projection should not be None when nested_key is
configured."
+ )
+
+ for row in array:
+ if row is None:
+ continue
+ key = self.key_projection.apply(row)
+ if not self._apply_nested_key_null_strategy(key):
+ continue
+
+ exists = row_map.get(key)
+ if exists is not None:
+ if not self.has_sequence_field or self._compare_sequence(row,
exists) >= 0:
+ row_map[key] = row
+ elif not limit_new_keys or len(row_map) < self.count_limit:
+ row_map[key] = 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
+ )
+ )
+
+
# ---------------------------------------------------------------------------
# Registration. Each builder binds an identifier to a factory that
# optionally validates the column DataType before constructing the
@@ -366,3 +737,6 @@ register_aggregator(
register_aggregator(
NAME_LISTAGG, _build_field_options(FieldListaggAgg, NAME_LISTAGG)
)
+register_aggregator(
+ NAME_NESTED_UPDATE, _build_field_options(FieldNestedUpdateAgg,
NAME_NESTED_UPDATE)
+)
diff --git a/paimon-python/pypaimon/schema/data_types.py
b/paimon-python/pypaimon/schema/data_types.py
index f525a6020a..c645460017 100755
--- a/paimon-python/pypaimon/schema/data_types.py
+++ b/paimon-python/pypaimon/schema/data_types.py
@@ -394,6 +394,12 @@ class RowType(DataType):
null_suffix = "" if self.nullable else " NOT NULL"
return "ROW<{}>{}".format(', '.join(field_strs), null_suffix)
+ def get_field_index(self, field_name: str) -> int:
+ for index, field in enumerate(self.fields):
+ if field.name == field_name:
+ return index
+ raise ValueError("Field {} not found in {}".format(field_name, self))
+
def reassign_field_id(data_type: DataType, field_id: "AtomicInteger") ->
DataType:
"""Return a copy of *data_type* with every nested field id reassigned from
diff --git a/paimon-python/pypaimon/table/row/projected_row.py
b/paimon-python/pypaimon/table/row/projected_row.py
index 610d0648ee..5fc66b792e 100644
--- a/paimon-python/pypaimon/table/row/projected_row.py
+++ b/paimon-python/pypaimon/table/row/projected_row.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-from typing import Any, List
+from typing import Any, List, Tuple
from pypaimon.table.row.internal_row import InternalRow
from pypaimon.table.row.row_kind import RowKind
@@ -84,3 +84,13 @@ class ProjectedRow(InternalRow):
ProjectedRow instance
"""
return ProjectedRow(projection)
+
+ def to_tuple(self) -> Tuple[Any, ...]:
+ assert isinstance(self.row, InternalRow), (
+ f"Expected InternalRow, but got {type(self.row).__name__}"
+ )
+ return tuple(
+ self.row.get_field(index)
+ if index >= 0 else None
+ for index in self.index_mapping
+ )
diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py
b/paimon-python/pypaimon/tests/test_field_aggregators.py
index b995b1c4bc..9c9d5876fe 100644
--- a/paimon-python/pypaimon/tests/test_field_aggregators.py
+++ b/paimon-python/pypaimon/tests/test_field_aggregators.py
@@ -28,8 +28,10 @@ import datetime
import unittest
from decimal import Decimal
from functools import reduce
+from typing import List
from pypaimon.common.options import CoreOptions, Options
+from pypaimon.data import Timestamp
from pypaimon.read.reader.aggregate import create_field_aggregator
from pypaimon.read.reader.aggregate.aggregators import (
FieldBoolAndAgg,
@@ -43,8 +45,11 @@ from pypaimon.read.reader.aggregate.aggregators import (
FieldPrimaryKeyAgg,
FieldSumAgg,
FieldListaggAgg,
+ FieldNestedUpdateAgg,
)
-from pypaimon.schema.data_types import AtomicType
+from pypaimon.schema.data_types import AtomicType, DataField, RowType,
ArrayType
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.internal_row import InternalRow
def _make(identifier, sql_type, options: CoreOptions = None):
@@ -545,6 +550,1014 @@ class FieldListaggAggTest(unittest.TestCase):
)
+class FieldNestedUpdateAggTest(unittest.TestCase):
+ IDENTIFIER = "nested_update"
+
+ DEFAULT_FIELDS = [
+ DataField(0, "k0", AtomicType("INT")),
+ DataField(1, "k1", AtomicType("INT")),
+ DataField(2, "v", AtomicType("STRING")),
+ ]
+
+ SEQUENCE_FIELDS = [
+ DataField(0, "k0", AtomicType("INT")),
+ DataField(1, "k1", AtomicType("INT")),
+ DataField(2, "v", AtomicType("STRING")),
+ DataField(3, "seq", AtomicType("INT")),
+ ]
+
+ 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_update(self):
+ agg = self._make(
+ self._make_data_type(),
+ CoreOptions(Options(
+ {
+ 'fields.field0.nested-key': 'k0,k1'
+ }
+ ))
+ )
+ self.assertIsInstance(agg, FieldNestedUpdateAgg)
+
+ accumulator = None
+
+ current: InternalRow = self.row(0, 0, "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, 0, "A"),
+ self.row(0, 1, "B"),
+ ])
+
+ current = self.row(0, 1, "b")
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 0, "A"),
+ self.row(0, 1, "b"),
+ ])
+
+ accumulator = agg.retract(accumulator, [self.row(0, 1, "b")])
+ self.assertCountEqual(accumulator, [self.row(0, 0, "A")])
+
+ def test_field_nested_append(self):
+ agg = self._make(self._make_data_type())
+
+ accumulator = None
+
+ current = self.row(0, 1, "B")
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [self.row(0, 1, "B")])
+
+ current = self.row(0, 1, "b")
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B"),
+ self.row(0, 1, "b"),
+ ])
+
+ accumulator = agg.retract(accumulator, [self.row(0, 1, "b")])
+ self.assertCountEqual(accumulator, [self.row(0, 1, "B")])
+
+ def test_field_nested_update_with_sequence_field_prerequisite(self):
+ # nested-sequence-field without nested-key should fail
+ with self.assertRaisesRegex(
+ ValueError,
+ "Option 'fields.<field-name>.nested-sequence-field' requires "
+ "'fields.<field-name>.nested-key' to be configured.",
+ ):
+ self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-sequence-field": "seq"
+ }
+ )
+ )
+ )
+
+ seq_agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-sequence-field": "seq",
+ }
+ )
+ )
+ )
+
+ self.assertIsInstance(seq_agg, FieldNestedUpdateAgg)
+
+ accumulator = None
+
+ accumulator = seq_agg.agg(accumulator, [self.row(0, 1, "A", 1)])
+ accumulator = seq_agg.agg(accumulator, [self.row(0, 1, "B", 2)])
+ self.assertCountEqual(accumulator, [self.row(0, 1, "B", 2)])
+
+ # older sequence value should be ignored
+ accumulator = seq_agg.agg(accumulator, [self.row(0, 1, "b_Late", 1)])
+ self.assertCountEqual(accumulator, [self.row(0, 1, "B", 2)])
+
+ def
test_field_nested_update_with_nested_key_null_strategy_prerequisite(self):
+ # nested-key-null-strategy requires nested-key
+ with self.assertRaisesRegex(
+ ValueError,
+ "Option 'fields.<field-name>.nested-key-null-strategy'
requires "
+ "'fields.<field-name>.nested-key' to be configured.",
+ ):
+ self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key-null-strategy": "merge"
+ }
+ )
+ )
+ )
+
+ # merge strategy
+ merge_agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-key-null-strategy": "merge",
+ }
+ )
+ )
+ )
+
+ merge_accumulator = None
+
+ merge_accumulator = merge_agg.agg(merge_accumulator, [self.row(0,
None, "A", 1)])
+ self.assertCountEqual(merge_accumulator, [self.row(0, None, "A", 1)])
+
+ # ignore strategy
+ ignore_agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-key-null-strategy": "ignore",
+ }
+ )
+ )
+ )
+
+ ignore_accumulator = None
+
+ ignore_accumulator = ignore_agg.agg(ignore_accumulator, [self.row(0,
1, "A", 1)])
+ ignore_accumulator = ignore_agg.agg(ignore_accumulator, [self.row(0,
None, "B", 2)])
+ self.assertCountEqual(ignore_accumulator, [self.row(0, 1, "A", 1)])
+
+ # error strategy
+ error_agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-key-null-strategy": "error",
+ }
+ )
+ )
+ )
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ error_agg.agg(None, [self.row(0, None, "B", 2)])
+
+ def test_field_nested_append_with_count_limit(self):
+ agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.count-limit": "2"
+ }
+ )
+ )
+ )
+
+ accumulator = None
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B")])
+ self.assertCountEqual(accumulator, [self.row(0, 1, "B")])
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "b")])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B"),
+ self.row(0, 1, "b"),
+ ])
+
+ # count limit = 2
+ # third element should be dropped
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "C")])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B"),
+ self.row(0, 1, "b"),
+ ])
+
+ def test_field_nested_append_with_count_limit_on_first_input_array(self):
+ agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.count-limit": "2"
+ }
+ )
+ )
+ )
+
+ accumulator = agg.agg(
+ None,
+ [
+ self.row(0, 1, "B"),
+ None,
+ self.row(0, 1, "b"),
+ self.row(0, 1, "C"),
+ ],
+ )
+
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B"),
+ self.row(0, 1, "b"),
+ ])
+
+ def
test_field_nested_update_with_count_limit_updates_existing_key_at_limit_without_sequence(self):
+ agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.count-limit": "2"
+ }
+ )
+ )
+ )
+
+ accumulator = None
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B")])
+ accumulator = agg.agg(accumulator, [self.row(1, 2, "C")])
+
+ # update existing key when count limit reached
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B_updated")])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated"),
+ self.row(1, 2, "C"),
+ ])
+
+ # new key exceeds limit, should be ignored
+ accumulator = agg.agg(accumulator, [self.row(2, 3, "D")])
+
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated"),
+ self.row(1, 2, "C"),
+ ])
+
+ def
test_field_nested_update_with_count_limit_on_first_input_array_without_sequence(self):
+ agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.count-limit": "2",
+ }
+ )
+ )
+ )
+
+ accumulator = agg.agg(
+ None,
+ [
+ self.row(0, 1, "B"),
+ self.row(1, 2, "C"),
+ self.row(2, 3, "D"),
+ self.row(0, 1, "B_updated"),
+ ],
+ )
+
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated"),
+ self.row(1, 2, "C"),
+ ])
+
+ def test_field_nested_update_with_sequence_field(self):
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-sequence-field": "seq",
+ }
+ )
+ )
+ )
+
+ accumulator = None
+
+ current = self.row(0, 0, "A", 1)
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [current])
+
+ current = self.row(0, 1, "B", 2)
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 0, "A", 1),
+ self.row(0, 1, "B", 2),
+ ])
+
+ current = self.row(0, 1, "b", 3)
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 0, "A", 1),
+ self.row(0, 1, "b", 3),
+ ])
+
+ # lower sequence should be ignored
+ current = self.row(0, 1, "B_late", 2)
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 0, "A", 1),
+ self.row(0, 1, "b", 3),
+ ])
+
+ accumulator = agg.retract(accumulator, [self.row(0, 1, "b", 3)])
+ self.assertCountEqual(accumulator, [self.row(0, 0, "A", 1), ])
+
+ def test_field_nested_update_with_multiple_sequence_fields(self):
+ fields = self.DEFAULT_FIELDS + [
+ DataField(3, "seq", AtomicType("INT")),
+ DataField(4, "ts", AtomicType("TIMESTAMP(3)"))
+ ]
+ agg = self._make(
+ self._make_data_type(fields=fields),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-sequence-field": "seq,ts",
+ }
+ )
+ )
+ )
+
+ accumulator = None
+
+ ts1 = Timestamp.from_epoch_millis(1000)
+ ts2 = Timestamp.from_epoch_millis(2000)
+ ts3 = Timestamp.from_epoch_millis(3000)
+
+ accumulator = agg.agg(accumulator, [self.row(1, 0, "A", 1, ts2)])
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B", 2, ts1)])
+ self.assertCountEqual(accumulator, [
+ self.row(1, 0, "A", 1, ts2),
+ self.row(0, 1, "B", 2, ts1),
+ ])
+
+ accumulator = agg.agg(accumulator, [self.row(1, 1, "C", 1, ts2)])
+ self.assertCountEqual(accumulator, [
+ self.row(1, 0, "A", 1, ts2),
+ self.row(0, 1, "B", 2, ts1),
+ self.row(1, 1, "C", 1, ts2),
+ ])
+
+ # smaller second sequence should be ignored
+ accumulator = agg.agg(accumulator, [self.row(1, 0,
"A_late_updated_by_ts", 1, ts1)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(1, 0, "A", 1, ts2),
+ self.row(0, 1, "B", 2, ts1),
+ self.row(1, 1, "C", 1, ts2),
+ ]
+ )
+
+ # same seq, larger ts should update
+ accumulator = agg.agg(accumulator, [self.row(1, 0, "A_updated_by_ts",
1, ts3)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(1, 0, "A_updated_by_ts", 1, ts3),
+ self.row(0, 1, "B", 2, ts1),
+ self.row(1, 1, "C", 1, ts2),
+ ]
+ )
+
+ # smaller first sequence ignored even with larger ts
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "b_ignored", 1,
ts3)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(1, 0, "A_updated_by_ts", 1, ts3),
+ self.row(0, 1, "B", 2, ts1),
+ self.row(1, 1, "C", 1, ts2),
+ ]
+ )
+
+ # same seq, larger ts
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B_updated_by_ts",
2, ts2)])
+
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(1, 0, "A_updated_by_ts", 1, ts3),
+ self.row(0, 1, "B_updated_by_ts", 2, ts2),
+ self.row(1, 1, "C", 1, ts2),
+ ]
+ )
+
+ # larger first sequence wins
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B_updated_by_seq",
3, ts1)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(1, 0, "A_updated_by_ts", 1, ts3),
+ self.row(0, 1, "B_updated_by_seq", 3, ts1),
+ self.row(1, 1, "C", 1, ts2),
+ ]
+ )
+
+ accumulator = agg.retract(accumulator, [self.row(0, 1,
"B_updated_by_seq", 3, ts1)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(1, 0, "A_updated_by_ts", 1, ts3),
+ self.row(1, 1, "C", 1, ts2),
+ ]
+ )
+
+ def
test_field_nested_update_with_count_limit_with_sequence_field_without_nested_key(self):
+ with self.assertRaisesRegex(
+ ValueError,
+ "Option 'fields.<field-name>.nested-sequence-field' requires "
+ "'fields.<field-name>.nested-key' to be configured.",
+ ):
+ self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-sequence-field": "seq",
+ "fields.field0.count-limit": "2",
+ }
+ )
+ )
+ )
+
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-sequence-field": "seq",
+ "fields.field0.count-limit": "2",
+ }
+ )
+ )
+ )
+
+ accumulator = None
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "A", 1)])
+ accumulator = agg.agg(accumulator, [self.row(0, 2, "B", 2)])
+ accumulator = agg.agg(accumulator, [self.row(0, 3, "C", 3)])
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "A_Update", 4)])
+ accumulator = agg.agg(accumulator, [self.row(0, 2, "B_Late", 1)])
+
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "A_Update", 4),
+ self.row(0, 2, "B", 2),
+ ])
+
+ def test_field_nested_update_with_count_limit_with_sequence_field(self):
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-sequence-field": "seq",
+ "fields.field0.count-limit": "2",
+ }
+ )
+ )
+ )
+
+ accumulator = None
+
+ current = self.row(0, 1, "B", 1)
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [self.row(0, 1, "B", 1)])
+
+ current = self.row(0, 1, "B_updated", 2)
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [self.row(0, 1, "B_updated", 2)])
+
+ current = self.row(1, 2, "C", 3)
+ accumulator = agg.agg(accumulator, [current])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated", 2),
+ self.row(1, 2, "C", 3),
+ ])
+
+ current = self.row(0, 3, "D", 4)
+ accumulator = agg.agg(accumulator, [current])
+
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated", 2),
+ self.row(1, 2, "C", 3),
+ ])
+
+ def
test_field_nested_update_with_count_limit_updates_existing_key_at_limit(self):
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-sequence-field": "seq",
+ "fields.field0.count-limit": "2",
+ }
+ )
+ )
+ )
+
+ accumulator = None
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B", 1)])
+ accumulator = agg.agg(accumulator, [self.row(1, 2, "C", 3)])
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B_updated", 4)])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated", 4),
+ self.row(1, 2, "C", 3),
+ ])
+
+ accumulator = agg.agg(accumulator, [self.row(2, 3, "D", 5)])
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated", 4),
+ self.row(1, 2, "C", 3),
+ ])
+
+ def
test_field_nested_update_with_count_limit_on_first_input_array_with_sequence(self):
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-sequence-field": "seq",
+ "fields.field0.count-limit": "2",
+ }
+ )
+ )
+ )
+
+ accumulator = agg.agg(
+ None,
+ [
+ self.row(0, 1, "B", 1),
+ self.row(1, 2, "C", 3),
+ self.row(2, 3, "D", 5),
+ self.row(0, 1, "B_updated", 4),
+ ],
+ )
+
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(0, 1, "B_updated", 4),
+ self.row(1, 2, "C", 3),
+ ],
+ )
+
+ def test_field_nested_update_when_nested_key_null_use_merge_strategy(self):
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-key-null-strategy": "merge",
+ }
+ )
+ )
+ )
+
+ current = self.row(0, None, "C", 3)
+ accumulator = agg.agg(None, [current])
+ self.assertCountEqual(accumulator, [current])
+
+ current = self.row(None, None, "D", 4)
+ accumulator = agg.agg(None, [current])
+ self.assertCountEqual(accumulator, [current])
+
+ accumulator = agg.agg(None, [self.row(0, 0, "A", 1)])
+ self.assertCountEqual(accumulator, [self.row(0, 0, "A", 1)])
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B", 2)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(0, 0, "A", 1),
+ self.row(0, 1, "B", 2),
+ ],
+ )
+
+ accumulator = agg.agg(accumulator, [self.row(0, None, "C", 3)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(0, 0, "A", 1),
+ self.row(0, 1, "B", 2),
+ self.row(0, None, "C", 3),
+ ],
+ )
+
+ accumulator = agg.agg(accumulator, [self.row(None, None, "D", 4)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(0, 0, "A", 1),
+ self.row(0, 1, "B", 2),
+ self.row(0, None, "C", 3),
+ self.row(None, None, "D", 4),
+ ],
+ )
+
+ def
test_field_nested_update_when_nested_key_null_use_ignore_strategy(self):
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-key-null-strategy": "ignore",
+ }
+ )
+ )
+ )
+
+ accumulator = agg.agg(None, [self.row(0, None, "C", 3)])
+ self.assertCountEqual(accumulator, [])
+
+ accumulator = agg.agg(None, [self.row(None, None, "D", 4)])
+ self.assertCountEqual(accumulator, [])
+
+ accumulator = agg.agg(None, [self.row(0, 0, "A", 1)])
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B", 2)])
+
+ accumulator = agg.agg(accumulator, [self.row(0, None, "C", 3)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(0, 0, "A", 1),
+ self.row(0, 1, "B", 2),
+ ],
+ )
+
+ accumulator = agg.agg(accumulator, [self.row(None, None, "D", 4)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(0, 0, "A", 1),
+ self.row(0, 1, "B", 2),
+ ],
+ )
+
+ def
test_field_nested_update_when_nested_key_null_use_throw_error_strategy(self):
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-key-null-strategy": "error",
+ }
+ )
+ )
+ )
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ agg.agg(None, [self.row(0, None, "C", 3)])
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ agg.agg(None, [self.row(None, None, "D", 4)])
+
+ accumulator = agg.agg(None, [self.row(0, 0, "A", 1)])
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B", 2)])
+ self.assertCountEqual(
+ accumulator,
+ [
+ self.row(0, 0, "A", 1),
+ self.row(0, 1, "B", 2),
+ ],
+ )
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ agg.agg(accumulator, [self.row(0, None, "C", 3)])
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ agg.agg(accumulator, [self.row(None, None, "D", 4)])
+
+ def
test_field_nested_update_with_count_limit_when_nested_key_null_use_merge_strategy(self):
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-sequence-field": "seq",
+ "fields.field0.nested-key-null-strategy": "merge",
+ "fields.field0.count-limit": "3",
+ }
+ )
+ )
+ )
+
+ accumulator = None
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B", 1)])
+ accumulator = agg.agg(accumulator, [self.row(None, 2, "NULL_2", 2)])
+ accumulator = agg.agg(accumulator, [self.row(None, None, "NULL_NULL",
3)])
+ accumulator = agg.agg(accumulator, [self.row(1, 2, "C", 5)])
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B_updated", 4)])
+
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated", 4),
+ self.row(None, 2, "NULL_2", 2),
+ self.row(None, None, "NULL_NULL", 3),
+ ])
+
+ def
test_field_nested_update_with_count_limit_when_nested_key_null_use_ignore_strategy(self):
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-sequence-field": "seq",
+ "fields.field0.nested-key-null-strategy": "ignore",
+ "fields.field0.count-limit": "3",
+ }
+ )
+ )
+ )
+
+ accumulator = None
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B", 1)])
+ accumulator = agg.agg(accumulator, [self.row(None, 2, "NULL_2", 2)])
+ accumulator = agg.agg(accumulator, [self.row(None, None, "NULL_NULL",
3)])
+ accumulator = agg.agg(accumulator, [self.row(1, 2, "C", 3)])
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B_updated", 4)])
+
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated", 4),
+ self.row(1, 2, "C", 3),
+ ])
+
+ accumulator = agg.agg(accumulator, [self.row(2, 3, "D", 5)])
+
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated", 4),
+ self.row(1, 2, "C", 3),
+ self.row(2, 3, "D", 5),
+ ])
+
+ def
test_field_nested_update_with_count_limit_when_nested_key_null_use_throw_error_strategy(self):
+ agg = self._make(
+ self._make_data_type(fields=self.SEQUENCE_FIELDS),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-sequence-field": "seq",
+ "fields.field0.nested-key-null-strategy": "error",
+ "fields.field0.count-limit": "3",
+ }
+ )
+ )
+ )
+
+ accumulator = None
+
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B", 1)])
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ agg.agg(accumulator, [self.row(None, 2, "NULL_2", 2)])
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ agg.agg(accumulator, [self.row(None, None, "NULL_NULL", 3)])
+
+ accumulator = agg.agg(accumulator, [self.row(1, 2, "C", 3)])
+ accumulator = agg.agg(accumulator, [self.row(0, 1, "B_updated", 4)])
+
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated", 4),
+ self.row(1, 2, "C", 3),
+ ])
+
+ accumulator = agg.agg(accumulator, [self.row(2, 3, "D", 5)])
+
+ self.assertCountEqual(accumulator, [
+ self.row(0, 1, "B_updated", 4),
+ self.row(1, 2, "C", 3),
+ self.row(2, 3, "D", 5),
+ ])
+
+ def
test_field_nested_update_retract_applies_nested_key_null_strategy_to_accumulator(self):
+ merge_agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ }
+ )
+ )
+ )
+
+ accumulator = None
+ accumulator = merge_agg.agg(accumulator, [self.row(0, None, "A")])
+ accumulator = merge_agg.agg(accumulator, [self.row(1, 0, "B")])
+ accumulator = merge_agg.agg(accumulator, [self.row(1, 1, "C")])
+
+ ignore_agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-key-null-strategy": "IGNORE",
+ }
+ )
+ )
+ )
+
+ result = ignore_agg.retract(accumulator, [self.row(1, 0, "B")])
+ self.assertCountEqual(result, [self.row(1, 1, "C")])
+
+ error_agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-key-null-strategy": "ERROR",
+ }
+ )
+ )
+ )
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ error_agg.retract(accumulator, [self.row(1, 0, "B")])
+
+ def
test_field_nested_update_retract_applies_nested_key_null_strategy_to_retract_input(self):
+ merge_agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ }
+ )
+ )
+ )
+
+ accumulator = None
+ accumulator = merge_agg.agg(accumulator, [self.row(0, 0, "A")])
+ accumulator = merge_agg.agg(accumulator, [self.row(1, 1, "B")])
+
+ ignore_agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-key-null-strategy": "IGNORE",
+ }
+ )
+ )
+ )
+
+ result = ignore_agg.retract(accumulator, [self.row(0, None, "X")])
+
+ self.assertCountEqual(result, [
+ self.row(0, 0, "A"),
+ self.row(1, 1, "B"),
+ ])
+
+ error_agg = self._make(
+ self._make_data_type(),
+ CoreOptions(
+ Options(
+ {
+ "fields.field0.nested-key": "k0,k1",
+ "fields.field0.nested-key-null-strategy": "ERROR",
+ }
+ )
+ )
+ )
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "Nested key contains null values. Primary key fields must not
be null.",
+ ):
+ error_agg.retract(accumulator, [self.row(0, None, "X")])
+
+ def test_field_nested_update_with_non_sequential_field_ids(self):
+ agg = self._make(
+ self._make_data_type(fields=[
+ DataField(10, "k0", AtomicType("INT")),
+ DataField(11, "k1", AtomicType("INT")),
+ DataField(12, "v", AtomicType("STRING")),
+ DataField(23, "seq", AtomicType("INT")),
+ ]),
+ CoreOptions(Options(
+ {
+ 'fields.field0.nested-key': 'k0,k1',
+ "fields.field0.nested-sequence-field": "seq",
+ }
+ ))
+ )
+
+ accumulator = None
+
+ accumulator = agg.agg(
+ accumulator,
+ [
+ self.row(0, 0, "A", 1),
+ self.row(0, 1, "B", 2),
+ self.row(0, 1, "b", 3),
+ self.row(0, 1, "B_late", 2)
+ ]
+ )
+ accumulator = agg.retract(accumulator, [self.row(0, 1, "b", 3)])
+ self.assertCountEqual(accumulator, [self.row(0, 0, "A", 1), ])
+
+
class RegistrationTest(unittest.TestCase):
"""Sanity check that all 10 expected aggregators (the primary-key
placeholder plus 9 value aggregators) are registered when the
diff --git a/paimon-python/pypaimon/tests/test_sequence_field_read.py
b/paimon-python/pypaimon/tests/test_sequence_field_read.py
index ed32768c2e..ea6df67f3e 100644
--- a/paimon-python/pypaimon/tests/test_sequence_field_read.py
+++ b/paimon-python/pypaimon/tests/test_sequence_field_read.py
@@ -383,8 +383,10 @@ class SequenceFieldReadE2ETest(unittest.TestCase):
extra_options={'sequence.field': 'ts',
'fields.val.nested-sequence-field': 'ts2'})
self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
- with self.assertRaises(NotImplementedError):
- self._read(table)
+ self.assertEqual(
+ self._read(table),
+ [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}],
+ )
def test_trailing_comma_sequence_field_tolerated(self):
"""A trailing comma (``'ts,'``) must be tolerated, matching Java
