[python][torch] Add lazy contiguous window dataset

docs/docs/pypaimon/multimodal-api.mdx

@@ -866,6 +866,68 @@ Notes:
   few large reads); scattered point reads coalesce less.
 - Blob reads are available only on `scan()`, not on the `search()` queries.
 
+### 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`.
+
+```shell
+pip install 'pypaimon[torch]'
+```
+
+```python
+import torch
+
+
+def float32_window(values):
+    return torch.tensor(values, dtype=torch.float32)
+
+
+windows = (
+    frames.scan()
+    .where("split = 'train'")
+    .to_contiguous_window_dataset(
+        window_size=16,
+        columns=["state", "action"],
+        group_key="episode_index",
+        order_key="frame_index",
+        tail="pad",
+        column_transforms={
+            "state": float32_window,
+            "action": float32_window,
+        },
+    )
+)
+
+sample = windows[0]
+assert sample["action"].shape == (16, action_size)
+assert sample["is_pad"].shape == (16,)
+```
+
+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.
+
+`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.
+
+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.
+
 ### Distributed BLOB processing with Ray
 
 For larger jobs, read descriptors with `to_ray()`, then fetch and process BLOB

docs/docs/pypaimon/pytorch.md

@@ -157,7 +157,53 @@ embedded frame ordinals keep frame mapping out of the normal data file. Use
 physical video ranges and cache decoder sessions per worker. See
 [Multimodal API: Video Frame Storage](multimodal-api#video-frame-storage)
 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
+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 torch.utils.data import DataLoader
+
+dataset = (
+    frames.scan()
+    .to_contiguous_window_dataset(
+        window_size=16,
+        columns=["state", "image"],
+        anchor_columns=["image"],
+        group_key="episode_index",
+        order_key="frame_index",
+        tail="pad",
+    )
+)
+
+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.
+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
+snapshot is pinned for the lifetime of the dataset, so later commits cannot
+change its index or sample contents.
+
+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.
 ## File Format Metadata Cache
 
 Reusable PyArrow Dataset metadata is cached across reads. Configure its estimated

paimon-python/pypaimon/multimodal/query.py

@@ -164,6 +164,66 @@ def to_torch(
             max_buffer_input_splits=max_buffer_input_splits,
         )
 
+    def to_contiguous_window_dataset(
+            self,
+            *,
+            window_size,
+            columns=None,
+            anchor_columns=None,
+            group_key="episode_index",
+            order_key="frame_index",
+            stride=1,
+            tail="drop",
+            column_transforms=None,
+            pad_values=None,
+            adapter=None,
+            blob_parallelism=64):
+        """Build a snapshot-pinned, map-style Dataset of contiguous rows.
+
+        The Dataset indexes only ``group_key``, ``order_key``, and Paimon row
+        IDs, then reads projected values on demand. Columns listed in
+        ``anchor_columns`` are provided to ``column_transforms`` as one-element
+        lists read from the first row of each window; ``adapter`` receives the
+        transformed values. ``order_key`` must contain non-null integers that
+        increase by exactly one within each group. The Dataset sorts rows within
+        each group and never creates a window across groups.
+
+        Args:
+            window_size: Number of rows in a complete window.
+            columns: Value columns to return, excluding the group and order
+                keys. The scan projection is used when omitted.
+            anchor_columns: Subset of ``columns`` read only from the window's
+                first row.
+            group_key: Column identifying an independent row sequence.
+            order_key: Integer position column within each group.
+            stride: Distance between scheduled window starts.
+            tail: Handling for incomplete final windows: ``drop``, ``pad``, or
+                ``error``.
+            column_transforms: Per-column callables applied to value lists.
+            pad_values: Optional replacement values used by ``tail='pad'``.
+            adapter: Callable that converts the complete sample mapping.
+            blob_parallelism: Maximum concurrent BLOB body reads per fetch.
+
+        Returns:
+            A snapshot-pinned ``ContiguousWindowDataset``. See that class for
+            padding, mask, transform, and adapter result semantics.
+        """
+        from pypaimon.multimodal.window_dataset import ContiguousWindowDataset
+        return ContiguousWindowDataset(
+            self,
+            window_size=window_size,
+            columns=columns,
+            anchor_columns=anchor_columns,
+            group_key=group_key,
+            order_key=order_key,
+            stride=stride,
+            tail=tail,
+            column_transforms=column_transforms,
+            pad_values=pad_values,
+            adapter=adapter,
+            blob_parallelism=blob_parallelism,
+        )
+
     def to_ray(
             self,
             *,
@@ -418,6 +478,12 @@ def to_arrow_batch_reader(self, *args, **kwargs):
             "not search queries."
         )
 
+    def to_contiguous_window_dataset(self, *args, **kwargs):
+        raise TypeError(
+            "to_contiguous_window_dataset is only supported on scan(), "
+            "not search queries."
+        )
+
 
 class VectorQuery(_PreFilterQuery):
     """Chainable query wrapper for vector global-index search."""