From 4ae25a8fbb4123da26e3b726bbd7c01be03bb974 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 9 Oct 2026 20:59:43 -0700 Subject: [PATCH 1/9] [python] Configure LeRobot video decoder cache capacity --- docs/docs/pypaimon/lerobot.md | 5 ++ .../pypaimon/multimodal/lerobot/dataset.py | 13 +++++ paimon-python/pypaimon/multimodal/video.py | 2 +- .../pypaimon/tests/multimodal_lerobot_test.py | 55 +++++++++++++++++++ 4 files changed, 74 insertions(+), 1 deletion(-) diff --git a/docs/docs/pypaimon/lerobot.md b/docs/docs/pypaimon/lerobot.md index 518a5d067400..dbe05e658211 100644 --- a/docs/docs/pypaimon/lerobot.md +++ b/docs/docs/pypaimon/lerobot.md @@ -232,6 +232,11 @@ on `index`; payloads remain lazy. Indexed videos prefetch metadata and target GO uncached PyAV reads on demand. Unindexed videos use the TorchCodec/PyAV scan path. Set `video_backend` to force either decoder. +Set `max_open_videos` on `PaimonLeRobotDataset` (or a custom +`PaimonDatasetReader`) to bound the decoder cache per camera per worker. +The default is 16; increasing it reduces decoder reopening at the cost of +more memory and open streams. + Subclass `PaimonDatasetReader` for a custom logical frame layout: ```python diff --git a/paimon-python/pypaimon/multimodal/lerobot/dataset.py b/paimon-python/pypaimon/multimodal/lerobot/dataset.py index 665ceb147863..f98bfe24c465 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -104,6 +104,7 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, + max_open_videos=16, video_backend=None, return_uint8=False, _resolved_meta=False): @@ -120,6 +121,7 @@ def __init__( delta_timestamps, tolerance_s, blob_parallelism, + max_open_videos, video_backend, return_uint8, ) @@ -154,6 +156,7 @@ def _init_dataset( delta_timestamps, tolerance_s, blob_parallelism, + max_open_videos, video_backend, return_uint8): self.meta = metadata @@ -163,6 +166,10 @@ def _init_dataset( self.tolerance_s = float(tolerance_s) if not math.isfinite(self.tolerance_s) or self.tolerance_s < 0: raise ValueError("tolerance_s must be finite and non-negative.") + if isinstance(max_open_videos, bool): + raise ValueError("max_open_videos must be a positive integer.") + self.max_open_videos = _positive_int( + max_open_videos, "max_open_videos") self.blob_parallelism = _positive_int( blob_parallelism, "blob_parallelism") if video_backend not in (None, "torchcodec", "pyav"): @@ -277,6 +284,7 @@ def _set_frame_rows( output_column=key, collate_fn=_identity, range_parallelism=self.blob_parallelism, + max_open_videos=self.max_open_videos, ) for key in self._video_keys ] @@ -588,6 +596,7 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, + max_open_videos=16, video_backend=None, return_uint8=False): self._frames_table, meta = _load_dataset(table, tag_name) @@ -600,6 +609,7 @@ def __init__( delta_timestamps=delta_timestamps, tolerance_s=tolerance_s, blob_parallelism=blob_parallelism, + max_open_videos=max_open_videos, video_backend=video_backend, return_uint8=return_uint8, _resolved_meta=True, @@ -634,6 +644,7 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, + max_open_videos=16, video_backend=None, return_uint8=False): _require_dataset_python() @@ -645,6 +656,7 @@ def __init__( or delta_timestamps is not None or tolerance_s != 1e-4 or blob_parallelism != 16 + or max_open_videos != 16 or video_backend is not None or return_uint8 ): @@ -660,6 +672,7 @@ def __init__( delta_timestamps=delta_timestamps, tolerance_s=tolerance_s, blob_parallelism=blob_parallelism, + max_open_videos=max_open_videos, video_backend=video_backend, return_uint8=return_uint8, ) diff --git a/paimon-python/pypaimon/multimodal/video.py b/paimon-python/pypaimon/multimodal/video.py index 05e70619d1aa..90489a0b0a78 100644 --- a/paimon-python/pypaimon/multimodal/video.py +++ b/paimon-python/pypaimon/multimodal/video.py @@ -150,7 +150,7 @@ def __init__( decoder_factory, decode_fn=None, output_column="frame", - max_open_videos=8, + max_open_videos=16, collate_fn=None, decode_batch_fn=None, range_parallelism=4): diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py index f39d7e1add02..ec5712cc66cf 100644 --- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py +++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py @@ -669,6 +669,61 @@ def test_indexed_default_backend_falls_back_on_ffmpeg_error(self): self.assertIs(expected, decoder.get_frames_at(indices=[3])) stream.seek.assert_called_once_with(0) + def test_video_decoder_cache_capacity(self): + metadata = { + "repo_id": "test/cache-capacity", + "info": { + "codebase_version": "v3.0", "fps": 10, + "total_frames": 1, "total_episodes": 1, "total_tasks": 1, + "features": { + "index": {"dtype": "int64", "shape": [1]}, + "episode_index": {"dtype": "int64", "shape": [1]}, + "frame_index": {"dtype": "int64", "shape": [1]}, + "timestamp": {"dtype": "float32", "shape": [1]}, + "task_index": {"dtype": "int64", "shape": [1]}, + "camera": {"dtype": "video", "shape": [2, 2, 3], + "names": ["height", "width", "channels"]}, + "camera_b": {"dtype": "video", "shape": [2, 2, 3], + "names": ["height", "width", "channels"]}, + }, + }, + "episodes": [{"episode_index": 0, "dataset_from_index": 0, + "dataset_to_index": 1, "length": 1, + "tasks": ["pick"]}], + "tasks": ["pick"], + } + from pypaimon.filesystem.local_file_io import LocalFileIO + + for options, expected in (({}, 16), ({"max_open_videos": 2}, 2)): + with self.subTest(options=options): + reader = _ManualDatasetReader( + metadata, file_io=LocalFileIO(), **options) + self.assertEqual(2, len(reader._video_collators)) + self.assertEqual([expected, expected], [ + c.max_open_videos for c in reader._video_collators]) + restored = pickle.loads(pickle.dumps(reader)) + self.assertEqual([expected, expected], [ + c.max_open_videos for c in restored._video_collators]) + dataset = pmm.PaimonLeRobotDataset(reader) + self.assertIs(reader, dataset.reader) + with self.assertRaisesRegex(ValueError, "Configure Dataset options"): + pmm.PaimonLeRobotDataset(reader, max_open_videos=3) + reader.close() + restored.close() + + for value in (0, -1, True, 1.5, None): + with self.subTest(value=value): + with self.assertRaisesRegex(ValueError, "max_open_videos"): + _ManualDatasetReader( + metadata, file_io=LocalFileIO(), max_open_videos=value) + + def test_table_dataset_forwards_video_cache_capacity(self): + module = "pypaimon.multimodal.lerobot.dataset." + table = Mock() + with patch(module + "_PaimonTableDatasetReader") as reader: + pmm.PaimonLeRobotDataset(table, max_open_videos=3) + self.assertEqual(3, reader.call_args.kwargs["max_open_videos"]) + def test_video_batches_include_delta_frames_and_preserve_backends(self): try: import torch From 3d8a2eea77e2fc068c855de3656d3f8da7d062ca Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 9 Oct 2026 21:04:01 -0700 Subject: [PATCH 2/9] [docs] Simplify decoder cache configuration description --- docs/docs/pypaimon/lerobot.md | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/docs/docs/pypaimon/lerobot.md b/docs/docs/pypaimon/lerobot.md index dbe05e658211..94fef683104c 100644 --- a/docs/docs/pypaimon/lerobot.md +++ b/docs/docs/pypaimon/lerobot.md @@ -232,10 +232,9 @@ on `index`; payloads remain lazy. Indexed videos prefetch metadata and target GO uncached PyAV reads on demand. Unindexed videos use the TorchCodec/PyAV scan path. Set `video_backend` to force either decoder. -Set `max_open_videos` on `PaimonLeRobotDataset` (or a custom -`PaimonDatasetReader`) to bound the decoder cache per camera per worker. -The default is 16; increasing it reduces decoder reopening at the cost of -more memory and open streams. +`max_open_videos` (default: 16) limits cached video decoders per camera per worker +in `PaimonLeRobotDataset` and `PaimonDatasetReader`. Larger values improve reuse +but consume more resources. Subclass `PaimonDatasetReader` for a custom logical frame layout: From 22935ed173d4d620b2bcecc9866f548dd112af1b Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 9 Oct 2026 21:10:20 -0700 Subject: [PATCH 3/9] [python] Limit decoder cache change to the default capacity --- docs/docs/pypaimon/lerobot.md | 4 -- .../pypaimon/multimodal/lerobot/dataset.py | 13 ----- .../pypaimon/tests/multimodal_lerobot_test.py | 55 ------------------- 3 files changed, 72 deletions(-) diff --git a/docs/docs/pypaimon/lerobot.md b/docs/docs/pypaimon/lerobot.md index 94fef683104c..518a5d067400 100644 --- a/docs/docs/pypaimon/lerobot.md +++ b/docs/docs/pypaimon/lerobot.md @@ -232,10 +232,6 @@ on `index`; payloads remain lazy. Indexed videos prefetch metadata and target GO uncached PyAV reads on demand. Unindexed videos use the TorchCodec/PyAV scan path. Set `video_backend` to force either decoder. -`max_open_videos` (default: 16) limits cached video decoders per camera per worker -in `PaimonLeRobotDataset` and `PaimonDatasetReader`. Larger values improve reuse -but consume more resources. - Subclass `PaimonDatasetReader` for a custom logical frame layout: ```python diff --git a/paimon-python/pypaimon/multimodal/lerobot/dataset.py b/paimon-python/pypaimon/multimodal/lerobot/dataset.py index f98bfe24c465..665ceb147863 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -104,7 +104,6 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, - max_open_videos=16, video_backend=None, return_uint8=False, _resolved_meta=False): @@ -121,7 +120,6 @@ def __init__( delta_timestamps, tolerance_s, blob_parallelism, - max_open_videos, video_backend, return_uint8, ) @@ -156,7 +154,6 @@ def _init_dataset( delta_timestamps, tolerance_s, blob_parallelism, - max_open_videos, video_backend, return_uint8): self.meta = metadata @@ -166,10 +163,6 @@ def _init_dataset( self.tolerance_s = float(tolerance_s) if not math.isfinite(self.tolerance_s) or self.tolerance_s < 0: raise ValueError("tolerance_s must be finite and non-negative.") - if isinstance(max_open_videos, bool): - raise ValueError("max_open_videos must be a positive integer.") - self.max_open_videos = _positive_int( - max_open_videos, "max_open_videos") self.blob_parallelism = _positive_int( blob_parallelism, "blob_parallelism") if video_backend not in (None, "torchcodec", "pyav"): @@ -284,7 +277,6 @@ def _set_frame_rows( output_column=key, collate_fn=_identity, range_parallelism=self.blob_parallelism, - max_open_videos=self.max_open_videos, ) for key in self._video_keys ] @@ -596,7 +588,6 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, - max_open_videos=16, video_backend=None, return_uint8=False): self._frames_table, meta = _load_dataset(table, tag_name) @@ -609,7 +600,6 @@ def __init__( delta_timestamps=delta_timestamps, tolerance_s=tolerance_s, blob_parallelism=blob_parallelism, - max_open_videos=max_open_videos, video_backend=video_backend, return_uint8=return_uint8, _resolved_meta=True, @@ -644,7 +634,6 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, - max_open_videos=16, video_backend=None, return_uint8=False): _require_dataset_python() @@ -656,7 +645,6 @@ def __init__( or delta_timestamps is not None or tolerance_s != 1e-4 or blob_parallelism != 16 - or max_open_videos != 16 or video_backend is not None or return_uint8 ): @@ -672,7 +660,6 @@ def __init__( delta_timestamps=delta_timestamps, tolerance_s=tolerance_s, blob_parallelism=blob_parallelism, - max_open_videos=max_open_videos, video_backend=video_backend, return_uint8=return_uint8, ) diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py index ec5712cc66cf..f39d7e1add02 100644 --- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py +++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py @@ -669,61 +669,6 @@ def test_indexed_default_backend_falls_back_on_ffmpeg_error(self): self.assertIs(expected, decoder.get_frames_at(indices=[3])) stream.seek.assert_called_once_with(0) - def test_video_decoder_cache_capacity(self): - metadata = { - "repo_id": "test/cache-capacity", - "info": { - "codebase_version": "v3.0", "fps": 10, - "total_frames": 1, "total_episodes": 1, "total_tasks": 1, - "features": { - "index": {"dtype": "int64", "shape": [1]}, - "episode_index": {"dtype": "int64", "shape": [1]}, - "frame_index": {"dtype": "int64", "shape": [1]}, - "timestamp": {"dtype": "float32", "shape": [1]}, - "task_index": {"dtype": "int64", "shape": [1]}, - "camera": {"dtype": "video", "shape": [2, 2, 3], - "names": ["height", "width", "channels"]}, - "camera_b": {"dtype": "video", "shape": [2, 2, 3], - "names": ["height", "width", "channels"]}, - }, - }, - "episodes": [{"episode_index": 0, "dataset_from_index": 0, - "dataset_to_index": 1, "length": 1, - "tasks": ["pick"]}], - "tasks": ["pick"], - } - from pypaimon.filesystem.local_file_io import LocalFileIO - - for options, expected in (({}, 16), ({"max_open_videos": 2}, 2)): - with self.subTest(options=options): - reader = _ManualDatasetReader( - metadata, file_io=LocalFileIO(), **options) - self.assertEqual(2, len(reader._video_collators)) - self.assertEqual([expected, expected], [ - c.max_open_videos for c in reader._video_collators]) - restored = pickle.loads(pickle.dumps(reader)) - self.assertEqual([expected, expected], [ - c.max_open_videos for c in restored._video_collators]) - dataset = pmm.PaimonLeRobotDataset(reader) - self.assertIs(reader, dataset.reader) - with self.assertRaisesRegex(ValueError, "Configure Dataset options"): - pmm.PaimonLeRobotDataset(reader, max_open_videos=3) - reader.close() - restored.close() - - for value in (0, -1, True, 1.5, None): - with self.subTest(value=value): - with self.assertRaisesRegex(ValueError, "max_open_videos"): - _ManualDatasetReader( - metadata, file_io=LocalFileIO(), max_open_videos=value) - - def test_table_dataset_forwards_video_cache_capacity(self): - module = "pypaimon.multimodal.lerobot.dataset." - table = Mock() - with patch(module + "_PaimonTableDatasetReader") as reader: - pmm.PaimonLeRobotDataset(table, max_open_videos=3) - self.assertEqual(3, reader.call_args.kwargs["max_open_videos"]) - def test_video_batches_include_delta_frames_and_preserve_backends(self): try: import torch From 42effe205b1feb146ff06c843011066324dedca5 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 9 Oct 2026 23:28:07 -0700 Subject: [PATCH 4/9] [python] Make LeRobot video decoder cache configurable --- docs/docs/pypaimon/lerobot.md | 4 ++ .../pypaimon/multimodal/lerobot/dataset.py | 13 +++++ .../pypaimon/tests/multimodal_lerobot_test.py | 55 +++++++++++++++++++ 3 files changed, 72 insertions(+) diff --git a/docs/docs/pypaimon/lerobot.md b/docs/docs/pypaimon/lerobot.md index 518a5d067400..94fef683104c 100644 --- a/docs/docs/pypaimon/lerobot.md +++ b/docs/docs/pypaimon/lerobot.md @@ -232,6 +232,10 @@ on `index`; payloads remain lazy. Indexed videos prefetch metadata and target GO uncached PyAV reads on demand. Unindexed videos use the TorchCodec/PyAV scan path. Set `video_backend` to force either decoder. +`max_open_videos` (default: 16) limits cached video decoders per camera per worker +in `PaimonLeRobotDataset` and `PaimonDatasetReader`. Larger values improve reuse +but consume more resources. + Subclass `PaimonDatasetReader` for a custom logical frame layout: ```python diff --git a/paimon-python/pypaimon/multimodal/lerobot/dataset.py b/paimon-python/pypaimon/multimodal/lerobot/dataset.py index 665ceb147863..f98bfe24c465 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -104,6 +104,7 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, + max_open_videos=16, video_backend=None, return_uint8=False, _resolved_meta=False): @@ -120,6 +121,7 @@ def __init__( delta_timestamps, tolerance_s, blob_parallelism, + max_open_videos, video_backend, return_uint8, ) @@ -154,6 +156,7 @@ def _init_dataset( delta_timestamps, tolerance_s, blob_parallelism, + max_open_videos, video_backend, return_uint8): self.meta = metadata @@ -163,6 +166,10 @@ def _init_dataset( self.tolerance_s = float(tolerance_s) if not math.isfinite(self.tolerance_s) or self.tolerance_s < 0: raise ValueError("tolerance_s must be finite and non-negative.") + if isinstance(max_open_videos, bool): + raise ValueError("max_open_videos must be a positive integer.") + self.max_open_videos = _positive_int( + max_open_videos, "max_open_videos") self.blob_parallelism = _positive_int( blob_parallelism, "blob_parallelism") if video_backend not in (None, "torchcodec", "pyav"): @@ -277,6 +284,7 @@ def _set_frame_rows( output_column=key, collate_fn=_identity, range_parallelism=self.blob_parallelism, + max_open_videos=self.max_open_videos, ) for key in self._video_keys ] @@ -588,6 +596,7 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, + max_open_videos=16, video_backend=None, return_uint8=False): self._frames_table, meta = _load_dataset(table, tag_name) @@ -600,6 +609,7 @@ def __init__( delta_timestamps=delta_timestamps, tolerance_s=tolerance_s, blob_parallelism=blob_parallelism, + max_open_videos=max_open_videos, video_backend=video_backend, return_uint8=return_uint8, _resolved_meta=True, @@ -634,6 +644,7 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, + max_open_videos=16, video_backend=None, return_uint8=False): _require_dataset_python() @@ -645,6 +656,7 @@ def __init__( or delta_timestamps is not None or tolerance_s != 1e-4 or blob_parallelism != 16 + or max_open_videos != 16 or video_backend is not None or return_uint8 ): @@ -660,6 +672,7 @@ def __init__( delta_timestamps=delta_timestamps, tolerance_s=tolerance_s, blob_parallelism=blob_parallelism, + max_open_videos=max_open_videos, video_backend=video_backend, return_uint8=return_uint8, ) diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py index f39d7e1add02..ec5712cc66cf 100644 --- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py +++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py @@ -669,6 +669,61 @@ def test_indexed_default_backend_falls_back_on_ffmpeg_error(self): self.assertIs(expected, decoder.get_frames_at(indices=[3])) stream.seek.assert_called_once_with(0) + def test_video_decoder_cache_capacity(self): + metadata = { + "repo_id": "test/cache-capacity", + "info": { + "codebase_version": "v3.0", "fps": 10, + "total_frames": 1, "total_episodes": 1, "total_tasks": 1, + "features": { + "index": {"dtype": "int64", "shape": [1]}, + "episode_index": {"dtype": "int64", "shape": [1]}, + "frame_index": {"dtype": "int64", "shape": [1]}, + "timestamp": {"dtype": "float32", "shape": [1]}, + "task_index": {"dtype": "int64", "shape": [1]}, + "camera": {"dtype": "video", "shape": [2, 2, 3], + "names": ["height", "width", "channels"]}, + "camera_b": {"dtype": "video", "shape": [2, 2, 3], + "names": ["height", "width", "channels"]}, + }, + }, + "episodes": [{"episode_index": 0, "dataset_from_index": 0, + "dataset_to_index": 1, "length": 1, + "tasks": ["pick"]}], + "tasks": ["pick"], + } + from pypaimon.filesystem.local_file_io import LocalFileIO + + for options, expected in (({}, 16), ({"max_open_videos": 2}, 2)): + with self.subTest(options=options): + reader = _ManualDatasetReader( + metadata, file_io=LocalFileIO(), **options) + self.assertEqual(2, len(reader._video_collators)) + self.assertEqual([expected, expected], [ + c.max_open_videos for c in reader._video_collators]) + restored = pickle.loads(pickle.dumps(reader)) + self.assertEqual([expected, expected], [ + c.max_open_videos for c in restored._video_collators]) + dataset = pmm.PaimonLeRobotDataset(reader) + self.assertIs(reader, dataset.reader) + with self.assertRaisesRegex(ValueError, "Configure Dataset options"): + pmm.PaimonLeRobotDataset(reader, max_open_videos=3) + reader.close() + restored.close() + + for value in (0, -1, True, 1.5, None): + with self.subTest(value=value): + with self.assertRaisesRegex(ValueError, "max_open_videos"): + _ManualDatasetReader( + metadata, file_io=LocalFileIO(), max_open_videos=value) + + def test_table_dataset_forwards_video_cache_capacity(self): + module = "pypaimon.multimodal.lerobot.dataset." + table = Mock() + with patch(module + "_PaimonTableDatasetReader") as reader: + pmm.PaimonLeRobotDataset(table, max_open_videos=3) + self.assertEqual(3, reader.call_args.kwargs["max_open_videos"]) + def test_video_batches_include_delta_frames_and_preserve_backends(self): try: import torch From f67699e497707cd17f865d106eda672cce2dcf30 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 9 Oct 2026 23:44:14 -0700 Subject: [PATCH 5/9] [python] Configure LeRobot decoder cache through table options --- docs/docs/pypaimon/lerobot.md | 8 +- .../pypaimon/multimodal/lerobot/dataset.py | 38 ++++--- paimon-python/pypaimon/multimodal/table.py | 6 ++ .../pypaimon/tests/multimodal_lerobot_test.py | 101 +++++++++--------- .../pypaimon/tests/multimodal_table_test.py | 23 ++++ 5 files changed, 111 insertions(+), 65 deletions(-) diff --git a/docs/docs/pypaimon/lerobot.md b/docs/docs/pypaimon/lerobot.md index 94fef683104c..f5d76f7a9c59 100644 --- a/docs/docs/pypaimon/lerobot.md +++ b/docs/docs/pypaimon/lerobot.md @@ -213,8 +213,9 @@ For map-style training, pass an image- or video-backed table group created by ```python from pypaimon.multimodal import PaimonLeRobotDataset +table = conn.get_table("robot_data") dataset = PaimonLeRobotDataset( - conn.get_table("robot_data"), + table, tag_name=tag, ) loader = dataset.to_dataloader( @@ -232,9 +233,8 @@ on `index`; payloads remain lazy. Indexed videos prefetch metadata and target GO uncached PyAV reads on demand. Unindexed videos use the TorchCodec/PyAV scan path. Set `video_backend` to force either decoder. -`max_open_videos` (default: 16) limits cached video decoders per camera per worker -in `PaimonLeRobotDataset` and `PaimonDatasetReader`. Larger values improve reuse -but consume more resources. +Set `table = table.copy({"pypaimon.lerobot.max-open-videos": "8"})` before +constructing the dataset to limit cached decoders per camera per worker (default: 16). Subclass `PaimonDatasetReader` for a custom logical frame layout: diff --git a/paimon-python/pypaimon/multimodal/lerobot/dataset.py b/paimon-python/pypaimon/multimodal/lerobot/dataset.py index f98bfe24c465..f21f6965b485 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -34,6 +34,7 @@ import pyarrow as pa +from pypaimon.common.options.config_options import ConfigOptions from pypaimon.multimodal.lerobot.metadata import ( _companion_table_identifiers, _restore_pandas_metadata, @@ -53,6 +54,13 @@ from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex +_VIDEO_DECODER_CACHE_SIZE = ( + ConfigOptions.key("pypaimon.lerobot.max-open-videos") + .int_type() + .default_value(16) +) + + _TORCH_DTYPE_NAMES = { "bool": "bool", "boolean": "bool", @@ -104,7 +112,6 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, - max_open_videos=16, video_backend=None, return_uint8=False, _resolved_meta=False): @@ -121,7 +128,6 @@ def __init__( delta_timestamps, tolerance_s, blob_parallelism, - max_open_videos, video_backend, return_uint8, ) @@ -156,7 +162,6 @@ def _init_dataset( delta_timestamps, tolerance_s, blob_parallelism, - max_open_videos, video_backend, return_uint8): self.meta = metadata @@ -166,10 +171,6 @@ def _init_dataset( self.tolerance_s = float(tolerance_s) if not math.isfinite(self.tolerance_s) or self.tolerance_s < 0: raise ValueError("tolerance_s must be finite and non-negative.") - if isinstance(max_open_videos, bool): - raise ValueError("max_open_videos must be a positive integer.") - self.max_open_videos = _positive_int( - max_open_videos, "max_open_videos") self.blob_parallelism = _positive_int( blob_parallelism, "blob_parallelism") if video_backend not in (None, "torchcodec", "pyav"): @@ -274,6 +275,7 @@ def _set_frame_rows( if self._video_keys and self._file_io is None: raise ValueError( "A video-backed PaimonDatasetReader must expose file_io.") + cache_size = self._video_decoder_cache_size() self._video_collators = [ VideoFrameCollator( access, @@ -284,12 +286,15 @@ def _set_frame_rows( output_column=key, collate_fn=_identity, range_parallelism=self.blob_parallelism, - max_open_videos=self.max_open_videos, + max_open_videos=cache_size, ) for key in self._video_keys ] self._init_delta_projection(validation_context, subtasks) + def _video_decoder_cache_size(self): + return _VIDEO_DECODER_CACHE_SIZE.default_value() + def _init_frame_contract(self, target_schema, info, validate_metadata): tasks = _metadata_member(self.meta, "tasks") subtasks = _metadata_member(self.meta, "subtasks") @@ -596,7 +601,6 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, - max_open_videos=16, video_backend=None, return_uint8=False): self._frames_table, meta = _load_dataset(table, tag_name) @@ -609,7 +613,6 @@ def __init__( delta_timestamps=delta_timestamps, tolerance_s=tolerance_s, blob_parallelism=blob_parallelism, - max_open_videos=max_open_videos, video_backend=video_backend, return_uint8=return_uint8, _resolved_meta=True, @@ -618,6 +621,18 @@ def __init__( def read_indices(self, indices, columns): return self._frame_rows.read_indices(indices, columns) + def _video_decoder_cache_size(self): + key = _VIDEO_DECODER_CACHE_SIZE.key() + raw = self._frames_table.options.options.to_map().get(key) + if isinstance(raw, (bool, float)): + raise ValueError("%s must be a positive integer." % key) + try: + size = self._frames_table.options.options.get( + _VIDEO_DECODER_CACHE_SIZE) + except (TypeError, ValueError) as error: + raise ValueError("%s must be a positive integer." % key) from error + return _positive_int(size, key) + def _validate_physical_metadata(self): return True @@ -644,7 +659,6 @@ def __init__( delta_timestamps=None, tolerance_s=1e-4, blob_parallelism=16, - max_open_videos=16, video_backend=None, return_uint8=False): _require_dataset_python() @@ -656,7 +670,6 @@ def __init__( or delta_timestamps is not None or tolerance_s != 1e-4 or blob_parallelism != 16 - or max_open_videos != 16 or video_backend is not None or return_uint8 ): @@ -672,7 +685,6 @@ def __init__( delta_timestamps=delta_timestamps, tolerance_s=tolerance_s, blob_parallelism=blob_parallelism, - max_open_videos=max_open_videos, video_backend=video_backend, return_uint8=return_uint8, ) diff --git a/paimon-python/pypaimon/multimodal/table.py b/paimon-python/pypaimon/multimodal/table.py index 00f1c03c4e60..859a9036dd80 100644 --- a/paimon-python/pypaimon/multimodal/table.py +++ b/paimon-python/pypaimon/multimodal/table.py @@ -102,6 +102,12 @@ def __init__(self, catalog, identifier: str, raw_table): self.raw_table = raw_table self.table = raw_table + def copy(self, options: dict) -> 'MultimodalTable': + """Return a table view with dynamic Paimon options.""" + raw_table = self.raw_table.copy(options) + return MultimodalTable( + self.catalog, raw_table.identifier.get_full_name(), raw_table) + def add(self, data): arrow_table = _to_arrow_table(data, _target_schema(self.raw_table)) write_builder = self.raw_table.new_batch_write_builder() diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py index ec5712cc66cf..8a548aa3ffbd 100644 --- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py +++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py @@ -56,6 +56,7 @@ _image_tensor, _index_names, _open_video_decoder, + _reader_metadata, _selected_episodes, _stack_visual_windows, _torch_row, @@ -669,60 +670,64 @@ def test_indexed_default_backend_falls_back_on_ffmpeg_error(self): self.assertIs(expected, decoder.get_frames_at(indices=[3])) stream.seek.assert_called_once_with(0) - def test_video_decoder_cache_capacity(self): - metadata = { - "repo_id": "test/cache-capacity", - "info": { - "codebase_version": "v3.0", "fps": 10, - "total_frames": 1, "total_episodes": 1, "total_tasks": 1, - "features": { - "index": {"dtype": "int64", "shape": [1]}, - "episode_index": {"dtype": "int64", "shape": [1]}, - "frame_index": {"dtype": "int64", "shape": [1]}, - "timestamp": {"dtype": "float32", "shape": [1]}, - "task_index": {"dtype": "int64", "shape": [1]}, - "camera": {"dtype": "video", "shape": [2, 2, 3], - "names": ["height", "width", "channels"]}, - "camera_b": {"dtype": "video", "shape": [2, 2, 3], - "names": ["height", "width", "channels"]}, - }, + def test_table_option_limits_video_decoders(self): + from pypaimon.filesystem.local_file_io import LocalFileIO + + info = { + "codebase_version": "v3.0", "fps": 10, + "total_frames": 1, "total_episodes": 1, "total_tasks": 1, + "features": { + "index": {"dtype": "int64", "shape": [1]}, + "episode_index": {"dtype": "int64", "shape": [1]}, + "frame_index": {"dtype": "int64", "shape": [1]}, + "timestamp": {"dtype": "float32", "shape": [1]}, + "task_index": {"dtype": "int64", "shape": [1]}, + "camera": {"dtype": "video", "shape": [2, 2, 3], + "names": ["height", "width", "channels"]}, + "camera_b": {"dtype": "video", "shape": [2, 2, 3], + "names": ["height", "width", "channels"]}, }, + } + metadata = _reader_metadata({ + "repo_id": "test/video-cache-option", "info": info, "episodes": [{"episode_index": 0, "dataset_from_index": 0, "dataset_to_index": 1, "length": 1, "tasks": ["pick"]}], "tasks": ["pick"], - } - from pypaimon.filesystem.local_file_io import LocalFileIO - - for options, expected in (({}, 16), ({"max_open_videos": 2}, 2)): - with self.subTest(options=options): - reader = _ManualDatasetReader( - metadata, file_io=LocalFileIO(), **options) - self.assertEqual(2, len(reader._video_collators)) - self.assertEqual([expected, expected], [ - c.max_open_videos for c in reader._video_collators]) - restored = pickle.loads(pickle.dumps(reader)) - self.assertEqual([expected, expected], [ - c.max_open_videos for c in restored._video_collators]) - dataset = pmm.PaimonLeRobotDataset(reader) - self.assertIs(reader, dataset.reader) - with self.assertRaisesRegex(ValueError, "Configure Dataset options"): - pmm.PaimonLeRobotDataset(reader, max_open_videos=3) - reader.close() - restored.close() - - for value in (0, -1, True, 1.5, None): - with self.subTest(value=value): - with self.assertRaisesRegex(ValueError, "max_open_videos"): - _ManualDatasetReader( - metadata, file_io=LocalFileIO(), max_open_videos=value) - - def test_table_dataset_forwards_video_cache_capacity(self): + }) + rows = SimpleNamespace( + num_rows=1, file_io=LocalFileIO(), close=lambda: None) module = "pypaimon.multimodal.lerobot.dataset." - table = Mock() - with patch(module + "_PaimonTableDatasetReader") as reader: - pmm.PaimonLeRobotDataset(table, max_open_videos=3) - self.assertEqual(3, reader.call_args.kwargs["max_open_videos"]) + option = "pypaimon.lerobot.max-open-videos" + + for configured, expected in ((None, 16), ("2", 2)): + with self.subTest(configured=configured): + options = {} if configured is None else {option: configured} + frames = SimpleNamespace( + options=SimpleNamespace(options=Options(options))) + with patch(module + "_load_dataset", return_value=(frames, metadata)), \ + patch(module + "_target_schema", + return_value=_schema_from_info(info)), \ + patch(module + "_PaimonTableFrameReader", + return_value=rows): + dataset = pmm.PaimonLeRobotDataset(Mock()) + try: + self.assertEqual([expected, expected], [ + c.max_open_videos for c in dataset.reader._video_collators]) + finally: + dataset.reader.close() + + for configured in ("0", "bad", True, 1.5): + with self.subTest(configured=configured): + frames = SimpleNamespace(options=SimpleNamespace( + options=Options({option: configured}))) + with patch(module + "_load_dataset", return_value=(frames, metadata)), \ + patch(module + "_target_schema", + return_value=_schema_from_info(info)), \ + patch(module + "_PaimonTableFrameReader", + return_value=rows): + with self.assertRaisesRegex(ValueError, option): + pmm.PaimonLeRobotDataset(Mock()) def test_video_batches_include_delta_frames_and_preserve_backends(self): try: diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py b/paimon-python/pypaimon/tests/multimodal_table_test.py index 00bd51e7d214..1ebb0bffadb1 100644 --- a/paimon-python/pypaimon/tests/multimodal_table_test.py +++ b/paimon-python/pypaimon/tests/multimodal_table_test.py @@ -98,6 +98,29 @@ def test_connect_rejects_positional_warehouse(self): with self.assertRaises(TypeError): pmm.connect(self.warehouse) + def test_copy_applies_dynamic_table_options(self): + table = self.conn.create_table( + "frames", schema=_schema({"id": pa.int32()}), + options=_PARQUET_OPTIONS) + copied = table.copy({"pypaimon.lerobot.max-open-videos": "8"}) + + self.assertEqual(table.identifier, copied.identifier) + self.assertIs(table.catalog, copied.catalog) + self.assertNotIn( + "pypaimon.lerobot.max-open-videos", + table.raw_table.table_schema.options) + self.assertEqual( + "8", copied.raw_table.table_schema.options[ + "pypaimon.lerobot.max-open-videos"]) + + table.add([{"id": 1}]) + snapshot_id = table.raw_table.snapshot_manager().get_latest_snapshot().id + table.raw_table.create_tag("training", snapshot_id=snapshot_id) + tagged = copied.raw_table.copy({"scan.tag-name": "training"}) + self.assertEqual( + "8", tagged.table_schema.options[ + "pypaimon.lerobot.max-open-videos"]) + def test_create_table_defaults_data_evolution_options(self): table = self.conn.create_table( "docs", From 0be489ccc97310f4fd0568eba907fb6da0a13456 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 9 Oct 2026 23:47:17 -0700 Subject: [PATCH 6/9] [python] Use read option name for decoder cache limit --- docs/docs/pypaimon/lerobot.md | 2 +- paimon-python/pypaimon/multimodal/lerobot/dataset.py | 2 +- paimon-python/pypaimon/tests/multimodal_lerobot_test.py | 2 +- paimon-python/pypaimon/tests/multimodal_table_test.py | 8 ++++---- 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/docs/docs/pypaimon/lerobot.md b/docs/docs/pypaimon/lerobot.md index f5d76f7a9c59..0dcf4f64f2de 100644 --- a/docs/docs/pypaimon/lerobot.md +++ b/docs/docs/pypaimon/lerobot.md @@ -233,7 +233,7 @@ on `index`; payloads remain lazy. Indexed videos prefetch metadata and target GO uncached PyAV reads on demand. Unindexed videos use the TorchCodec/PyAV scan path. Set `video_backend` to force either decoder. -Set `table = table.copy({"pypaimon.lerobot.max-open-videos": "8"})` before +Set `table = table.copy({"read.video.max-open-decoders": "8"})` before constructing the dataset to limit cached decoders per camera per worker (default: 16). Subclass `PaimonDatasetReader` for a custom logical frame layout: diff --git a/paimon-python/pypaimon/multimodal/lerobot/dataset.py b/paimon-python/pypaimon/multimodal/lerobot/dataset.py index f21f6965b485..d85bbbfea16f 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -55,7 +55,7 @@ _VIDEO_DECODER_CACHE_SIZE = ( - ConfigOptions.key("pypaimon.lerobot.max-open-videos") + ConfigOptions.key("read.video.max-open-decoders") .int_type() .default_value(16) ) diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py index 8a548aa3ffbd..24dc3b3c2b48 100644 --- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py +++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py @@ -698,7 +698,7 @@ def test_table_option_limits_video_decoders(self): rows = SimpleNamespace( num_rows=1, file_io=LocalFileIO(), close=lambda: None) module = "pypaimon.multimodal.lerobot.dataset." - option = "pypaimon.lerobot.max-open-videos" + option = "read.video.max-open-decoders" for configured, expected in ((None, 16), ("2", 2)): with self.subTest(configured=configured): diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py b/paimon-python/pypaimon/tests/multimodal_table_test.py index 1ebb0bffadb1..e2f665e8b1a6 100644 --- a/paimon-python/pypaimon/tests/multimodal_table_test.py +++ b/paimon-python/pypaimon/tests/multimodal_table_test.py @@ -102,16 +102,16 @@ def test_copy_applies_dynamic_table_options(self): table = self.conn.create_table( "frames", schema=_schema({"id": pa.int32()}), options=_PARQUET_OPTIONS) - copied = table.copy({"pypaimon.lerobot.max-open-videos": "8"}) + copied = table.copy({"read.video.max-open-decoders": "8"}) self.assertEqual(table.identifier, copied.identifier) self.assertIs(table.catalog, copied.catalog) self.assertNotIn( - "pypaimon.lerobot.max-open-videos", + "read.video.max-open-decoders", table.raw_table.table_schema.options) self.assertEqual( "8", copied.raw_table.table_schema.options[ - "pypaimon.lerobot.max-open-videos"]) + "read.video.max-open-decoders"]) table.add([{"id": 1}]) snapshot_id = table.raw_table.snapshot_manager().get_latest_snapshot().id @@ -119,7 +119,7 @@ def test_copy_applies_dynamic_table_options(self): tagged = copied.raw_table.copy({"scan.tag-name": "training"}) self.assertEqual( "8", tagged.table_schema.options[ - "pypaimon.lerobot.max-open-videos"]) + "read.video.max-open-decoders"]) def test_create_table_defaults_data_evolution_options(self): table = self.conn.create_table( From e48b09ebe8ef8ff52cf9a058b27963ad0301e8d1 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 9 Oct 2026 23:58:51 -0700 Subject: [PATCH 7/9] [python] Define LeRobot decoder cache option in CoreOptions --- .../pypaimon/common/options/core_options.py | 12 ++++++++++++ .../pypaimon/multimodal/lerobot/dataset.py | 16 ++++------------ .../pypaimon/tests/multimodal_lerobot_test.py | 7 ++++--- 3 files changed, 20 insertions(+), 15 deletions(-) diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 525b16f70892..102d7cb23b2b 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -1163,6 +1163,15 @@ class CoreOptions: .with_description("Read batch size for any file format if it supports.") ) + READ_VIDEO_MAX_OPEN_DECODERS: ConfigOption[int] = ( + ConfigOptions.key("read.video.max-open-decoders") + .int_type() + .default_value(16) + .with_description( + "Maximum cached video decoders per video feature in a PyPaimon " + "LeRobot DataLoader worker.") + ) + PARQUET_COLUMN_INDEX_ENABLED: ConfigOption[bool] = ( ConfigOptions.key("parquet.filter.columnindex.enabled") .boolean_type() @@ -1927,6 +1936,9 @@ def local_cache_whitelist(self) -> str: def read_batch_size(self, default=None) -> int: return self.options.get(CoreOptions.READ_BATCH_SIZE, default or 1024) + def read_video_max_open_decoders(self) -> int: + return self.options.get(CoreOptions.READ_VIDEO_MAX_OPEN_DECODERS) + def parquet_column_index_enabled(self) -> bool: return self.options.get(CoreOptions.PARQUET_COLUMN_INDEX_ENABLED) diff --git a/paimon-python/pypaimon/multimodal/lerobot/dataset.py b/paimon-python/pypaimon/multimodal/lerobot/dataset.py index d85bbbfea16f..da6c56454e9a 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -34,7 +34,7 @@ import pyarrow as pa -from pypaimon.common.options.config_options import ConfigOptions +from pypaimon.common.options.core_options import CoreOptions from pypaimon.multimodal.lerobot.metadata import ( _companion_table_identifiers, _restore_pandas_metadata, @@ -54,13 +54,6 @@ from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex -_VIDEO_DECODER_CACHE_SIZE = ( - ConfigOptions.key("read.video.max-open-decoders") - .int_type() - .default_value(16) -) - - _TORCH_DTYPE_NAMES = { "bool": "bool", "boolean": "bool", @@ -293,7 +286,7 @@ def _set_frame_rows( self._init_delta_projection(validation_context, subtasks) def _video_decoder_cache_size(self): - return _VIDEO_DECODER_CACHE_SIZE.default_value() + return CoreOptions.READ_VIDEO_MAX_OPEN_DECODERS.default_value() def _init_frame_contract(self, target_schema, info, validate_metadata): tasks = _metadata_member(self.meta, "tasks") @@ -622,13 +615,12 @@ def read_indices(self, indices, columns): return self._frame_rows.read_indices(indices, columns) def _video_decoder_cache_size(self): - key = _VIDEO_DECODER_CACHE_SIZE.key() + key = CoreOptions.READ_VIDEO_MAX_OPEN_DECODERS.key() raw = self._frames_table.options.options.to_map().get(key) if isinstance(raw, (bool, float)): raise ValueError("%s must be a positive integer." % key) try: - size = self._frames_table.options.options.get( - _VIDEO_DECODER_CACHE_SIZE) + size = self._frames_table.options.read_video_max_open_decoders() except (TypeError, ValueError) as error: raise ValueError("%s must be a positive integer." % key) from error return _positive_int(size, key) diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py index 24dc3b3c2b48..6473b53584ef 100644 --- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py +++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py @@ -42,6 +42,7 @@ import pypaimon.multimodal as pmm from pypaimon.common.identifier import Identifier from pypaimon.common.options import Options +from pypaimon.common.options.core_options import CoreOptions from pypaimon.multimodal.source_utils import _SourceFileIO from pypaimon.multimodal.connection import MultimodalConnection from pypaimon.multimodal.lerobot import load_from_lerobot @@ -704,7 +705,7 @@ def test_table_option_limits_video_decoders(self): with self.subTest(configured=configured): options = {} if configured is None else {option: configured} frames = SimpleNamespace( - options=SimpleNamespace(options=Options(options))) + options=CoreOptions(Options(options))) with patch(module + "_load_dataset", return_value=(frames, metadata)), \ patch(module + "_target_schema", return_value=_schema_from_info(info)), \ @@ -719,8 +720,8 @@ def test_table_option_limits_video_decoders(self): for configured in ("0", "bad", True, 1.5): with self.subTest(configured=configured): - frames = SimpleNamespace(options=SimpleNamespace( - options=Options({option: configured}))) + frames = SimpleNamespace( + options=CoreOptions(Options({option: configured}))) with patch(module + "_load_dataset", return_value=(frames, metadata)), \ patch(module + "_target_schema", return_value=_schema_from_info(info)), \ From c47dbd6c13e3a7bee0c27a6d0251eadb3b9be8ca Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Sat, 10 Oct 2026 00:04:24 -0700 Subject: [PATCH 8/9] [python] Read LeRobot decoder limit when creating collators --- .../pypaimon/common/options/core_options.py | 13 +++++++++++- .../pypaimon/multimodal/lerobot/dataset.py | 21 ++++++------------- .../pypaimon/tests/multimodal_lerobot_test.py | 8 +++++-- 3 files changed, 24 insertions(+), 18 deletions(-) diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 102d7cb23b2b..46322f8fc073 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -1937,7 +1937,18 @@ def read_batch_size(self, default=None) -> int: return self.options.get(CoreOptions.READ_BATCH_SIZE, default or 1024) def read_video_max_open_decoders(self) -> int: - return self.options.get(CoreOptions.READ_VIDEO_MAX_OPEN_DECODERS) + option = CoreOptions.READ_VIDEO_MAX_OPEN_DECODERS + raw = self.options.to_map().get(option.key()) + if isinstance(raw, (bool, float)): + raise ValueError("%s must be a positive integer." % option.key()) + try: + value = self.options.get(option) + except (TypeError, ValueError) as error: + raise ValueError( + "%s must be a positive integer." % option.key()) from error + if value <= 0: + raise ValueError("%s must be a positive integer." % option.key()) + return value def parquet_column_index_enabled(self) -> bool: return self.options.get(CoreOptions.PARQUET_COLUMN_INDEX_ENABLED) diff --git a/paimon-python/pypaimon/multimodal/lerobot/dataset.py b/paimon-python/pypaimon/multimodal/lerobot/dataset.py index da6c56454e9a..38793e26da4e 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -268,7 +268,12 @@ def _set_frame_rows( if self._video_keys and self._file_io is None: raise ValueError( "A video-backed PaimonDatasetReader must expose file_io.") - cache_size = self._video_decoder_cache_size() + table_options = getattr(self._read_table, "options", None) + cache_size = ( + table_options.read_video_max_open_decoders() + if isinstance(table_options, CoreOptions) + else CoreOptions.READ_VIDEO_MAX_OPEN_DECODERS.default_value() + ) self._video_collators = [ VideoFrameCollator( access, @@ -285,9 +290,6 @@ def _set_frame_rows( ] self._init_delta_projection(validation_context, subtasks) - def _video_decoder_cache_size(self): - return CoreOptions.READ_VIDEO_MAX_OPEN_DECODERS.default_value() - def _init_frame_contract(self, target_schema, info, validate_metadata): tasks = _metadata_member(self.meta, "tasks") subtasks = _metadata_member(self.meta, "subtasks") @@ -614,17 +616,6 @@ def __init__( def read_indices(self, indices, columns): return self._frame_rows.read_indices(indices, columns) - def _video_decoder_cache_size(self): - key = CoreOptions.READ_VIDEO_MAX_OPEN_DECODERS.key() - raw = self._frames_table.options.options.to_map().get(key) - if isinstance(raw, (bool, float)): - raise ValueError("%s must be a positive integer." % key) - try: - size = self._frames_table.options.read_video_max_open_decoders() - except (TypeError, ValueError) as error: - raise ValueError("%s must be a positive integer." % key) from error - return _positive_int(size, key) - def _validate_physical_metadata(self): return True diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py index 6473b53584ef..492b9792e1e8 100644 --- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py +++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py @@ -696,8 +696,6 @@ def test_table_option_limits_video_decoders(self): "tasks": ["pick"]}], "tasks": ["pick"], }) - rows = SimpleNamespace( - num_rows=1, file_io=LocalFileIO(), close=lambda: None) module = "pypaimon.multimodal.lerobot.dataset." option = "read.video.max-open-decoders" @@ -706,6 +704,9 @@ def test_table_option_limits_video_decoders(self): options = {} if configured is None else {option: configured} frames = SimpleNamespace( options=CoreOptions(Options(options))) + rows = SimpleNamespace( + _table=frames, num_rows=1, file_io=LocalFileIO(), + close=lambda: None) with patch(module + "_load_dataset", return_value=(frames, metadata)), \ patch(module + "_target_schema", return_value=_schema_from_info(info)), \ @@ -722,6 +723,9 @@ def test_table_option_limits_video_decoders(self): with self.subTest(configured=configured): frames = SimpleNamespace( options=CoreOptions(Options({option: configured}))) + rows = SimpleNamespace( + _table=frames, num_rows=1, file_io=LocalFileIO(), + close=lambda: None) with patch(module + "_load_dataset", return_value=(frames, metadata)), \ patch(module + "_target_schema", return_value=_schema_from_info(info)), \ From 9d87790fee379541aa60dcc754992ec5a3105211 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Sat, 10 Oct 2026 02:33:07 -0700 Subject: [PATCH 9/9] [python] Test LeRobot read option across tag and pickle --- docs/docs/pypaimon/lerobot.md | 1 + .../pypaimon/tests/multimodal_lerobot_test.py | 12 ++++++++++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/docs/docs/pypaimon/lerobot.md b/docs/docs/pypaimon/lerobot.md index 0dcf4f64f2de..3499a703b76a 100644 --- a/docs/docs/pypaimon/lerobot.md +++ b/docs/docs/pypaimon/lerobot.md @@ -235,6 +235,7 @@ uncached PyAV reads on demand. Unindexed videos use the TorchCodec/PyAV scan pat Set `table = table.copy({"read.video.max-open-decoders": "8"})` before constructing the dataset to limit cached decoders per camera per worker (default: 16). +Each open decoder uses memory and a file handle, so usage grows with cameras and workers. Subclass `PaimonDatasetReader` for a custom logical frame layout: diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py index 492b9792e1e8..a9eb32ef00fe 100644 --- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py +++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py @@ -3540,10 +3540,18 @@ def test_table_dataset_pickle_preserves_episode_metadata_and_reads(self): import torch self.connection.load_from_lerobot("worker_pickle", self.image_source) - table = self.connection.get_table("worker_pickle") - dataset = pmm.PaimonLeRobotDataset(table, return_uint8=True) + snapshots = self.connection.create_lerobot_tag( + "worker_pickle", "training") + table = self.connection.get_table("worker_pickle").copy({ + "read.video.max-open-decoders": "2"}) + dataset = pmm.PaimonLeRobotDataset( + table, tag_name="training", return_uint8=True) restored = pickle.loads(pickle.dumps(dataset)) + for reader in (dataset.reader, restored.reader): + self.assertEqual(snapshots["frames"], reader._snapshot_id) + self.assertEqual( + 2, reader._read_table.options.read_video_max_open_decoders()) self.assertEqual(dataset.meta.episodes[:], restored.meta.episodes[:]) self.assertEqual( dataset.meta.episodes._fingerprint,