[python] Add paired HDF5/Paimon ACT benchmark

docs/docs/pypaimon/multimodal-api.mdx

@@ -720,6 +720,64 @@ 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_id",
+        order_key="step_idx",
+        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.
+
 ### Distributed BLOB processing with Ray
 
 For larger jobs, read descriptors with `to_ray()`, then fetch and process BLOB

docs/docs/pypaimon/pytorch.md

@@ -106,7 +106,49 @@ 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_id",
+        order_key="step_idx",
+        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.
 ## File Format Metadata Cache
 
 Reusable PyArrow Dataset metadata is cached across reads. Configure its estimated

docs/docs/pypaimon/robomind-act-benchmark.md

@@ -0,0 +1,83 @@
+---
+title: "RoboMIND Paired ACT Benchmark"
+sidebar_position: 8
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# RoboMIND Paired ACT Benchmark
+
+The paired benchmark compares original RoboMIND AgileX HDF5 with an already
+ingested and canonical-action-backfilled Paimon warehouse. It does not include
+ingestion or backfill time. Install the ACT and HDF5 extras, run the
+[RoboMIND AgileX pipeline](robomind-agilex), and then execute:
+
+```shell
+pip install 'pypaimon[act,hdf5]'
+python -m pypaimon.benchmark.paired_act \
+  --input /data/RoboMIND/h5_agilex_3rgb \
+  --warehouse /data/warehouse \
+  --report /data/results/paired-act.json
+```
+
+A successful run prints a compact `SUCCEEDED` result and writes the full JSON
+report. A source, parity, or configuration mismatch raises an error and does
+not write a successful report.
+
+One immutable configuration controls both paths. The runner computes train-only
+normalization once, verifies its canonical action values against the requested
+version in `feature_stats_agilex`, and passes the same object to both adapters.
+A seeded window plan fixes every warmup, loader, training, and validation
+anchor. Before training, the runner compares `sample_id`, `episode_id`, and
+`step_idx` by value and requires exact `torch.equal` parity for state, action,
+image, and padding tensors.
+
+The Paimon adapter uses `ContiguousWindowDataset`, not an ACT-specific table
+reader. Dataset construction indexes only episode, frame, and row IDs. Window
+payloads remain lazy until `__getitem__`, and all train and validation reads are
+pinned to the exact frames snapshot recorded by the normalization statistics.
+PyTorch batch access coalesces overlapping row IDs into one payload read. The
+image columns are marked as anchor-only, so each sample loads the observation
+images once rather than once per action-horizon row. The adapter maps each
+generic window to the same tensor contract as the HDF5 adapter without
+materializing episodes in memory.
+
+Each backend then uses the same CPU LeRobot ACT policy, initial seed, AdamW
+optimizer, batch size, window sequence, and optimizer step count. At least
+three rounds run in alternating order (`HDF5 → Paimon`, then
+`Paimon → HDF5`) to expose ordering effects. The benchmark does not drop the OS
+page cache and records `cache_control=uncontrolled`.
+
+The JSON report contains:
+
+- input manifest, table snapshot, normalization, configuration, and window
+  sequence digests;
+- exact tensor, train-loss, and validation-loss parity gates;
+- first-batch latency, DataLoader samples per second, fixed-step time, and a
+  separate dataset-build-plus-first-batch Python allocation replay for every
+  run;
+- per-backend median, minimum, and maximum across rounds;
+- explicit unverified scope, including native-memory completeness, GPU,
+  multi-worker loading, distributed training, recovery, and policy quality.
+
+Python peak allocation uses `tracemalloc` after wall-clock measurement so its
+overhead does not distort throughput. The replay covers dataset construction
+and one first batch; it does not include every native Arrow or Torch allocation.
+Treat it as a reproducible engineering diagnostic, not total process RSS.