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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
