From 443119284c053ba35a9728904d383d1326685340 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 06:59:02 -0700
Subject: [PATCH 01/13] [python] Add temporal window aggregation

---
 paimon-python/pypaimon/multimodal/__init__.py |   2 +
 paimon-python/pypaimon/multimodal/temporal.py | 195 ++++++++++++++++
 .../tests/multimodal_temporal_test.py         | 209 ++++++++++++++++++
 3 files changed, 406 insertions(+)

diff --git a/paimon-python/pypaimon/multimodal/__init__.py b/paimon-python/pypaimon/multimodal/__init__.py
index 1807ba4d6f7c..2bebe2c0afa7 100644
--- a/paimon-python/pypaimon/multimodal/__init__.py
+++ b/paimon-python/pypaimon/multimodal/__init__.py
@@ -44,6 +44,7 @@
 )
 from pypaimon.multimodal.temporal import (
     TemporalAlignment,
+    aggregate_window,
     interpolate,
     join_asof,
 )
@@ -76,6 +77,7 @@
     "VectorRoute",
     "VideoFrameCollator",
     "VideoFrameDescriptor",
+    "aggregate_window",
     "connect",
     "interpolate",
     "join_asof",
diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py
index 43de0a8ed108..6a77c3eefa67 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -74,6 +74,21 @@ def interpolate(left, right, *, on, by, tolerance=None,
     )
 
 
+def aggregate_window(left, right, *, on, by, preceding, aggregations,
+                     following=None, closed="both", right_on=None,
+                     suffix="_right") -> "TemporalAlignment":
+    """Aggregate right values in a time window around each left row."""
+    return TemporalAlignment(left, on=on, by=by).aggregate_window(
+        right,
+        preceding=preceding,
+        following=following,
+        aggregations=aggregations,
+        closed=closed,
+        right_on=right_on,
+        suffix=suffix,
+    )
+
+
 def _normalize_temporal_keys(on, by):
     if not isinstance(on, str) or not on:
         raise ValueError("on must be a non-empty column name.")
@@ -140,6 +155,25 @@ def interpolate(self, right, *, tolerance=None, right_on=None,
         )
         return self._append(source)
 
+    def aggregate_window(self, right, *, preceding, aggregations,
+                         following=None, closed="both", right_on=None,
+                         suffix="_right") -> "TemporalAlignment":
+        """Append aggregation of a right-side time window."""
+        position = len(self._sources) + 1
+        source = _WindowAggregationRight(
+            "right source %d" % position,
+            right,
+            self._on,
+            self._by,
+            preceding,
+            following,
+            aggregations,
+            closed,
+            right_on,
+            suffix,
+        )
+        return self._append(source)
+
     def _append(self, source):
         result = object.__new__(TemporalAlignment)
         result._anchor = self._anchor
@@ -484,6 +518,167 @@ def build_arrays(self, anchor_rows, fetcher):
         return arrays
 
 
+class _WindowAggregationRight(_AsOfJoinRight):
+
+    _SUPPORTED_AGGREGATIONS = {
+        "count", "first", "last", "max", "mean", "min",
+    }
+
+    def __init__(self, label, query, anchor_on, by, preceding, following,
+                 aggregations, closed, right_on, suffix):
+        super().__init__(
+            label, query, anchor_on, by, "nearest", None,
+            right_on, suffix)
+        self._preceding_key = _window_bound_key(
+            "preceding", preceding, self.time_type)
+        if following is None:
+            following = (
+                timedelta(0) if pa.types.is_timestamp(self.time_type) else 0)
+        self._following_key = _window_bound_key(
+            "following", following, self.time_type)
+        if closed not in ("both", "left", "neither", "right"):
+            raise ValueError(
+                "closed must be 'both', 'left', 'right', or 'neither'.")
+        self.closed = closed
+        self.aggregations = _normalize_aggregations(
+            aggregations, self.payload_schema, self.label,
+            self._SUPPORTED_AGGREGATIONS)
+        self.payload_schema = pa.schema([
+            field for field in self.payload_schema
+            if field.name in self.aggregations
+        ], metadata=self.payload_schema.metadata)
+
+    def output_field(self, field, effective=True):
+        try:
+            output_type = _aggregate_output_type(
+                field.type, self.aggregations[field.name])
+        except TypeError:
+            if effective:
+                raise
+            output_type = field.type
+        return pa.field(
+            field.name, output_type, nullable=True,
+            metadata=field.metadata)
+
+    def match(self, anchor_row):
+        key = tuple(anchor_row[name] for name in self.by)
+        bounds = self._index.get(key)
+        if bounds is None:
+            return []
+        target = anchor_row[_TIME_KEY]
+        start, end = bounds
+        left = target - self._preceding_key
+        right = target + self._following_key
+        first = (
+            bisect_left(self._time_keys, left, start, end)
+            if self.closed in ("both", "left")
+            else bisect_right(self._time_keys, left, start, end)
+        )
+        last = (
+            bisect_right(self._time_keys, right, first, end)
+            if self.closed in ("both", "right")
+            else bisect_left(self._time_keys, right, first, end)
+        )
+        return [self._row_ids[index].as_py()
+                for index in range(first, last)]
+
+    def build_arrays(self, anchor_rows, fetcher):
+        matches = [self.match(row) for row in anchor_rows]
+        unique_ids = list(dict.fromkeys(
+            row_id for match in matches for row_id in match))
+        values = fetcher.fetch(unique_ids)
+        positions = {
+            row_id: index for index, row_id in enumerate(unique_ids)
+        }
+        indices = [
+            [positions[row_id] for row_id in match]
+            for match in matches
+        ]
+        arrays = []
+        for field in self.payload_schema:
+            effective = fetcher.schema.field(field.name)
+            aggregation = self.aggregations[field.name]
+            output_type = _aggregate_output_type(
+                effective.type, aggregation)
+            arrays.append(pa.array([
+                _aggregate_values(
+                    values[field.name], row_indices, aggregation)
+                for row_indices in indices
+            ], type=output_type))
+        return arrays
+
+
+def _normalize_aggregations(aggregations, schema, label, supported):
+    if not isinstance(aggregations, dict) or not aggregations:
+        raise ValueError("aggregations must be a non-empty dict.")
+    unknown = [name for name in aggregations if name not in schema.names]
+    if unknown:
+        raise ValueError(
+            "%s is missing aggregation columns %r." % (label, unknown))
+    for name, aggregation in aggregations.items():
+        if not isinstance(aggregation, str) or aggregation not in supported:
+            raise ValueError(
+                "Unsupported aggregation %r for column %r; expected one of "
+                "%r." % (aggregation, name, sorted(supported)))
+    return dict(aggregations)
+
+
+def _aggregate_output_type(data_type, aggregation):
+    if not (pa.types.is_integer(data_type)
+            or pa.types.is_floating(data_type)):
+        raise TypeError(
+            "Window aggregation requires integer or floating-point scalar "
+            "columns; got %s." % data_type)
+    if aggregation == "mean":
+        return pa.float64()
+    if aggregation == "count":
+        return pa.int64()
+    return data_type
+
+
+def _aggregate_values(values, indices, aggregation):
+    if not indices:
+        return 0 if aggregation == "count" else None
+    selected = pc.take(values, pa.array(indices, type=pa.int64()))
+    if aggregation == "count":
+        return pc.count(selected).as_py()
+    if aggregation == "mean":
+        items = [item for item in selected.to_pylist()
+                 if item is not None]
+        if not items:
+            return None
+        if pa.types.is_integer(values.type):
+            return sum(items) / len(items)
+        if not all(math.isfinite(item) for item in items):
+            return pc.mean(selected).as_py()
+        scale = max(abs(item) for item in items)
+        if scale == 0:
+            return 0.0
+        return (math.fsum(item / scale for item in items) / len(items)) * scale
+    if aggregation == "min":
+        return pc.min(selected).as_py()
+    if aggregation == "max":
+        return pc.max(selected).as_py()
+    items = selected.to_pylist()
+    if aggregation == "first":
+        return next((item for item in items if item is not None), None)
+    return next((item for item in reversed(items) if item is not None), None)
+
+
+def _window_bound_key(name, value, data_type):
+    if isinstance(value, bool) or not isinstance(value, (Real, timedelta)):
+        raise TypeError(
+            "%s must be numeric or datetime.timedelta." % name)
+    if (isinstance(value, Real) and not isinstance(value, Integral)
+            and not math.isfinite(value)):
+        raise ValueError("%s must be finite." % name)
+    zero = timedelta(0) if isinstance(value, timedelta) else 0
+    if value < zero:
+        raise ValueError("%s must be non-negative." % name)
+    _validate_tolerance(value, data_type)
+    return _time_tolerance_key(value, data_type)
+
+
 def _validate_join_options(direction, tolerance, right_on, suffix):
     if direction not in ("backward", "forward", "nearest"):
         raise ValueError(
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index 174352bf4239..e796f264d0ee 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -188,6 +188,215 @@ def test_linear_interpolation_stays_in_group_without_extrapolation(self):
             row["value"] for row in rows
         ])
 
+    def test_window_aggregation_stays_in_group_and_skips_nulls(self):
+        anchors = self._table("window_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        samples = self._table("window_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "average": pa.int32(),
+            "minimum": pa.int32(),
+            "maximum": pa.int32(),
+            "first_value": pa.int32(),
+            "last_value": pa.int32(),
+            "valid_count": pa.int32(),
+        })
+        anchors.add([
+            {"episode_id": 1, "event_time": 10},
+            {"episode_id": 2, "event_time": 10},
+            {"episode_id": 3, "event_time": 10},
+        ])
+        samples.add([
+            dict({"episode_id": 1, "event_time": time}, **{
+                name: value for name in (
+                    "average", "minimum", "maximum", "first_value",
+                    "last_value", "valid_count")
+            })
+            for time, value in ((5, 1), (10, None), (15, 5))
+        ] + [dict({"episode_id": 2, "event_time": 10}, **{
+            name: 100 for name in (
+                "average", "minimum", "maximum", "first_value",
+                "last_value", "valid_count")
+        })])
+
+        result = pmm.aggregate_window(
+            anchors.scan(),
+            samples.scan().select([
+                "average", "minimum", "maximum", "first_value",
+                "last_value", "valid_count",
+            ]),
+            on="event_time",
+            by="episode_id",
+            preceding=5,
+            following=5,
+            aggregations={
+                "average": "mean",
+                "minimum": "min",
+                "maximum": "max",
+                "first_value": "first",
+                "last_value": "last",
+                "valid_count": "count",
+            },
+        )
+        rows = sorted(result.to_list(), key=lambda row: row["episode_id"])
+
+        self.assertIsInstance(result, pmm.TemporalAlignment)
+        self.assertEqual(pa.float64(), result.schema.field("average").type)
+        self.assertEqual(pa.int64(), result.schema.field("valid_count").type)
+        self.assertEqual(
+            (3.0, 1, 5, 1, 5, 2),
+            tuple(rows[0][name] for name in (
+                "average", "minimum", "maximum", "first_value",
+                "last_value", "valid_count")),
+        )
+        self.assertEqual(100.0, rows[1]["average"])
+        self.assertIsNone(rows[2]["average"])
+        self.assertEqual(0, rows[2]["valid_count"])
+
+    def test_window_aggregation_supports_asymmetric_timestamp_bounds(self):
+        anchors = self._table("window_timestamp_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.timestamp("ms"),
+        })
+        samples = self._table("window_timestamp_samples", {
+            "episode_id": pa.int32(),
+            "captured_at": pa.timestamp("ms"),
+            "value": pa.float32(),
+        })
+        anchor = datetime(2026, 9, 1, 12, 0, 0)
+        anchors.add([{"episode_id": 1, "event_time": anchor}])
+        samples.add([
+            {"episode_id": 1,
+             "captured_at": anchor + timedelta(milliseconds=offset),
+             "value": value}
+            for offset, value in ((-11, 100.0), (-10, 1.0),
+                                  (0, 2.0), (5, 3.0), (6, 100.0))
+        ])
+
+        row = pmm.aggregate_window(
+            anchors.scan(), samples.scan().select("value"),
+            on="event_time", right_on="captured_at", by="episode_id",
+            preceding=timedelta(milliseconds=10),
+            following=timedelta(milliseconds=5),
+            aggregations={"value": "mean"},
+        ).to_list()[0]
+
+        self.assertEqual(2.0, row["value"])
+
+        right_closed = pmm.aggregate_window(
+            anchors.scan(), samples.scan().select("value"),
+            on="event_time", right_on="captured_at", by="episode_id",
+            preceding=timedelta(milliseconds=10),
+            following=timedelta(milliseconds=5),
+            aggregations={"value": "mean"},
+            closed="right",
+        ).to_list()[0]
+        self.assertEqual(2.5, right_closed["value"])
+
+    def test_window_mean_avoids_numeric_overflow_and_integer_rounding(self):
+        anchors = self._table("window_numeric_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        samples = self._table("window_numeric_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "integer_value": pa.int64(),
+            "float_value": pa.float64(),
+        })
+        anchors.add([{"episode_id": 1, "event_time": 1}])
+        samples.add([
+            {"episode_id": 1, "event_time": 0,
+             "integer_value": -(1 << 63),
+             "float_value": sys.float_info.max},
+            {"episode_id": 1, "event_time": 2,
+             "integer_value": (1 << 63) - 1,
+             "float_value": sys.float_info.max},
+        ])
+
+        row = pmm.aggregate_window(
+            anchors.scan(),
+            samples.scan().select(["integer_value", "float_value"]),
+            on="event_time", by="episode_id", preceding=1, following=1,
+            aggregations={
+                "integer_value": "mean",
+                "float_value": "mean",
+            },
+        ).to_list()[0]
+
+        self.assertEqual(-0.5, row["integer_value"])
+        self.assertEqual(sys.float_info.max, row["float_value"])
+
+    def test_window_aggregation_can_follow_an_asof_join(self):
+        anchors = self._table("window_chain_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        images = self._table("window_chain_images", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "image": pa.string(),
+        })
+        imu = self._table("window_chain_imu", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "acceleration": pa.float32(),
+        })
+        anchors.add([{"episode_id": 1, "event_time": 10}])
+        images.add([{"episode_id": 1, "event_time": 9, "image": "frame"}])
+        imu.add([
+            {"episode_id": 1, "event_time": 8, "acceleration": 1.0},
+            {"episode_id": 1, "event_time": 10, "acceleration": 3.0},
+        ])
+
+        row = pmm.join_asof(
+            anchors.scan(), images.scan().select("image"),
+            on="event_time", by="episode_id", direction="nearest",
+        ).aggregate_window(
+            imu.scan().select("acceleration"),
+            preceding=2,
+            aggregations={"acceleration": "mean"},
+        ).to_list()[0]
+
+        self.assertEqual("frame", row["image"])
+        self.assertEqual(2.0, row["acceleration"])
+
+    def test_window_aggregation_validates_options(self):
+        table = self._table("window_validation", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "value": pa.int32(),
+            "text": pa.string(),
+        })
+
+        def scan():
+            return table.scan().select("value")
+
+        with self.assertRaisesRegex(ValueError, "non-negative"):
+            pmm.aggregate_window(
+                scan(), scan(), on="event_time", by="episode_id",
+                preceding=-1, aggregations={"value": "mean"})
+        with self.assertRaisesRegex(ValueError, "Unsupported aggregation"):
+            pmm.aggregate_window(
+                scan(), scan(), on="event_time", by="episode_id",
+                preceding=1, aggregations={"value": "median"})
+        with self.assertRaisesRegex(ValueError, "closed must be"):
+            pmm.aggregate_window(
+                scan(), scan(), on="event_time", by="episode_id",
+                preceding=1, aggregations={"value": "mean"},
+                closed="middle")
+        with self.assertRaisesRegex(ValueError, "missing aggregation columns"):
+            pmm.aggregate_window(
+                scan(), scan(), on="event_time", by="episode_id",
+                preceding=1, aggregations={"missing": "mean"})
+        with self.assertRaisesRegex(TypeError, "requires integer or floating"):
+            pmm.aggregate_window(
+                scan(), table.scan().select("text"),
+                on="event_time", by="episode_id", preceding=1,
+                aggregations={"text": "first"}).to_arrow()
+
     def test_linear_interpolation_preserves_an_exact_infinite_float(self):
         anchors = self._table("linear_exact_anchors", {
             "episode_id": pa.int32(),

From 68860e06205eaa799d38b0afc92a421f945e3028 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 06:59:08 -0700
Subject: [PATCH 02/13] [docs] Document temporal window aggregation

---
 docs/docs/pypaimon/multimodal-reading.md | 24 ++++++++++++++++++++++++
 1 file changed, 24 insertions(+)

diff --git a/docs/docs/pypaimon/multimodal-reading.md b/docs/docs/pypaimon/multimodal-reading.md
index adb32b75b8f1..1aafeae08c6f 100644
--- a/docs/docs/pypaimon/multimodal-reading.md
+++ b/docs/docs/pypaimon/multimodal-reading.md
@@ -102,6 +102,30 @@ states_at_steps = aligned.interpolate(
 )
 ```
 
+Use `aggregate_window` to reduce higher-frequency numeric scalar samples
+around each left timestamp. Bounds are inclusive by default; `following`
+defaults to zero.
+Supported aggregations are `mean`, `min`, `max`, `first`, `last`, and `count`;
+nulls are skipped. Set `closed` to `left`, `right`, or `neither` to exclude
+endpoints. Empty windows return null, except `count` returns zero.
+
+```python
+from pypaimon.multimodal import aggregate_window
+
+steps_with_imu = aggregate_window(
+    steps.scan(),
+    imu.scan().select(["acceleration", "angular_velocity"]),
+    on="event_time",
+    by="episode_id",
+    preceding=timedelta(milliseconds=50),
+    following=timedelta(milliseconds=50),
+    aggregations={
+        "acceleration": "mean",
+        "angular_velocity": "mean",
+    },
+)
+```
+
 ### Reading BLOB columns
 
 `scan().read_blobs(column)` bulk-fetches a BLOB column's bytes for the filtered

From c8afcd2f4cb4e91c1d909c75db49936d56d45343 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 07:05:15 -0700
Subject: [PATCH 03/13] [docs] Shorten window aggregation example

---
 docs/docs/pypaimon/multimodal-reading.md | 26 ++++++++----------------
 1 file changed, 9 insertions(+), 17 deletions(-)

diff --git a/docs/docs/pypaimon/multimodal-reading.md b/docs/docs/pypaimon/multimodal-reading.md
index 1aafeae08c6f..90d31492aef9 100644
--- a/docs/docs/pypaimon/multimodal-reading.md
+++ b/docs/docs/pypaimon/multimodal-reading.md
@@ -102,27 +102,19 @@ states_at_steps = aligned.interpolate(
 )
 ```
 
-Use `aggregate_window` to reduce higher-frequency numeric scalar samples
-around each left timestamp. Bounds are inclusive by default; `following`
-defaults to zero.
-Supported aggregations are `mean`, `min`, `max`, `first`, `last`, and `count`;
-nulls are skipped. Set `closed` to `left`, `right`, or `neither` to exclude
-endpoints. Empty windows return null, except `count` returns zero.
+Use `aggregate_window(left, right, ...)` directly or chain
+`aligned.aggregate_window(...)`. It reduces numeric right rows in each `by`
+group and `[left - preceding, left + following]` window. `following` defaults
+to zero; `closed` controls endpoints. Supported aggregations are `mean`, `min`,
+`max`, `first`, `last`, and `count`. Nulls are skipped; empty windows return
+null, except `count` returns zero.
 
 ```python
-from pypaimon.multimodal import aggregate_window
-
-steps_with_imu = aggregate_window(
-    steps.scan(),
-    imu.scan().select(["acceleration", "angular_velocity"]),
-    on="event_time",
-    by="episode_id",
+steps_with_imu = aligned.aggregate_window(
+    imu.scan().select("acceleration"),
     preceding=timedelta(milliseconds=50),
     following=timedelta(milliseconds=50),
-    aggregations={
-        "acceleration": "mean",
-        "angular_velocity": "mean",
-    },
+    aggregations={"acceleration": "mean"},
 )
 ```
 

From cbb84e1895ab572a89f373cad3c05e4230c07904 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 07:10:44 -0700
Subject: [PATCH 04/13] [python] Rename window aggregation API to join_window

---
 docs/docs/pypaimon/multimodal-reading.md      | 15 +++++-----
 paimon-python/pypaimon/multimodal/__init__.py |  4 +--
 paimon-python/pypaimon/multimodal/temporal.py | 18 ++++++------
 .../tests/multimodal_temporal_test.py         | 28 +++++++++----------
 4 files changed, 33 insertions(+), 32 deletions(-)

diff --git a/docs/docs/pypaimon/multimodal-reading.md b/docs/docs/pypaimon/multimodal-reading.md
index 90d31492aef9..13e141f66a07 100644
--- a/docs/docs/pypaimon/multimodal-reading.md
+++ b/docs/docs/pypaimon/multimodal-reading.md
@@ -102,15 +102,16 @@ states_at_steps = aligned.interpolate(
 )
 ```
 
-Use `aggregate_window(left, right, ...)` directly or chain
-`aligned.aggregate_window(...)`. It reduces numeric right rows in each `by`
-group and `[left - preceding, left + following]` window. `following` defaults
-to zero; `closed` controls endpoints. Supported aggregations are `mean`, `min`,
-`max`, `first`, `last`, and `count`. Nulls are skipped; empty windows return
-null, except `count` returns zero.
+Use `join_window(left, right, ...)` directly or chain
+`aligned.join_window(...)`. Unlike single-table `rolling().agg()`, it joins
+each left row to right rows in the same `by` group and
+`[left - preceding, left + following]` time window, then aggregates them.
+`following` defaults to zero; `closed` controls endpoints. Supported
+aggregations are `mean`, `min`, `max`, `first`, `last`, and `count`. Nulls are
+skipped; empty windows return null, except `count` returns zero.
 
 ```python
-steps_with_imu = aligned.aggregate_window(
+steps_with_imu = aligned.join_window(
     imu.scan().select("acceleration"),
     preceding=timedelta(milliseconds=50),
     following=timedelta(milliseconds=50),
diff --git a/paimon-python/pypaimon/multimodal/__init__.py b/paimon-python/pypaimon/multimodal/__init__.py
index 2bebe2c0afa7..00d59d5c7184 100644
--- a/paimon-python/pypaimon/multimodal/__init__.py
+++ b/paimon-python/pypaimon/multimodal/__init__.py
@@ -44,9 +44,9 @@
 )
 from pypaimon.multimodal.temporal import (
     TemporalAlignment,
-    aggregate_window,
     interpolate,
     join_asof,
+    join_window,
 )
 from pypaimon.multimodal.video import VideoFrameCollator
 from pypaimon.table.row.blob import Blob, BlobDescriptor, VideoFrameDescriptor
@@ -77,10 +77,10 @@
     "VectorRoute",
     "VideoFrameCollator",
     "VideoFrameDescriptor",
-    "aggregate_window",
     "connect",
     "interpolate",
     "join_asof",
+    "join_window",
     "lit",
     "source_col",
     "target_col",
diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py
index 6a77c3eefa67..79b33d69c391 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -74,11 +74,11 @@ def interpolate(left, right, *, on, by, tolerance=None,
     )
 
 
-def aggregate_window(left, right, *, on, by, preceding, aggregations,
-                     following=None, closed="both", right_on=None,
-                     suffix="_right") -> "TemporalAlignment":
-    """Aggregate right values in a time window around each left row."""
-    return TemporalAlignment(left, on=on, by=by).aggregate_window(
+def join_window(left, right, *, on, by, preceding, aggregations,
+                following=None, closed="both", right_on=None,
+                suffix="_right") -> "TemporalAlignment":
+    """Join and aggregate right values in each left row's time window."""
+    return TemporalAlignment(left, on=on, by=by).join_window(
         right,
         preceding=preceding,
         following=following,
@@ -155,10 +155,10 @@ def interpolate(self, right, *, tolerance=None, right_on=None,
         )
         return self._append(source)
 
-    def aggregate_window(self, right, *, preceding, aggregations,
-                         following=None, closed="both", right_on=None,
-                         suffix="_right") -> "TemporalAlignment":
-        """Append aggregation of a right-side time window."""
+    def join_window(self, right, *, preceding, aggregations,
+                    following=None, closed="both", right_on=None,
+                    suffix="_right") -> "TemporalAlignment":
+        """Append a right-side window join with aggregation."""
         position = len(self._sources) + 1
         source = _WindowAggregationRight(
             "right source %d" % position,
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index e796f264d0ee..6f2d1ec9efc3 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -188,7 +188,7 @@ def test_linear_interpolation_stays_in_group_without_extrapolation(self):
             row["value"] for row in rows
         ])
 
-    def test_window_aggregation_stays_in_group_and_skips_nulls(self):
+    def test_window_join_stays_in_group_and_skips_nulls(self):
         anchors = self._table("window_anchors", {
             "episode_id": pa.int32(),
             "event_time": pa.int64(),
@@ -221,7 +221,7 @@ def test_window_aggregation_stays_in_group_and_skips_nulls(self):
                 "last_value", "valid_count")
         })])
 
-        result = pmm.aggregate_window(
+        result = pmm.join_window(
             anchors.scan(),
             samples.scan().select([
                 "average", "minimum", "maximum", "first_value",
@@ -255,7 +255,7 @@ def test_window_aggregation_stays_in_group_and_skips_nulls(self):
         self.assertIsNone(rows[2]["average"])
         self.assertEqual(0, rows[2]["valid_count"])
 
-    def test_window_aggregation_supports_asymmetric_timestamp_bounds(self):
+    def test_window_join_supports_asymmetric_timestamp_bounds(self):
         anchors = self._table("window_timestamp_anchors", {
             "episode_id": pa.int32(),
             "event_time": pa.timestamp("ms"),
@@ -275,7 +275,7 @@ def test_window_aggregation_supports_asymmetric_timestamp_bounds(self):
                                   (0, 2.0), (5, 3.0), (6, 100.0))
         ])
 
-        row = pmm.aggregate_window(
+        row = pmm.join_window(
             anchors.scan(), samples.scan().select("value"),
             on="event_time", right_on="captured_at", by="episode_id",
             preceding=timedelta(milliseconds=10),
@@ -285,7 +285,7 @@ def test_window_aggregation_supports_asymmetric_timestamp_bounds(self):
 
         self.assertEqual(2.0, row["value"])
 
-        right_closed = pmm.aggregate_window(
+        right_closed = pmm.join_window(
             anchors.scan(), samples.scan().select("value"),
             on="event_time", right_on="captured_at", by="episode_id",
             preceding=timedelta(milliseconds=10),
@@ -316,7 +316,7 @@ def test_window_mean_avoids_numeric_overflow_and_integer_rounding(self):
              "float_value": sys.float_info.max},
         ])
 
-        row = pmm.aggregate_window(
+        row = pmm.join_window(
             anchors.scan(),
             samples.scan().select(["integer_value", "float_value"]),
             on="event_time", by="episode_id", preceding=1, following=1,
@@ -329,7 +329,7 @@ def test_window_mean_avoids_numeric_overflow_and_integer_rounding(self):
         self.assertEqual(-0.5, row["integer_value"])
         self.assertEqual(sys.float_info.max, row["float_value"])
 
-    def test_window_aggregation_can_follow_an_asof_join(self):
+    def test_window_join_can_follow_an_asof_join(self):
         anchors = self._table("window_chain_anchors", {
             "episode_id": pa.int32(),
             "event_time": pa.int64(),
@@ -354,7 +354,7 @@ def test_window_aggregation_can_follow_an_asof_join(self):
         row = pmm.join_asof(
             anchors.scan(), images.scan().select("image"),
             on="event_time", by="episode_id", direction="nearest",
-        ).aggregate_window(
+        ).join_window(
             imu.scan().select("acceleration"),
             preceding=2,
             aggregations={"acceleration": "mean"},
@@ -363,7 +363,7 @@ def test_window_aggregation_can_follow_an_asof_join(self):
         self.assertEqual("frame", row["image"])
         self.assertEqual(2.0, row["acceleration"])
 
-    def test_window_aggregation_validates_options(self):
+    def test_window_join_validates_options(self):
         table = self._table("window_validation", {
             "episode_id": pa.int32(),
             "event_time": pa.int64(),
@@ -375,24 +375,24 @@ def scan():
             return table.scan().select("value")
 
         with self.assertRaisesRegex(ValueError, "non-negative"):
-            pmm.aggregate_window(
+            pmm.join_window(
                 scan(), scan(), on="event_time", by="episode_id",
                 preceding=-1, aggregations={"value": "mean"})
         with self.assertRaisesRegex(ValueError, "Unsupported aggregation"):
-            pmm.aggregate_window(
+            pmm.join_window(
                 scan(), scan(), on="event_time", by="episode_id",
                 preceding=1, aggregations={"value": "median"})
         with self.assertRaisesRegex(ValueError, "closed must be"):
-            pmm.aggregate_window(
+            pmm.join_window(
                 scan(), scan(), on="event_time", by="episode_id",
                 preceding=1, aggregations={"value": "mean"},
                 closed="middle")
         with self.assertRaisesRegex(ValueError, "missing aggregation columns"):
-            pmm.aggregate_window(
+            pmm.join_window(
                 scan(), scan(), on="event_time", by="episode_id",
                 preceding=1, aggregations={"missing": "mean"})
         with self.assertRaisesRegex(TypeError, "requires integer or floating"):
-            pmm.aggregate_window(
+            pmm.join_window(
                 scan(), table.scan().select("text"),
                 on="event_time", by="episode_id", preceding=1,
                 aggregations={"text": "first"}).to_arrow()

From 39c82bc2159b98fefb6347016a31102e6d53b539 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 07:15:22 -0700
Subject: [PATCH 05/13] [python] Align window join internal naming

---
 paimon-python/pypaimon/multimodal/temporal.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py
index 79b33d69c391..cef1ea13ff7b 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -160,7 +160,7 @@ def join_window(self, right, *, preceding, aggregations,
                     suffix="_right") -> "TemporalAlignment":
         """Append a right-side window join with aggregation."""
         position = len(self._sources) + 1
-        source = _WindowAggregationRight(
+        source = _WindowJoinRight(
             "right source %d" % position,
             right,
             self._on,
@@ -518,7 +518,7 @@ def build_arrays(self, anchor_rows, fetcher):
         return arrays
 
 
-class _WindowAggregationRight(_AsOfJoinRight):
+class _WindowJoinRight(_AsOfJoinRight):
 
     _SUPPORTED_AGGREGATIONS = {
         "count", "first", "last", "max", "mean", "min",

From f1c9ad7e6dcc457a3e746663ea7e9d7128c53b08 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 07:30:12 -0700
Subject: [PATCH 06/13] [python] Prune window join payload reads

---
 docs/docs/pypaimon/multimodal-reading.md      |  8 ++--
 paimon-python/pypaimon/multimodal/temporal.py |  6 +++
 .../tests/multimodal_temporal_test.py         | 45 +++++++++++++++++++
 3 files changed, 56 insertions(+), 3 deletions(-)

diff --git a/docs/docs/pypaimon/multimodal-reading.md b/docs/docs/pypaimon/multimodal-reading.md
index 13e141f66a07..08a74915143a 100644
--- a/docs/docs/pypaimon/multimodal-reading.md
+++ b/docs/docs/pypaimon/multimodal-reading.md
@@ -106,9 +106,11 @@ Use `join_window(left, right, ...)` directly or chain
 `aligned.join_window(...)`. Unlike single-table `rolling().agg()`, it joins
 each left row to right rows in the same `by` group and
 `[left - preceding, left + following]` time window, then aggregates them.
-`following` defaults to zero; `closed` controls endpoints. Supported
-aggregations are `mean`, `min`, `max`, `first`, `last`, and `count`. Nulls are
-skipped; empty windows return null, except `count` returns zero.
+`following` defaults to zero. Aggregation columns must be integer or
+floating-point scalars; `closed` is `both`, `left`, `right`, or `neither` and
+refers to the interval endpoints. Supported aggregations are `mean`, `min`,
+`max`, `first`, `last`, and `count`. Nulls are skipped; empty windows return
+null, except `count` returns zero.
 
 ```python
 steps_with_imu = aligned.join_window(
diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py
index cef1ea13ff7b..862f7707115e 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -543,6 +543,12 @@ def __init__(self, label, query, anchor_on, by, preceding, following,
         self.aggregations = _normalize_aggregations(
             aggregations, self.payload_schema, self.label,
             self._SUPPORTED_AGGREGATIONS)
+        query_schema, paths = _query_schema_and_paths(self.query)
+        self.query.select([
+            ".".join(path)
+            for field, path in zip(query_schema, paths)
+            if field.name in self.aggregations
+        ])
         self.payload_schema = pa.schema([
             field for field in self.payload_schema
             if field.name in self.aggregations
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index 6f2d1ec9efc3..303b3be12473 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -295,6 +295,51 @@ def test_window_join_supports_asymmetric_timestamp_bounds(self):
         ).to_list()[0]
         self.assertEqual(2.5, right_closed["value"])
 
+    def test_window_join_prunes_unaggregated_right_columns(self):
+        anchors = self._table("window_projection_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        samples = self._table("window_projection_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "payload": pa.struct([
+                pa.field("unused_nested", pa.int32()),
+                pa.field("value", pa.int32()),
+            ]),
+            "unused": pa.string(),
+        })
+        anchors.add([{"episode_id": 1, "event_time": 10}])
+        samples.add([{
+            "episode_id": 1,
+            "event_time": 10,
+            "payload": {"value": 7, "unused_nested": 8},
+            "unused": "not read",
+        }])
+        payload_reads = []
+        original = FormatPyArrowReader._read_parquet_row_group_batches
+
+        def tracked(reader, row_group, columns):
+            if columns is not None and "payload" in columns:
+                payload_reads.append(tuple(columns))
+            yield from original(reader, row_group, columns)
+
+        with mock.patch.object(
+                FormatPyArrowReader,
+                "_read_parquet_row_group_batches", tracked):
+            row = pmm.join_window(
+                anchors.scan(),
+                samples.scan().select(["payload.value", "unused"]),
+                on="event_time", by="episode_id", preceding=0,
+                aggregations={"payload_value": "mean"},
+            ).to_list()[0]
+
+        self.assertEqual(7.0, row["payload_value"])
+        self.assertTrue(payload_reads)
+        self.assertNotIn("unused", {
+            name for columns in payload_reads for name in columns
+        })
+
     def test_window_mean_avoids_numeric_overflow_and_integer_rounding(self):
         anchors = self._table("window_numeric_anchors", {
             "episode_id": pa.int32(),

From b14304a022fe5d9cfe33afd7a8d2027cef168c94 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 07:37:05 -0700
Subject: [PATCH 07/13] [python] Support named window aggregations

---
 docs/docs/pypaimon/multimodal-reading.md      | 16 +--
 paimon-python/pypaimon/multimodal/temporal.py | 99 ++++++++++++-------
 .../tests/multimodal_temporal_test.py         | 57 +++++------
 3 files changed, 100 insertions(+), 72 deletions(-)

diff --git a/docs/docs/pypaimon/multimodal-reading.md b/docs/docs/pypaimon/multimodal-reading.md
index 08a74915143a..234dd4f6e2ff 100644
--- a/docs/docs/pypaimon/multimodal-reading.md
+++ b/docs/docs/pypaimon/multimodal-reading.md
@@ -106,18 +106,22 @@ Use `join_window(left, right, ...)` directly or chain
 `aligned.join_window(...)`. Unlike single-table `rolling().agg()`, it joins
 each left row to right rows in the same `by` group and
 `[left - preceding, left + following]` time window, then aggregates them.
-`following` defaults to zero. Aggregation columns must be integer or
-floating-point scalars; `closed` is `both`, `left`, `right`, or `neither` and
-refers to the interval endpoints. Supported aggregations are `mean`, `min`,
-`max`, `first`, `last`, and `count`. Nulls are skipped; empty windows return
-null, except `count` returns zero.
+`following` defaults to zero. `mean`, `min`, and `max` require integer or
+floating-point scalars; `count`, `first`, and `last` also accept non-numeric
+values. `closed` is `both`, `left`, `right`, or `neither` and refers to the
+interval endpoints. Nulls are skipped; empty windows return null, except
+`count` returns zero. Use `(source, operation)` pairs to aggregate one source
+column more than once; `{"value": "mean"}` remains valid shorthand.
 
 ```python
 steps_with_imu = aligned.join_window(
     imu.scan().select("acceleration"),
     preceding=timedelta(milliseconds=50),
     following=timedelta(milliseconds=50),
-    aggregations={"acceleration": "mean"},
+    aggregations={
+        "acceleration_mean": ("acceleration", "mean"),
+        "acceleration_max": ("acceleration", "max"),
+    },
 )
 ```
 
diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py
index 862f7707115e..c992dedebf9c 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -284,11 +284,9 @@ def _output_schema(self, anchor_schema=None, source_fetchers=None):
                 source.payload_schema if source_fetchers is None
                 else source_fetchers[position].schema
             )
-            for name in source.payload_schema.names:
-                field = source.output_field(
-                    payload_schema.field(name),
-                    effective=source_fetchers is not None,
-                )
+            for field in source.output_fields(
+                    payload_schema,
+                    effective=source_fetchers is not None):
                 output_name = field.name
                 if output_name in names:
                     output_name += source.suffix
@@ -407,6 +405,12 @@ def match(self, anchor_row):
     def output_field(field, effective=True):
         return field
 
+    def output_fields(self, payload_schema, effective=True):
+        return [
+            self.output_field(payload_schema.field(name), effective)
+            for name in self.payload_schema.names
+        ]
+
     def build_arrays(self, anchor_rows, fetcher):
         matches = [self.match(row) for row in anchor_rows]
         matched_ids = [match for match in matches if match is not None]
@@ -543,28 +547,35 @@ def __init__(self, label, query, anchor_on, by, preceding, following,
         self.aggregations = _normalize_aggregations(
             aggregations, self.payload_schema, self.label,
             self._SUPPORTED_AGGREGATIONS)
+        source_names = {
+            specification[1] for specification in self.aggregations
+        }
         query_schema, paths = _query_schema_and_paths(self.query)
         self.query.select([
             ".".join(path)
             for field, path in zip(query_schema, paths)
-            if field.name in self.aggregations
+            if field.name in source_names
         ])
         self.payload_schema = pa.schema([
             field for field in self.payload_schema
-            if field.name in self.aggregations
+            if field.name in source_names
         ], metadata=self.payload_schema.metadata)
 
-    def output_field(self, field, effective=True):
-        try:
-            output_type = _aggregate_output_type(
-                field.type, self.aggregations[field.name])
-        except TypeError:
-            if effective:
-                raise
-            output_type = field.type
-        return pa.field(
-            field.name, output_type, nullable=True,
-            metadata=field.metadata)
+    def output_fields(self, payload_schema, effective=True):
+        fields = []
+        for output_name, source_name, aggregation in self.aggregations:
+            source = payload_schema.field(source_name)
+            try:
+                output_type = _aggregate_output_type(
+                    source.type, aggregation)
+            except TypeError:
+                if effective:
+                    raise
+                output_type = source.type
+            fields.append(pa.field(
+                output_name, output_type, nullable=True,
+                metadata=source.metadata))
+        return fields
 
     def match(self, anchor_row):
         key = tuple(anchor_row[name] for name in self.by)
@@ -601,14 +612,13 @@ def build_arrays(self, anchor_rows, fetcher):
             for match in matches
         ]
         arrays = []
-        for field in self.payload_schema:
-            effective = fetcher.schema.field(field.name)
-            aggregation = self.aggregations[field.name]
+        for _, source_name, aggregation in self.aggregations:
+            effective = fetcher.schema.field(source_name)
             output_type = _aggregate_output_type(
                 effective.type, aggregation)
             arrays.append(pa.array([
                 _aggregate_values(
-                    values[field.name], row_indices, aggregation)
+                    values[source_name], row_indices, aggregation)
                 for row_indices in indices
             ], type=output_type))
         return arrays
@@ -617,28 +627,49 @@ def build_arrays(self, anchor_rows, fetcher):
 def _normalize_aggregations(aggregations, schema, label, supported):
     if not isinstance(aggregations, dict) or not aggregations:
         raise ValueError("aggregations must be a non-empty dict.")
-    unknown = [name for name in aggregations if name not in schema.names]
-    if unknown:
-        raise ValueError(
-            "%s is missing aggregation columns %r." % (label, unknown))
-    for name, aggregation in aggregations.items():
+    normalized = []
+    missing = []
+    for output_name, specification in aggregations.items():
+        if not isinstance(output_name, str) or not output_name:
+            raise ValueError(
+                "Aggregation output names must be non-empty strings.")
+        if isinstance(specification, str):
+            source_name = output_name
+            aggregation = specification
+        elif isinstance(specification, tuple) and len(specification) == 2:
+            source_name, aggregation = specification
+        else:
+            raise ValueError(
+                "Aggregation %r must be an operation or a "
+                "(source column, operation) pair." % output_name)
+        if not isinstance(source_name, str) or not source_name:
+            raise ValueError(
+                "Aggregation source columns must be non-empty strings.")
+        if source_name not in schema.names:
+            missing.append(source_name)
         if not isinstance(aggregation, str) or aggregation not in supported:
             raise ValueError(
-                "Unsupported aggregation %r for column %r; expected one of "
-                "%r." % (aggregation, name, sorted(supported)))
-    return dict(aggregations)
+                "Unsupported aggregation %r for output %r; expected one of "
+                "%r." % (aggregation, output_name, sorted(supported)))
+        normalized.append((output_name, source_name, aggregation))
+    if missing:
+        raise ValueError(
+            "%s is missing aggregation columns %r." % (label, missing))
+    return tuple(normalized)
 
 
 def _aggregate_output_type(data_type, aggregation):
+    if aggregation == "count":
+        return pa.int64()
+    if aggregation in ("first", "last"):
+        return data_type
     if not (pa.types.is_integer(data_type)
             or pa.types.is_floating(data_type)):
         raise TypeError(
-            "Window aggregation requires integer or floating-point scalar "
-            "columns; got %s." % data_type)
+            "Window %s aggregation requires an integer or floating-point "
+            "scalar column; got %s." % (aggregation, data_type))
     if aggregation == "mean":
         return pa.float64()
-    if aggregation == "count":
-        return pa.int64()
     return data_type
 
 
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index 303b3be12473..74b001d09b8c 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -196,12 +196,8 @@ def test_window_join_stays_in_group_and_skips_nulls(self):
         samples = self._table("window_samples", {
             "episode_id": pa.int32(),
             "event_time": pa.int64(),
-            "average": pa.int32(),
-            "minimum": pa.int32(),
-            "maximum": pa.int32(),
-            "first_value": pa.int32(),
-            "last_value": pa.int32(),
-            "valid_count": pa.int32(),
+            "value": pa.int32(),
+            "label": pa.string(),
         })
         anchors.add([
             {"episode_id": 1, "event_time": 10},
@@ -209,47 +205,43 @@ def test_window_join_stays_in_group_and_skips_nulls(self):
             {"episode_id": 3, "event_time": 10},
         ])
         samples.add([
-            dict({"episode_id": 1, "event_time": time}, **{
-                name: value for name in (
-                    "average", "minimum", "maximum", "first_value",
-                    "last_value", "valid_count")
-            })
-            for time, value in ((5, 1), (10, None), (15, 5))
-        ] + [dict({"episode_id": 2, "event_time": 10}, **{
-            name: 100 for name in (
-                "average", "minimum", "maximum", "first_value",
-                "last_value", "valid_count")
-        })])
+            {"episode_id": 1, "event_time": 5,
+             "value": 1, "label": "a"},
+            {"episode_id": 1, "event_time": 10,
+             "value": None, "label": None},
+            {"episode_id": 1, "event_time": 15,
+             "value": 5, "label": "c"},
+            {"episode_id": 2, "event_time": 10,
+             "value": 100, "label": "z"},
+        ])
 
         result = pmm.join_window(
             anchors.scan(),
-            samples.scan().select([
-                "average", "minimum", "maximum", "first_value",
-                "last_value", "valid_count",
-            ]),
+            samples.scan().select(["value", "label"]),
             on="event_time",
             by="episode_id",
             preceding=5,
             following=5,
             aggregations={
-                "average": "mean",
-                "minimum": "min",
-                "maximum": "max",
-                "first_value": "first",
-                "last_value": "last",
-                "valid_count": "count",
+                "average": ("value", "mean"),
+                "minimum": ("value", "min"),
+                "maximum": ("value", "max"),
+                "first_label": ("label", "first"),
+                "last_label": ("label", "last"),
+                "valid_count": ("label", "count"),
             },
         )
         rows = sorted(result.to_list(), key=lambda row: row["episode_id"])
 
         self.assertIsInstance(result, pmm.TemporalAlignment)
         self.assertEqual(pa.float64(), result.schema.field("average").type)
+        self.assertEqual(pa.string(), result.schema.field("first_label").type)
         self.assertEqual(pa.int64(), result.schema.field("valid_count").type)
         self.assertEqual(
-            (3.0, 1, 5, 1, 5, 2),
+            (3.0, 1, 5, "a", "c", 2),
             tuple(rows[0][name] for name in (
-                "average", "minimum", "maximum", "first_value",
-                "last_value", "valid_count")),
+                "average", "minimum", "maximum", "first_label",
+                "last_label", "valid_count")),
         )
         self.assertEqual(100.0, rows[1]["average"])
         self.assertIsNone(rows[2]["average"])
@@ -436,11 +428,12 @@ def scan():
             pmm.join_window(
                 scan(), scan(), on="event_time", by="episode_id",
                 preceding=1, aggregations={"missing": "mean"})
-        with self.assertRaisesRegex(TypeError, "requires integer or floating"):
+        with self.assertRaisesRegex(
+                TypeError, "requires an integer or floating"):
             pmm.join_window(
                 scan(), table.scan().select("text"),
                 on="event_time", by="episode_id", preceding=1,
-                aggregations={"text": "first"}).to_arrow()
+                aggregations={"text": "mean"}).to_arrow()
 
     def test_linear_interpolation_preserves_an_exact_infinite_float(self):
         anchors = self._table("linear_exact_anchors", {

From b501c6612e7193be85d8ddfd79b6fddb5ba1ca71 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 08:16:28 -0700
Subject: [PATCH 08/13] [python] Fix window join precision and projection

---
 paimon-python/pypaimon/multimodal/temporal.py | 54 ++++++++++---
 .../tests/multimodal_temporal_test.py         | 77 +++++++++++++++++++
 2 files changed, 120 insertions(+), 11 deletions(-)

diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py
index c992dedebf9c..f5ef06197864 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -18,6 +18,7 @@
 
 from bisect import bisect_left, bisect_right
 from datetime import timedelta
+from fractions import Fraction
 import json
 import math
 from numbers import Integral, Real
@@ -202,7 +203,8 @@ def to_arrow_batch_reader(self, *, batch_size=1024):
         for source in self._sources:
             source.plan()
             source_fetchers.append(
-                _RowIdFetcher(source.query, row_group_cache))
+                _RowIdFetcher(
+                    source.query, row_group_cache, source._fetch_names))
         schema = self._output_schema(anchor_fetcher.schema, source_fetchers)
         self.schema = schema
 
@@ -365,6 +367,7 @@ def __init__(self, label, query, anchor_on, by, direction, tolerance,
             field for field, path in zip(schema, paths)
             if tuple(path) not in excluded
         ])
+        self._fetch_names = None
         self._index = None
 
     def plan(self):
@@ -550,12 +553,9 @@ def __init__(self, label, query, anchor_on, by, preceding, following,
         source_names = {
             specification[1] for specification in self.aggregations
         }
-        query_schema, paths = _query_schema_and_paths(self.query)
-        self.query.select([
-            ".".join(path)
-            for field, path in zip(query_schema, paths)
-            if field.name in source_names
-        ])
+        self._fetch_names = tuple(
+            field.name for field in self.payload_schema
+            if field.name in source_names)
         self.payload_schema = pa.schema([
             field for field in self.payload_schema
             if field.name in source_names
@@ -688,6 +688,10 @@ def _aggregate_values(values, indices, aggregation):
             return sum(items) / len(items)
         if not all(math.isfinite(item) for item in items):
             return pc.mean(selected).as_py()
+        try:
+            return math.fsum(items) / len(items)
+        except OverflowError:
+            pass
         scale = max(abs(item) for item in items)
         if scale == 0:
             return 0.0
@@ -713,6 +717,14 @@ def _window_bound_key(name, value, data_type):
     if value < zero:
         raise ValueError("%s must be non-negative." % name)
     _validate_tolerance(value, data_type)
+    if pa.types.is_integer(data_type) and not isinstance(value, Integral):
+        try:
+            exact = Fraction(value)
+        except TypeError:
+            exact = Fraction(float(value))
+        if exact.denominator == 1:
+            return exact.numerator
+        return exact
     return _time_tolerance_key(value, data_type)
 
 
@@ -1001,13 +1013,29 @@ def _validate_metadata(query, metadata, key_columns):
 
 class _RowIdFetcher:
 
-    def __init__(self, query, row_group_cache):
+    def __init__(self, query, row_group_cache, output_names=None):
         _validate_pinned_tag(query)
-        self._schema = _query_schema(query)
+        query_schema, query_paths = _query_schema_and_paths(query)
+        visible_projection = query._effective_projection()
+        if output_names is None:
+            self._schema = query_schema
+        else:
+            output_names = set(output_names)
+            selected = [
+                (field, path)
+                for field, path in zip(query_schema, query_paths)
+                if field.name in output_names
+            ]
+            self._schema = pa.schema(
+                [field for field, unused in selected],
+                metadata=query_schema.metadata,
+            )
+            visible_projection = [
+                ".".join(path) for unused, path in selected
+            ]
         table = query._table.copy_without_time_travel({
             CoreOptions.BLOB_AS_DESCRIPTOR.key(): "true",
         })
-        visible_projection = query._effective_projection()
         plan_builder = table.new_read_builder()
         if visible_projection is not None:
             plan_projection = visible_projection
@@ -1168,7 +1196,11 @@ def fetch(self, row_ids):
             )
         take = pa.array(
             [positions[row_id] for row_id in row_ids], type=pa.int64())
-        return arrow.select(self._schema.names).take(take)
+        visible = pa.Table.from_arrays(
+            [arrow.column(index) for index in range(len(self._schema))],
+            schema=self._schema,
+        )
+        return visible.take(take)
 
     def _find_splits(self, ranges):
         split_indices = set()
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index 74b001d09b8c..bec6d6dbb8c9 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -287,6 +287,32 @@ def test_window_join_supports_asymmetric_timestamp_bounds(self):
         ).to_list()[0]
         self.assertEqual(2.5, right_closed["value"])
 
+    def test_window_join_keeps_float_bounds_exact_for_integer_time(self):
+        anchors = self._table("window_integer_bound_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        samples = self._table("window_integer_bound_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "value": pa.int32(),
+        })
+        timestamp = 1_700_000_000_000_000_001
+        anchors.add([{"episode_id": 1, "event_time": timestamp}])
+        samples.add([
+            {"episode_id": 1, "event_time": timestamp - 1, "value": 1},
+            {"episode_id": 1, "event_time": timestamp, "value": 2},
+        ])
+
+        row = pmm.join_window(
+            anchors.scan(), samples.scan().select("value"),
+            on="event_time", by="episode_id",
+            preceding=0.0, following=0.0,
+            aggregations={"matches": ("value", "count")},
+        ).to_list()[0]
+
+        self.assertEqual(1, row["matches"])
+
     def test_window_join_prunes_unaggregated_right_columns(self):
         anchors = self._table("window_projection_anchors", {
             "episode_id": pa.int32(),
@@ -332,6 +358,33 @@ def tracked(reader, row_group, columns):
             name for columns in payload_reads for name in columns
         })
 
+    def test_window_join_preserves_nested_projection_aliases_when_pruned(self):
+        anchors = self._table("window_alias_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        samples = self._table("window_alias_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "a_b": pa.int32(),
+            "a": pa.struct([pa.field("b", pa.int32())]),
+        })
+        anchors.add([{"episode_id": 1, "event_time": 10}])
+        samples.add([{
+            "episode_id": 1,
+            "event_time": 10,
+            "a_b": 3,
+            "a": {"b": 7},
+        }])
+
+        row = pmm.join_window(
+            anchors.scan(), samples.scan().select(["a_b", "a.b"]),
+            on="event_time", by="episode_id", preceding=0,
+            aggregations={"nested_mean": ("a_b__0", "mean")},
+        ).to_list()[0]
+
+        self.assertEqual(7.0, row["nested_mean"])
+
     def test_window_mean_avoids_numeric_overflow_and_integer_rounding(self):
         anchors = self._table("window_numeric_anchors", {
             "episode_id": pa.int32(),
@@ -366,6 +419,30 @@ def test_window_mean_avoids_numeric_overflow_and_integer_rounding(self):
         self.assertEqual(-0.5, row["integer_value"])
         self.assertEqual(sys.float_info.max, row["float_value"])
 
+    def test_window_mean_preserves_finite_float_cancellation(self):
+        anchors = self._table("window_float_mean_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        samples = self._table("window_float_mean_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "value": pa.float64(),
+        })
+        anchors.add([{"episode_id": 1, "event_time": 1}])
+        samples.add([
+            {"episode_id": 1, "event_time": 0, "value": 1e16},
+            {"episode_id": 1, "event_time": 2, "value": -1e16 + 2},
+        ])
+
+        row = pmm.join_window(
+            anchors.scan(), samples.scan().select("value"),
+            on="event_time", by="episode_id", preceding=1, following=1,
+            aggregations={"value": "mean"},
+        ).to_list()[0]
+
+        self.assertEqual(1.0, row["value"])
+
     def test_window_join_can_follow_an_asof_join(self):
         anchors = self._table("window_chain_anchors", {
             "episode_id": pa.int32(),

From 1537d24d9591182308ed3fcafccbb9579afaf253 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 08:28:00 -0700
Subject: [PATCH 09/13] [python] Fix fractional integer window bounds

---
 paimon-python/pypaimon/multimodal/temporal.py | 38 ++++++++++++++-----
 .../tests/multimodal_temporal_test.py         | 17 +++++----
 2 files changed, 38 insertions(+), 17 deletions(-)

diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py
index f5ef06197864..a5c9fc9f57f5 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -586,16 +586,34 @@ def match(self, anchor_row):
         start, end = bounds
         left = target - self._preceding_key
         right = target + self._following_key
-        first = (
-            bisect_left(self._time_keys, left, start, end)
-            if self.closed in ("both", "left")
-            else bisect_right(self._time_keys, left, start, end)
-        )
-        last = (
-            bisect_right(self._time_keys, right, first, end)
-            if self.closed in ("both", "right")
-            else bisect_left(self._time_keys, right, first, end)
-        )
+        if pa.types.is_integer(self.time_type):
+            first_key = (
+                math.ceil(left)
+                if self.closed in ("both", "left")
+                else math.floor(left) + 1
+            )
+            last_key = (
+                math.floor(right)
+                if self.closed in ("both", "right")
+                else math.ceil(right) - 1
+            )
+            if first_key > last_key:
+                return []
+            first = bisect_left(
+                self._time_keys, first_key, start, end)
+            last = bisect_right(
+                self._time_keys, last_key, first, end)
+        else:
+            first = (
+                bisect_left(self._time_keys, left, start, end)
+                if self.closed in ("both", "left")
+                else bisect_right(self._time_keys, left, start, end)
+            )
+            last = (
+                bisect_right(self._time_keys, right, first, end)
+                if self.closed in ("both", "right")
+                else bisect_left(self._time_keys, right, first, end)
+            )
         return [self._row_ids[index].as_py()
                 for index in range(first, last)]
 
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index bec6d6dbb8c9..cf0cee9e47e2 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -302,16 +302,19 @@ def test_window_join_keeps_float_bounds_exact_for_integer_time(self):
         samples.add([
             {"episode_id": 1, "event_time": timestamp - 1, "value": 1},
             {"episode_id": 1, "event_time": timestamp, "value": 2},
+            {"episode_id": 1, "event_time": timestamp + 1, "value": 3},
         ])
 
-        row = pmm.join_window(
-            anchors.scan(), samples.scan().select("value"),
-            on="event_time", by="episode_id",
-            preceding=0.0, following=0.0,
-            aggregations={"matches": ("value", "count")},
-        ).to_list()[0]
+        for width in (0.0, 0.125):
+            with self.subTest(width=width):
+                row = pmm.join_window(
+                    anchors.scan(), samples.scan().select("value"),
+                    on="event_time", by="episode_id",
+                    preceding=width, following=width,
+                    aggregations={"matches": ("value", "count")},
+                ).to_list()[0]
 
-        self.assertEqual(1, row["matches"])
+                self.assertEqual(1, row["matches"])
 
     def test_window_join_prunes_unaggregated_right_columns(self):
         anchors = self._table("window_projection_anchors", {

From 38ef500dfd3f70c62623065f9d3d99f0c097687d Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 08:35:36 -0700
Subject: [PATCH 10/13] [python] Fix integer window boundary comparisons

---
 paimon-python/pypaimon/multimodal/temporal.py |  8 +-
 .../tests/multimodal_temporal_test.py         | 77 ++++++++++++++++---
 2 files changed, 73 insertions(+), 12 deletions(-)

diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py
index a5c9fc9f57f5..08387dbbdf20 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -586,7 +586,8 @@ def match(self, anchor_row):
         start, end = bounds
         left = target - self._preceding_key
         right = target + self._following_key
-        if pa.types.is_integer(self.time_type):
+        if (pa.types.is_integer(self.time_type)
+                or pa.types.is_timestamp(self.time_type)):
             first_key = (
                 math.ceil(left)
                 if self.closed in ("both", "left")
@@ -597,6 +598,9 @@ def match(self, anchor_row):
                 if self.closed in ("both", "right")
                 else math.ceil(right) - 1
             )
+            # Avoid comparing NumPy keys with out-of-range Python integers.
+            first_key = max(first_key, _python_scalar(self._time_keys[start]))
+            last_key = min(last_key, _python_scalar(self._time_keys[end - 1]))
             if first_key > last_key:
                 return []
             first = bisect_left(
@@ -728,6 +732,8 @@ def _window_bound_key(name, value, data_type):
     if isinstance(value, bool) or not isinstance(value, (Real, timedelta)):
         raise TypeError(
             "%s must be numeric or datetime.timedelta." % name)
+    if isinstance(value, Integral):
+        value = int(value)
     if (isinstance(value, Real) and not isinstance(value, Integral)
             and not math.isfinite(value)):
         raise ValueError("%s must be finite." % name)
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index cf0cee9e47e2..0ebb5b880fd2 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -23,6 +23,7 @@
 from datetime import datetime, timedelta
 from unittest import mock
 
+import numpy as np
 import pyarrow as pa
 import pypaimon.multimodal as pmm
 from pypaimon.multimodal import temporal
@@ -287,7 +288,7 @@ def test_window_join_supports_asymmetric_timestamp_bounds(self):
         ).to_list()[0]
         self.assertEqual(2.5, right_closed["value"])
 
-    def test_window_join_keeps_float_bounds_exact_for_integer_time(self):
+    def test_window_join_keeps_numeric_bounds_exact_for_integer_time(self):
         anchors = self._table("window_integer_bound_anchors", {
             "episode_id": pa.int32(),
             "event_time": pa.int64(),
@@ -298,23 +299,77 @@ def test_window_join_keeps_float_bounds_exact_for_integer_time(self):
             "value": pa.int32(),
         })
         timestamp = 1_700_000_000_000_000_001
-        anchors.add([{"episode_id": 1, "event_time": timestamp}])
-        samples.add([
-            {"episode_id": 1, "event_time": timestamp - 1, "value": 1},
-            {"episode_id": 1, "event_time": timestamp, "value": 2},
-            {"episode_id": 1, "event_time": timestamp + 1, "value": 3},
-        ])
+        anchors.add(pa.Table.from_pydict({
+            "episode_id": [1], "event_time": [timestamp],
+        }))
+        samples.add(pa.Table.from_pydict({
+            "episode_id": [1, 1, 1],
+            "event_time": [timestamp - 1, timestamp, timestamp + 1],
+            "value": [1, 2, 3],
+        }))
 
-        for width in (0.0, 0.125):
-            with self.subTest(width=width):
+        for preceding, following in (
+                (0.0, 0.0), (0.125, 0.125),
+                (np.int64(0), 0), (0, np.int64(0)),
+                (np.int64(0), np.int64(0))):
+            with self.subTest(preceding=preceding, following=following):
                 row = pmm.join_window(
                     anchors.scan(), samples.scan().select("value"),
                     on="event_time", by="episode_id",
-                    preceding=width, following=width,
-                    aggregations={"matches": ("value", "count")},
+                    preceding=preceding, following=following,
+                    aggregations={
+                        "matches": ("value", "count"),
+                        "first_value": ("value", "first"),
+                    },
                 ).to_list()[0]
 
                 self.assertEqual(1, row["matches"])
+                self.assertEqual(2, row["first_value"])
+
+    def test_window_join_clips_bounds_to_integer_time_range(self):
+        anchors = self._table("window_range_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        samples = self._table("window_range_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "value": pa.int32(),
+        })
+        minimum, maximum = -(1 << 63), (1 << 63) - 1
+        anchors.add(pa.Table.from_pydict({
+            "episode_id": [1, 2], "event_time": [minimum, maximum],
+        }))
+        times = {1: [minimum, minimum + 1], 2: [maximum - 1, maximum]}
+        samples.add(pa.Table.from_pydict({
+            "episode_id": [1, 1, 2, 2],
+            "event_time": times[1] + times[2],
+            "value": [1, 2, 3, 4],
+        }))
+
+        for preceding, following in ((0, 0), (0, 1), (1, 0), (1, 1)):
+            for closed in ("both", "left", "right", "neither"):
+                with self.subTest(
+                        preceding=preceding, following=following,
+                        closed=closed):
+                    rows = pmm.join_window(
+                        anchors.scan(), samples.scan().select("value"),
+                        on="event_time", by="episode_id",
+                        preceding=preceding, following=following,
+                        closed=closed,
+                        aggregations={"matches": ("value", "count")},
+                    ).to_list()
+                    for row in rows:
+                        left = row["event_time"] - preceding
+                        right = row["event_time"] + following
+                        expected = sum(
+                            (time >= left if closed in ("both", "left")
+                             else time > left)
+                            and (time <= right if closed in ("both", "right")
+                                 else time < right)
+                            for time in times[row["episode_id"]]
+                        )
+                        self.assertEqual(expected, row["matches"])
 
     def test_window_join_prunes_unaggregated_right_columns(self):
         anchors = self._table("window_projection_anchors", {

From 3b9ec626d3eb53284e472f648f93f6033f33dd9b Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 08:56:55 -0700
Subject: [PATCH 11/13] [python] Preserve masked alias types when pruning
 window columns

---
 paimon-python/pypaimon/multimodal/temporal.py | 10 +---
 .../tests/multimodal_temporal_test.py         | 56 +++++++++++++++++++
 2 files changed, 59 insertions(+), 7 deletions(-)

diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py
index 08387dbbdf20..a45fd0dbcc95 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -1043,6 +1043,7 @@ def __init__(self, query, row_group_cache, output_names=None):
         visible_projection = query._effective_projection()
         if output_names is None:
             self._schema = query_schema
+            visible_paths = query_paths
         else:
             output_names = set(output_names)
             selected = [
@@ -1054,9 +1055,8 @@ def __init__(self, query, row_group_cache, output_names=None):
                 [field for field, unused in selected],
                 metadata=query_schema.metadata,
             )
-            visible_projection = [
-                ".".join(path) for unused, path in selected
-            ]
+            visible_paths = [path for unused, path in selected]
+            visible_projection = [".".join(path) for path in visible_paths]
         table = query._table.copy_without_time_travel({
             CoreOptions.BLOB_AS_DESCRIPTOR.key(): "true",
         })
@@ -1131,10 +1131,6 @@ def __init__(self, query, row_group_cache, output_names=None):
             builder.read_type())
         effective_schema = _effective_masked_schema(
             physical_schema, masking)
-        visible_paths = (
-            None if self._name_paths is None
-            else self._name_paths[:len(self._schema)]
-        )
         self._schema = _project_effective_schema(
             self._schema, visible_paths, effective_schema, masking)
         self._fetch_schema = _project_effective_schema(
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index 0ebb5b880fd2..4ebf57a25227 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -443,6 +443,62 @@ def test_window_join_preserves_nested_projection_aliases_when_pruned(self):
 
         self.assertEqual(7.0, row["nested_mean"])
 
+    def test_window_join_preserves_masked_alias_types_when_pruned(self):
+        anchors = self._table("window_masked_alias_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        samples = self._table("window_masked_alias_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "a_b": pa.int32(),
+            "a": pa.struct([pa.field("b", pa.int32())]),
+        })
+        anchors.add(pa.Table.from_pydict({
+            "episode_id": [1, 1], "event_time": [10, 20],
+        }))
+        samples.add(pa.Table.from_pydict({
+            "episode_id": [1], "event_time": [10],
+            "a_b": [3],
+            "a": pa.array([{"b": 7}],
+                          type=pa.struct([pa.field("b", pa.int32())])),
+        }))
+        auth = TableQueryAuthResult(
+            filter=None,
+            column_masking={"a_b": json.dumps({
+                "name": "CAST",
+                "fieldRef": {"index": 2, "name": "a_b", "type": "INT"},
+                "type": "STRING",
+            })},
+        )
+        samples.raw_table.catalog_environment.table_query_auth = (
+            lambda options, identifier: lambda select: auth)
+
+        for projection, source in (
+                (["a_b"], "a_b"), (["a.b", "a_b"], "a_b__0")):
+            with self.subTest(projection=projection):
+                result = pmm.join_window(
+                    anchors.scan(), samples.scan().select(projection),
+                    on="event_time", by="episode_id", preceding=0,
+                    aggregations={
+                        "first_value": (source, "first"),
+                        "last_value": (source, "last"),
+                    },
+                ).to_arrow()
+                for name in ("first_value", "last_value"):
+                    self.assertEqual(pa.string(), result[name].type)
+                    self.assertEqual(["3", None], result[name].to_pylist())
+
+            for operation in ("mean", "min", "max"):
+                with self.subTest(projection=projection, operation=operation):
+                    with self.assertRaisesRegex(
+                            TypeError, "requires an integer or floating"):
+                        pmm.join_window(
+                            anchors.scan(), samples.scan().select(projection),
+                            on="event_time", by="episode_id", preceding=0,
+                            aggregations={"value": (source, operation)},
+                        ).to_arrow()
+
     def test_window_mean_avoids_numeric_overflow_and_integer_rounding(self):
         anchors = self._table("window_numeric_anchors", {
             "episode_id": pa.int32(),

From 8edc1a0d7cda9248d75f5ca7fd4c091d9e3768c6 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sat, 12 Sep 2026 09:00:08 -0700
Subject: [PATCH 12/13] [python] Cover numeric masking of pruned window aliases

---
 .../tests/multimodal_temporal_test.py         | 42 +++++++++++++++++++
 1 file changed, 42 insertions(+)

diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index 4ebf57a25227..e5aae8cbbb32 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -499,6 +499,48 @@ def test_window_join_preserves_masked_alias_types_when_pruned(self):
                             aggregations={"value": (source, operation)},
                         ).to_arrow()
 
+    def test_window_mean_uses_masked_numeric_alias_type_when_pruned(self):
+        anchors = self._table("window_numeric_alias_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        samples = self._table("window_numeric_alias_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "a_b": pa.string(),
+            "a": pa.struct([pa.field("b", pa.int32())]),
+        })
+        anchors.add(pa.Table.from_pydict({
+            "episode_id": [1, 1], "event_time": [10, 20],
+        }))
+        samples.add(pa.Table.from_pydict({
+            "episode_id": [1, 1], "event_time": [9, 11],
+            "a_b": ["6", "10"],
+            "a": pa.array([{"b": 1}, {"b": 2}],
+                          type=pa.struct([pa.field("b", pa.int32())])),
+        }))
+        auth = TableQueryAuthResult(
+            filter=None,
+            column_masking={"a_b": json.dumps({
+                "name": "CAST",
+                "fieldRef": {"index": 2, "name": "a_b", "type": "STRING"},
+                "type": "DOUBLE",
+            })},
+        )
+        samples.raw_table.catalog_environment.table_query_auth = (
+            lambda options, identifier: lambda select: auth)
+
+        for projection, source in (
+                (["a_b"], "a_b"), (["a.b", "a_b"], "a_b__0")):
+            with self.subTest(projection=projection):
+                result = pmm.join_window(
+                    anchors.scan(), samples.scan().select(projection),
+                    on="event_time", by="episode_id", preceding=1, following=1,
+                    aggregations={"value": (source, "mean")},
+                ).to_arrow()
+                self.assertEqual(pa.float64(), result["value"].type)
+                self.assertEqual([8.0, None], result["value"].to_pylist())
+
     def test_window_mean_avoids_numeric_overflow_and_integer_rounding(self):
         anchors = self._table("window_numeric_anchors", {
             "episode_id": pa.int32(),

From bf78e7d81a49efd76d1a6652d2d3128a40263db5 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Sun, 13 Sep 2026 14:12:27 +0800
Subject: [PATCH 13/13] python: fix temporal window boundary precision

---
 paimon-python/pypaimon/multimodal/temporal.py | 16 +++-
 .../tests/multimodal_temporal_test.py         | 93 +++++++++++++++++++
 2 files changed, 107 insertions(+), 2 deletions(-)

diff --git a/paimon-python/pypaimon/multimodal/temporal.py b/paimon-python/pypaimon/multimodal/temporal.py
index a45fd0dbcc95..bb42053ef399 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -732,6 +732,8 @@ def _window_bound_key(name, value, data_type):
     if isinstance(value, bool) or not isinstance(value, (Real, timedelta)):
         raise TypeError(
             "%s must be numeric or datetime.timedelta." % name)
+    if isinstance(value, Real):
+        value = _python_scalar(value)
     if isinstance(value, Integral):
         value = int(value)
     if (isinstance(value, Real) and not isinstance(value, Integral)
@@ -1088,8 +1090,10 @@ def __init__(self, query, row_group_cache, output_names=None):
             projected_builder.read_type())
         projected_paths = projected_builder._nested_name_paths()
         if projected_paths is not None:
+            table_names = set(_table_schema(query).names)
             for field, path in zip(projected_schema, projected_paths):
-                if field.name in masking and field.name != path[0]:
+                if (field.name in masking and field.name != path[0]
+                        and field.name not in table_names):
                     raise ValueError(
                         "Temporal alignment cannot safely apply column "
                         "masking to nested projection %r."
@@ -1481,7 +1485,15 @@ def _time_search_keys(values, data_type):
 def _time_tolerance_key(tolerance, data_type):
     if tolerance is None or not pa.types.is_timestamp(data_type):
         return tolerance
-    return pa.scalar(tolerance, type=pa.duration(data_type.unit)).value
+    microseconds = (
+        (tolerance.days * 24 * 60 * 60 + tolerance.seconds) * 1_000_000
+        + tolerance.microseconds
+    )
+    divisors = {"s": 1_000_000, "ms": 1_000, "us": 1}
+    if data_type.unit == "ns":
+        return microseconds * 1_000
+    exact = Fraction(microseconds, divisors[data_type.unit])
+    return exact.numerator if exact.denominator == 1 else exact
 
 
 def _python_scalar(value):
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index e5aae8cbbb32..5ee34556d4d1 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -288,6 +288,68 @@ def test_window_join_supports_asymmetric_timestamp_bounds(self):
         ).to_list()[0]
         self.assertEqual(2.5, right_closed["value"])
 
+    def test_window_join_preserves_subunit_timestamp_bounds(self):
+        anchors = self._table("window_subunit_timestamp_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.timestamp("ms"),
+        })
+        samples = self._table("window_subunit_timestamp_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.timestamp("ms"),
+            "value": pa.int32(),
+        })
+        anchor = datetime(2026, 9, 1, 12, 0, 0)
+        anchors.add([{"episode_id": 1, "event_time": anchor}])
+        samples.add([
+            {"episode_id": 1,
+             "event_time": anchor + timedelta(milliseconds=offset),
+             "value": offset}
+            for offset in (-1, 0, 1)
+        ])
+
+        exact = pmm.join_window(
+            anchors.scan(), samples.scan().select("value"),
+            on="event_time", by="episode_id",
+            preceding=timedelta(microseconds=500),
+            following=timedelta(0), closed="right",
+            aggregations={"matches": ("value", "count")},
+        ).to_list()[0]
+        open_window = pmm.join_window(
+            anchors.scan(), samples.scan().select("value"),
+            on="event_time", by="episode_id",
+            preceding=timedelta(microseconds=1_500),
+            following=timedelta(microseconds=1_500), closed="neither",
+            aggregations={"matches": ("value", "count")},
+        ).to_list()[0]
+
+        self.assertEqual(1, exact["matches"])
+        self.assertEqual(3, open_window["matches"])
+
+    def test_window_join_normalizes_numpy_float_bounds(self):
+        anchors = self._table("window_numpy_float_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.float64(),
+        })
+        samples = self._table("window_numpy_float_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.float64(),
+            "value": pa.int32(),
+        })
+        timestamp = 1_700_000_000.001
+        anchors.add([{"episode_id": 1, "event_time": timestamp}])
+        samples.add([{
+            "episode_id": 1, "event_time": timestamp, "value": 7,
+        }])
+
+        row = pmm.join_window(
+            anchors.scan(), samples.scan().select("value"),
+            on="event_time", by="episode_id",
+            preceding=0, following=np.float32(0),
+            aggregations={"matches": ("value", "count")},
+        ).to_list()[0]
+
+        self.assertEqual(1, row["matches"])
+
     def test_window_join_keeps_numeric_bounds_exact_for_integer_time(self):
         anchors = self._table("window_integer_bound_anchors", {
             "episode_id": pa.int32(),
@@ -499,6 +561,37 @@ def test_window_join_preserves_masked_alias_types_when_pruned(self):
                             aggregations={"value": (source, operation)},
                         ).to_arrow()
 
+    def test_window_join_matches_masks_by_original_nested_path(self):
+        anchors = self._table("window_nested_mask_anchors", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+        })
+        samples = self._table("window_nested_mask_samples", {
+            "episode_id": pa.int32(),
+            "event_time": pa.int64(),
+            "a_b": pa.int32(),
+            "a": pa.struct([pa.field("b", pa.int32())]),
+        })
+        anchors.add([{"episode_id": 1, "event_time": 10}])
+        samples.add([{
+            "episode_id": 1, "event_time": 10,
+            "a_b": 3, "a": {"b": 7},
+        }])
+        auth = TableQueryAuthResult(
+            filter=None,
+            column_masking={"a_b": json.dumps({"name": "NULL"})},
+        )
+        samples.raw_table.catalog_environment.table_query_auth = (
+            lambda options, identifier: lambda select: auth)
+
+        row = pmm.join_window(
+            anchors.scan(), samples.scan().select(["a_b", "a.b"]),
+            on="event_time", by="episode_id", preceding=0,
+            aggregations={"nested_mean": ("a_b__0", "mean")},
+        ).to_list()[0]
+
+        self.assertEqual(7.0, row["nested_mean"])
+
     def test_window_mean_uses_masked_numeric_alias_type_when_pruned(self):
         anchors = self._table("window_numeric_alias_anchors", {
             "episode_id": pa.int32(),