This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 196e28f3a7 [python] Add paired ACT storage benchmark (#9466)
196e28f3a7 is described below
commit 196e28f3a7b257daa9aded93814f98c3cd88fc88
Author: Yann Byron <[email protected]>
AuthorDate: Wed Sep 9 19:52:16 2026 +0800
[python] Add paired ACT storage benchmark (#9466)
---
docs/docs/pypaimon/robomind-act-benchmark.md | 185 ++++++
docs/docs/pypaimon/robomind-agilex.md | 16 +-
paimon-python/pypaimon/benchmark/act/__init__.py | 17 +
paimon-python/pypaimon/benchmark/act/__main__.py | 200 ++++++
paimon-python/pypaimon/benchmark/act/compare.py | 231 +++++++
.../pypaimon/benchmark/act/default_experiment.json | 22 +
paimon-python/pypaimon/benchmark/act/experiment.py | 42 ++
paimon-python/pypaimon/benchmark/act/harness.py | 577 ++++++++++++++++
paimon-python/pypaimon/benchmark/act/hdf5.py | 201 ++++++
paimon-python/pypaimon/benchmark/act/paimon.py | 153 +++++
paimon-python/pypaimon/benchmark/act/runner.py | 740 +++++++++++++++++++++
paimon-python/pypaimon/sample/robomind_agilex.py | 28 +-
paimon-python/pypaimon/tests/act_benchmark_test.py | 208 ++++++
paimon-python/pypaimon/tests/act_runner_test.py | 709 ++++++++++++++++++++
.../tests/robomind_agilex_pipeline_test.py | 19 +-
paimon-python/setup.py | 28 +-
16 files changed, 3338 insertions(+), 38 deletions(-)
diff --git a/docs/docs/pypaimon/robomind-act-benchmark.md
b/docs/docs/pypaimon/robomind-act-benchmark.md
new file mode 100644
index 0000000000..b9c070b769
--- /dev/null
+++ b/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; see [RoboMIND AgileX](./robomind-agilex) to build the
warehouse.
+
+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.
diff --git a/docs/docs/pypaimon/robomind-agilex.md
b/docs/docs/pypaimon/robomind-agilex.md
index ee8aa65a10..035438018f 100644
--- a/docs/docs/pypaimon/robomind-agilex.md
+++ b/docs/docs/pypaimon/robomind-agilex.md
@@ -45,11 +45,11 @@ normalization statistics.
## Run the local pipeline
-After downloading RoboMIND, use Python 3.11 or later, install the HDF5 and
-Vortex extras, and provide the source and warehouse directories to one command:
+After downloading RoboMIND, install the HDF5 extra and provide the source and
+warehouse directories to one command:
```bash
-pip install 'pypaimon[hdf5,vortex]'
+pip install 'pypaimon[hdf5]'
python -m pypaimon.sample.robomind_agilex \
--input /data/RoboMIND/h5_agilex_3rgb \
--warehouse /data/warehouse
@@ -133,10 +133,10 @@ time with that versioned row.
The tables are non-primary-key append tables. Repeating ingestion therefore
appends duplicate rows by design; it does not mean row-level update/delete is
-disabled. The sample keeps deletion vectors enabled, stores vectors with
-Vortex, and sets `blob-as-descriptor=false` because its transforms emit raw
-image/depth bytes rather than external BLOB descriptors. Parquet data format,
-dynamic bucket mode, and global-index search mode are inherited defaults and
-are not repeated in the sample options.
+disabled. The sample keeps deletion vectors enabled and sets
+`blob-as-descriptor=false` because its transforms emit raw image/depth bytes
+rather than external BLOB descriptors. Parquet data format, dynamic bucket
+mode, and global-index search mode are inherited defaults and are not repeated
+in the sample options.
Run local and Ray modes against separate new warehouses when comparing them.
diff --git a/paimon-python/pypaimon/benchmark/act/__init__.py
b/paimon-python/pypaimon/benchmark/act/__init__.py
new file mode 100644
index 0000000000..f6224db904
--- /dev/null
+++ b/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."""
diff --git a/paimon-python/pypaimon/benchmark/act/__main__.py
b/paimon-python/pypaimon/benchmark/act/__main__.py
new file mode 100644
index 0000000000..d1d8560ae2
--- /dev/null
+++ b/paimon-python/pypaimon/benchmark/act/__main__.py
@@ -0,0 +1,200 @@
+# 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.
+
+"""Command-line entry point for ACT benchmark preparation, runs, and
reports."""
+
+# ruff: noqa: E402
+
+import sys
+
+
+def _require_supported_python(version_info):
+ """Reject runtimes older than the ACT dependencies support."""
+ if tuple(version_info[:2]) < (3, 10):
+ raise RuntimeError("ACT benchmark requires Python 3.10 or newer.")
+
+
+_require_supported_python(sys.version_info)
+
+import argparse
+import copy
+import json
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+from pypaimon.benchmark.act.compare import (
+ compare_results,
+ load_result_documents,
+)
+from pypaimon.benchmark.act.experiment import load_experiment
+from pypaimon.benchmark.act.runner import prepare_experiment, run_experiment
+
+
+_CONFIG_ARGUMENTS = (
+ ("seed", int),
+ ("action_horizon", int),
+ ("batch_size", int),
+ ("optimizer_steps", int),
+ ("image_height", int),
+ ("image_width", int),
+ ("learning_rate", float),
+ ("weight_decay", float),
+ ("warmup_batches", int),
+ ("timed_batches", int),
+ ("fetch_batches", int),
+ ("rounds", int),
+)
+
+
+def main(argv=None):
+ """Parse an ACT benchmark subcommand and write its JSON artifact."""
+ parser = _parser()
+ args = parser.parse_args(argv)
+ if args.command == "prepare":
+ definition = copy.deepcopy(load_experiment(args.experiment))
+ for name, _ in _CONFIG_ARGUMENTS:
+ value = getattr(args, name)
+ if value is not None:
+ definition["config"][name] = value
+ for name in (
+ "statistics_version", "train_episode_id",
+ "validation_episode_id"):
+ value = getattr(args, name)
+ if value is not None:
+ definition[name] = value
+ output = Path(args.output)
+ experiment = prepare_experiment(
+ args.input,
+ args.warehouse,
+ output,
+ definition=definition,
+ database=args.database,
+ )
+ _print_artifact("experiment", output, experiment["schema_version"])
+ return 0
+ if args.command == "run":
+ experiment = load_experiment(args.experiment)
+ output = (
+ Path(args.output)
+ if args.output else _artifact_path(
+ args.results_dir,
+ "%s-%s" % (experiment["benchmark_id"], args.backend),
+ )
+ )
+ result = run_experiment(
+ args.backend,
+ args.experiment,
+ output,
+ input_root=args.input,
+ warehouse=args.warehouse,
+ )
+ _print_artifact("result", output, result["status"])
+ return 0
+ results_dir = args.results_dir
+ if not args.results and results_dir is None:
+ results_dir = "act-results"
+ results = load_result_documents(args.results, results_dir=results_dir)
+ comparison = compare_results(results)
+ output = (
+ Path(args.output)
+ if args.output else _artifact_path(
+ results_dir or "act-results", "comparison")
+ )
+ output.parent.mkdir(parents=True, exist_ok=True)
+ output.write_text(
+ json.dumps(comparison, indent=2, sort_keys=True) + "\n",
+ encoding="utf-8",
+ )
+ _print_artifact("comparison", output, comparison["status"])
+ return 0 if comparison["status"] == "SUCCEEDED" else 1
+
+
+def _parser():
+ parser = argparse.ArgumentParser(
+ description=__doc__,
+ formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+ )
+ commands = parser.add_subparsers(dest="command", required=True)
+
+ prepare = commands.add_parser(
+ "prepare",
+ help="Resolve a shared experiment against matching HDF5 and Paimon
data.",
+ formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+ )
+ prepare.add_argument("--input", required=True, help="RoboMIND HDF5 root.")
+ prepare.add_argument("--warehouse", required=True, help="Paimon
warehouse.")
+ prepare.add_argument(
+ "--experiment",
+ help="Input experiment JSON; packaged defaults are used when omitted.",
+ )
+ prepare.add_argument(
+ "--output", default="act-results/experiment.json",
+ help="Resolved experiment JSON path.")
+ prepare.add_argument("--database", default="robomind")
+ prepare.add_argument("--statistics-version")
+ prepare.add_argument("--train-episode-id")
+ prepare.add_argument("--validation-episode-id")
+ for name, argument_type in _CONFIG_ARGUMENTS:
+ prepare.add_argument(
+ "--" + name.replace("_", "-"), type=argument_type, default=None)
+
+ run = commands.add_parser(
+ "run",
+ help="Run one storage backend using a resolved experiment.",
+ formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+ )
+ run.add_argument("--backend", required=True, choices=("hdf5", "paimon"))
+ run.add_argument("--experiment", required=True)
+ run.add_argument("--input", help="HDF5 root; required for backend=hdf5.")
+ run.add_argument(
+ "--warehouse", help="Paimon warehouse; required for backend=paimon.")
+ run.add_argument("--output", help="Explicit result JSON path.")
+ run.add_argument(
+ "--results-dir", default="act-results",
+ help="Directory for an automatically named result.")
+
+ compare = commands.add_parser(
+ "compare",
+ help=(
+ "Group results by experiment and aggregate compatible repeats."
+ ),
+ formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+ )
+ compare.add_argument("results", nargs="*", help="Explicit result JSON
files.")
+ compare.add_argument(
+ "--results-dir",
+ help="Also discover ACT result JSON files in this directory.")
+ compare.add_argument("--output", help="Comparison JSON path.")
+ return parser
+
+
+def _artifact_path(directory, prefix):
+ timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+ return Path(directory).expanduser() / (
+ "%s-%s-%s.json" % (prefix, timestamp, uuid.uuid4().hex[:8]))
+
+
+def _print_artifact(kind, path, status):
+ print(json.dumps({
+ "artifact": str(Path(path).expanduser().resolve()),
+ "kind": kind,
+ "status": status,
+ }, sort_keys=True))
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/paimon-python/pypaimon/benchmark/act/compare.py
b/paimon-python/pypaimon/benchmark/act/compare.py
new file mode 100644
index 0000000000..ab53f4dd67
--- /dev/null
+++ b/paimon-python/pypaimon/benchmark/act/compare.py
@@ -0,0 +1,231 @@
+# 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.
+
+"""Validate and aggregate independently produced ACT benchmark results."""
+
+import hashlib
+import json
+from pathlib import Path
+
+
+_METRICS = {
+ "batch_fetch_samples_per_s": "higher",
+ "dataset_build_s": "lower",
+ "first_batch_s": "lower",
+ "fixed_steps_s": "lower",
+ "python_peak_allocated_bytes": "lower",
+ "wall_time_s": "lower",
+}
+
+
+def canonical_sha256(value):
+ """Return the SHA-256 of a JSON value using canonical serialization."""
+ payload = json.dumps(value, sort_keys=True, separators=(",", ":"))
+ return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def load_result_documents(paths, results_dir=None):
+ """Load explicit result files plus ACT results discovered in a directory.
+
+ Explicit paths must contain result documents. Directory discovery ignores
+ experiment and prior comparison JSON files. A path found both ways is read
+ once, preserving explicit-path order followed by sorted directory entries.
+ """
+ candidates = [Path(path).expanduser().resolve() for path in paths]
+ explicit = set(candidates)
+ if results_dir is not None:
+ directory = Path(results_dir).expanduser().resolve()
+ candidates.extend(sorted(directory.glob("*.json")))
+ seen = set()
+ results = []
+ for path in candidates:
+ path = path.resolve()
+ if path in seen:
+ continue
+ seen.add(path)
+ with path.open(encoding="utf-8") as result_file:
+ document = json.load(result_file)
+ if document.get("schema_version") != "act-benchmark-result@1":
+ if path in explicit:
+ raise ValueError("Not an ACT benchmark result: %s." % path)
+ continue
+ results.append(document)
+ if not results:
+ raise ValueError("No ACT benchmark result files were found.")
+ return results
+
+
+def compare_results(results):
+ """Group result documents by experiment and compare compatible backends.
+
+ Results from different experiment definitions remain separate. Results in
+ one experiment group must report the same runtime environment; otherwise
+ the group is marked incompatible and no performance ratios are produced.
+
+ Args:
+ results: Iterable of decoded ``act-benchmark-result@1`` documents.
+
+ Returns:
+ A JSON-compatible comparison document with one entry per experiment.
+ """
+ groups = {}
+ for result in results:
+ if result.get("schema_version") != "act-benchmark-result@1":
+ raise ValueError("Unsupported ACT benchmark result schema.")
+ experiment = result.get("experiment")
+ if not isinstance(experiment, dict):
+ raise ValueError("ACT benchmark result has no experiment object.")
+ experiment_sha256 = canonical_sha256(experiment)
+ if result.get("experiment_sha256") != experiment_sha256:
+ raise ValueError("ACT result experiment SHA-256 differs.")
+ if result.get("status") != "SUCCEEDED":
+ raise ValueError("ACT comparison requires successful results.")
+ if result.get("backend") not in ("hdf5", "paimon"):
+ raise ValueError("ACT result has an unsupported backend.")
+ groups.setdefault(experiment_sha256, []).append(result)
+
+ experiments = [
+ _compare_experiment(experiment_sha256, grouped)
+ for experiment_sha256, grouped in sorted(groups.items())
+ ]
+ statuses = {item["status"] for item in experiments}
+ if statuses == {"SUCCEEDED"}:
+ status = "SUCCEEDED"
+ elif "FAILED" in statuses:
+ status = "FAILED"
+ else:
+ status = "INCOMPATIBLE"
+ return {
+ "schema_version": "act-benchmark-comparison@1",
+ "status": status,
+ "experiments": experiments,
+ }
+
+
+def _compare_experiment(experiment_sha256, results):
+ environments = {
+ canonical_sha256(result.get("environment", {})) for result in results
+ }
+ by_backend = {}
+ for result in results:
+ by_backend.setdefault(result["backend"], []).append(result)
+ if set(by_backend) != {"hdf5", "paimon"}:
+ return {
+ "experiment_sha256": experiment_sha256,
+ "experiment": results[0]["experiment"],
+ "status": "INCOMPATIBLE",
+ "reason": "both hdf5 and paimon results are required",
+ "backends": sorted(by_backend),
+ "result_count": len(results),
+ "metrics": {},
+ }
+ if len(environments) != 1:
+ return {
+ "experiment_sha256": experiment_sha256,
+ "experiment": results[0]["experiment"],
+ "status": "INCOMPATIBLE",
+ "reason": "runtime environments differ",
+ "environment_sha256s": sorted(environments),
+ "backends": sorted(by_backend),
+ "metrics": {},
+ }
+ models = {canonical_sha256(result.get("model")) for result in results}
+ if len(models) != 1:
+ return _failed_group(
+ experiment_sha256, by_backend, results, "models differ")
+ fingerprints = {
+ result.get("tensor_fingerprint", {}).get("sha256")
+ for result in results
+ }
+ if len(fingerprints) != 1 or None in fingerprints:
+ return _failed_group(
+ experiment_sha256,
+ by_backend,
+ results,
+ "tensor fingerprints differ",
+ )
+ loss_traces = {canonical_sha256([{
+ "round": run["round"],
+ "train_loss": run["train_loss"],
+ "validation_loss": run["validation_loss"],
+ } for run in result.get("runs", [])]) for result in results}
+ if len(loss_traces) != 1:
+ return _failed_group(
+ experiment_sha256, by_backend, results, "loss traces differ")
+
+ medians = {
+ backend: _aggregate_backend(items)
+ for backend, items in by_backend.items()
+ }
+ metrics = {}
+ for name, preferred in _METRICS.items():
+ values = {
+ backend: summary[name]
+ for backend, summary in medians.items()
+ if name in summary
+ }
+ if values:
+ metric = dict(values)
+ metric["preferred"] = preferred
+ if set(values) == {"hdf5", "paimon"}:
+ if preferred == "higher" and values["hdf5"]:
+ metric["paimon_over_hdf5"] = (
+ values["paimon"] / values["hdf5"])
+ elif preferred == "lower" and values["paimon"]:
+ metric["hdf5_over_paimon"] = (
+ values["hdf5"] / values["paimon"])
+ metrics[name] = metric
+ return {
+ "experiment_sha256": experiment_sha256,
+ "experiment": results[0]["experiment"],
+ "status": "SUCCEEDED",
+ "environment": results[0]["environment"],
+ "environment_sha256": next(iter(environments)),
+ "backends": sorted(by_backend),
+ "result_count": len(results),
+ "metrics": metrics,
+ }
+
+
+def _failed_group(experiment_sha256, by_backend, results, reason):
+ return {
+ "experiment_sha256": experiment_sha256,
+ "experiment": results[0]["experiment"],
+ "status": "FAILED",
+ "reason": reason,
+ "backends": sorted(by_backend),
+ "result_count": len(results),
+ "metrics": {},
+ }
+
+
+def _aggregate_backend(results):
+ names = set.intersection(*(
+ set(result.get("summary", {})) for result in results
+ ))
+ aggregated = {}
+ for name in names:
+ if name not in _METRICS:
+ continue
+ values = [result["summary"][name]["median"] for result in results]
+ values.sort()
+ middle = len(values) // 2
+ aggregated[name] = (
+ values[middle]
+ if len(values) % 2
+ else (values[middle - 1] + values[middle]) / 2.0
+ )
+ return aggregated
diff --git a/paimon-python/pypaimon/benchmark/act/default_experiment.json
b/paimon-python/pypaimon/benchmark/act/default_experiment.json
new file mode 100644
index 0000000000..91c2ffe291
--- /dev/null
+++ b/paimon-python/pypaimon/benchmark/act/default_experiment.json
@@ -0,0 +1,22 @@
+{
+ "benchmark_id": "robomind-act",
+ "config": {
+ "action_horizon": 32,
+ "batch_size": 2,
+ "fetch_batches": 8,
+ "image_height": 64,
+ "image_width": 80,
+ "learning_rate": 0.0001,
+ "optimizer_steps": 2,
+ "rounds": 3,
+ "seed": 20260825,
+ "timed_batches": 32,
+ "warmup_batches": 1,
+ "weight_decay": 0.0001
+ },
+ "dataset": "RoboMIND AgileX",
+ "schema_version": "act-benchmark-experiment@1",
+ "statistics_version": "robomind-agilex-joint-position@1",
+ "train_episode_id": null,
+ "validation_episode_id": null
+}
diff --git a/paimon-python/pypaimon/benchmark/act/experiment.py
b/paimon-python/pypaimon/benchmark/act/experiment.py
new file mode 100644
index 0000000000..f1bb675ca2
--- /dev/null
+++ b/paimon-python/pypaimon/benchmark/act/experiment.py
@@ -0,0 +1,42 @@
+# 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.
+
+"""Load the declarative parameters shared by ACT benchmark runs."""
+
+import json
+from pathlib import Path
+
+
+DEFAULT_EXPERIMENT = Path(__file__).with_name("default_experiment.json")
+
+
+def load_experiment(path=None):
+ """Load an ACT experiment definition from JSON or the packaged default.
+
+ Args:
+ path: Optional JSON path. When omitted, the packaged RoboMIND ACT
+ benchmark defaults are loaded.
+
+ Returns:
+ A dictionary containing the benchmark identity, normalization version,
+ episode selection, and shared ACT/measurement configuration.
+ """
+ source = DEFAULT_EXPERIMENT if path is None else Path(path)
+ with source.expanduser().open(encoding="utf-8") as experiment_file:
+ experiment = json.load(experiment_file)
+ if not isinstance(experiment, dict):
+ raise ValueError("ACT experiment must be a JSON object.")
+ return experiment
diff --git a/paimon-python/pypaimon/benchmark/act/harness.py
b/paimon-python/pypaimon/benchmark/act/harness.py
new file mode 100644
index 0000000000..17be12ea14
--- /dev/null
+++ b/paimon-python/pypaimon/benchmark/act/harness.py
@@ -0,0 +1,577 @@
+# 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.
+
+"""Shared deterministic ACT model, trainer, and window plan for benchmarks."""
+
+import gc
+import hashlib
+import json
+import math
+import random
+import time
+import tracemalloc
+from dataclasses import asdict, dataclass
+from io import BytesIO
+
+import numpy as np
+import torch
+import torch.nn.functional as functional
+from PIL import Image
+from torch.utils.data import default_collate
+
+
+CAMERA_KEYS = (
+ "observation.images.front",
+ "observation.images.left_wrist",
+ "observation.images.right_wrist",
+)
+
+
+@dataclass(frozen=True)
+class BenchmarkConfig:
+ """Immutable model, sampling, training, and measurement parameters.
+
+ Every backend reconstructs this configuration from the resolved experiment
+ so tensor shapes, optimizer behavior, random seeds, and metric boundaries
+ remain comparable.
+ """
+
+ seed: int = 20260825
+ action_horizon: int = 32
+ batch_size: int = 2
+ optimizer_steps: int = 2
+ image_height: int = 64
+ image_width: int = 80
+ learning_rate: float = 1e-4
+ weight_decay: float = 1e-4
+ warmup_batches: int = 1
+ timed_batches: int = 32
+ fetch_batches: int = 8
+ rounds: int = 3
+
+ def __post_init__(self):
+ positive_ints = (
+ "action_horizon",
+ "batch_size",
+ "optimizer_steps",
+ "image_height",
+ "image_width",
+ "warmup_batches",
+ "timed_batches",
+ "fetch_batches",
+ )
+ for name in positive_ints:
+ value = getattr(self, name)
+ if (
+ isinstance(value, bool)
+ or not isinstance(value, int)
+ or value <= 0):
+ raise ValueError("%s must be a positive int." % name)
+ if isinstance(self.seed, bool) or not isinstance(self.seed, int):
+ raise ValueError("seed must be an int.")
+ if isinstance(self.rounds, bool) or not isinstance(self.rounds, int):
+ raise ValueError("rounds must be an int.")
+ if self.rounds < 3:
+ raise ValueError("rounds must be at least 3.")
+ if self.learning_rate <= 0:
+ raise ValueError("learning_rate must be positive.")
+ if self.weight_decay < 0:
+ raise ValueError("weight_decay must not be negative.")
+
+ def to_dict(self):
+ return asdict(self)
+
+
+@dataclass(frozen=True)
+class WindowPlan:
+ """Logical dataset-window indices consumed by one experiment.
+
+ Measurement indices cover warm-up and timed reads, train indices cover
+ fixed optimizer steps, and validation indices cover the final loss. These
+ are map-style dataset indices, not Paimon row IDs. ``sha256`` identifies
+ the exact plan across independent backend processes.
+ """
+
+ seed: int
+ measurement_indices: tuple
+ train_indices: tuple
+ validation_indices: tuple
+
+ @property
+ def sha256(self):
+ payload = json.dumps(
+ self.to_dict(), sort_keys=True, separators=(",", ":"))
+ return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+ def to_dict(self):
+ return {
+ "seed": self.seed,
+ "measurement_indices": list(self.measurement_indices),
+ "train_indices": list(self.train_indices),
+ "validation_indices": list(self.validation_indices),
+ }
+
+
+def build_window_plan(train_window_count, validation_window_count, config):
+ """Build deterministic measurement, training, and validation indices.
+
+ Args:
+ train_window_count: Number of complete windows in the train dataset.
+ validation_window_count: Number of complete validation windows.
+ config: Shared benchmark configuration supplying counts and the seed.
+
+ Returns:
+ A :class:`WindowPlan`. When more samples are needed than a dataset
+ contains, consecutive seeded permutations are concatenated; sampling
+ does not become independent sampling with replacement.
+ """
+ train_window_count = _positive_int(
+ train_window_count, "train_window_count")
+ validation_window_count = _positive_int(
+ validation_window_count, "validation_window_count")
+ batch_fetch_count = (
+ config.warmup_batches + config.timed_batches) * config.batch_size
+ train_count = config.optimizer_steps * config.batch_size
+ return WindowPlan(
+ seed=config.seed,
+ measurement_indices=tuple(_repeat_permutations(
+ train_window_count, batch_fetch_count, config.seed + 1)),
+ train_indices=tuple(_repeat_permutations(
+ train_window_count, train_count, config.seed + 2)),
+ validation_indices=tuple(_repeat_permutations(
+ validation_window_count, config.batch_size, config.seed + 3)),
+ )
+
+
+def decode_rgb_image(payload):
+ """Decode JPEG/PNG bytes into an ``H x W x 3`` RGB NumPy array.
+
+ Raises:
+ ValueError: If Pillow cannot decode the payload as an image.
+ """
+ try:
+ return np.asarray(Image.open(BytesIO(payload)).convert("RGB"))
+ except Exception as error:
+ raise ValueError("Cannot decode ACT RGB image bytes.") from error
+
+
+def decode_image_tensor(value):
+ """Decode bytes or an HDF5 uint8 value into normalized ``C x H x W``.
+
+ The returned NumPy array is float32 with values in ``[0, 1]``. Both
+ storage backends call this function so image conversion is not part of the
+ performance difference being measured.
+ """
+ payload = (
+ bytes(value)
+ if isinstance(value, (bytes, bytearray, memoryview))
+ else np.asarray(value, dtype=np.uint8).tobytes()
+ )
+ image = decode_rgb_image(payload)
+ return np.transpose(image, (2, 0, 1)).astype(np.float32) / 255.0
+
+
+def validate_act_batch(batch, config):
+ """Validate a collated batch against the shared ACT tensor contract.
+
+ Successful validation returns ``None``. It checks exact fields, tensor
+ shapes and dtypes, finite values, image range, complete unpadded windows,
+ and ``sample_id == episode_id#frame_index`` identity.
+ """
+ required = {
+ "sample_id", "episode_id", "frame_index", "qpos", "action",
+ "images", "is_pad",
+ }
+ if set(batch) != required:
+ raise ValueError(
+ "ACT batch fields differ: expected %s, got %s."
+ % (sorted(required), sorted(batch)))
+ batch_size = len(batch["sample_id"])
+ expected = {
+ "qpos": ((batch_size, 14), torch.float32),
+ "action": ((batch_size, config.action_horizon, 14), torch.float32),
+ "images": (
+ (batch_size, len(CAMERA_KEYS), 3)
+ + tuple(batch["images"].shape[-2:]),
+ torch.float32,
+ ),
+ "is_pad": ((batch_size, config.action_horizon), torch.bool),
+ "frame_index": ((batch_size,), torch.int64),
+ }
+ for name, (shape, dtype) in expected.items():
+ value = batch[name]
+ if not isinstance(value, torch.Tensor):
+ raise ValueError("%s must be a torch.Tensor." % name)
+ if tuple(value.shape) != shape:
+ raise ValueError(
+ "%s has shape %s; expected %s."
+ % (name, tuple(value.shape), shape))
+ if value.dtype != dtype:
+ raise ValueError(
+ "%s has dtype %s; expected %s." % (name, value.dtype, dtype))
+ for name in ("qpos", "action", "images"):
+ if not torch.isfinite(batch[name]).all():
+ raise ValueError("%s contains NaN or Inf." % name)
+ if torch.any(batch["images"] < 0) or torch.any(batch["images"] > 1):
+ raise ValueError("images must be normalized to [0, 1].")
+ if batch["is_pad"].any():
+ raise ValueError("ACT benchmark windows must be complete and
unpadded.")
+ for sample_id, episode_id, frame_index in zip(
+ batch["sample_id"], batch["episode_id"],
+ batch["frame_index"].tolist()):
+ if sample_id != "%s#%s" % (episode_id, frame_index):
+ raise ValueError(
+ "sample_id is not aligned with episode_id and frame_index.")
+
+
+def build_lerobot_batch(batch, config):
+ """Map a shared ACT batch to LeRobot ``ACTPolicy`` feature names.
+
+ Images are resized bilinearly to the configured height and width when
+ necessary. State, action, and padding retain their original semantics.
+ """
+ validate_act_batch(batch, config)
+ images = batch["images"]
+ target_size = (config.image_height, config.image_width)
+ if tuple(images.shape[-2:]) != target_size:
+ flat = images.flatten(0, 1)
+ flat = functional.interpolate(
+ flat, size=target_size, mode="bilinear", align_corners=False)
+ images = flat.reshape(images.shape[:3] + target_size)
+ result = {
+ "observation.state": batch["qpos"],
+ "action": batch["action"],
+ "action_is_pad": batch["is_pad"],
+ }
+ for index, name in enumerate(CAMERA_KEYS):
+ result[name] = images[:, index]
+ return result
+
+
+def build_act_policy(config):
+ """Build the reduced CPU ACT policy used only by this benchmark.
+
+ Returns:
+ ``(policy, metadata)`` containing the LeRobot policy and a
+ JSON-compatible description of its architecture and parameter counts.
+ Pretrained weights are disabled, so this function performs no model
+ download and does not represent a production training configuration.
+ """
+ try:
+ import importlib.metadata
+ from lerobot.configs.types import FeatureType, PolicyFeature
+ from lerobot.policies.act.configuration_act import ACTConfig
+ from lerobot.policies.act.modeling_act import ACTPolicy
+ except ImportError as error:
+ raise ImportError(
+ "ACT benchmark requires: "
+ "pip install -e '.[act]'.") from error
+
+ inputs = {
+ "observation.state": PolicyFeature(FeatureType.STATE, (14,)),
+ }
+ inputs.update({
+ name: PolicyFeature(
+ FeatureType.VISUAL,
+ (3, config.image_height, config.image_width),
+ )
+ for name in CAMERA_KEYS
+ })
+ act_config = ACTConfig(
+ input_features=inputs,
+ output_features={
+ "action": PolicyFeature(FeatureType.ACTION, (14,)),
+ },
+ device="cpu",
+ chunk_size=config.action_horizon,
+ n_action_steps=config.action_horizon,
+ vision_backbone="resnet18",
+ pretrained_backbone_weights=None,
+ dim_model=64,
+ n_heads=4,
+ dim_feedforward=256,
+ n_encoder_layers=1,
+ n_decoder_layers=1,
+ use_vae=True,
+ latent_dim=16,
+ n_vae_encoder_layers=1,
+ kl_weight=10.0,
+ )
+ policy = ACTPolicy(act_config)
+ return policy, {
+ "implementation": "lerobot.ACTPolicy",
+ "lerobot_version": importlib.metadata.version("lerobot"),
+ "vision_backbone": act_config.vision_backbone,
+ "pretrained_backbone_weights": act_config.pretrained_backbone_weights,
+ "chunk_size": act_config.chunk_size,
+ "dim_model": act_config.dim_model,
+ "n_heads": act_config.n_heads,
+ "n_encoder_layers": act_config.n_encoder_layers,
+ "n_decoder_layers": act_config.n_decoder_layers,
+ "n_vae_encoder_layers": act_config.n_vae_encoder_layers,
+ "latent_dim": act_config.latent_dim,
+ "kl_weight": act_config.kl_weight,
+ "parameter_count": sum(
+ parameter.numel() for parameter in policy.parameters()),
+ "trainable_parameter_count": sum(
+ parameter.numel()
+ for parameter in policy.parameters() if parameter.requires_grad),
+ }
+
+
+def run_backend(
+ backend,
+ round_number,
+ dataset_factory,
+ plan,
+ config,
+ sample_sequence_sha256,
+ policy_factory=None):
+ """Measure one backend with the shared plan, model, and trainer.
+
+ ``backend`` is a result label and ``round_number`` identifies the repeat.
+ ``dataset_factory`` must return ``(train_dataset, validation_dataset)`` and
+ must be reusable: it is called for the timed run and again by the separate
+ Python-memory replay. ``policy_factory`` is an optional test hook returning
+ ``(policy, model_metadata)``.
+
+ Returns:
+ A JSON-compatible metrics dictionary covering dataset construction,
+ first batch, timed batch fetch, fixed optimizer steps, validation loss,
+ and a separate ``tracemalloc`` peak replay. ``fixed_steps_s`` includes
+ dataset fetch, while each ``train_trace.step_time_s`` starts after its
+ batch is fetched and covers conversion, forward/backward, and optimizer
+ update. OS page cache is not controlled and native Arrow/Torch
+ allocations are outside tracemalloc.
+ """
+ _seed_everything(config.seed)
+ policy_factory = policy_factory or build_act_policy
+ started = time.monotonic()
+ dataset_started = time.monotonic()
+ train_dataset, validation_dataset = dataset_factory()
+ dataset_build_s = time.monotonic() - dataset_started
+
+ warmup_sample_count = config.warmup_batches * config.batch_size
+ warmup_iterator = _iter_logical_batches(
+ train_dataset,
+ plan.measurement_indices[:warmup_sample_count],
+ logical_batch_size=config.batch_size,
+ fetch_batches=1,
+ )
+ first_batch_started = time.monotonic()
+ first_batch = next(warmup_iterator)
+ first_batch_s = time.monotonic() - first_batch_started
+ validate_act_batch(first_batch, config)
+ for _ in range(config.warmup_batches - 1):
+ validate_act_batch(next(warmup_iterator), config)
+
+ batch_fetch_iterator = _iter_logical_batches(
+ train_dataset,
+ plan.measurement_indices[warmup_sample_count:],
+ logical_batch_size=config.batch_size,
+ fetch_batches=config.fetch_batches,
+ )
+ batch_fetch_seconds = 0.0
+ batch_fetch_sample_count = 0
+ for _ in range(config.timed_batches):
+ batch_fetch_started = time.monotonic()
+ batch = next(batch_fetch_iterator)
+ batch_fetch_seconds += time.monotonic() - batch_fetch_started
+ validate_act_batch(batch, config)
+ batch_fetch_sample_count += len(batch["sample_id"])
+
+ _seed_everything(config.seed)
+ policy, model = policy_factory(config)
+ parameters = (
+ policy.get_optim_params()
+ if hasattr(policy, "get_optim_params") else policy.parameters())
+ optimizer = torch.optim.AdamW(
+ parameters,
+ lr=config.learning_rate,
+ weight_decay=config.weight_decay,
+ )
+ policy.train()
+ train_started = time.monotonic()
+ losses = []
+ for step, batch in enumerate(_iter_logical_batches(
+ train_dataset,
+ plan.train_indices,
+ logical_batch_size=config.batch_size,
+ fetch_batches=config.fetch_batches,
+ ), 1):
+ step_started = time.monotonic()
+ model_batch = build_lerobot_batch(batch, config)
+ optimizer.zero_grad(set_to_none=True)
+ loss, components = policy(model_batch)
+ if loss.ndim != 0 or not torch.isfinite(loss):
+ raise FloatingPointError(
+ "ACT produced a non-finite scalar loss at step %d." % step)
+ loss.backward()
+ optimizer.step()
+ losses.append({
+ "step": step,
+ "total": float(loss.detach()),
+ "components": {
+ name: _finite_float(value, name)
+ for name, value in components.items()
+ },
+ "step_time_s": time.monotonic() - step_started,
+ })
+ fixed_steps_s = time.monotonic() - train_started
+ if len(losses) != config.optimizer_steps:
+ raise AssertionError(
+ "Expected %d optimizer steps, got %d."
+ % (config.optimizer_steps, len(losses)))
+
+ # ACTPolicy only constructs the VAE posterior needed by its supervised
+ # loss while the module is in training mode. Keep that mode for validation
+ # but disable gradients and parameter updates below.
+ policy.train()
+ _seed_everything(config.seed + 4)
+ validation_batch = next(_iter_logical_batches(
+ validation_dataset,
+ plan.validation_indices,
+ logical_batch_size=config.batch_size,
+ fetch_batches=config.fetch_batches,
+ ))
+ with torch.no_grad():
+ validation_loss, _ = policy(build_lerobot_batch(
+ validation_batch, config))
+ validation_value = _finite_float(validation_loss, "validation_loss")
+ wall_time_s = time.monotonic() - started
+ python_peak = _measure_python_peak(dataset_factory, plan, config)
+
+ return {
+ "round": round_number,
+ "backend": backend,
+ "sample_sequence_sha256": sample_sequence_sha256,
+ "model": model,
+ "optimizer": {
+ "name": "AdamW",
+ "learning_rate": config.learning_rate,
+ "weight_decay": config.weight_decay,
+ },
+ "warmup_batches": config.warmup_batches,
+ "first_batch_s": first_batch_s,
+ "dataset_build_s": dataset_build_s,
+ "batch_fetch_samples": batch_fetch_sample_count,
+ "batch_fetch_s": batch_fetch_seconds,
+ "batch_fetch_samples_per_s": (
+ batch_fetch_sample_count / batch_fetch_seconds),
+ "fixed_steps_s": fixed_steps_s,
+ "train_loss": [item["total"] for item in losses],
+ "train_trace": losses,
+ "validation_loss": validation_value,
+ "python_peak_allocated_bytes": python_peak,
+ "peak_memory_measurement": (
+ "python-tracemalloc-separate-dataset-first-batch"),
+ "wall_time_s": wall_time_s,
+ }
+
+
+def _measure_python_peak(dataset_factory, plan, config):
+ """Measure Python allocation peak in a separate dataset-first-batch replay.
+
+ The factory is called again so tracing overhead cannot distort the main
+ throughput timings. The returned integer is the tracemalloc peak in bytes.
+ """
+ gc.collect()
+ tracemalloc.start()
+ try:
+ train_dataset, _ = dataset_factory()
+ indices = plan.measurement_indices[
+ :config.batch_size * config.fetch_batches
+ ]
+ next(_iter_logical_batches(
+ train_dataset,
+ indices,
+ logical_batch_size=config.batch_size,
+ fetch_batches=config.fetch_batches,
+ ))
+ _, peak = tracemalloc.get_traced_memory()
+ return peak
+ finally:
+ tracemalloc.stop()
+
+
+def _repeat_permutations(size, count, seed):
+ """Return ``count`` indices by concatenating seeded permutations."""
+ values = []
+ generator = np.random.RandomState(seed)
+ while len(values) < count:
+ values.extend(generator.permutation(size).tolist())
+ return values[:count]
+
+
+def _seed_everything(seed):
+ """Reset Python, NumPy, and Torch RNGs and enable deterministic Torch
ops."""
+ random.seed(seed)
+ np.random.seed(seed)
+ torch.manual_seed(seed)
+ torch.use_deterministic_algorithms(True)
+
+
+def _finite_float(value, name):
+ if isinstance(value, torch.Tensor):
+ if value.numel() != 1:
+ raise ValueError("%s must be scalar." % name)
+ value = float(value.detach())
+ else:
+ value = float(value)
+ if not math.isfinite(value):
+ raise FloatingPointError("%s is NaN or Inf." % name)
+ return value
+
+
+def _positive_int(value, name):
+ if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+ raise ValueError("%s must be a positive int." % name)
+ return value
+
+
+def _iter_logical_batches(
+ dataset, indices, *, logical_batch_size, fetch_batches):
+ """Yield collated model batches while coalescing physical dataset reads.
+
+ Args:
+ dataset: Map-style dataset implementing ``__getitem__`` and optionally
+ plural ``__getitems__(indices)`` access.
+ indices: Explicit ordered logical-window indices. Their count must be
+ divisible by ``logical_batch_size``.
+ logical_batch_size: Number of samples consumed by one model step.
+ fetch_batches: Logical batches combined into one physical dataset read.
+
+ Yields:
+ Collated logical batches in the exact input-index order. A plural
+ dataset method is preferred when available; otherwise samples are read
+ individually and split back into the same logical batches.
+ """
+ logical_batch_size = _positive_int(
+ logical_batch_size, "logical_batch_size")
+ fetch_batches = _positive_int(fetch_batches, "fetch_batches")
+ if len(indices) % logical_batch_size:
+ raise ValueError("indices must contain complete logical batches.")
+ physical_size = logical_batch_size * fetch_batches
+ getitems = getattr(dataset, "__getitems__", None)
+ for offset in range(0, len(indices), physical_size):
+ physical_indices = list(indices[offset:offset + physical_size])
+ if getitems is None:
+ samples = [dataset[index] for index in physical_indices]
+ else:
+ samples = getitems(physical_indices)
+ for logical_offset in range(0, len(samples), logical_batch_size):
+ yield default_collate(
+ samples[logical_offset:logical_offset + logical_batch_size])
diff --git a/paimon-python/pypaimon/benchmark/act/hdf5.py
b/paimon-python/pypaimon/benchmark/act/hdf5.py
new file mode 100644
index 0000000000..4e37e461cd
--- /dev/null
+++ b/paimon-python/pypaimon/benchmark/act/hdf5.py
@@ -0,0 +1,201 @@
+# 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.
+
+"""HDF5 dataset adapter for the RoboMIND ACT benchmark."""
+
+import numpy as np
+import torch
+from torch.utils.data import Dataset
+
+from pypaimon.benchmark.act.harness import decode_image_tensor
+
+
+QPOS_FIELDS = (
+ "puppet/joint_position_left",
+ "puppet/joint_position_right",
+)
+ACTION_FIELDS = (
+ "master/joint_position_left",
+ "master/joint_position_right",
+)
+IMAGE_FIELDS = (
+ "observations/rgb_images/camera_front",
+ "observations/rgb_images/camera_left_wrist",
+ "observations/rgb_images/camera_right_wrist",
+)
+
+
+class Hdf5ACTWindowDataset(Dataset):
+ """Read complete ACT windows lazily from one HDF5 episode.
+
+ ``episode`` supplies the file path, logical episode ID, and frame count.
+ For a window anchor, state and three camera images come from the anchor
+ frame while actions cover ``[anchor, anchor + action_horizon)``. Each
+ access opens and closes the HDF5 file and returns the shared ACT sample
+ mapping consumed by :mod:`pypaimon.benchmark.act.harness`.
+ """
+
+ def __init__(self, episode, normalization, action_horizon):
+ self.episode = episode
+ self.normalization = normalization
+ self.action_horizon = action_horizon
+ self.window_count = episode.frame_count - action_horizon + 1
+ if self.window_count <= 0:
+ raise ValueError(
+ "Episode %s is shorter than action horizon %d."
+ % (episode.episode_id, action_horizon))
+
+ def __len__(self):
+ return self.window_count
+
+ def __getitem__(self, anchor):
+ """Return the ACT window whose first frame is ``anchor``.
+
+ Negative anchors follow Python sequence semantics. State and images
+ come from the anchor frame, while action contains the complete horizon.
+ """
+ if anchor < 0:
+ anchor += self.window_count
+ if anchor < 0 or anchor >= self.window_count:
+ raise IndexError(anchor)
+ import h5py
+
+ with h5py.File(str(self.episode.path), "r") as h5:
+ qpos = _read_vectors(h5, QPOS_FIELDS, anchor)
+ action = _read_vectors(
+ h5,
+ ACTION_FIELDS,
+ slice(anchor, anchor + self.action_horizon),
+ )
+ images = np.stack([
+ decode_image_tensor(h5[field][anchor]) for field in
IMAGE_FIELDS
+ ])
+ qpos = (
+ (qpos - self.normalization["qpos_mean"])
+ / self.normalization["qpos_std"])
+ action = (
+ (action - self.normalization["action_mean"])
+ / self.normalization["action_std"])
+ return {
+ "sample_id": "%s#%d" % (self.episode.episode_id, anchor),
+ "episode_id": self.episode.episode_id,
+ "frame_index": anchor,
+ "qpos": torch.from_numpy(np.ascontiguousarray(qpos)),
+ "action": torch.from_numpy(np.ascontiguousarray(action)),
+ "images": torch.from_numpy(np.ascontiguousarray(images)),
+ "is_pad": torch.zeros(self.action_horizon, dtype=torch.bool),
+ }
+
+
+def create_datasets(train_episode, validation_episode, normalization, config):
+ """Create HDF5 datasets for the experiment's selected episodes.
+
+ Args:
+ train_episode: Selected training episode with its HDF5 path and frame
+ count.
+ validation_episode: Selected validation episode with the same fields.
+ normalization: Shared state and action normalization arrays.
+ config: Benchmark configuration containing the action horizon.
+
+ Returns:
+ ``(train_dataset, validation_dataset)`` in that order.
+ """
+ return (
+ Hdf5ACTWindowDataset(
+ train_episode, normalization, config.action_horizon),
+ Hdf5ACTWindowDataset(
+ validation_episode, normalization, config.action_horizon),
+ )
+
+
+def compute_normalization(episodes):
+ """Compute train-only HDF5 state and action normalization.
+
+ Args:
+ episodes: Discovered episodes carrying ``path``, ``split``, and
+ ``success`` attributes.
+
+ Returns:
+ ``(normalization, metadata)`` where normalization contains float32
+ arrays used by training. Metadata retains float64-accumulated moments
+ over the canonical float32 actions and the frame count used to validate
+ Paimon statistics. Standard deviations use a ``1e-2`` floor.
+ """
+ train = [
+ episode for episode in episodes
+ if episode.split == "train" and episode.success
+ ]
+ if not train:
+ raise ValueError("No successful train episodes are available.")
+ qpos = _Moments(14)
+ action = _Moments(14)
+ import h5py
+
+ for episode in sorted(train, key=lambda item: item.episode_id):
+ with h5py.File(str(episode.path), "r") as h5:
+ qpos.update(_read_vectors(
+ h5, QPOS_FIELDS, slice(None), dtype=np.float64))
+ action.update(_read_vectors(
+ h5, ACTION_FIELDS, slice(None), dtype=np.float32))
+ qpos_mean, qpos_std = qpos.finish()
+ action_mean, action_std = action.finish()
+ return ({
+ "qpos_mean": qpos_mean.astype(np.float32),
+ "qpos_std": qpos_std.astype(np.float32),
+ "action_mean": action_mean.astype(np.float32),
+ "action_std": action_std.astype(np.float32),
+ }, {
+ "action_mean": action_mean,
+ "action_std": action_std,
+ "frame_count": action.count,
+ })
+
+
+def _read_vectors(h5, fields, selection, dtype=np.float32):
+ value = np.concatenate([
+ np.asarray(h5[field][selection], dtype=dtype) for field in fields
+ ], axis=-1)
+ if not np.isfinite(value).all():
+ raise ValueError("ACT vector contains NaN or Inf.")
+ return value
+
+
+class _Moments(object):
+ """Accumulate float64 population moments with a ``1e-2`` std floor."""
+
+ def __init__(self, width):
+ self.count = 0
+ self.total = np.zeros(width, dtype=np.float64)
+ self.total_square = np.zeros(width, dtype=np.float64)
+
+ def update(self, value):
+ value = np.asarray(value, dtype=np.float64)
+ if value.ndim != 2 or value.shape[1] != len(self.total):
+ raise ValueError(
+ "Unexpected normalization shape %s." % (value.shape,))
+ if not np.isfinite(value).all():
+ raise ValueError("Normalization input contains NaN or Inf.")
+ self.count += value.shape[0]
+ self.total += value.sum(axis=0)
+ self.total_square += np.square(value).sum(axis=0)
+
+ def finish(self):
+ if self.count == 0:
+ raise ValueError("Cannot compute normalization from no frames.")
+ mean = self.total / self.count
+ variance = np.maximum(
+ self.total_square / self.count - np.square(mean), 0.0)
+ return mean, np.maximum(np.sqrt(variance), 1e-2)
diff --git a/paimon-python/pypaimon/benchmark/act/paimon.py
b/paimon-python/pypaimon/benchmark/act/paimon.py
new file mode 100644
index 0000000000..18b50664b6
--- /dev/null
+++ b/paimon-python/pypaimon/benchmark/act/paimon.py
@@ -0,0 +1,153 @@
+# 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.
+
+"""Paimon dataset adapter for the RoboMIND ACT benchmark."""
+
+import numpy as np
+import torch
+
+from pypaimon.benchmark.act.harness import decode_image_tensor
+from pypaimon.sample import robomind_agilex as agilex
+
+
+QPOS_COLUMNS = (
+ "state_joint_position_left",
+ "state_joint_position_right",
+)
+ACTION_COLUMNS = ("action",)
+IMAGE_COLUMNS = (
+ "observation_images_rgb_front",
+ "observation_images_rgb_wrist_left",
+ "observation_images_rgb_wrist_right",
+)
+
+
+class PaimonACTAdapter:
+ """Convert a contiguous Paimon row window to the shared ACT sample.
+
+ State and camera columns are taken from the anchor row. The action column
+ covers the full horizon. The returned mapping has the same IDs, tensors,
+ shapes, and normalization as :class:`Hdf5ACTWindowDataset`.
+ """
+
+ def __init__(self, normalization):
+ self.normalization = normalization
+
+ def __call__(self, sample):
+ """Convert the generic window mapping into ACT tensors and identity.
+
+ 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.
+ """
+ qpos = np.concatenate([
+ np.asarray(sample[name][0], dtype=np.float32)
+ for name in QPOS_COLUMNS
+ ])
+ action = np.concatenate([
+ np.asarray(sample[name], dtype=np.float32)
+ for name in ACTION_COLUMNS
+ ], axis=-1)
+ images = np.stack([
+ decode_image_tensor(sample[name][0]) for name in IMAGE_COLUMNS
+ ])
+ qpos = (
+ (qpos - self.normalization["qpos_mean"])
+ / self.normalization["qpos_std"])
+ action = (
+ (action - self.normalization["action_mean"])
+ / self.normalization["action_std"])
+ episode_id = sample["episode_id"]
+ frame_index = sample["frame_index"]
+ return {
+ "sample_id": "%s#%d" % (episode_id, frame_index),
+ "episode_id": episode_id,
+ "frame_index": frame_index,
+ "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"],
+ }
+
+
+def create_datasets(
+ frames,
+ snapshot_id,
+ train_episode_id,
+ validation_episode_id,
+ normalization,
+ config):
+ """Create lazy train and validation windows pinned to one snapshot.
+
+ State and image columns are anchor-only, so one sample reads the initial
+ joint position and three observation images once rather than once per
+ action-horizon row.
+
+ Args:
+ frames: Paimon frames table used to create both scans.
+ snapshot_id: Snapshot pinned by experiment preparation. Both returned
+ datasets reject any different resolved snapshot.
+ train_episode_id: Episode selected for training windows.
+ validation_episode_id: Episode selected for validation windows.
+ normalization: Shared state and action normalization arrays.
+ config: Benchmark configuration containing the action horizon.
+
+ Returns:
+ ``(train_dataset, validation_dataset)`` in that order, as lazy
+ ``ContiguousWindowDataset`` instances pinned to ``snapshot_id``.
+ """
+ datasets = tuple(
+ frames.scan(snapshot_id=snapshot_id).where(
+ "episode_id = '%s'" % episode_id.replace("'", "''")
+ ).to_contiguous_window_dataset(
+ window_size=config.action_horizon,
+ 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",
+ adapter=PaimonACTAdapter(normalization),
+ )
+ for episode_id in (train_episode_id, validation_episode_id)
+ )
+ actual_snapshot_ids = {dataset.snapshot_id for dataset in datasets}
+ if actual_snapshot_ids != {snapshot_id}:
+ raise RuntimeError(
+ "Paimon ACT windows must remain pinned to frames snapshot %s; "
+ "got %s." % (snapshot_id, sorted(actual_snapshot_ids)))
+ return datasets
+
+
+def statistics_row(connection, statistics_version):
+ """Return the unique versioned action-statistics row."""
+ escaped = statistics_version.replace("'", "''")
+ rows = (connection.get_table(agilex.FEATURE_STATS_TABLE).scan()
+ .where("statistics_version = '%s'" % escaped).to_list())
+ if len(rows) != 1:
+ raise ValueError(
+ "Expected one normalization row for %r, got %d."
+ % (statistics_version, len(rows)))
+ return rows[0]
+
+
+def latest_snapshot_id(table):
+ """Return the table's latest snapshot ID or fail for an empty table."""
+ snapshot = table.raw_table.snapshot_manager().get_latest_snapshot()
+ if snapshot is None:
+ raise ValueError("Paimon frames table has no snapshot.")
+ return snapshot.id
diff --git a/paimon-python/pypaimon/benchmark/act/runner.py
b/paimon-python/pypaimon/benchmark/act/runner.py
new file mode 100644
index 0000000000..657aba0b72
--- /dev/null
+++ b/paimon-python/pypaimon/benchmark/act/runner.py
@@ -0,0 +1,740 @@
+# 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.
+
+"""Prepare and run RoboMIND ACT benchmarks over HDF5 or Paimon.
+
+Both adapters consume one immutable :class:`BenchmarkConfig`, one train-only
+normalization object, and one explicit window plan. The runner resets the same
+seed before constructing the same LeRobot ACT policy and AdamW trainer for each
+backend. Each backend runs independently without attempting OS cache control
+and writes its tensor fingerprint, loss trace, timing metrics, and Python
+allocation metrics to one result JSON document.
+Ingestion and canonical-action backfill are deliberately outside the benchmark.
+"""
+
+import gc
+import hashlib
+import json
+import os
+import platform
+import subprocess
+import time
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+import PIL
+import h5py
+import numpy as np
+import pyarrow as pa
+import torch
+import pypaimon.multimodal as pmm
+from pypaimon import build_info
+from pypaimon.benchmark.act.hdf5 import (
+ compute_normalization as compute_hdf5_normalization,
+ create_datasets as create_hdf5_datasets,
+)
+from pypaimon.benchmark.act.experiment import load_experiment
+from pypaimon.benchmark.act.harness import (
+ BenchmarkConfig,
+ WindowPlan,
+ build_window_plan,
+ run_backend,
+)
+from pypaimon.benchmark.act.compare import canonical_sha256
+from pypaimon.benchmark.act.paimon import (
+ create_datasets as create_paimon_datasets,
+ latest_snapshot_id,
+ statistics_row,
+)
+from pypaimon.sample import robomind_agilex as agilex
+
+
+@dataclass(frozen=True)
+class _BenchmarkEpisode:
+ path: Path
+ source_key: str
+ episode_id: str
+ split: str
+ success: bool
+ frame_count: int
+
+
+def prepare_experiment(
+ input_root,
+ warehouse,
+ output_path,
+ *,
+ definition=None,
+ database=agilex.DEFAULT_DATABASE):
+ """Resolve a benchmark definition against matching HDF5 and Paimon data.
+
+ Preparation is outside timed benchmark execution. It verifies source
+ identity and Paimon statistics, selects eligible train/validation episodes,
+ computes train-only normalization, and fixes every logical window index.
+
+ Args:
+ input_root: RoboMIND AgileX HDF5 root used as the source of episode
+ files and raw normalization moments.
+ warehouse: Existing Paimon warehouse containing the matching ingested
+ and canonical-action-backfilled dataset.
+ output_path: Destination for the resolved experiment JSON document.
+ definition: Optional decoded experiment definition. The packaged
+ defaults are used when omitted.
+ database: Paimon database containing the RoboMIND tables.
+
+ Returns:
+ The resolved, JSON-compatible experiment dictionary written to
+ ``output_path``.
+ """
+ definition = load_experiment() if definition is None else definition
+ if definition.get("schema_version") != "act-benchmark-experiment@1":
+ raise ValueError("Unsupported ACT benchmark experiment schema.")
+ config = BenchmarkConfig(**definition["config"])
+ statistics_version = definition["statistics_version"]
+ input_root = Path(input_root).expanduser().resolve()
+ warehouse = Path(warehouse).expanduser().resolve()
+ output_path = Path(output_path).expanduser().resolve()
+
+ discovered = agilex.discover_episodes(input_root)
+ connection = pmm.connect(
+ database=database, options={"warehouse": str(warehouse)})
+ source_episodes, source_sha256 = _validate_source_identity(
+ discovered, _episode_rows(connection))
+ source_by_id = {episode.episode_id: episode for episode in source_episodes}
+ frames = connection.get_table(agilex.FRAMES_TABLE)
+ frames_snapshot_id = latest_snapshot_id(frames)
+ normalization, normalization_metadata = _shared_normalization(
+ source_episodes,
+ connection,
+ frames_snapshot_id,
+ statistics_version,
+ )
+ del normalization
+ train_episode = _select_episode(
+ source_by_id,
+ split="train",
+ requested=definition.get("train_episode_id"),
+ action_horizon=config.action_horizon,
+ )
+ validation_episode = _select_episode(
+ source_by_id,
+ split="val",
+ requested=definition.get("validation_episode_id"),
+ action_horizon=config.action_horizon,
+ )
+ plan = build_window_plan(
+ train_episode.frame_count - config.action_horizon + 1,
+ validation_episode.frame_count - config.action_horizon + 1,
+ config,
+ )
+ sequence_sha256 = _sample_sequence_sha256(
+ train_episode.episode_id, validation_episode.episode_id, plan)
+ episodes = sorted(({
+ "episode_id": episode.episode_id,
+ "source_key": episode.source_key,
+ "split": episode.split,
+ "success": episode.success,
+ "frame_count": episode.frame_count,
+ } for episode in source_episodes), key=lambda item: item["episode_id"])
+ experiment = {
+ "schema_version": "act-benchmark-experiment@1",
+ "benchmark_id": definition.get("benchmark_id", "robomind-act"),
+ "dataset": definition.get("dataset", "RoboMIND AgileX"),
+ "config": config.to_dict(),
+ "statistics_version": statistics_version,
+ "train_episode_id": train_episode.episode_id,
+ "validation_episode_id": validation_episode.episode_id,
+ "source": {
+ "sha256": source_sha256,
+ "episodes": episodes,
+ },
+ "normalization": normalization_metadata,
+ "window_plan": {
+ **plan.to_dict(),
+ "sha256": plan.sha256,
+ "sample_sequence_sha256": sequence_sha256,
+ },
+ "paimon": {
+ "database": database,
+ "frames_table": agilex.FRAMES_TABLE,
+ "frames_snapshot_id": frames_snapshot_id,
+ },
+ }
+ output_path.parent.mkdir(parents=True, exist_ok=True)
+ output_path.write_text(
+ json.dumps(experiment, indent=2, sort_keys=True) + "\n",
+ encoding="utf-8",
+ )
+ return experiment
+
+
+def run_experiment(
+ backend,
+ experiment_path,
+ output_path,
+ *,
+ input_root=None,
+ warehouse=None,
+ policy_factory=None):
+ """Run one storage backend against a resolved ACT experiment.
+
+ Args:
+ backend: Result label and dataset implementation, either ``hdf5`` or
+ ``paimon``.
+ experiment_path: Resolved JSON produced by :func:`prepare_experiment`.
+ output_path: Destination JSON result path.
+ input_root: Required only for the HDF5 backend.
+ warehouse: Required only for the Paimon backend.
+ policy_factory: Optional test hook returning ``(policy, metadata)``.
+
+ Returns:
+ A JSON-compatible single-backend result containing the resolved
+ experiment, runtime environment, tensor fingerprint, per-round raw
+ metrics, and median/min/max summary.
+ """
+ if backend not in ("hdf5", "paimon"):
+ raise ValueError("backend must be 'hdf5' or 'paimon'.")
+ experiment = load_experiment(experiment_path)
+ _validate_resolved_experiment(experiment)
+ config = BenchmarkConfig(**experiment["config"])
+ plan = _window_plan_from_experiment(experiment)
+ normalization = {
+ name: np.asarray(value, dtype=np.float32)
+ for name, value in experiment["normalization"]["values"].items()
+ }
+ sequence_sha256 = experiment["window_plan"]["sample_sequence_sha256"]
+ if backend == "hdf5":
+ if input_root is None:
+ raise ValueError("input_root is required for the HDF5 backend.")
+ episodes = _hdf5_episodes_from_experiment(input_root, experiment)
+ by_id = {episode.episode_id: episode for episode in episodes}
+ train_episode = by_id[experiment["train_episode_id"]]
+ validation_episode = by_id[experiment["validation_episode_id"]]
+
+ def dataset_factory():
+ return create_hdf5_datasets(
+ train_episode, validation_episode, normalization, config)
+
+ source = {"input_root": str(Path(input_root).expanduser().resolve())}
+ else:
+ if warehouse is None:
+ raise ValueError("warehouse is required for the Paimon backend.")
+ dataset_factory, source = _paimon_factory_from_experiment(
+ warehouse, experiment, normalization, config)
+
+ started_at = _utc_now()
+ started = time.monotonic()
+ runs = []
+ for round_number in range(1, config.rounds + 1):
+ runs.append(run_backend(
+ backend,
+ round_number,
+ dataset_factory,
+ plan,
+ config,
+ sequence_sha256,
+ policy_factory=policy_factory,
+ ))
+ gc.collect()
+ # Fingerprinting scans samples, so keep it after the timed rounds.
+ fingerprint = _tensor_fingerprint(dataset_factory(), plan)
+ result = {
+ "schema_version": "act-benchmark-result@1",
+ "benchmark_id": experiment["benchmark_id"],
+ "run_id": "%s-%s" % (
+ started_at.replace(":", "").replace("-", ""),
+ uuid.uuid4().hex[:8],
+ ),
+ "status": "SUCCEEDED",
+ "backend": backend,
+ "experiment": experiment,
+ "experiment_sha256": canonical_sha256(experiment),
+ "source": source,
+ "tensor_fingerprint": fingerprint,
+ "model": runs[0]["model"],
+ "runs": runs,
+ "summary": _summarize(runs),
+ "environment": _runtime_environment(
+ Path(__file__).resolve().parents[4]),
+ "command": _command_argv(),
+ "timing": {"wall_time_s": time.monotonic() - started},
+ "unverified": [
+ "OS page cache is uncontrolled; no cache dropping was attempted.",
+ "CPU fixed-step loss parity proves engineering equivalence, "
+ "not policy quality.",
+ "GPU, multi-worker dataset loading, distributed training, and "
+ "recovery are unverified.",
+ "Python tracemalloc excludes native Arrow and Torch allocations.",
+ ],
+ "started_at": started_at,
+ "finished_at": _utc_now(),
+ }
+ output_path = Path(output_path).expanduser().resolve()
+ output_path.parent.mkdir(parents=True, exist_ok=True)
+ output_path.write_text(
+ json.dumps(result, indent=2, sort_keys=True) + "\n",
+ encoding="utf-8",
+ )
+ return result
+
+
+def _validate_resolved_experiment(experiment):
+ """Reject incomplete or internally inconsistent resolved experiments."""
+ required = {
+ "schema_version", "benchmark_id", "dataset", "config",
+ "statistics_version", "train_episode_id", "validation_episode_id",
+ "source", "normalization", "window_plan", "paimon",
+ }
+ if experiment.get("schema_version") != "act-benchmark-experiment@1":
+ raise ValueError("Unsupported ACT benchmark experiment schema.")
+ missing = required - set(experiment)
+ if missing:
+ raise ValueError(
+ "Resolved ACT experiment is missing: %s."
+ % ", ".join(sorted(missing)))
+ source = experiment["source"]
+ if canonical_sha256(source["episodes"]) != source["sha256"]:
+ raise ValueError("ACT experiment source-manifest hash differs.")
+ normalization = experiment["normalization"]
+ if canonical_sha256(normalization["values"]) != normalization["sha256"]:
+ raise ValueError("ACT experiment normalization hash differs.")
+ plan = _window_plan_from_experiment(experiment)
+ if plan.sha256 != experiment["window_plan"]["sha256"]:
+ raise ValueError("ACT experiment window-plan hash differs.")
+ episodes = {
+ item["episode_id"]: item for item in source["episodes"]
+ }
+ try:
+ train = episodes[experiment["train_episode_id"]]
+ validation = episodes[experiment["validation_episode_id"]]
+ except KeyError as error:
+ raise ValueError(
+ "ACT experiment selected episode is absent from the source."
+ ) from error
+ config = BenchmarkConfig(**experiment["config"])
+ expected_plan = build_window_plan(
+ train["frame_count"] - config.action_horizon + 1,
+ validation["frame_count"] - config.action_horizon + 1,
+ config,
+ )
+ if expected_plan.to_dict() != plan.to_dict():
+ raise ValueError(
+ "ACT experiment window plan was not built from its config and "
+ "selected episodes.")
+ expected_sequence = _sample_sequence_sha256(
+ train["episode_id"], validation["episode_id"], plan)
+ if expected_sequence != experiment["window_plan"][
+ "sample_sequence_sha256"]:
+ raise ValueError("ACT experiment sample-sequence hash differs.")
+
+
+def _window_plan_from_experiment(experiment):
+ """Reconstruct immutable logical-window indices from JSON values."""
+ value = experiment["window_plan"]
+ return WindowPlan(
+ seed=value["seed"],
+ measurement_indices=tuple(value["measurement_indices"]),
+ train_indices=tuple(value["train_indices"]),
+ validation_indices=tuple(value["validation_indices"]),
+ )
+
+
+def _hdf5_episodes_from_experiment(input_root, experiment):
+ """Validate HDF5 episode identity and attach manifest frame counts."""
+ discovered =
agilex.discover_episodes(Path(input_root).expanduser().resolve())
+ by_id = {episode.episode_id: episode for episode in discovered}
+ expected = experiment["source"]["episodes"]
+ actual_identity = sorted(({
+ "episode_id": episode.episode_id,
+ "source_key": episode.source_key,
+ "split": episode.split,
+ "success": episode.success,
+ } for episode in discovered), key=lambda item: item["episode_id"])
+ expected_identity = [{
+ "episode_id": item["episode_id"],
+ "source_key": item["source_key"],
+ "split": item["split"],
+ "success": item["success"],
+ } for item in expected]
+ if actual_identity != expected_identity:
+ raise ValueError("HDF5 source differs from the ACT experiment.")
+ return [
+ _BenchmarkEpisode(
+ path=by_id[item["episode_id"]].path,
+ source_key=item["source_key"],
+ episode_id=item["episode_id"],
+ split=item["split"],
+ success=item["success"],
+ frame_count=item["frame_count"],
+ )
+ for item in expected
+ ]
+
+
+def _paimon_factory_from_experiment(
+ warehouse, experiment, normalization, config):
+ """Validate Paimon source/statistics and return a pinned dataset
factory."""
+ warehouse = Path(warehouse).expanduser().resolve()
+ paimon = experiment["paimon"]
+ connection = pmm.connect(
+ database=paimon["database"],
+ options={"warehouse": str(warehouse)},
+ )
+ frames = connection.get_table(paimon["frames_table"])
+ snapshot_id = paimon["frames_snapshot_id"]
+ expected_episodes = experiment["source"]["episodes"]
+ actual_episodes = sorted(_episode_rows(connection),
+ key=lambda item: item["episode_id"])
+ if actual_episodes != expected_episodes:
+ raise ValueError("Paimon source differs from the ACT experiment.")
+ row = statistics_row(connection, experiment["statistics_version"])
+ expected_normalization = experiment["normalization"]
+ action_mean = np.asarray(row["action_mean"], dtype=np.float32)
+ action_std = np.asarray(row["action_std"], dtype=np.float32)
+ if (
+ row["source_snapshot_id"] != snapshot_id
+ or row["source_split"] != "train"
+ or row["frame_count"] != expected_normalization["frame_count"]
+ or row["feature_name"] != "action"
+ or row["standard_deviation_floor"] != 1e-2
+ or not np.array_equal(
+ action_mean, normalization["action_mean"])
+ or not np.array_equal(action_std, normalization["action_std"])):
+ raise ValueError(
+ "Paimon normalization differs from the ACT experiment.")
+
+ def factory():
+ return create_paimon_datasets(
+ frames,
+ snapshot_id,
+ experiment["train_episode_id"],
+ experiment["validation_episode_id"],
+ normalization,
+ config,
+ )
+
+ return factory, {
+ "warehouse": str(warehouse),
+ "database": paimon["database"],
+ "frames_table": paimon["frames_table"],
+ "frames_snapshot_id": snapshot_id,
+ }
+
+
+def _tensor_fingerprint(datasets, plan):
+ """Hash the exact planned sample IDs and tensors outside timed
execution."""
+ comparisons = (
+ ("train", datasets[0],
+ sorted(set(plan.measurement_indices + plan.train_indices))),
+ ("validation", datasets[1],
+ sorted(set(plan.validation_indices))),
+ )
+ digest = hashlib.sha256()
+ count = 0
+ for split, dataset, indices in comparisons:
+ for index in indices:
+ sample = dataset[index]
+ identity = {
+ "split": split,
+ "index": index,
+ "sample_id": sample["sample_id"],
+ "episode_id": sample["episode_id"],
+ "frame_index": sample["frame_index"],
+ }
+ digest.update(json.dumps(
+ identity, sort_keys=True, separators=(",", ":")
+ ).encode("utf-8"))
+ for name in ("qpos", "action", "images", "is_pad"):
+ tensor = sample[name].detach().cpu().contiguous()
+ digest.update(name.encode("utf-8"))
+ digest.update(str(tensor.dtype).encode("ascii"))
+ digest.update(str(tuple(tensor.shape)).encode("ascii"))
+ digest.update(tensor.numpy().tobytes())
+ count += 1
+ return {
+ "sha256": digest.hexdigest(),
+ "checked_window_count": count,
+ "fields": [
+ "sample_id", "episode_id", "frame_index", "qpos", "action",
+ "images", "is_pad",
+ ],
+ }
+
+
+def _shared_normalization(
+ episodes,
+ connection,
+ frames_snapshot_id,
+ statistics_version):
+ """Build one train-only normalization contract for both backends.
+
+ HDF5 supplies state moments and float64-accumulated moments over canonical
+ float32 actions from successful train episodes. Versioned Paimon action
+ statistics must match those moments, train scope, frame count, source
+ snapshot, feature name, and ``1e-2`` standard-deviation floor.
+
+ Returns:
+ ``(arrays, metadata)`` where arrays are float32 training values and
+ metadata is JSON-compatible and includes their canonical SHA-256.
+ """
+ normalization, hdf5_metadata = compute_hdf5_normalization(episodes)
+ action_mean = hdf5_metadata["action_mean"]
+ action_std = hdf5_metadata["action_std"]
+ action_count = hdf5_metadata["frame_count"]
+ row = statistics_row(connection, statistics_version)
+ if row["source_snapshot_id"] != frames_snapshot_id:
+ raise ValueError(
+ "Normalization source snapshot %s differs from frames "
+ "snapshot %s."
+ % (row["source_snapshot_id"], frames_snapshot_id))
+ if row["source_split"] != "train" or row["frame_count"] != action_count:
+ raise ValueError(
+ "Versioned action normalization has the wrong train scope.")
+ if row["feature_name"] != "action":
+ raise ValueError("Versioned normalization feature must be action.")
+ if row["standard_deviation_floor"] != 1e-2:
+ raise ValueError(
+ "Versioned normalization must use the 1e-2 std floor.")
+ stored_mean = np.asarray(row["action_mean"], dtype=np.float64)
+ stored_std = np.asarray(row["action_std"], dtype=np.float64)
+ if not (
+ np.allclose(stored_mean, action_mean, rtol=1e-10, atol=1e-10)
+ and np.allclose(stored_std, action_std, rtol=1e-10, atol=1e-10)):
+ raise ValueError(
+ "Versioned Paimon action normalization differs from HDF5 source.")
+ normalization["action_mean"] = stored_mean.astype(np.float32)
+ normalization["action_std"] = stored_std.astype(np.float32)
+ serializable = {
+ name: value.tolist() for name, value in normalization.items()
+ }
+ digest = hashlib.sha256(json.dumps(
+ serializable, sort_keys=True, separators=(",", ":")
+ ).encode("utf-8")).hexdigest()
+ return normalization, {
+ "statistics_version": statistics_version,
+ "source_split": "train",
+ "frame_count": action_count,
+ "standard_deviation_floor": 1e-2,
+ "values": serializable,
+ "sha256": digest,
+ }
+
+
+def _episode_rows(connection):
+ return connection.get_table(agilex.EPISODES_TABLE).scan().select([
+ "episode_id",
+ "source_key",
+ "split",
+ "success",
+ "frame_count",
+ ]).to_list()
+
+
+def _validate_source_identity(episodes, rows):
+ """Match HDF5 discovery to Paimon episodes and return a manifest hash.
+
+ Episode ID, source key, split, and success must match exactly. Paimon's
+ versioned episode rows contribute frame counts used to build complete
+ windows. The returned records retain the local HDF5 paths while the hash
+ covers only portable source metadata.
+ """
+ expected = {
+ item.episode_id: {
+ "episode_id": item.episode_id,
+ "source_key": item.source_key,
+ "split": item.split,
+ "success": item.success,
+ }
+ for item in episodes
+ }
+ actual = {
+ item["episode_id"]: {
+ "episode_id": item["episode_id"],
+ "source_key": item["source_key"],
+ "split": item["split"],
+ "success": item["success"],
+ }
+ for item in rows
+ }
+ if actual != expected or len(actual) != len(rows):
+ raise ValueError(
+ "HDF5 and Paimon source identity differ; rebuild or select "
+ "matching inputs.")
+ rows_by_id = {item["episode_id"]: item for item in rows}
+ enriched = [
+ _BenchmarkEpisode(
+ path=item.path,
+ source_key=item.source_key,
+ episode_id=item.episode_id,
+ split=item.split,
+ success=item.success,
+ frame_count=rows_by_id[item.episode_id]["frame_count"],
+ )
+ for item in episodes
+ ]
+ manifest = sorted([
+ {
+ "episode_id": item.episode_id,
+ "source_key": item.source_key,
+ "split": item.split,
+ "success": item.success,
+ "frame_count": rows_by_id[item.episode_id]["frame_count"],
+ }
+ for item in episodes
+ ], key=lambda item: item["episode_id"])
+ payload = json.dumps(manifest, sort_keys=True, separators=(",", ":"))
+ return enriched, hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _select_episode(source_by_id, split, requested, action_horizon):
+ """Select a successful, split-matching episode long enough for one window.
+
+ An explicit episode is honored when eligible; otherwise the
+ lexicographically first eligible episode ID is selected.
+ """
+ eligible = {
+ episode_id: episode
+ for episode_id, episode in source_by_id.items()
+ if episode.split == split
+ and episode.success
+ and episode.frame_count >= action_horizon
+ }
+ if not eligible:
+ raise ValueError(
+ "No successful %s episode is long enough for horizon %d."
+ % (split, action_horizon))
+ selected = requested or min(eligible)
+ if selected not in eligible:
+ raise ValueError(
+ "Requested %s episode is missing, unsuccessful, or too short: %s."
+ % (split, selected))
+ return eligible[selected]
+
+
+def _summarize(runs):
+ """Return median, minimum, and maximum metrics across backend repeats."""
+ metrics = (
+ "dataset_build_s",
+ "first_batch_s",
+ "batch_fetch_samples_per_s",
+ "fixed_steps_s",
+ "validation_loss",
+ "python_peak_allocated_bytes",
+ "wall_time_s",
+ )
+ result = {"round_count": len(runs)}
+ for name in metrics:
+ values = [item[name] for item in runs]
+ result[name] = {
+ "median": float(np.median(values)),
+ "min": float(np.min(values)),
+ "max": float(np.max(values)),
+ }
+ return result
+
+
+def _sample_sequence_sha256(train_episode_id, validation_episode_id, plan):
+ """Hash episode-qualified sample IDs in measurement/train/validation
order."""
+ value = {
+ "batch_fetch": [
+ "%s#%d" % (train_episode_id, index)
+ for index in plan.measurement_indices
+ ],
+ "train": [
+ "%s#%d" % (train_episode_id, index)
+ for index in plan.train_indices
+ ],
+ "validation": [
+ "%s#%d" % (validation_episode_id, index)
+ for index in plan.validation_indices
+ ],
+ }
+ return hashlib.sha256(json.dumps(
+ value, sort_keys=True, separators=(",", ":")
+ ).encode("utf-8")).hexdigest()
+
+
+def _git_head(repository):
+ try:
+ return subprocess.check_output(
+ ["git", "-C", str(repository), "rev-parse", "HEAD"],
+ stderr=subprocess.DEVNULL,
+ universal_newlines=True,
+ ).strip()
+ except (OSError, subprocess.CalledProcessError):
+ return "UNKNOWN"
+
+
+def _runtime_environment(repository):
+ """Return dependency, CPU, thread, and source identity for comparison."""
+ source_commit = _git_head(repository)
+ package_build = build_info.full_version()
+ if source_commit == "UNKNOWN" and package_build == "UNKNOWN":
+ raise RuntimeError(
+ "ACT benchmark cannot determine its source identity.")
+ return {
+ "python": platform.python_version(),
+ "os": platform.platform(),
+ "machine": platform.machine(),
+ "cpu_identity": _cpu_identity(),
+ "cpu_count": os.cpu_count() or 1,
+ "torch_threads": torch.get_num_threads(),
+ "torch_interop_threads": torch.get_num_interop_threads(),
+ "pypaimon_build": package_build,
+ "numpy": np.__version__,
+ "pyarrow": pa.__version__,
+ "h5py": h5py.__version__,
+ "pillow": PIL.__version__,
+ "torch": torch.__version__,
+ "source_commit": source_commit,
+ }
+
+
+def _cpu_identity():
+ """Return the most specific CPU model available from the local OS."""
+ if platform.system() == "Darwin":
+ try:
+ return subprocess.check_output(
+ ["sysctl", "-n", "machdep.cpu.brand_string"],
+ stderr=subprocess.DEVNULL,
+ universal_newlines=True,
+ ).strip()
+ except (OSError, subprocess.CalledProcessError):
+ pass
+ identity = platform.processor().strip()
+ if identity:
+ return identity
+ if platform.system() == "Linux":
+ try:
+ for line in Path("/proc/cpuinfo").read_text().splitlines():
+ if line.startswith(("model name", "Hardware")):
+ return line.partition(":")[2].strip()
+ except OSError:
+ pass
+ return platform.machine()
+
+
+def _command_argv():
+ """Return the invoked Python basename and command-line arguments."""
+ import sys
+ return [os.path.basename(sys.executable)] + list(sys.argv)
+
+
+def _utc_now():
+ return datetime.now(timezone.utc).isoformat(
+ timespec="seconds").replace("+00:00", "Z")
diff --git a/paimon-python/pypaimon/sample/robomind_agilex.py
b/paimon-python/pypaimon/sample/robomind_agilex.py
index 5a4b83a170..03805914a3 100644
--- a/paimon-python/pypaimon/sample/robomind_agilex.py
+++ b/paimon-python/pypaimon/sample/robomind_agilex.py
@@ -37,7 +37,6 @@ DEFAULT_STATISTICS_VERSION =
"robomind-agilex-joint-position@1"
TABLE_OPTIONS = {
"deletion-vectors.enabled": "true",
"blob-as-descriptor": "false",
- "vector.file.format": "vortex",
}
NUMERIC_FIELDS = (
@@ -59,12 +58,27 @@ NUMERIC_FIELDS = (
("action_joint_velocity_right", "master/joint_velocity_right"),
)
IMAGE_FIELDS = (
- ("rgb_front", "observations/rgb_images/camera_front"),
- ("rgb_left_wrist", "observations/rgb_images/camera_left_wrist"),
- ("rgb_right_wrist", "observations/rgb_images/camera_right_wrist"),
- ("depth_front", "observations/depth_images/camera_front"),
- ("depth_left_wrist", "observations/depth_images/camera_left_wrist"),
- ("depth_right_wrist", "observations/depth_images/camera_right_wrist"),
+ ("observation_images_rgb_front", "observations/rgb_images/camera_front"),
+ (
+ "observation_images_rgb_wrist_left",
+ "observations/rgb_images/camera_left_wrist",
+ ),
+ (
+ "observation_images_rgb_wrist_right",
+ "observations/rgb_images/camera_right_wrist",
+ ),
+ (
+ "observation_images_depth_front",
+ "observations/depth_images/camera_front",
+ ),
+ (
+ "observation_images_depth_wrist_left",
+ "observations/depth_images/camera_left_wrist",
+ ),
+ (
+ "observation_images_depth_wrist_right",
+ "observations/depth_images/camera_right_wrist",
+ ),
)
_ACTION_LEFT = "action_joint_position_left"
diff --git a/paimon-python/pypaimon/tests/act_benchmark_test.py
b/paimon-python/pypaimon/tests/act_benchmark_test.py
new file mode 100644
index 0000000000..9f9fb4357c
--- /dev/null
+++ b/paimon-python/pypaimon/tests/act_benchmark_test.py
@@ -0,0 +1,208 @@
+# 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.
+
+import json
+
+import pytest
+
+from pypaimon.benchmark.act.compare import (
+ canonical_sha256,
+ compare_results,
+ load_result_documents,
+)
+from pypaimon.benchmark.act.experiment import load_experiment
+
+
+def test_packaged_experiment_contains_a_valid_benchmark_config():
+ _require_act_runtime()
+ from pypaimon.benchmark.act.harness import BenchmarkConfig
+
+ experiment = load_experiment()
+
+ assert experiment["schema_version"] == "act-benchmark-experiment@1"
+ assert experiment["benchmark_id"] == "robomind-act"
+ assert BenchmarkConfig(**experiment["config"]).to_dict() == (
+ experiment["config"])
+ assert experiment["statistics_version"] == (
+ "robomind-agilex-joint-position@1")
+
+
+def test_compare_reports_ratio_for_matching_backend_results():
+ experiment = {"schema_version": "act-benchmark-experiment@1"}
+ environment = {"python": "3.10", "machine": "arm64"}
+ hdf5 = _result("hdf5", experiment, environment, throughput=10.0)
+ paimon = _result("paimon", experiment, environment, throughput=15.0)
+ hdf5["summary"]["first_batch_s"] = {
+ "median": 2.0, "min": 2.0, "max": 2.0}
+ paimon["summary"]["first_batch_s"] = {
+ "median": 1.0, "min": 1.0, "max": 1.0}
+
+ comparison = compare_results([hdf5, paimon])
+
+ assert comparison["status"] == "SUCCEEDED"
+ assert len(comparison["experiments"]) == 1
+ group = comparison["experiments"][0]
+ assert group["backends"] == ["hdf5", "paimon"]
+ assert group["metrics"]["batch_fetch_samples_per_s"] == {
+ "hdf5": 10.0,
+ "paimon": 15.0,
+ "paimon_over_hdf5": 1.5,
+ "preferred": "higher",
+ }
+ assert group["metrics"]["first_batch_s"] == {
+ "hdf5": 2.0,
+ "paimon": 1.0,
+ "hdf5_over_paimon": 2.0,
+ "preferred": "lower",
+ }
+
+
+def test_compare_rejects_different_tensor_fingerprints():
+ experiment = {"schema_version": "act-benchmark-experiment@1"}
+ environment = {"python": "3.10", "machine": "arm64"}
+ hdf5 = _result("hdf5", experiment, environment, throughput=10.0)
+ paimon = _result("paimon", experiment, environment, throughput=15.0)
+ paimon["tensor_fingerprint"]["sha256"] = "different"
+
+ comparison = compare_results([hdf5, paimon])
+
+ assert comparison["status"] == "FAILED"
+ group = comparison["experiments"][0]
+ assert group["status"] == "FAILED"
+ assert group["reason"] == "tensor fingerprints differ"
+ assert group["metrics"] == {}
+
+
+def test_compare_requires_results_from_both_backends():
+ experiment = {"schema_version": "act-benchmark-experiment@1"}
+ environment = {"python": "3.10", "machine": "arm64"}
+
+ comparison = compare_results([
+ _result("hdf5", experiment, environment, throughput=10.0),
+ ])
+
+ assert comparison["status"] == "INCOMPATIBLE"
+ group = comparison["experiments"][0]
+ assert group["status"] == "INCOMPATIBLE"
+ assert group["reason"] == "both hdf5 and paimon results are required"
+ assert group["metrics"] == {}
+
+
+def test_load_results_combines_explicit_files_and_directory(tmp_path):
+ experiment = {"schema_version": "act-benchmark-experiment@1"}
+ environment = {"python": "3.10", "machine": "arm64"}
+ hdf5_path = tmp_path / "hdf5.json"
+ paimon_path = tmp_path / "paimon.json"
+ ignored_path = tmp_path / "experiment.json"
+ hdf5_path.write_text(json.dumps(
+ _result("hdf5", experiment, environment, throughput=10.0)))
+ paimon_path.write_text(json.dumps(
+ _result("paimon", experiment, environment, throughput=15.0)))
+ ignored_path.write_text(json.dumps(experiment))
+
+ results = load_result_documents(
+ [hdf5_path], results_dir=tmp_path)
+
+ assert [result["backend"] for result in results] == ["hdf5", "paimon"]
+
+
+def test_compare_groups_multiple_experiments_without_cross_comparing():
+ environment = {"python": "3.10", "machine": "arm64"}
+ results = []
+ for seed in (1, 2):
+ experiment = {
+ "schema_version": "act-benchmark-experiment@1",
+ "config": {"seed": seed},
+ }
+ results.extend([
+ _result("hdf5", experiment, environment, throughput=10.0),
+ _result("paimon", experiment, environment, throughput=15.0),
+ ])
+
+ comparison = compare_results(results)
+
+ assert comparison["status"] == "SUCCEEDED"
+ assert len(comparison["experiments"]) == 2
+ assert all(
+ group["backends"] == ["hdf5", "paimon"]
+ for group in comparison["experiments"]
+ )
+
+
+def test_compare_reports_incompatible_runtime_environments():
+ experiment = {"schema_version": "act-benchmark-experiment@1"}
+ hdf5 = _result(
+ "hdf5", experiment,
+ {"python": "3.10", "machine": "arm64"}, throughput=10.0)
+ paimon = _result(
+ "paimon", experiment,
+ {"python": "3.11", "machine": "arm64"}, throughput=15.0)
+
+ comparison = compare_results([hdf5, paimon])
+
+ assert comparison["status"] == "INCOMPATIBLE"
+ group = comparison["experiments"][0]
+ assert group["reason"] == "runtime environments differ"
+ assert len(group["environment_sha256s"]) == 2
+ assert group["metrics"] == {}
+
+
+def test_compare_rejects_tampered_result_experiment_hash():
+ experiment = {"schema_version": "act-benchmark-experiment@1"}
+ result = _result(
+ "hdf5",
+ experiment,
+ {"python": "3.10", "machine": "arm64"},
+ throughput=10.0,
+ )
+ result["experiment_sha256"] = "tampered"
+
+ with pytest.raises(ValueError, match="experiment SHA-256 differs"):
+ compare_results([result])
+
+
+def _require_act_runtime():
+ pytest.importorskip("torch")
+ pytest.importorskip("PIL.Image")
+
+
+def _result(backend, experiment, environment, throughput):
+ return {
+ "schema_version": "act-benchmark-result@1",
+ "status": "SUCCEEDED",
+ "backend": backend,
+ "experiment": experiment,
+ "experiment_sha256": canonical_sha256(experiment),
+ "environment": environment,
+ "model": {"implementation": "test-policy", "parameter_count": 1},
+ "tensor_fingerprint": {
+ "sha256": "same-tensors",
+ "checked_window_count": 2,
+ },
+ "runs": [{
+ "round": round_number,
+ "train_loss": [1.0, 0.5],
+ "validation_loss": 0.25,
+ } for round_number in range(1, 4)],
+ "summary": {
+ "round_count": 3,
+ "batch_fetch_samples_per_s": {
+ "median": throughput,
+ "min": throughput,
+ "max": throughput,
+ },
+ },
+ }
diff --git a/paimon-python/pypaimon/tests/act_runner_test.py
b/paimon-python/pypaimon/tests/act_runner_test.py
new file mode 100644
index 0000000000..57d64bf8c1
--- /dev/null
+++ b/paimon-python/pypaimon/tests/act_runner_test.py
@@ -0,0 +1,709 @@
+# 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.
+
+# ruff: noqa: E402
+
+import json
+import tracemalloc
+from io import BytesIO
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import numpy as np
+import pytest
+
+
+torch = pytest.importorskip("torch")
+Image = pytest.importorskip("PIL.Image")
+h5py = pytest.importorskip("h5py")
+
+import pypaimon.multimodal as pmm
+import pypaimon.multimodal.window_dataset as window_dataset
+import pypaimon.benchmark.act.harness as act_harness
+import pypaimon.benchmark.act.__main__ as act_cli
+import pypaimon.benchmark.act.runner as act_runner
+from pypaimon.benchmark.act.runner import (
+ BenchmarkConfig,
+ prepare_experiment,
+ run_experiment,
+)
+from pypaimon.benchmark.act.experiment import load_experiment
+from pypaimon.benchmark.act.compare import canonical_sha256, compare_results
+from pypaimon.benchmark.act.harness import build_window_plan, run_backend
+from pypaimon.benchmark.act.hdf5 import Hdf5ACTWindowDataset
+from pypaimon.benchmark.act.paimon import (
+ ACTION_COLUMNS,
+ IMAGE_COLUMNS,
+ QPOS_COLUMNS,
+ create_datasets as create_paimon_datasets,
+ latest_snapshot_id,
+)
+from pypaimon.multimodal.window_dataset import ContiguousWindowDataset
+from pypaimon.sample import robomind_agilex as agilex
+
+
+def test_logical_batches_coalesce_one_physical_fetch():
+ class BatchDataset:
+ def __init__(self):
+ self.calls = []
+
+ def __getitems__(self, indices):
+ self.calls.append(list(indices))
+ return [{"value": torch.tensor(index)} for index in indices]
+
+ dataset = BatchDataset()
+
+ batches = list(act_harness._iter_logical_batches(
+ dataset,
+ tuple(range(8)),
+ logical_batch_size=2,
+ fetch_batches=4,
+ ))
+
+ assert dataset.calls == [list(range(8))]
+ assert [batch["value"].tolist() for batch in batches] == [
+ [0, 1], [2, 3], [4, 5], [6, 7],
+ ]
+
+
+def test_logical_batches_reject_incomplete_batch_tail():
+ class BatchDataset:
+ def __getitems__(self, indices):
+ return [{"value": torch.tensor(index)} for index in indices]
+
+ with pytest.raises(ValueError, match="complete logical batches"):
+ list(act_harness._iter_logical_batches(
+ BatchDataset(),
+ tuple(range(9)),
+ logical_batch_size=2,
+ fetch_batches=4,
+ ))
+
+
+def test_backend_times_without_tracemalloc_and_measures_memory_separately():
+ states = []
+ config = BenchmarkConfig(
+ seed=11,
+ action_horizon=1,
+ batch_size=1,
+ optimizer_steps=1,
+ image_height=2,
+ image_width=2,
+ warmup_batches=1,
+ timed_batches=1,
+ rounds=3,
+ )
+
+ class TracingDataset(torch.utils.data.Dataset):
+ def __len__(self):
+ return 2
+
+ def __getitem__(self, index):
+ states.append(tracemalloc.is_tracing())
+ return {
+ "sample_id": "episode-a#%d" % index,
+ "episode_id": "episode-a",
+ "frame_index": index,
+ "qpos": torch.zeros(14),
+ "action": torch.zeros((1, 14)),
+ "images": torch.zeros((3, 3, 2, 2)),
+ "is_pad": torch.zeros(1, dtype=torch.bool),
+ }
+
+ dataset = TracingDataset()
+ plan = build_window_plan(len(dataset), len(dataset), config)
+ result = run_backend(
+ "test",
+ 1,
+ lambda: (dataset, dataset),
+ plan,
+ config,
+ "sequence-sha256",
+ policy_factory=_policy_factory,
+ )
+
+ assert states[0] is False
+ assert states[-1] is True
+ assert result["peak_memory_measurement"] == (
+ "python-tracemalloc-separate-dataset-first-batch")
+
+
+def test_backend_coalesces_timed_batch_fetches():
+ config = BenchmarkConfig(
+ seed=11,
+ action_horizon=1,
+ batch_size=2,
+ optimizer_steps=1,
+ image_height=2,
+ image_width=2,
+ warmup_batches=1,
+ timed_batches=4,
+ fetch_batches=4,
+ rounds=3,
+ )
+
+ clock = SimpleNamespace(value=0.0)
+
+ class BatchDataset(torch.utils.data.Dataset):
+ def __init__(self):
+ self.calls = []
+
+ def __len__(self):
+ return 16
+
+ def __getitem__(self, index):
+ return {
+ "sample_id": "episode-a#%d" % index,
+ "episode_id": "episode-a",
+ "frame_index": index,
+ "qpos": torch.zeros(14),
+ "action": torch.zeros((1, 14)),
+ "images": torch.zeros((3, 3, 2, 2)),
+ "is_pad": torch.zeros(1, dtype=torch.bool),
+ }
+
+ def __getitems__(self, indices):
+ clock.value += 0.25
+ self.calls.append(list(indices))
+ return [self[index] for index in indices]
+
+ dataset = BatchDataset()
+ plan = build_window_plan(len(dataset), len(dataset), config)
+
+ def validate(_batch, _config):
+ clock.value += 1.0
+
+ with (
+ patch.object(act_harness, "_measure_python_peak", return_value=0),
+ patch.object(
+ act_harness.time, "monotonic", side_effect=lambda: clock.value),
+ patch.object(act_harness, "validate_act_batch", side_effect=validate),
+ ):
+ result = run_backend(
+ "test",
+ 1,
+ lambda: (dataset, dataset),
+ plan,
+ config,
+ "sequence-sha256",
+ policy_factory=_policy_factory,
+ )
+
+ assert dataset.calls == [
+ list(plan.measurement_indices[:2]),
+ list(plan.measurement_indices[2:10]),
+ list(plan.train_indices),
+ list(plan.validation_indices),
+ ]
+ assert result["batch_fetch_s"] == 0.25
+
+
+def _jpeg(value):
+ buffer = BytesIO()
+ Image.fromarray(np.full((8, 10, 3), value, dtype=np.uint8)).save(
+ buffer, format="JPEG")
+ return np.frombuffer(buffer.getvalue(), dtype=np.uint8)
+
+
+def _write_episode(root, split, name, offset, frames=6):
+ path = (root / "13_packbowl" / "success_episodes" / split / name
+ / "data" / "trajectory.hdf5")
+ path.parent.mkdir(parents=True)
+ with h5py.File(path, "w") as h5:
+ h5.create_dataset("language_raw", data=[b"pack the bowl"])
+ h5.create_dataset(
+ "language_distilbert",
+ data=np.zeros((1, 1, 768), dtype=np.float16),
+ )
+ for index, (_, hdf5_path) in enumerate(agilex.NUMERIC_FIELDS):
+ values = np.arange(frames * 7, dtype=np.float64).reshape(frames, 7)
+ h5.create_dataset(hdf5_path, data=values + offset + index * 100)
+ variable = h5py.vlen_dtype(np.dtype("uint8"))
+ for image_index, (_, hdf5_path) in enumerate(agilex.IMAGE_FIELDS):
+ dataset = h5.create_dataset(hdf5_path, (frames,), dtype=variable)
+ for frame_index in range(frames):
+ dataset[frame_index] = _jpeg(
+ offset + image_index + frame_index)
+ return path
+
+
[email protected]
+def benchmark_input(tmp_path, monkeypatch):
+ root = tmp_path / "input"
+ _write_episode(root, "train", "train-a", 1)
+ _write_episode(root, "train", "train-b", 11.1)
+ _write_episode(root, "val", "val-a", 21)
+ warehouse = tmp_path / "warehouse"
+ monkeypatch.setattr(agilex, "TABLE_OPTIONS", {
+ **agilex.TABLE_OPTIONS,
+ "vector.file.format": "parquet",
+ })
+ agilex.ingest_local(root, warehouse, batch_size=2)
+ agilex.backfill_canonical_action(
+ warehouse, statistics_version="act-test@1")
+ return root, warehouse
+
+
+class _Policy(torch.nn.Module):
+
+ def __init__(self):
+ super().__init__()
+ self.scale = torch.nn.Parameter(torch.tensor(0.0))
+
+ def forward(self, batch):
+ assert self.training
+ target = batch["action"].mean() + batch["observation.state"].mean()
+ loss = (self.scale - target).square()
+ return loss, {
+ "l1_loss": loss.detach(),
+ "kld_loss": torch.tensor(0.0),
+ }
+
+
+def _policy_factory(config):
+ return _Policy(), {
+ "implementation": "test-policy",
+ "chunk_size": config.action_horizon,
+ "parameter_count": 1,
+ }
+
+
+def test_prepare_writes_resolved_experiment(benchmark_input, tmp_path):
+ input_root, warehouse = benchmark_input
+ definition = load_experiment()
+ definition["statistics_version"] = "act-test@1"
+ definition["config"].update({
+ "seed": 17,
+ "action_horizon": 3,
+ "batch_size": 2,
+ "optimizer_steps": 2,
+ "image_height": 8,
+ "image_width": 10,
+ "timed_batches": 2,
+ })
+ output = tmp_path / "experiment.json"
+
+ experiment = prepare_experiment(
+ input_root, warehouse, output, definition=definition)
+
+ assert json.loads(output.read_text()) == experiment
+ assert experiment["schema_version"] == "act-benchmark-experiment@1"
+ assert experiment["train_episode_id"] == "train-a"
+ assert experiment["validation_episode_id"] == "val-a"
+ assert experiment["source"]["episodes"][0]["frame_count"] == 6
+ assert len(experiment["source"]["sha256"]) == 64
+ assert len(experiment["normalization"]["sha256"]) == 64
+ assert len(experiment["window_plan"]["sha256"]) == 64
+ assert experiment["paimon"]["frames_snapshot_id"] > 0
+
+
+def test_independent_backend_results_preserve_tensor_and_loss_parity(
+ benchmark_input, tmp_path):
+ input_root, warehouse = benchmark_input
+ definition = load_experiment()
+ definition["statistics_version"] = "act-test@1"
+ definition["config"].update({
+ "seed": 17,
+ "action_horizon": 3,
+ "batch_size": 2,
+ "optimizer_steps": 2,
+ "image_height": 8,
+ "image_width": 10,
+ "timed_batches": 2,
+ })
+ experiment_path = tmp_path / "experiment.json"
+ prepare_experiment(
+ input_root, warehouse, experiment_path, definition=definition)
+
+ hdf5_path = tmp_path / "hdf5-result.json"
+ events = []
+ real_run_backend = act_runner.run_backend
+ real_tensor_fingerprint = act_runner._tensor_fingerprint
+
+ def record_run(*args, **kwargs):
+ events.append("round")
+ return real_run_backend(*args, **kwargs)
+
+ def record_fingerprint(*args, **kwargs):
+ events.append("fingerprint")
+ return real_tensor_fingerprint(*args, **kwargs)
+
+ with (
+ patch.object(act_runner, "run_backend", side_effect=record_run),
+ patch.object(
+ act_runner,
+ "_tensor_fingerprint",
+ side_effect=record_fingerprint,
+ ),
+ ):
+ hdf5_result = run_experiment(
+ "hdf5",
+ experiment_path,
+ hdf5_path,
+ input_root=input_root,
+ policy_factory=_policy_factory,
+ )
+ assert events == ["round"] * definition["config"]["rounds"] + [
+ "fingerprint"]
+ paimon_path = tmp_path / "paimon-result.json"
+ paimon_result = run_experiment(
+ "paimon",
+ experiment_path,
+ paimon_path,
+ warehouse=warehouse,
+ policy_factory=_policy_factory,
+ )
+
+ assert json.loads(hdf5_path.read_text()) == hdf5_result
+ assert json.loads(paimon_path.read_text()) == paimon_result
+ assert hdf5_result["schema_version"] == "act-benchmark-result@1"
+ assert paimon_result["schema_version"] == "act-benchmark-result@1"
+ assert hdf5_result["experiment"] == paimon_result["experiment"]
+ assert hdf5_result["tensor_fingerprint"] == (
+ paimon_result["tensor_fingerprint"])
+ assert [run["train_loss"] for run in hdf5_result["runs"]] == [
+ run["train_loss"] for run in paimon_result["runs"]
+ ]
+ assert [run["validation_loss"] for run in hdf5_result["runs"]] == [
+ run["validation_loss"] for run in paimon_result["runs"]
+ ]
+ comparison = compare_results([hdf5_result, paimon_result])
+ assert comparison["status"] == "SUCCEEDED"
+ assert comparison["experiments"][0]["backends"] == ["hdf5", "paimon"]
+
+
+def test_backends_match_the_golden_act_window_contract(benchmark_input):
+ input_root, warehouse = benchmark_input
+ normalization = {
+ "qpos_mean": np.zeros(14, dtype=np.float32),
+ "qpos_std": np.ones(14, dtype=np.float32),
+ "action_mean": np.zeros(14, dtype=np.float32),
+ "action_std": np.ones(14, dtype=np.float32),
+ }
+ hdf5 = Hdf5ACTWindowDataset(
+ SimpleNamespace(
+ path=(input_root / "13_packbowl" / "success_episodes" / "train"
+ / "train-a" / "data" / "trajectory.hdf5"),
+ episode_id="train-a",
+ frame_count=6,
+ ),
+ normalization,
+ action_horizon=3,
+ )
+ connection = pmm.connect(
+ database=agilex.DEFAULT_DATABASE,
+ options={"warehouse": str(warehouse)},
+ )
+ frames = connection.get_table(agilex.FRAMES_TABLE)
+ paimon, _ = create_paimon_datasets(
+ frames,
+ latest_snapshot_id(frames),
+ "train-a",
+ "val-a",
+ normalization,
+ BenchmarkConfig(action_horizon=3),
+ )
+
+ expected = hdf5[1]
+ actual = paimon[1]
+
+ assert set(expected) == {
+ "sample_id", "episode_id", "frame_index", "qpos", "action",
+ "images", "is_pad",
+ }
+ assert expected["sample_id"] == "train-a#1"
+ assert expected["episode_id"] == "train-a"
+ assert expected["frame_index"] == 1
+ assert torch.equal(expected["qpos"], torch.tensor(
+ list(range(408, 415)) + list(range(508, 515)),
+ dtype=torch.float32,
+ ))
+ assert torch.equal(expected["action"], torch.tensor([
+ list(range(1208, 1215)) + list(range(1308, 1315)),
+ list(range(1215, 1222)) + list(range(1315, 1322)),
+ list(range(1222, 1229)) + list(range(1322, 1329)),
+ ], dtype=torch.float32))
+ assert torch.allclose(
+ expected["images"][:, :, 0, 0],
+ torch.tensor([[2 / 255] * 3, [3 / 255] * 3, [4 / 255] * 3]),
+ )
+ assert not expected["is_pad"].any()
+ for name in ("qpos", "action", "images", "is_pad"):
+ assert torch.equal(expected[name], actual[name])
+ for name in ("sample_id", "episode_id", "frame_index"):
+ assert expected[name] == actual[name]
+
+
+def test_hdf5_window_index_bounds(tmp_path):
+ path = _write_episode(tmp_path, "train", "train-a", 1)
+ dataset = Hdf5ACTWindowDataset(
+ SimpleNamespace(
+ path=path,
+ episode_id="train-a",
+ frame_count=6,
+ ),
+ {
+ "qpos_mean": np.zeros(14, dtype=np.float32),
+ "qpos_std": np.ones(14, dtype=np.float32),
+ "action_mean": np.zeros(14, dtype=np.float32),
+ "action_std": np.ones(14, dtype=np.float32),
+ },
+ action_horizon=3,
+ )
+
+ assert dataset[-1]["sample_id"] == "train-a#3"
+ with pytest.raises(IndexError):
+ dataset[-len(dataset) - 1]
+ with pytest.raises(IndexError):
+ dataset[len(dataset)]
+
+
+def test_paimon_run_rejects_normalization_not_recorded_in_statistics(
+ benchmark_input, tmp_path):
+ input_root, warehouse = benchmark_input
+ definition = load_experiment()
+ definition["statistics_version"] = "act-test@1"
+ definition["config"].update({
+ "action_horizon": 3,
+ "batch_size": 1,
+ "optimizer_steps": 1,
+ "image_height": 8,
+ "image_width": 10,
+ })
+ experiment_path = tmp_path / "experiment.json"
+ experiment = prepare_experiment(
+ input_root, warehouse, experiment_path, definition=definition)
+ experiment["normalization"]["values"]["action_mean"][0] += 1
+ experiment["normalization"]["sha256"] = canonical_sha256(
+ experiment["normalization"]["values"])
+ experiment_path.write_text(json.dumps(experiment))
+
+ with pytest.raises(ValueError, match="normalization differs"):
+ run_experiment(
+ "paimon",
+ experiment_path,
+ tmp_path / "must-not-exist.json",
+ warehouse=warehouse,
+ policy_factory=_policy_factory,
+ )
+
+
+def test_run_rejects_tampered_source_manifest(benchmark_input, tmp_path):
+ input_root, warehouse = benchmark_input
+ definition = load_experiment()
+ definition["statistics_version"] = "act-test@1"
+ definition["config"].update({
+ "action_horizon": 3,
+ "batch_size": 1,
+ "optimizer_steps": 1,
+ "image_height": 8,
+ "image_width": 10,
+ })
+ experiment_path = tmp_path / "experiment.json"
+ experiment = prepare_experiment(
+ input_root, warehouse, experiment_path, definition=definition)
+ experiment["source"]["episodes"][0]["frame_count"] += 1
+ experiment_path.write_text(json.dumps(experiment))
+
+ with pytest.raises(ValueError, match="source-manifest hash differs"):
+ run_experiment(
+ "hdf5",
+ experiment_path,
+ tmp_path / "must-not-exist.json",
+ input_root=input_root,
+ policy_factory=_policy_factory,
+ )
+
+
+def test_run_rejects_config_not_used_to_build_window_plan(
+ benchmark_input, tmp_path):
+ input_root, warehouse = benchmark_input
+ definition = load_experiment()
+ definition["statistics_version"] = "act-test@1"
+ definition["config"].update({
+ "action_horizon": 3,
+ "batch_size": 1,
+ "optimizer_steps": 1,
+ "image_height": 8,
+ "image_width": 10,
+ })
+ experiment_path = tmp_path / "experiment.json"
+ experiment = prepare_experiment(
+ input_root, warehouse, experiment_path, definition=definition)
+ experiment["config"]["seed"] += 1
+ experiment_path.write_text(json.dumps(experiment))
+
+ with pytest.raises(ValueError, match="window plan was not built"):
+ run_experiment(
+ "hdf5",
+ experiment_path,
+ tmp_path / "must-not-exist.json",
+ input_root=input_root,
+ policy_factory=_policy_factory,
+ )
+
+
+def test_paimon_windows_are_lazy_snapshot_pinned_and_vortex_independent(
+ benchmark_input, tmp_path):
+ input_root, warehouse = benchmark_input
+ definition = load_experiment()
+ definition["statistics_version"] = "act-test@1"
+ definition["config"]["action_horizon"] = 3
+ experiment = prepare_experiment(
+ input_root,
+ warehouse,
+ tmp_path / "experiment.json",
+ definition=definition,
+ )
+ normalization = {
+ name: np.asarray(value, dtype=np.float32)
+ for name, value in experiment["normalization"]["values"].items()
+ }
+ connection = pmm.connect(
+ database=agilex.DEFAULT_DATABASE,
+ options={"warehouse": str(warehouse)},
+ )
+ frames = connection.get_table(agilex.FRAMES_TABLE)
+ assert frames.raw_table.table_schema.options["vector.file.format"] == (
+ "parquet")
+ snapshot_id = latest_snapshot_id(frames)
+
+ with patch(
+ "pypaimon.multimodal.window_dataset.fetch_blob_bodies",
+ side_effect=window_dataset.fetch_blob_bodies) as fetch:
+ train, validation = create_paimon_datasets(
+ frames,
+ snapshot_id,
+ "train-a",
+ "val-a",
+ normalization,
+ BenchmarkConfig(
+ action_horizon=3,
+ batch_size=1,
+ optimizer_steps=1,
+ image_height=8,
+ image_width=10,
+ rounds=3,
+ ),
+ )
+ assert fetch.call_count == 0
+ assert isinstance(train, ContiguousWindowDataset)
+ assert isinstance(validation, ContiguousWindowDataset)
+ with patch.object(
+ train, "_read_rows", wraps=train._read_rows) as read_rows:
+ sample_before_append = train[0]
+ assert [call.args[1] for call in read_rows.call_args_list] == [
+ list(ACTION_COLUMNS),
+ list(QPOS_COLUMNS + IMAGE_COLUMNS),
+ ]
+ assert [len(call.args[0]) for call in read_rows.call_args_list] == [3,
1]
+ assert fetch.call_count == 1
+ assert {
+ name: len(fetch.call_args.args[1][name])
+ for name in IMAGE_COLUMNS
+ } == {name: 1 for name in IMAGE_COLUMNS}
+
+ scalar, blobs = frames.scan().where(
+ "episode_id = 'train-a' AND frame_index = 5"
+ ).read_blobs(IMAGE_COLUMNS)
+ appended = scalar.to_pylist()[0]
+ appended["frame_index"] = 6
+ for name in IMAGE_COLUMNS:
+ appended[name] = blobs[name][0]
+ frames.add([appended])
+
+ assert train.snapshot_id == snapshot_id
+ assert validation.snapshot_id == snapshot_id
+ assert latest_snapshot_id(frames) != snapshot_id
+ assert len(train) == 4
+ sample_after_append = train[0]
+ for name in ("qpos", "action", "images", "is_pad"):
+ assert torch.equal(
+ sample_before_append[name], sample_after_append[name])
+
+
+def test_requires_at_least_three_measurement_rounds():
+ with pytest.raises(ValueError, match="rounds must be at least 3"):
+ BenchmarkConfig(rounds=2)
+
+
+def test_fetch_batches_must_be_positive():
+ assert BenchmarkConfig(fetch_batches=4).fetch_batches == 4
+ with pytest.raises(ValueError, match="fetch_batches must be a positive
int"):
+ BenchmarkConfig(fetch_batches=0)
+
+
+def test_cli_exposes_prepare_run_and_compare_contracts(capsys):
+ with pytest.raises(SystemExit):
+ act_cli.main(["prepare", "--help"])
+
+ prepare_help = capsys.readouterr().out
+ assert "--experiment" in prepare_help
+ assert "--fetch-batches" in prepare_help
+
+ with pytest.raises(SystemExit):
+ act_cli.main(["run", "--help"])
+
+ run_help = capsys.readouterr().out
+ assert "--backend" in run_help
+ assert "--experiment" in run_help
+ assert "--results-dir" in run_help
+
+ with pytest.raises(SystemExit):
+ act_cli.main(["compare", "--help"])
+
+ assert "--results-dir" in capsys.readouterr().out
+
+
+def test_cli_requires_python_3_10_or_newer():
+ with pytest.raises(RuntimeError, match="Python 3.10 or newer"):
+ act_cli._require_supported_python((3, 9))
+
+ act_cli._require_supported_python((3, 10))
+
+
+def test_automatic_artifact_paths_do_not_overwrite_same_second(tmp_path):
+ with patch.object(act_cli, "datetime") as now:
+ now.now.return_value.strftime.return_value = "20260901T120000Z"
+
+ first = act_cli._artifact_path(tmp_path, "robomind-act-hdf5")
+ second = act_cli._artifact_path(tmp_path, "robomind-act-hdf5")
+
+ assert first != second
+ assert first.parent == tmp_path
+ assert second.parent == tmp_path
+
+
+def test_runtime_environment_uses_package_identity_outside_git_checkout(
+ tmp_path):
+ with patch.object(act_runner, "_git_head", return_value="UNKNOWN"):
+ environment = act_runner._runtime_environment(tmp_path)
+
+ assert environment["source_commit"] == "UNKNOWN"
+ assert environment["pypaimon_build"] != "UNKNOWN"
+ assert environment["cpu_identity"]
+ assert environment["cpu_count"] > 0
+ assert environment["torch_threads"] > 0
+ assert environment["torch_interop_threads"] > 0
+ assert all(environment[name] for name in (
+ "numpy", "pyarrow", "h5py", "pillow"))
+
+ with (
+ patch.object(act_runner, "_git_head", return_value="UNKNOWN"),
+ patch.object(act_runner.build_info, "full_version",
return_value="UNKNOWN"),
+ pytest.raises(RuntimeError, match="source identity"),
+ ):
+ act_runner._runtime_environment(tmp_path)
diff --git a/paimon-python/pypaimon/tests/robomind_agilex_pipeline_test.py
b/paimon-python/pypaimon/tests/robomind_agilex_pipeline_test.py
index f77e146515..4f677e42ac 100644
--- a/paimon-python/pypaimon/tests/robomind_agilex_pipeline_test.py
+++ b/paimon-python/pypaimon/tests/robomind_agilex_pipeline_test.py
@@ -14,7 +14,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import importlib.util
import inspect
import json
import subprocess
@@ -31,12 +30,6 @@ from pypaimon.sample import robomind_agilex as agilex
h5py = pytest.importorskip("h5py")
-requires_vortex = pytest.mark.skipif(
- importlib.util.find_spec("vortex") is None,
- reason="RoboMIND ingestion uses Vortex, which requires Python >= 3.11",
-)
-
-
_NUMERIC_PATHS = [path for _, path in agilex.NUMERIC_FIELDS]
_IMAGE_PATHS = [path for _, path in agilex.IMAGE_FIELDS]
@@ -87,7 +80,6 @@ def customer_agilex_input(request):
return value
-@requires_vortex
def test_explicit_customer_input_uses_downloaded_episodes(
customer_agilex_input, tmp_path):
episodes = agilex.discover_episodes(customer_agilex_input)
@@ -149,8 +141,10 @@ def
test_shared_transform_streams_complete_agilex_business_schema(
frames = pa.Table.from_batches(batches)
assert frames["episode_id"].to_pylist() == ["train-a"] * 3
assert frames["frame_index"].to_pylist() == [0, 1, 2]
- assert frames["rgb_front"][0].as_py() == b"train-a:0:0"
- assert frames["depth_right_wrist"][2].as_py() == b"train-a:5:2"
+ assert frames["observation_images_rgb_front"][0].as_py() == (
+ b"train-a:0:0")
+ assert frames["observation_images_depth_wrist_right"][2].as_py() == (
+ b"train-a:5:2")
assert frames["action_joint_position_left"][1].as_py() == [
float(value) for value in range(1207, 1214)
]
@@ -253,7 +247,6 @@ def
test_transform_rejects_empty_episode_but_accepts_one_frame(tmp_path):
assert [batch.num_rows for batch in batches] == [1]
-@requires_vortex
def test_local_ingest_and_backfill_materialize_only_canonical_action(
agilex_input, tmp_path):
root, paths = agilex_input
@@ -302,7 +295,7 @@ def
test_local_ingest_and_backfill_materialize_only_canonical_action(
for table in (episodes, frames, stats):
options = table.raw_table.table_schema.options
assert options["deletion-vectors.enabled"] == "true"
- assert options["vector.file.format"] == "vortex"
+ assert options["vector.file.format"] == "parquet"
assert options["blob-as-descriptor"] == "false"
assert "file.format" not in agilex.TABLE_OPTIONS
assert "action" in frames.raw_table.field_names
@@ -352,7 +345,6 @@ def
test_local_ingest_and_backfill_materialize_only_canonical_action(
assert refreshed_snapshot > backfill["statistics_snapshot_id"]
-@requires_vortex
def test_canonical_action_backfill_resumes_after_schema_change(
agilex_input, tmp_path, monkeypatch):
root, _ = agilex_input
@@ -387,7 +379,6 @@ def
test_canonical_action_backfill_resumes_after_schema_change(
np.testing.assert_array_equal(row["action"], expected)
-@requires_vortex
def test_ray_ingest_matches_local_schema_rows_and_backfill(
agilex_input, tmp_path):
ray = pytest.importorskip("ray")
diff --git a/paimon-python/setup.py b/paimon-python/setup.py
index 12e24632bc..e2cfcc33a8 100644
--- a/paimon-python/setup.py
+++ b/paimon-python/setup.py
@@ -216,6 +216,15 @@ def read_requirements():
install_requires = read_requirements()
+LEROBOT_DEPENDENCIES = [
+ # datasets 4.1+ may select PyArrow 21+, while PyPaimon currently
+ # supports PyArrow <20. Pandas 2.2.2+ supports NumPy 2.x selected
+ # by LeRobot's media dependencies.
+ 'datasets>=4,<4.1; python_version>="3.10"',
+ 'pandas>=2.2.2,<3; python_version>="3.10"',
+ 'lerobot>=0.4.4,<0.5; python_version>="3.10"',
+]
+
long_description = "See Apache Paimon Python API \
[Doc](https://paimon.apache.org/docs/master/pypaimon/python-api/) for usage."
@@ -224,7 +233,12 @@ setup(
version=VERSION,
packages=PACKAGES,
include_package_data=True,
- package_data={"pypaimon": ["_full_version"]},
+ package_data={
+ "pypaimon": [
+ "_full_version",
+ "benchmark/act/default_experiment.json",
+ ],
+ },
cmdclass={"build_py": PaimonBuildPy, "sdist": PaimonSdist},
install_requires=install_requires,
entry_points={
@@ -241,20 +255,16 @@ setup(
# rosbags is pure Python and does not require a ROS installation.
'rosbags>=0.11.5,<0.12; python_version>="3.10"',
],
- 'lerobot': [
- # datasets 4.1+ may select PyArrow 21+, while PyPaimon currently
- # supports PyArrow <20. Pandas 2.2.2+ supports NumPy 2.x selected
- # by LeRobot's media dependencies.
- 'datasets>=4,<4.1; python_version>="3.10"',
- 'pandas>=2.2.2,<3; python_version>="3.10"',
- 'lerobot>=0.4.4,<0.5; python_version>="3.10"',
- ],
+ 'lerobot': LEROBOT_DEPENDENCIES,
'ray': [
'ray>=2.10,<3; python_version>="3.8"',
],
'torch': [
'torch',
],
+ 'act': LEROBOT_DEPENDENCIES + [
+ 'Pillow; python_version>="3.10"',
+ ],
'daft': [
'daft>=0.7.6; python_version>="3.10"',
],
