[python] Extend ContiguousWindowDataset for frame-level training

docs/docs/pypaimon/multimodal-reading.md

@@ -182,61 +182,124 @@ Notes:
 
 ### Contiguous windows for PyTorch
 
-Install the `torch` extra, then use `to_contiguous_window_dataset` to expose
-map-style windows without loading the selected rows or BLOB payloads into Python
-memory up front. The Dataset builds a compact index from the group column, order
-column, and Paimon row IDs. Each `__getitem__` call fetches only that window from
-the snapshot recorded in `dataset.snapshot_id`.
+Use `to_contiguous_window_dataset` to build map-style training samples from a
+single frame table. Each row represents one time step; task text, labels, and
+other context must already be materialized in that table. The Dataset does not
+read companion tables. It builds a compact index of group/order values and
+Paimon row IDs, then fetches requested columns and BLOB payloads on demand from
+the snapshot recorded in `dataset.snapshot_id`. Row tracking must be enabled.
 
 ```shell
 pip install 'pypaimon[torch]'
 ```
 
 ```python
-import torch
-
+from functools import partial
 
-def float32_window(values):
-    return torch.tensor(values, dtype=torch.float32)
+import torch
+from pypaimon.multimodal.window_transforms import images_to_tensor, to_tensor
 
+float32_window = partial(to_tensor, dtype=torch.float32)
 
 windows = (
-    frames.scan()
+    frames.scan(tag_name="train-v1")
     .where("split = 'train'")
     .to_contiguous_window_dataset(
-        window_size=16,
-        columns=["state", "action"],
+        columns=["state", "image", "action"],
         group_key="episode_index",
         order_key="frame_index",
-        tail="pad",
+        frame_offsets={"state": [-2, -1, 0], "action": list(range(16))},
+        boundary="pad",
         column_transforms={
             "state": float32_window,
+            "image": images_to_tensor,
             "action": float32_window,
         },
     )
 )
 
 sample = windows[0]
 assert sample["action"].shape == (16, action_size)
-assert sample["is_pad"].shape == (16,)
+assert sample["action_is_pad"].shape == (16,)
+assert sample["state"].shape == (3, state_size)
+assert sample["image"].shape == (1, channels, height, width)
 ```
 
-The group and order keys in a sample identify the window anchor. Every projected
-column contains the whole window. With `tail="drop"`, only full windows are
-exposed. With `tail="pad"`, every real row is an anchor; missing suffix values
-repeat the last real value by default and `is_pad` is `True` exactly at those
-positions. With `tail="error"`, construction fails if any scheduled anchor is
-incomplete. Use `pad_values` to override the repeated value for individual
-columns. Anchors advance by `stride`, which defaults to one row.
+The group and order keys identify the anchor. `frame_offsets` defines each
+column's relative frame positions, including history, future, and sparse
+sampling. Offset order and duplicates are preserved. A selected column omitted
+from the mapping uses `[0]`, so the image above is read only at the anchor.
+Without any window definition, every selected column uses `[0]`. Each column
+has its own Boolean `<column>_is_pad` tensor, where `True` marks an out-of-group
+position. These generated names must not collide with other output fields.
+
+Anchors start at each group's first row and advance by `stride` (default one).
+The stride controls anchor spacing, not the spacing inside a window:
+
+| `boundary` | Behavior at either end of a group |
+|------------|----------------------------------|
+| `drop` (default) | Keep only anchors for which all requested offsets are valid. |
+| `pad` | Keep scheduled anchors and repeat the nearest endpoint for missing positions. |
+| `error` | Fail construction if any scheduled anchor has an incomplete window. |
+
+`pad_values={"action": [0.0, 0.0]}` overrides endpoint repetition for that
+column. Supply values in the same raw representation as a table cell. Padding
+happens before column transforms, so normalization also affects constant padding.
+Use the mask when excluding padded actions from a loss.
+
+As an alternative, specify seconds with `delta_timestamps`, an explicit positive
+`fps`, and optional `tolerance_s` (default `1e-4`). For example,
+`delta_timestamps={"state": [-0.1, 0.0]}, fps=10` selects offsets `[-1, 0]`.
+Both `fps` and `tolerance_s` are only accepted with `delta_timestamps`.
+Offsets must align to the regular frame grid within the tolerance; this API
+does not search actual timestamps or interpolate missing frames. Do not combine
+`delta_timestamps`, `frame_offsets`, and `window_size`.
 
 `column_transforms` receive one padded Python list per projected column. This is
-where applications define tensor dtype and shape or decode BLOB bytes. The
-optional `adapter` receives the resulting sample mapping and can rename or
-combine fields for a model-specific batch contract. The core Dataset does not
-know model field names, image formats, or normalization rules. Top-level
-functions and callable classes are recommended for transforms and adapters so
-the Dataset remains picklable by multi-worker `torch.utils.data.DataLoader`
-instances.
+where applications define tensor dtype and shape or decode BLOB bytes. Even
+singleton windows retain the time dimension. `images_to_tensor` returns TCHW
+images with EXIF orientation applied. Eight-bit pixels become float32 in
+`[0, 1]`; use `partial(images_to_tensor, return_uint8=True)` to keep uint8.
+Higher-bit-depth images become float32 in their original units. All images in
+a column window must have the same shape.
+
+The optional `adapter` receives the transformed sample and can normalize values,
+rename/combine fields, remove singleton time dimensions, or coordinate random
+augmentation across cameras and time. Statistics must be supplied explicitly;
+the Dataset never computes or discovers normalization statistics. For example:
+
+```python
+class NormalizeAction:
+    def __init__(self, mean, std):
+        self.mean = torch.as_tensor(mean, dtype=torch.float32)
+        self.std = torch.as_tensor(std, dtype=torch.float32)
+        if not torch.isfinite(self.std).all() or not (self.std > 0).all():
+            raise ValueError("Action standard deviations must be finite and positive.")
+
+    def __call__(self, sample):
+        sample["action"] = (sample["action"] - self.mean) / self.std
+        return sample
+```
+
+Pass `adapter=NormalizeAction(train_action_mean, train_action_std)` when
+constructing the Dataset. Use training-set statistics, with the same values for
+validation. Keep callbacks picklable (top-level functions, `partial`, or callable
+classes) for multi-worker DataLoader use. Physical reads are coalesced in
+`__getitems__`, while samples and transforms retain their requested order and
+duplicates. Negative indices and slices follow Python sequence semantics.
+
+Rows are sorted by group and order. Order values must be non-null integers
+increasing by exactly one within each group. Duplicate steps and internal gaps,
+including gaps introduced by a scan filter, fail validation. Windows never
+cross groups. A snapshot pin fixes the data being read; use retained tags for
+long-running training. Sampler progress and augmentation RNG state remain the
+training application's responsibility.
+
+Existing `window_size=16` calls remain supported, with `anchor_columns` selecting
+singleton fields and `tail="drop"/"pad"/"error"` controlling the old forward
+window. These calls preserve their single `is_pad` mask and previous shapes.
+`anchor_columns` and `tail` require `window_size`; do not combine `tail` and
+`boundary`. New applications should use field offsets and per-column masks.
 
 Columns configured by `video-frame-field` are rejected: a window read would drop
 the `frame_index` and other metadata carried by their `VideoFrameDescriptor`

docs/docs/pypaimon/pytorch.md

@@ -159,42 +159,55 @@ physical video ranges and cache decoder sessions per worker. See
 for the write path and a complete decoder example.
 ## Contiguous Windows
 
-Use a map-style `ContiguousWindowDataset` when training samples are fixed-size
-windows which must not cross a sequence boundary. The dataset builds an index
+Use a map-style `ContiguousWindowDataset` to construct training samples from
+one materialized frame table. Each field can have its own history or future
+window, and windows never cross a sequence boundary. The dataset builds an index
 from only the group column, order column, and Paimon row IDs. Projected values,
 including BLOB payloads, are read from the pinned snapshot when a sample is
 requested; they are not retained in the index.
 
 ```python
+from functools import partial
+
+import torch
 from torch.utils.data import DataLoader
+from pypaimon.multimodal.window_transforms import images_to_tensor, to_tensor
 
 dataset = (
     frames.scan()
     .to_contiguous_window_dataset(
-        window_size=16,
-        columns=["state", "image"],
-        anchor_columns=["image"],
+        columns=["state", "image", "action"],
+        frame_offsets={"state": [-2, -1, 0], "action": list(range(16))},
         group_key="episode_index",
         order_key="frame_index",
-        tail="pad",
+        boundary="pad",
+        column_transforms={
+            "state": partial(to_tensor, dtype=torch.float32),
+            "action": partial(to_tensor, dtype=torch.float32),
+            "image": images_to_tensor,
+        },
     )
 )
 
 loader = DataLoader(dataset, batch_size=32, num_workers=4, shuffle=True)
 ```
 
-Each item contains the group and order keys, one list for each requested
-column, and a boolean `is_pad` tensor where `True` marks padding. Padding
-repeats the final real value by default; `pad_values` can override individual
-columns. Columns named in `anchor_columns` contain only the first row's value,
-which is useful when an observation applies to a full action window. Use
-`column_transforms` to convert column lists to tensors and
-`adapter` to produce a model-specific sample mapping. Keep these callbacks
-picklable when using multiple DataLoader workers.
-
-Scheduled anchors start at row zero and advance by `stride` (default `1`).
-`tail="drop"` omits incomplete windows, `tail="pad"` includes and pads them,
-and `tail="error"` rejects a sequence with any scheduled incomplete window.
+Each raw item contains scalar group/order keys, a list for every selected column,
+and a Boolean `<column>_is_pad` tensor where `True` marks padding. Unspecified
+field offsets default to `[0]`. The example produces state/action tensors with
+time lengths 3/16 and an image tensor of shape `(1, C, H, W)`; DataLoader adds the
+batch dimension. `pad_values` can replace endpoint repetition for individual
+columns, before any transforms. `adapter` can normalize or rename fields and
+produce a model-specific mapping. Keep callbacks picklable for multiple workers.
+
+Scheduled anchors start at each group's first row and advance by `stride`
+(default `1`). `boundary="drop"` (default) omits anchors incomplete for any
+field, `boundary="pad"` pads either end, and `boundary="error"` rejects any
+scheduled incomplete window. `delta_timestamps` with an explicit `fps` is an
+alternative to integer offsets, with frame-grid alignment checked against
+`tolerance_s`. Existing `window_size`/`anchor_columns`/`tail` calls retain their
+single `is_pad` output. See [multimodal reading](multimodal-reading#contiguous-windows-for-pytorch)
+for the full sample and conversion contract.
 Rows are sorted by `order_key` inside each `group_key` value. Order values must
 be integers which increase by exactly one; duplicates and missing steps are
 rejected, and windows never cross groups. The resolved Paimon
@@ -203,6 +216,12 @@ change its index or sample contents. A dataset pinned through `tag_name` fails
 its reads if the tag is moved to another snapshot, rather than mixing rows from
 the two snapshots.
 
+Use normal PyTorch samplers for shuffling and distributed training. For DDP,
+pass `DistributedSampler(dataset)` to DataLoader, omit `shuffle=True`, and call
+the sampler's `set_epoch(epoch)` each epoch. Its padding or drop behavior still
+applies when the sample count is not divisible by the number of ranks. Dataset
+version pinning does not save sampler progress or random augmentation state.
+
 Columns configured by `video-frame-field` are rejected: a window read would drop
 the `frame_index` and other metadata carried by their `VideoFrameDescriptor`
 values. Read those columns with `to_torch()` instead.

paimon-python/pypaimon/benchmark/act/paimon.py

@@ -52,7 +52,7 @@ def __call__(self, sample):
         The persisted ``frame_index`` is forwarded as the shared ACT sample
         position.
         State and image columns are singleton lists; action retains the full
-        horizon and ``is_pad`` is forwarded unchanged.
+        horizon and ``action_is_pad`` becomes the ACT ``is_pad`` mask.
         """
         qpos = np.concatenate([
             np.asarray(sample[name][0], dtype=np.float32)
@@ -80,7 +80,7 @@ def __call__(self, sample):
             "qpos": torch.from_numpy(np.ascontiguousarray(qpos)),
             "action": torch.from_numpy(np.ascontiguousarray(action)),
             "images": torch.from_numpy(np.ascontiguousarray(images)),
-            "is_pad": sample["is_pad"],
+            "is_pad": sample["action_is_pad"],
         }
 
 
@@ -114,13 +114,14 @@ def create_datasets(
         frames.scan(snapshot_id=snapshot_id).where(
             "episode_id = '%s'" % episode_id.replace("'", "''")
         ).to_contiguous_window_dataset(
-            window_size=config.action_horizon,
+            frame_offsets={
+                name: range(config.action_horizon) for name in ACTION_COLUMNS
+            },
             columns=QPOS_COLUMNS + ACTION_COLUMNS + IMAGE_COLUMNS,
-            anchor_columns=QPOS_COLUMNS + IMAGE_COLUMNS,
             group_key="episode_id",
             order_key="frame_index",
             stride=1,
-            tail="drop",
+            boundary="drop",
             adapter=PaimonACTAdapter(normalization),
         )
         for episode_id in (train_episode_id, validation_episode_id)

paimon-python/pypaimon/multimodal/lerobot/dataset.py

@@ -18,7 +18,6 @@
 """LeRobot-compatible map-style reads from a multimodal Paimon table."""
 
 import bisect
-import io
 import json
 import math
 import operator
@@ -1332,14 +1331,10 @@ def _torch_row(row, features, return_uint8=False):
 def _image_tensor(payload, feature, return_uint8=False):
     if payload is None:
         raise ValueError("LeRobot image feature contains a null frame.")
-    import numpy as np
-    import torch
-    try:
-        from PIL import Image, ImageOps
-    except ImportError as error:
-        raise ImportError(
-            "PaimonLeRobotDataset requires Pillow from "
-            "'pypaimon[lerobot]'.") from error
+    from pypaimon.multimodal.window_transforms import (
+        _decode_image,
+        _image_array_to_tensor,
+    )
 
     expected_shape = _feature_shape(feature, "image")
     if len(expected_shape) != 3:
@@ -1349,21 +1344,12 @@ def _image_tensor(payload, feature, return_uint8=False):
     payload_shape = expected_shape[1:] + expected_shape[:1] \
         if names and names[0] in ("channel", "channels") \
         else expected_shape
-    with Image.open(io.BytesIO(payload)) as image:
-        array = np.array(ImageOps.exif_transpose(image), copy=True)
-    if array.ndim == 2:
-        array = array[:, :, None]
+    array = _decode_image(payload)
     if array.shape != payload_shape:
         raise ValueError(
             "LeRobot image payload has shape %s, expected %s."
             % (array.shape, payload_shape))
-    normalize = array.dtype == np.uint8
-    tensor = torch.from_numpy(array).permute(2, 0, 1)
-    if normalize and return_uint8:
-        return tensor
-    # Preserve high-bit-depth and floating-point images in native units.
-    tensor = tensor.float()
-    return tensor.div_(255) if normalize else tensor
+    return _image_array_to_tensor(array, return_uint8)
 
 
 def _video_tensor(frame, feature, return_uint8=False):