This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 19f0453a09e2d0897d46e51a6b66c5a55c197e17 Author: QuakeWang <[email protected]> AuthorDate: Thu Sep 17 16:51:21 2026 +0800 [python] Support allow_exact_matches in temporal as-of joins (#9915) --- docs/docs/pypaimon/multimodal-reading.md | 15 ++- paimon-python/pypaimon/multimodal/temporal.py | 47 +++++--- .../pypaimon/tests/multimodal_temporal_test.py | 128 +++++++++++++++++++++ 3 files changed, 175 insertions(+), 15 deletions(-) diff --git a/docs/docs/pypaimon/multimodal-reading.md b/docs/docs/pypaimon/multimodal-reading.md index d3df05a9bd..f9c3730968 100644 --- a/docs/docs/pypaimon/multimodal-reading.md +++ b/docs/docs/pypaimon/multimodal-reading.md @@ -77,11 +77,22 @@ for batch in aligned.to_arrow_batch_reader(batch_size=128): ``` `direction` is `backward`, `forward`, or `nearest`; tolerance is inclusive and -zero means exact. Nearest ties use the earlier time. For duplicate timestamps, -backward uses the last row and forward uses the first. Nearest uses the last +zero permits only exact matches. Nearest ties use the earlier time. For duplicate +timestamps, backward uses the last row and forward uses the first. Nearest uses the last row for an exact match; otherwise it uses the backward or forward candidate's rule. Misses return null. +Both top-level and chained `join_asof` calls accept `allow_exact_matches`, which +defaults to `True`. Set it to `False` to exclude all right rows with the same +timestamp as the left row. Backward then matches strictly earlier timestamps, +forward matches strictly later timestamps, and nearest selects the closest +non-equal timestamp, still preferring the earlier time on ties. For example, +with a left time of `10` and right times `[9, 10]`, strict backward matches `9`. +This is useful when aligning an event with a state from strictly before it. +Combining `allow_exact_matches=False` with `tolerance=0` (or `timedelta(0)` for +timestamp keys) produces no matches: left rows are preserved with null right +values. + Keys must be non-null with matching types. Use `right_on` for a different right timestamp and `suffix` for conflicts. Select the right timestamp to compute the match delta. diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py index 7ccf119ca9..93f118d94b 100644 --- a/paimon-python/pypaimon/multimodal/temporal.py +++ b/paimon-python/pypaimon/multimodal/temporal.py @@ -53,14 +53,19 @@ _TEMPORAL_ROW_GROUP_CACHE_MAX_SIZE = 64 * 1024 * 1024 def join_asof(left, right, *, on, by, direction="backward", tolerance=None, - right_on=None, suffix="_right") -> "TemporalAlignment": - """Join each left row with at most one time-aligned right row.""" + right_on=None, suffix="_right", + allow_exact_matches=True) -> "TemporalAlignment": + """Join each left row with at most one time-aligned right row. + + Set allow_exact_matches=False to exclude equal timestamps. + """ return TemporalAlignment(left, on=on, by=by).join_asof( right, direction=direction, tolerance=tolerance, right_on=right_on, suffix=suffix, + allow_exact_matches=allow_exact_matches, ) @@ -125,8 +130,12 @@ class TemporalAlignment: self.schema = self._output_schema() def join_asof(self, right, *, direction="backward", tolerance=None, - right_on=None, suffix="_right") -> "TemporalAlignment": - """Append a right-side as-of join without materializing this scan.""" + right_on=None, suffix="_right", + allow_exact_matches=True) -> "TemporalAlignment": + """Append a right-side as-of join without materializing this scan. + + Set allow_exact_matches=False to exclude equal timestamps. + """ position = len(self._sources) + 1 label = "right source %d" % position source = _AsOfJoinRight( @@ -138,6 +147,7 @@ class TemporalAlignment: tolerance, right_on, suffix, + allow_exact_matches=allow_exact_matches, ) return self._append(source) @@ -342,11 +352,14 @@ class TemporalAlignment: class _AsOfJoinRight: def __init__(self, label, query, anchor_on, by, direction, tolerance, - right_on, suffix): + right_on, suffix, *, allow_exact_matches=True): _validate_join_options(direction, tolerance, right_on, suffix) + if not isinstance(allow_exact_matches, bool): + raise TypeError("allow_exact_matches must be a bool.") self.label = label self.query = _pin_scan_to_snapshot(_require_scan(query, label)) self.direction = direction + self.allow_exact_matches = allow_exact_matches self.suffix = suffix self.anchor_on = anchor_on self.on = anchor_on if right_on is None else right_on @@ -395,7 +408,8 @@ class _AsOfJoinRight: return None target_key = anchor_row[_TIME_KEY] index = _match_index( - self._time_keys, target_key, self.direction, *bounds) + self._time_keys, target_key, self.direction, *bounds, + allow_exact_matches=self.allow_exact_matches) if index is None: return None matched_key = _python_scalar(self._time_keys[index]) @@ -1464,24 +1478,31 @@ def _project_effective_schema( return pa.schema(fields, metadata=schema.metadata) -def _match_index(times, target, method, start=0, end=None): +def _match_index(times, target, method, start=0, end=None, *, + allow_exact_matches=True): end = len(times) if end is None else end if start >= end: return None position = bisect_left(times, target, start, end) if method == "backward": - position = bisect_right(times, target, start, end) + if allow_exact_matches: + position = bisect_right(times, target, start, end) return position - 1 if position > start else None if method == "forward": + if not allow_exact_matches: + position = bisect_right(times, target, position, end) return position if position < end else None if method == "nearest": - if position < end and times[position] == target: + if (allow_exact_matches + and position < end and times[position] == target): return bisect_right(times, target, position, end) - 1 - if position == start: - return start - if position == end: - return end - 1 before = position - 1 + if not allow_exact_matches: + position = bisect_right(times, target, position, end) + if before < start: + return position if position < end else None + if position == end: + return before before_value = _python_scalar(times[before]) after_value = _python_scalar(times[position]) if target - before_value <= after_value - target: diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py index 8a45cca0f3..50d8e92bcb 100644 --- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py +++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py @@ -120,6 +120,134 @@ class MultimodalTemporalTest(unittest.TestCase): ).to_list()[0] self.assertEqual(expected, row["value"]) + def test_asof_excludes_exact_matches_within_group(self): + anchors = self._table("strict_anchors", { + "episode_id": pa.int32(), "event_time": pa.int64(), + }) + samples = self._table("strict_samples", { + "episode_id": pa.int32(), "event_time": pa.int64(), + "value": pa.int32(), + }) + groups = [ + [9, 9, 10, 10, 11, 11], + [10, 10], + [10, 10, 11], + [9, 10, 10], + [], + [8, 11], + ] + anchors.add([ + {"episode_id": group, "event_time": 10} + for group in range(len(groups)) + ]) + samples.add([ + {"episode_id": group, "event_time": time, "value": index} + for group, times in enumerate(groups) + for index, time in enumerate(times) + ]) + + for direction, expected in ( + ("backward", [1, None, None, 0, None, 0]), + ("forward", [4, None, 2, None, None, 1]), + ("nearest", [1, None, 2, 0, None, 1])): + with self.subTest(direction=direction): + rows = pmm.join_asof( + anchors.scan(), samples.scan().select("value"), + on="event_time", by="episode_id", direction=direction, + allow_exact_matches=False, + ).to_list() + self.assertEqual(dict(enumerate(expected)), { + row["episode_id"]: row["value"] for row in rows + }) + + def test_asof_exact_match_options_are_independent_in_chain(self): + anchors = self._table("strict_chain_anchors", { + "episode_id": pa.int32(), "event_time": pa.int64(), + }) + samples = self._table("strict_chain_samples", { + "episode_id": pa.int32(), "sample_time": pa.int64(), + "value": pa.int32(), + }) + anchors.add([{"episode_id": 1, "event_time": 10}]) + samples.add([ + {"episode_id": 1, "sample_time": time, "value": time} + for time in [9, 10, 10, 11] + ]) + for direction, expected in ( + ("backward", 9), ("forward", 11), ("nearest", 9)): + for tolerance in (0, 0.5, 1, None): + with self.subTest(direction=direction, tolerance=tolerance): + aligned = pmm.join_asof( + anchors.scan(), samples.scan().select("value"), + on="event_time", by="episode_id", + right_on="sample_time", direction=direction, + tolerance=tolerance, allow_exact_matches=False, + ) + row = aligned.join_asof( + samples.scan().select("value"), + right_on="sample_time", allow_exact_matches=True, + ).join_asof( + samples.scan().select("value"), + right_on="sample_time", allow_exact_matches=False, + suffix="_strict", + ).to_list()[0] + self.assertEqual( + expected if tolerance in (1, None) else None, + row["value"], + ) + self.assertEqual(10, row["value_right"]) + self.assertEqual(9, row["value_strict"]) + + def test_asof_excludes_exact_nanosecond_timestamps(self): + anchors = self._table("strict_ns_anchors", { + "episode_id": pa.int32(), "event_time": pa.timestamp("ns"), + }) + samples = self._table("strict_ns_samples", { + "episode_id": pa.int32(), "event_time": pa.timestamp("ns"), + "value": pa.int32(), + }) + anchors.add(pa.table({ + "episode_id": pa.array([1], type=pa.int32()), + "event_time": pa.array( + [1_000_000_001], type=pa.int64()).cast(pa.timestamp("ns")), + })) + samples.add(pa.table({ + "episode_id": pa.array([1, 1, 1], type=pa.int32()), + "event_time": pa.array( + [1_000_000_000, 1_000_000_001, 1_000_000_002], + type=pa.int64()).cast(pa.timestamp("ns")), + "value": pa.array([9, 10, 11], type=pa.int32()), + })) + for direction, expected in ( + ("backward", 9), ("forward", 11), ("nearest", 9)): + for tolerance in (timedelta(0), timedelta(microseconds=1)): + with self.subTest(direction=direction, tolerance=tolerance): + row = pmm.join_asof( + anchors.scan(), samples.scan().select("value"), + on="event_time", by="episode_id", + direction=direction, tolerance=tolerance, + allow_exact_matches=False, + ).to_list()[0] + self.assertEqual( + expected if tolerance else None, row["value"]) + + def test_asof_validates_allow_exact_matches(self): + table = self._table("strict_options", { + "episode_id": pa.int32(), "event_time": pa.int64(), + }) + aligned = pmm.join_asof( + table.scan(), table.scan(), on="event_time", by="episode_id") + for value in (None, 0, 1, "false"): + with self.subTest(value=value): + with self.assertRaisesRegex(TypeError, "allow_exact_matches"): + pmm.join_asof( + table.scan(), table.scan(), + on="event_time", by="episode_id", + allow_exact_matches=value, + ) + with self.assertRaisesRegex(TypeError, "allow_exact_matches"): + aligned.join_asof(table.scan(), allow_exact_matches=value) + def test_nearest_uses_candidate_side_for_duplicate_timestamps(self): anchors = self._table("duplicate_nearest_anchors", { "episode_id": pa.int32(),
