You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "JunRuiLee (via GitHub)" <gi...@apache.org> on 2026/06/02 07:21:26 UTC
[PR] [python] Honor sequence.field on the primary-key read path [paimon]
JunRuiLee opened a new pull request, #8075:
URL: https://github.com/apache/paimon/pull/8075
### Purpose
The pypaimon merge read path broke key-ties using only the file-level sequence
number (`HeapEntry.__lt__`), ignoring the user-configured `sequence.field`. For
PK tables that set it, this silently returned the wrong "latest" value whenever
write/file order disagreed with the sequence-field order.
Mirrors Java's `SortMergeReaderWithMinHeap` + `UserDefinedSeqComparator`:
- **sort_merge_reader**: add `builtin_seq_comparator` so heap order becomes
key → sequence.field → file sequence number. Handles multiple fields,
descending sort order, and parameterized types (`TIMESTAMP(p)`, `DECIMAL(p,s)`).
- **table_read**: re-inject and project out a sequence field dropped by a
projection.
- **aggregation_merge_function**: don't aggregate the sequence column (force
`last_value`), matching Java `getAggFuncName`.
- **merge_engine_support**: add `check_sequence_field_valid` mirroring Java
`SchemaValidation.validateSequenceField` (pypaimon has no schema-creation
validation) — rejects a missing/duplicated sequence field, an aggregator on
the sequence column, `sequence.field` on FIRST_ROW, and cross-partition
update. These raise `ValueError`.
### Tests
`test_sequence_field_read.py` + additions to `test_aggregation_e2e.py`: covers
deduplicate / partial-update / aggregation ordering, multiple fields,
descending order, parameterized types, projection dropping the field, and all
rejection cases. Unset `sequence.field` keeps the existing file-sequence-number
fallback.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Honor sequence.field on the primary-key read path [paimon]
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on PR #8075:
URL: https://github.com/apache/paimon/pull/8075#issuecomment-4601715515
Thanks @JingsongLi — all three addressed in the latest commits, PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Honor sequence.field on the primary-key read path [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #8075:
URL: https://github.com/apache/paimon/pull/8075
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Honor sequence.field on the primary-key read path [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8075:
URL: https://github.com/apache/paimon/pull/8075#discussion_r3339659385
##########
paimon-python/pypaimon/read/reader/sort_merge_reader.py:
##########
@@ -168,51 +179,119 @@ def update(self) -> bool:
class HeapEntry:
- def __init__(self, key: InternalRow, element: Element, key_comparator):
+ def __init__(self, key: InternalRow, element: Element, key_comparator,
+ seq_comparator=None):
self.key = key
self.element = element
self.key_comparator = key_comparator
+ self.seq_comparator = seq_comparator
def __lt__(self, other):
+ # Heap order mirrors Java ``SortMergeReaderWithMinHeap``: user key
+ # -> user-defined sequence comparator (``sequence.field``) on the
+ # value row -> file-level sequence number.
result = self.key_comparator(self.key, other.key)
- if result < 0:
- return True
- elif result > 0:
- return False
-
- return self.element.kv.sequence_number < other.element.kv.sequence_number
-
-
-def builtin_key_comparator(key_schema: List[DataField]) -> Callable[[Any, Any], int]:
- # Precompute comparability flags to avoid repeated type checks
+ if result == 0 and self.seq_comparator is not None:
+ result = self.seq_comparator(
+ self.element.kv.value, other.element.kv.value)
+ if result == 0:
+ result = self.element.kv.sequence_number - other.element.kv.sequence_number
+ return result < 0
+
+
+def _base_type_name(field: DataField) -> str:
+ """Base type keyword of a field, stripping any ``(precision[, scale])``
+ parameters and the ``NOT NULL`` suffix. E.g. ``DECIMAL(10, 2)`` and
+ ``TIMESTAMP(6)`` map to ``DECIMAL`` / ``TIMESTAMP``.
+ """
+ return field.type.type.split('(')[0].split(' ')[0]
+
+
+def _row_field_comparator(
+ fields: List[DataField],
+ indices: List[int],
+ ascending: bool = True) -> Callable[[Any, Any], int]:
+ """Build a comparator over two rows on the given ``indices`` (positions
+ in ``fields`` / the row's ``get_field``), compared left-to-right.
+
+ Shared by :func:`builtin_key_comparator` (all key fields, ascending) and
+ :func:`builtin_seq_comparator` (the configured sequence fields, with
+ sort-order). Comparability is precomputed once; ``ascending=False``
+ reverses the overall result. ``None`` rows/values sort first under
+ ascending order (and the reversal flips that consistently).
+ """
comparable_types = {member.value for member in Keyword if member is not Keyword.VARIANT}
- comparable_flags = [field.type.type.split(' ')[0] in comparable_types for field in key_schema]
+ comparable_flags = [_base_type_name(fields[idx]) in comparable_types for idx in indices]
+ sign = 1 if ascending else -1
- def comparator(key1: InternalRow, key2: InternalRow) -> int:
- if key1 is None and key2 is None:
+ def comparator(row1: InternalRow, row2: InternalRow) -> int:
+ if row1 is None and row2 is None:
return 0
- if key1 is None:
- return -1
- if key2 is None:
- return 1
- for i, comparable in enumerate(comparable_flags):
- val1 = key1.get_field(i)
- val2 = key2.get_field(i)
+ if row1 is None:
+ return -sign
+ if row2 is None:
+ return sign
+ for pos, idx in enumerate(indices):
+ val1 = row1.get_field(idx)
+ val2 = row2.get_field(idx)
if val1 is None and val2 is None:
continue
if val1 is None:
- return -1
+ return -sign
Review Comment:
This reverses null ordering when `sequence.field.sort-order=descending`. Java builds the sequence comparator with `nullIsLast=false` (`CodeGeneratorImpl#getSortSpec`), and `GenerateUtils.generateRowCompare` applies descending only to the non-null comparison result, so null sequence values still sort first even for descending order. Here `-sign` / `sign` makes nulls sort last when `ascending=False`, which can make a null-sequence row become the last row fed into the merge function and therefore the winner. Could we keep null ordering independent from `ascending` and only flip non-null value comparisons? A regression test for descending sequence fields with nulls would catch this.
##########
paimon-python/pypaimon/common/options/core_options.py:
##########
@@ -808,6 +833,23 @@ def changelog_producer(self, default=None):
def merge_engine(self, default=None):
return self.options.get(CoreOptions.MERGE_ENGINE, default)
+ def sequence_field(self) -> List[str]:
+ """User-defined sequence fields, in declaration order. Empty list
+ when ``sequence.field`` is unset. Mirrors Java
+ ``CoreOptions.sequenceField()``.
+ """
+ raw = self.options.get(CoreOptions.SEQUENCE_FIELD)
+ if not raw:
+ return []
+ return [name.strip() for name in raw.split(",") if name.strip()]
Review Comment:
Java `CoreOptions.sequenceField()` trims each segment but does not drop empty ones (`Arrays.stream(s.split(',')).map(String::trim).collect(...)`), so a malformed value like `sequence.field='ts,,ts2'` leaves an empty field name and is rejected by schema validation. This comprehension filters empty segments out, so the same invalid config is silently accepted as `['ts', 'ts2']`. To stay aligned with Java, please preserve empty trimmed segments here and let `check_sequence_field_valid` reject the missing empty field.
##########
paimon-python/pypaimon/read/reader/sort_merge_reader.py:
##########
@@ -168,51 +179,119 @@ def update(self) -> bool:
class HeapEntry:
- def __init__(self, key: InternalRow, element: Element, key_comparator):
+ def __init__(self, key: InternalRow, element: Element, key_comparator,
+ seq_comparator=None):
self.key = key
self.element = element
self.key_comparator = key_comparator
+ self.seq_comparator = seq_comparator
def __lt__(self, other):
+ # Heap order mirrors Java ``SortMergeReaderWithMinHeap``: user key
+ # -> user-defined sequence comparator (``sequence.field``) on the
+ # value row -> file-level sequence number.
result = self.key_comparator(self.key, other.key)
- if result < 0:
- return True
- elif result > 0:
- return False
-
- return self.element.kv.sequence_number < other.element.kv.sequence_number
-
-
-def builtin_key_comparator(key_schema: List[DataField]) -> Callable[[Any, Any], int]:
- # Precompute comparability flags to avoid repeated type checks
+ if result == 0 and self.seq_comparator is not None:
+ result = self.seq_comparator(
+ self.element.kv.value, other.element.kv.value)
+ if result == 0:
+ result = self.element.kv.sequence_number - other.element.kv.sequence_number
+ return result < 0
+
+
+def _base_type_name(field: DataField) -> str:
+ """Base type keyword of a field, stripping any ``(precision[, scale])``
+ parameters and the ``NOT NULL`` suffix. E.g. ``DECIMAL(10, 2)`` and
+ ``TIMESTAMP(6)`` map to ``DECIMAL`` / ``TIMESTAMP``.
+ """
+ return field.type.type.split('(')[0].split(' ')[0]
Review Comment:
This assumes every comparable sequence field is an `AtomicType` with `field.type.type`. Java's `UserDefinedSeqComparator` delegates to `RecordComparator`, which supports ARRAY, VECTOR, MAP, MULTISET, and ROW as well as atomic types, and the docs say `sequence.field` supports fields of all data types. For a Java-valid table using a complex sequence field, pypaimon will currently fail with an attribute error or `Unsupported ... comparison`. If full recursive comparison is too much for this PR, I think we should explicitly validate and reject unsupported complex sequence-field types with a clear `NotImplementedError` instead of presenting this as aligned with Java.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org