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 65b265930d [python] Support allow_exact_matches in temporal as-of
joins (#9915)
65b265930d is described below
commit 65b265930de4f78bc7ef905e92695e4dd3e9893c
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(),
