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 ec7f508740 [python] Build temporal group indexes with Arrow run-end
encoding (#10466)
ec7f508740 is described below
commit ec7f5087404e46f7cd5fa7b257c544e4a4c93388
Author: zhigang <[email protected]>
AuthorDate: Fri Oct 9 21:54:09 2026 +0800
[python] Build temporal group indexes with Arrow run-end encoding (#10466)
---
.gitignore | 1 +
paimon-python/pypaimon/multimodal/temporal.py | 24 +++++++++
.../pypaimon/tests/multimodal_temporal_test.py | 62 ++++++++++++++++++++++
3 files changed, 87 insertions(+)
diff --git a/.gitignore b/.gitignore
index 9f1331ac3d..92b2b85097 100644
--- a/.gitignore
+++ b/.gitignore
@@ -22,6 +22,7 @@ paimon-python/.idea/
.claude
.cursor
.qoder
+.qwen
### Java / Maven ###
target
diff --git a/paimon-python/pypaimon/multimodal/temporal.py
b/paimon-python/pypaimon/multimodal/temporal.py
index c1052cd502..760fde10f3 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -390,6 +390,30 @@ class _AsOfJoinRight:
self._row_ids = metadata[_ROW_ID].combine_chunks()
self._index = {}
group_columns = [metadata[name].combine_chunks() for name in self.by]
+ encode = getattr(pc, "run_end_encode", None)
+ if len(group_columns) == 1 and encode is not None:
+ column = group_columns[0]
+ data_type = column.type
+ # Only use types whose Arrow runs have Python key equality.
+ # In particular, run-end encoding distinguishes -0.0 from +0.0.
+ if (pa.types.is_integer(data_type)
+ or pa.types.is_boolean(data_type)
+ or pa.types.is_string(data_type)
+ or pa.types.is_large_string(data_type)
+ or pa.types.is_binary(data_type)
+ or pa.types.is_large_binary(data_type)):
+ try:
+ groups = encode(column, run_end_type=pa.int64())
+ except pa.ArrowNotImplementedError:
+ pass
+ else:
+ start = 0
+ # Avoid Arrow scalar conversion per run for dense groups.
+ ends = groups.run_ends.to_numpy().tolist()
+ for key, end in zip(groups.values.to_pylist(), ends):
+ self._index[(key,)] = (start, end)
+ start = end
+ return
previous = None
start = 0
for position in range(len(metadata)):
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index f74b8fe245..69c650cf4b 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -122,6 +122,68 @@ class MultimodalTemporalTest(unittest.TestCase):
).to_list()[0]
self.assertEqual(expected, row["value"])
+ def test_alignment_treats_signed_zero_as_one_group(self):
+ schema = {"group": pa.float64(), "event_time": pa.int64()}
+ anchors = self._table("zero_anchors", schema)
+ samples = self._table("zero_samples", {**schema, "value": pa.int32()})
+ anchors.add([{"group": 0.0, "event_time": time} for time in (1, 2)])
+ samples.add(
+ [
+ {"group": -0.0, "event_time": 1, "value": 7},
+ {"group": 0.0, "event_time": 2, "value": 8},
+ ]
+ )
+ rows = pmm.join_asof(
+ anchors.scan(),
+ samples.scan().select("value"),
+ on="event_time",
+ by="group",
+ tolerance=0,
+ ).to_list()
+ self.assertEqual({1: 7, 2: 8}, {row["event_time"]: row["value"] for
row in rows})
+
+ def test_alignment_preserves_composite_group_boundaries(self):
+ schema = {"host": pa.string(), "sensor": pa.int32(), "event_time":
pa.int64()}
+ anchors = self._table("composite_anchors", schema)
+ samples = self._table("composite_samples", {**schema, "value":
pa.int32()})
+ keys = [("a", 1), ("a", 2), ("b", 1)]
+ anchors.add([{"host": host, "sensor": sensor, "event_time": 10} for
host, sensor in keys])
+ samples.add(
+ [
+ {"host": host, "sensor": sensor, "event_time": 9, "value":
index}
+ for index, (host, sensor) in enumerate(keys)
+ ]
+ )
+ rows = pmm.join_asof(
+ anchors.scan(),
+ samples.scan().select("value"),
+ on="event_time",
+ by=["host", "sensor"],
+ ).to_list()
+ self.assertEqual(dict(zip(keys, range(3))), {(row["host"],
row["sensor"]): row["value"] for row in rows})
+
+ def test_alignment_without_run_end_kernel(self):
+ schema = {"group": pa.string(), "event_time": pa.int64()}
+ anchors = self._table("fallback_anchors", schema)
+ samples = self._table("fallback_samples", {**schema, "value":
pa.int32()})
+ anchors.add([{"group": "a", "event_time": 1}])
+ samples.add(
+ [
+ {"group": "a", "event_time": 1, "value": 7},
+ {"group": "b", "event_time": 1, "value": 8},
+ ]
+ )
+ for kernel in (None,
mock.Mock(side_effect=pa.ArrowNotImplementedError("No kernel"))):
+ with self.subTest(kernel=kernel), mock.patch.object(temporal.pc,
"run_end_encode", kernel, create=True):
+ rows = pmm.join_asof(
+ anchors.scan(),
+ samples.scan().select("value"),
+ on="event_time",
+ by="group",
+ tolerance=0,
+ ).to_list()
+ self.assertEqual([7], [row["value"] for row in rows])
+
def test_asof_excludes_exact_matches_within_group(self):
anchors = self._table("strict_anchors", {
"episode_id": pa.int32(), "event_time": pa.int64(),
