[python] Add ACT storage benchmark and contiguous windows

docs/docs/pypaimon/multimodal-api.mdx

@@ -866,6 +866,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

@@ -157,7 +157,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,185 @@
+---
+title: "RoboMIND ACT Storage 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 ACT Storage Benchmark
+
+This benchmark measures the same CPU LeRobot ACT training workload over an
+original RoboMIND AgileX HDF5 dataset or an already ingested and
+canonical-action-backfilled Paimon warehouse. Ingestion and backfill are outside
+the timed scope.
+
+The backends run independently. A resolved experiment document preserves the
+shared configuration, normalization, seed, episode selection, Paimon snapshot,
+and logical window sequence. Result comparison verifies that contract before it
+calculates performance ratios.
+
+## Install
+
+Python 3.10 or newer is required.
+
+```shell
+pip install 'pypaimon[act,hdf5]'
+```
+
+## 1. Prepare the experiment
+
+```shell
+python -m pypaimon.benchmark.act prepare \
+  --input /data/RoboMIND/h5_agilex_3rgb \
+  --warehouse /data/warehouse \
+  --output /data/results/experiment.json
+```
+
+Preparation is not timed. It verifies that HDF5 discovery matches the Paimon
+episodes table, checks versioned action statistics against train-only HDF5
+moments, selects eligible train and validation episodes, pins the frames
+snapshot, and materializes deterministic measurement, training, and validation
+window indices.
+
+Without `--experiment`, preparation starts from the packaged
+`default_experiment.json`. `--experiment` replaces that definition, so a
+custom JSON file must contain every required field. Command-line options then
+override individual values:
+
+```shell
+python -m pypaimon.benchmark.act prepare \
+  --experiment my-experiment.json \
+  --input /data/RoboMIND/h5_agilex_3rgb \
+  --warehouse /data/warehouse \
+  --action-horizon 32 \
+  --batch-size 2 \
+  --fetch-batches 8 \
+  --rounds 3 \
+  --output /data/results/experiment.json
+```
+
+The resolved experiment embeds the effective parameters as well as:
+
+- portable source episode metadata and its SHA-256;
+- normalization values, scope, version, frame count, and SHA-256;
+- selected train and validation episode IDs;
+- every logical window index, the window-plan SHA-256, and the
+  episode-qualified sample-sequence SHA-256;
+- the Paimon database, frames table, and pinned snapshot ID.
+
+## 2. Run each backend
+
+```shell
+python -m pypaimon.benchmark.act run \
+  --backend hdf5 \
+  --experiment /data/results/experiment.json \
+  --input /data/RoboMIND/h5_agilex_3rgb \
+  --results-dir /data/results
+
+python -m pypaimon.benchmark.act run \
+  --backend paimon \
+  --experiment /data/results/experiment.json \
+  --warehouse /data/warehouse \
+  --results-dir /data/results
+```
+
+Use `--output` to choose an exact result path. Otherwise the command writes an
+automatically named JSON file below `--results-dir` and prints its absolute
+path as a compact JSON object.
+
+Each result contains the complete resolved experiment and experiment SHA-256,
+backend identity, runtime environment, model metadata, planned-sample tensor
+fingerprint, three or more raw measurement rounds, and median/minimum/maximum
+summary metrics.
+
+Both adapters produce the same shared sample contract. State and camera images
+come from the anchor frame; action covers the complete horizon. HDF5 reads a
+window on demand from one episode file. Paimon uses a lazy, snapshot-pinned
+`ContiguousWindowDataset`; image columns are anchor-only, and plural
+`__getitems__` access coalesces multiple logical batches into a physical
+fetch before splitting them back into the unchanged model batch size.
+
+## 3. Compare results
+
+Compare explicit files:
+
+```shell
+python -m pypaimon.benchmark.act compare \
+  /data/results/robomind-act-hdf5-20260901T010000Z-a1b2c3d4.json \
+  /data/results/robomind-act-paimon-20260901T011000Z-e5f6a7b8.json \
+  --output /data/results/comparison.json
+```
+
+Or discover all ACT result documents in a directory:
+
+```shell
+python -m pypaimon.benchmark.act compare \
+  --results-dir /data/results \
+  --output /data/results/comparison.json
+```
+
+Directory discovery ignores experiment and prior comparison JSON files.
+Results are grouped by experiment SHA-256. Different experiments remain
+separate entries in one comparison artifact; only compatible repeated results
+for the same experiment and backend are aggregated.
+
+Within one experiment group, comparison requires identical runtime environment,
+model metadata, tensor fingerprint, train-loss trace, and validation-loss trace.
+An environment mismatch marks the group `INCOMPATIBLE`; a model, tensor, or
+loss mismatch marks it `FAILED`. Neither case produces performance ratios.
+
+For compatible HDF5 and Paimon results, higher-is-better metrics report
+`paimon_over_hdf5`. Lower-is-better latency, time, and memory metrics report
+`hdf5_over_paimon`, which is the Paimon speedup or reduction factor.
+
+## Measurements
+
+Every backend repeat records:
+
+- dataset construction time;
+- first-batch latency after construction;
+- batch-fetch samples per second after warm-up;
+- end-to-end fixed ACT optimizer-step time, including dataset fetch;
+- per-step loss and compute time after each training batch has been fetched;
+- validation loss;
+- total measured wall time;
+- Python peak allocation from a separate dataset-first-batch replay.
+
+The shared harness resets Python, NumPy, and Torch random generators before
+model construction and enables deterministic Torch algorithms. The logical
+window plan is explicit rather than delegated to a streaming reader.
+
+Python peak allocation uses `tracemalloc` after wall-clock measurement so
+tracing overhead does not distort throughput. It does not include every native
+Arrow or Torch allocation. The benchmark does not drop the OS page cache.
+GPU, multi-worker loading, distributed training, recovery, and policy quality
+remain outside this benchmark.
+
+## Code organization
+
+- `benchmark.act.harness`: shared ACT tensors, model, trainer, window plan, and
+  measurement lifecycle;
+- `benchmark.act.hdf5`: HDF5 window dataset and train normalization moments;
+- `benchmark.act.paimon`: Paimon adapter, snapshot-pinned datasets, and
+  versioned statistics access;
+- `benchmark.act.runner`: experiment preparation and one-backend execution;
+- `benchmark.act.compare`: result discovery, compatibility checks, grouping by
+  experiment, and aggregation of compatible repeated runs;
+- `benchmark.act.__main__`: the `prepare`, `run`, and `compare`
+  command-line interface.

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

@@ -0,0 +1,17 @@
+# 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.
+
+"""ACT training benchmark backends and result comparison."""