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 606f230631 [python] Add ROSBag ingestion support (#9530)
606f230631 is described below
commit 606f230631e59500e35bd2975a73a3ac650dfba4
Author: Yann Byron <[email protected]>
AuthorDate: Wed Sep 2 14:39:35 2026 +0800
[python] Add ROSBag ingestion support (#9530)
---
.github/workflows/paimon-python-checks.yml | 4 +
docs/docs/pypaimon/multimodal-api.mdx | 93 +++
paimon-python/README.md | 63 ++
paimon-python/pypaimon/multimodal/__init__.py | 8 +
paimon-python/pypaimon/multimodal/connection.py | 24 +
paimon-python/pypaimon/multimodal/hdf5.py | 97 +--
.../pypaimon/multimodal/lerobot/source.py | 6 +-
.../pypaimon/multimodal/rosbag/__init__.py | 31 +
paimon-python/pypaimon/multimodal/rosbag/api.py | 160 ++++
paimon-python/pypaimon/multimodal/rosbag/loader.py | 377 +++++++++
paimon-python/pypaimon/multimodal/rosbag/source.py | 371 +++++++++
.../pypaimon/multimodal/rosbag/staging.py | 158 ++++
paimon-python/pypaimon/multimodal/source_utils.py | 93 ++-
paimon-python/pypaimon/ray/__init__.py | 2 +
paimon-python/pypaimon/ray/rosbag.py | 212 +++++
.../pypaimon/tests/multimodal_hdf5_test.py | 10 +-
.../pypaimon/tests/multimodal_lerobot_test.py | 8 +-
.../pypaimon/tests/multimodal_rosbag_test.py | 889 +++++++++++++++++++++
paimon-python/pypaimon/tests/ray_rosbag_test.py | 251 ++++++
paimon-python/setup.py | 4 +
20 files changed, 2757 insertions(+), 104 deletions(-)
diff --git a/.github/workflows/paimon-python-checks.yml
b/.github/workflows/paimon-python-checks.yml
index 53ad2cf2a3..b8ecc00913 100755
--- a/.github/workflows/paimon-python-checks.yml
+++ b/.github/workflows/paimon-python-checks.yml
@@ -139,6 +139,10 @@ jobs:
fi
python -m pip install 'h5py>=3,<4'
python -c "import h5py; print('h5py', h5py.__version__)"
+ if python -c "import sys; sys.exit(0 if sys.version_info >= (3,
10) else 1)"; then
+ python -m pip install './paimon-python[rosbag]'
+ python -c "import rosbags; print('rosbags installed')"
+ fi
if [[ "${{ matrix.python-version }}" == "3.10" ]]; then
python -m pip install './paimon-python[lerobot]'
python -c "import datasets, lerobot; print('datasets',
datasets.__version__, 'lerobot', lerobot.__version__)"
diff --git a/docs/docs/pypaimon/multimodal-api.mdx
b/docs/docs/pypaimon/multimodal-api.mdx
index e8732875b1..5e6d352181 100644
--- a/docs/docs/pypaimon/multimodal-api.mdx
+++ b/docs/docs/pypaimon/multimodal-api.mdx
@@ -520,6 +520,99 @@ HDF5 core itself has no Ray dependency.
provenance columns, maintain a source ledger, skip prior inputs, or detect
source drift. Calling it again with the same input appends the rows again.
+## Load ROSBag
+
+`MultimodalConnection.load_from_rosbag` imports ROS1 `.bag`, ROS2 SQLite3 or
+MCAP recording directories, and standalone ROS2 `.mcap` files through a user
+Arrow transform. It requires Python 3.10 or newer:
+
+```shell
+pip install 'pypaimon[rosbag]'
+```
+
+```python
+import pyarrow as pa
+
+
+schema = pa.schema([
+ pa.field("source", pa.string(), nullable=False),
+ pa.field("timestamp", pa.int64(), nullable=False),
+ pa.field("value", pa.string(), nullable=False),
+])
+conn.create_table("messages", schema=schema)
+
+
+def transform(reader, source):
+ rows = []
+ for connection, timestamp, rawdata in reader.messages():
+ message = reader.deserialize(rawdata, connection.msgtype)
+ rows.append({
+ "source": source.name,
+ "timestamp": timestamp,
+ "value": message.data,
+ })
+ return pa.Table.from_pylist(rows)
+
+
+result = conn.load_from_rosbag(
+ "messages",
+ "/data/recordings",
+ transform=transform,
+)
+print(result.source_count, result.row_count, result.snapshot_id)
+```
+
+A transform receives an open `rosbags.highlevel.AnyReader` and a
+`RosbagSource`. The source exposes the original normalized `uri`, the temporary
+or original `local_path`, `format`, `name`, `stem`, and `is_remote`. Reader and
+local staging paths are valid only while the transform is running. A transform
+returns an Arrow table, record batch, or an iterable of either.
+
+Directories are searched recursively, but a directory containing
+`metadata.yaml` is one ROS2 logical source and is not searched below that
+point. Standalone `.db3` files are rejected because they may be one split of a
+larger recording. Set `allow_storage_fragment=True` only when importing exactly
+that SQLite storage fragment is intentional; the loader performs a SQLite
+integrity check but cannot prove that other recording splits are absent.
+
+Local paths and `file://`, `oss://`, `s3://`, `hdfs://`, `viewfs://`, and
+`gs://` URIs use PyPaimon FileIO. Pass source credentials through
+`source_options`; target warehouse options are not inherited. Unlike HDF5,
+`rosbags` requires local paths, so remote files are copied in bounded chunks to
+an attempt-scoped temporary directory. ROS2 metadata member paths are checked
+for traversal and normalization collisions before copying.
+
+Before Paimon creates a writer, the loader:
+
+1. validates every source manifest and checks for active or recovery sidecars;
+2. scans every recording to EOF and compares declared and readable messages;
+3. executes each transform once and applies strict target Arrow schema checks;
+4. stores all validated output in a temporary Arrow IPC file.
+
+This front-loaded validation normally reads each recording twice and needs
+temporary disk capacity, but a source, transform, or schema failure does not
+create Paimon data files. Source size, modification time, and ROS2 directory
+members are compared during the call. These checks detect observable changes;
+without a versioned URI they are not a transactional snapshot of an object
+store directory.
+
+Use `RosbagStagingConfig` to choose the local temporary directory, reserve free
+space with `min_free_bytes`, cap serial temporary usage with `max_bytes`, and
+set the remote copy chunk size with `copy_buffer_bytes`. In Ray mode,
+`max_bytes` applies to each worker's raw ROSBag staging; Arrow output capacity
+is controlled by Ray object-store and spill settings.
+
+For Ray, install `pypaimon[ray,rosbag]` and call
+`pypaimon.ray.load_from_rosbag`. The driver discovers manifests, each worker
+materializes and validates complete sources locally, and the transformed Ray
+Dataset is explicitly `materialize()`d before `write_paimon`. Success uses one
+coordinator commit. Ray may retry a transform, so transforms must be
+deterministic and free of non-idempotent external side effects.
+
+Both APIs are append-only and not retry-safe after a commit exception. A
+successful call creates one snapshot; write-stage failures may leave
uncommitted
+orphan files for normal Paimon cleanup.
+
## Load LeRobot Dataset v3
`load_from_lerobot` imports a local directory, FileIO URI, or Hugging Face
diff --git a/paimon-python/README.md b/paimon-python/README.md
index a008dbe260..75ee07090d 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -131,6 +131,69 @@ source ledger, skip files, or detect drift. Repeating the
same call appends the
rows again. It is not retry-safe because an exception from the commit can have
an unknown result; inspect table state before deciding whether to retry.
+# ROSBag to multimodal tables
+
+ROSBag loading requires Python 3.10 or newer:
+
+```commandline
+pip install 'pypaimon[rosbag]'
+```
+
+Create the target table, then map ROS messages with a user transform:
+
+```python
+import pyarrow as pa
+
+
+schema = pa.schema([
+ pa.field("source", pa.string(), nullable=False),
+ pa.field("timestamp", pa.int64(), nullable=False),
+ pa.field("value", pa.string(), nullable=False),
+])
+connection.create_table("messages", schema=schema)
+
+
+def transform(reader, source):
+ rows = []
+ for connection, timestamp, rawdata in reader.messages():
+ message = reader.deserialize(rawdata, connection.msgtype)
+ rows.append({
+ "source": source.name,
+ "timestamp": timestamp,
+ "value": message.data,
+ })
+ return pa.Table.from_pylist(rows)
+
+result = connection.load_from_rosbag(
+ "messages",
+ "s3://robot-data/recordings",
+ transform=transform,
+ source_options={"fs.s3.endpoint": "https://s3.example.com"},
+)
+```
+
+ROS1 `.bag`, ROS2 SQLite3/MCAP directories, and standalone ROS2 `.mcap`
+files are supported. OSS, S3, HDFS, ViewFS, and GCS URI sources use FileIO
+and are copied in bounded chunks to a local temporary directory because
+`rosbags` requires local paths. Standalone `.db3` files are rejected by
+default; `allow_storage_fragment=True` imports the one SQLite fragment without
+claiming that the complete recording is present.
+
+Every source is scanned to EOF before its transform runs. Transform output is
+strictly checked against the target Arrow schema and stored in a temporary
+Arrow IPC file. Paimon writers are created only after every source passes, so
+source, transform, and schema errors do not create Paimon data files. This
+front-loaded validation reads each recording twice and requires temporary disk
+space. A successful call commits all sources in one snapshot.
+
+Ray uses the same validation contract. Install both extras and call
+`pypaimon.ray.load_from_rosbag`; transformed output is fully materialized in
+Ray before `write_paimon` starts:
+
+```commandline
+pip install 'pypaimon[ray,rosbag]'
+```
+
# HDFS without a local Hadoop install
`pypaimon` supports HDFS through a pure-protocol client based on
diff --git a/paimon-python/pypaimon/multimodal/__init__.py
b/paimon-python/pypaimon/multimodal/__init__.py
index b669267b41..53717d0155 100644
--- a/paimon-python/pypaimon/multimodal/__init__.py
+++ b/paimon-python/pypaimon/multimodal/__init__.py
@@ -29,6 +29,11 @@ from pypaimon.multimodal.hdf5 import (
Hdf5File,
Hdf5LoadResult,
)
+from pypaimon.multimodal.rosbag import (
+ RosbagLoadResult,
+ RosbagSource,
+ RosbagStagingConfig,
+)
from pypaimon.multimodal.table import (
MultimodalTable,
TextRoute,
@@ -56,6 +61,9 @@ __all__ = [
"NoSuchKey",
"ObjectInfo",
"PutObjectResult",
+ "RosbagLoadResult",
+ "RosbagSource",
+ "RosbagStagingConfig",
"TextRoute",
"VectorRoute",
"VideoFrameCollator",
diff --git a/paimon-python/pypaimon/multimodal/connection.py
b/paimon-python/pypaimon/multimodal/connection.py
index 3c55032b84..8f51cdaf97 100644
--- a/paimon-python/pypaimon/multimodal/connection.py
+++ b/paimon-python/pypaimon/multimodal/connection.py
@@ -136,6 +136,30 @@ class MultimodalConnection:
source_options=source_options,
)
+ def load_from_rosbag(
+ self,
+ table_name: str,
+ paths,
+ *,
+ transform,
+ default_typestore=None,
+ typestore_factory=None,
+ source_options=None,
+ staging=None,
+ allow_storage_fragment: bool = False):
+ """Validate and append ROS1/ROS2 transforms in one commit."""
+ from pypaimon.multimodal.rosbag import load_from_rosbag
+ return load_from_rosbag(
+ self.get_table(table_name),
+ paths,
+ transform=transform,
+ default_typestore=default_typestore,
+ typestore_factory=typestore_factory,
+ source_options=source_options,
+ staging=staging,
+ allow_storage_fragment=allow_storage_fragment,
+ )
+
def drop_table(self, name: str, ignore_if_not_exists: bool = False):
self.catalog.drop_table(
self._identifier(name),
diff --git a/paimon-python/pypaimon/multimodal/hdf5.py
b/paimon-python/pypaimon/multimodal/hdf5.py
index 25583726cd..e050d72b67 100644
--- a/paimon-python/pypaimon/multimodal/hdf5.py
+++ b/paimon-python/pypaimon/multimodal/hdf5.py
@@ -17,13 +17,12 @@
"""Strict append-only ingestion from seekable HDF5 sources."""
import os
-import re
import sys
from contextlib import closing
from dataclasses import dataclass
-from pathlib import Path, PurePosixPath, PureWindowsPath
+from pathlib import Path, PurePosixPath
from typing import Callable, Mapping, Optional
-from urllib.parse import quote, unquote, urlparse, urlunparse
+from urllib.parse import unquote, urlparse
import pyarrow as pa
import pyarrow.fs as pafs
@@ -31,9 +30,11 @@ import pyarrow.fs as pafs
from pypaimon.common.options import Options
from pypaimon.filesystem.local_file_io import _file_uri_path
from pypaimon.filesystem.pyarrow_file_io import LegacyOssDirectoryListingError
-from pypaimon.filesystem.resolving_file_io import ResolvingFileIO
from pypaimon.multimodal.arrow_utils import strict_arrow_table
from pypaimon.multimodal.source_utils import (
+ _SourceFileIO,
+ _normalize_source_path,
+ _qualified_status_path,
_source_path_text,
_validated_source_options,
_validate_source_kerberos,
@@ -43,6 +44,7 @@ from pypaimon.write.commit_callback import CommitCallback
_HDF5_SUFFIXES = (".h5", ".hdf5")
+_Hdf5SourceFileIO = _SourceFileIO
@dataclass(frozen=True)
@@ -98,38 +100,6 @@ class _SnapshotRecorder(CommitCallback):
self.snapshot_id = context.snapshot.id
-class _Hdf5SourceFileIO:
- """Resolve HDF5 URIs while keeping decoding local to external sources."""
-
- def __init__(self, options):
- self._resolver = ResolvingFileIO(options)
-
- def _resolve(self, path):
- file_io = self._resolver._get_fileio(path)
- native_path = file_io.to_filesystem_path(path)
- if urlparse(path).scheme.lower() != "file":
- native_path = unquote(native_path)
- return file_io, native_path
-
- def get_file_status(self, path):
- file_io, native_path = self._resolve(path)
- return file_io.get_file_status(native_path)
-
- def list_status(self, path):
- file_io, native_path = self._resolve(path)
- return file_io.list_status(native_path)
-
- def new_input_stream(self, path):
- file_io, native_path = self._resolve(path)
- return file_io.new_input_stream(native_path)
-
- def to_filesystem_path(self, path):
- return self._resolve(path)[1]
-
- def close(self):
- self._resolver.close()
-
-
def load_from_hdf5(
table,
paths,
@@ -384,61 +354,6 @@ def _raise_legacy_directory_listing_error(path, error):
% path) from error
-def _normalize_source_path(value):
- path = _source_path_text(value)
- parsed = urlparse(path)
- if _is_windows_drive_path(parsed):
- windows_path = PureWindowsPath(path)
- if not windows_path.is_absolute():
- raise ValueError("Windows source paths must be absolute: %s" %
path)
- return "file:///%s" % quote(windows_path.as_posix(), safe="/:")
- if not parsed.scheme:
- return Path(path).expanduser().resolve().as_uri()
- return _quote_uri_path(path)
-
-
-def _quote_uri_path(uri):
- match = re.match(r"^([A-Za-z][A-Za-z0-9+.-]*://[^/]*)(.*)$", uri)
- if match is None:
- return uri
- return match.group(1) + quote(match.group(2), safe="/:%")
-
-
-def _qualified_status_path(parent_path, status):
- status_path = str(status.path)
- status_uri = urlparse(status_path)
- if status_uri.scheme and not _is_windows_drive_path(status_uri):
- return _quote_uri_path(status_path)
-
- parent_uri = urlparse(parent_path)
- scheme = parent_uri.scheme.lower()
- if scheme == "file":
- return _normalize_source_path(status_path)
- if not scheme or _is_windows_drive_path(parent_uri):
- return _normalize_source_path(status_path)
-
- if scheme in ("hdfs", "viewfs"):
- return urlunparse((
- scheme,
- parent_uri.netloc,
- quote("/" + status_path.lstrip("/"), safe="/:"),
- "",
- "",
- "",
- ))
-
- key = status_path.lstrip("/")
- if parent_uri.netloc and not (
- key == parent_uri.netloc
- or key.startswith(parent_uri.netloc + "/")):
- key = parent_uri.netloc + "/" + key
- return "%s://%s" % (scheme, quote(key, safe="/:"))
-
-
-def _is_windows_drive_path(parsed):
- return len(parsed.scheme) == 1 and not parsed.netloc
-
-
def _hdf5_suffix(path):
parsed = urlparse(path)
return PurePosixPath(unquote(parsed.path)).suffix.lower() in _HDF5_SUFFIXES
diff --git a/paimon-python/pypaimon/multimodal/lerobot/source.py
b/paimon-python/pypaimon/multimodal/lerobot/source.py
index eaf995d69e..8d71e3eecb 100644
--- a/paimon-python/pypaimon/multimodal/lerobot/source.py
+++ b/paimon-python/pypaimon/multimodal/lerobot/source.py
@@ -31,8 +31,8 @@ import pyarrow.parquet as pq
from pypaimon.common.options import Options
from pypaimon.filesystem.pyarrow_file_io import LegacyOssDirectoryListingError
-from pypaimon.multimodal.hdf5 import (
- _Hdf5SourceFileIO,
+from pypaimon.multimodal.source_utils import (
+ _SourceFileIO,
_normalize_source_path,
_qualified_status_path,
)
@@ -80,7 +80,7 @@ def _resolved_source(source, source_options):
return
source_uri = _normalize_source_path(value).rstrip("/")
- source_file_io = _Hdf5SourceFileIO(Options(source_options))
+ source_file_io = _SourceFileIO(Options(source_options))
try:
try:
status = source_file_io.get_file_status(source_uri)
diff --git a/paimon-python/pypaimon/multimodal/rosbag/__init__.py
b/paimon-python/pypaimon/multimodal/rosbag/__init__.py
new file mode 100644
index 0000000000..5fbd2ad717
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/rosbag/__init__.py
@@ -0,0 +1,31 @@
+# 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.
+
+"""ROSBag ingestion for multimodal tables."""
+
+from pypaimon.multimodal.rosbag.api import (
+ RosbagLoadResult,
+ RosbagStagingConfig,
+ load_from_rosbag,
+)
+from pypaimon.multimodal.rosbag.source import RosbagSource
+
+__all__ = [
+ "RosbagLoadResult",
+ "RosbagSource",
+ "RosbagStagingConfig",
+ "load_from_rosbag",
+]
diff --git a/paimon-python/pypaimon/multimodal/rosbag/api.py
b/paimon-python/pypaimon/multimodal/rosbag/api.py
new file mode 100644
index 0000000000..03e64a8209
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/rosbag/api.py
@@ -0,0 +1,160 @@
+# 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.
+
+"""Public ROSBag ingestion API."""
+
+import os
+import sys
+from dataclasses import dataclass
+from typing import Callable, Mapping, Optional
+
+from pypaimon.common.options import Options
+from pypaimon.multimodal.rosbag.loader import _load_rosbag_manifests
+from pypaimon.multimodal.rosbag.source import _discover_rosbag_sources
+from pypaimon.multimodal.source_utils import (
+ _SourceFileIO,
+ _validated_source_options,
+ _validate_source_kerberos,
+)
+
+
+@dataclass(frozen=True)
+class RosbagStagingConfig:
+ """Local temporary-space controls for ROSBag ingestion.
+
+ In serial mode, ``max_bytes`` covers remote source copies and validated
+ Arrow IPC output. Ray applies it to each worker's remote source copy;
+ Ray object-store capacity is configured separately.
+ """
+
+ directory: Optional[str] = None
+ max_bytes: Optional[int] = None
+ min_free_bytes: int = 1 << 30
+ copy_buffer_bytes: int = 8 << 20
+
+
+@dataclass(frozen=True)
+class RosbagLoadResult:
+ """Counts and committed snapshot for one ROSBag load.
+
+ ``batch_count`` is the number of transform outputs in serial mode and
+ ``None`` in Ray mode, where Ray may reorganize output blocks.
+ """
+
+ source_count: int
+ batch_count: Optional[int]
+ row_count: int
+ snapshot_id: Optional[int]
+
+
+def load_from_rosbag(
+ table,
+ paths,
+ *,
+ transform: Callable,
+ default_typestore=None,
+ typestore_factory=None,
+ source_options: Optional[Mapping[str, object]] = None,
+ staging: Optional[RosbagStagingConfig] = None,
+ allow_storage_fragment: bool = False):
+ """Validate and append ROS1/ROS2 transforms in one commit.
+
+ ``transform(reader, source)`` returns Arrow tables or record batches.
+ Every source and transform result is validated before a Paimon writer is
+ created. The operation is append-only and a commit exception can have an
+ unknown result, so callers must inspect table state before retrying.
+ """
+ if sys.version_info < (3, 10):
+ raise RuntimeError(
+ "load_from_rosbag requires Python 3.10 or newer; the rosbag "
+ "extra is not available on older Python versions.")
+ if not callable(transform):
+ raise ValueError("transform must be callable.")
+ if default_typestore is not None and typestore_factory is not None:
+ raise ValueError(
+ "default_typestore and typestore_factory are mutually exclusive.")
+ if typestore_factory is not None and not callable(typestore_factory):
+ raise ValueError("typestore_factory must be callable.")
+ if staging is None:
+ staging = RosbagStagingConfig()
+ if not isinstance(staging, RosbagStagingConfig):
+ raise ValueError("staging must be a RosbagStagingConfig.")
+ _validate_staging_config(staging)
+
+ path_values = _path_values(paths)
+ options = _validated_source_options(source_options)
+ _validate_source_kerberos(path_values, options, source_name="ROSBag")
+ source_file_io = _SourceFileIO(Options(options))
+ try:
+ manifests = _discover_rosbag_sources(
+ path_values,
+ source_file_io,
+ allow_storage_fragment=allow_storage_fragment,
+ )
+ if not manifests:
+ return RosbagLoadResult(0, 0, 0, None)
+ try:
+ from rosbags.highlevel import AnyReader
+ except ImportError as error:
+ raise ImportError(
+ "load_from_rosbag requires rosbags; install "
+ "'pypaimon[rosbag]'.") from error
+ return _load_rosbag_manifests(
+ table,
+ manifests,
+ transform,
+ source_file_io,
+ AnyReader,
+ default_typestore=default_typestore,
+ typestore_factory=typestore_factory,
+ staging=staging,
+ )
+ finally:
+ source_file_io.close()
+
+
+def _path_values(paths):
+ if isinstance(paths, (str, os.PathLike)):
+ return [paths]
+ if isinstance(paths, bytes):
+ raise ValueError("paths must be a path or an iterable of paths.")
+ try:
+ return list(paths)
+ except TypeError as error:
+ raise ValueError(
+ "paths must be a path or an iterable of paths.") from error
+
+
+def _validate_staging_config(staging):
+ if (
+ staging.max_bytes is not None
+ and (
+ isinstance(staging.max_bytes, bool)
+ or not isinstance(staging.max_bytes, int)
+ or staging.max_bytes < 0)):
+ raise ValueError("staging.max_bytes must be a non-negative integer.")
+ if (
+ isinstance(staging.min_free_bytes, bool)
+ or not isinstance(staging.min_free_bytes, int)
+ or staging.min_free_bytes < 0):
+ raise ValueError(
+ "staging.min_free_bytes must be a non-negative integer.")
+ if (
+ isinstance(staging.copy_buffer_bytes, bool)
+ or not isinstance(staging.copy_buffer_bytes, int)
+ or staging.copy_buffer_bytes <= 0):
+ raise ValueError(
+ "staging.copy_buffer_bytes must be a positive integer.")
diff --git a/paimon-python/pypaimon/multimodal/rosbag/loader.py
b/paimon-python/pypaimon/multimodal/rosbag/loader.py
new file mode 100644
index 0000000000..7f123948a2
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/rosbag/loader.py
@@ -0,0 +1,377 @@
+# 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 ROSBag transforms before creating a Paimon writer."""
+
+import io
+from contextlib import contextmanager
+from pathlib import Path
+import shutil
+from tempfile import TemporaryDirectory
+from urllib.parse import quote, urlparse
+
+import pyarrow as pa
+
+from pypaimon.multimodal.arrow_utils import strict_arrow_table
+from pypaimon.multimodal.rosbag.source import RosbagSource
+from pypaimon.multimodal.rosbag.staging import (
+ _materialized_rosbag,
+ _verify_manifest_members,
+)
+from pypaimon.multimodal.table import _target_schema
+from pypaimon.write.commit_callback import CommitCallback
+
+
+class _SnapshotRecorder(CommitCallback):
+
+ def __init__(self):
+ self.snapshot_id = None
+
+ def call(self, context):
+ self.snapshot_id = context.snapshot.id
+
+
+class _BoundedStagingOutput(io.RawIOBase):
+ """File output that rejects a write before it exceeds its byte limit."""
+
+ def __init__(self, path, max_bytes):
+ super().__init__()
+ self._output = Path(path).open("wb")
+ self._max_bytes = max_bytes
+ self._reserved_bytes = 0
+
+ def set_reserved_bytes(self, byte_count):
+ required_bytes = self._output.tell() + byte_count
+ if required_bytes > self._max_bytes:
+ self._raise_limit(required_bytes)
+ self._reserved_bytes = byte_count
+
+ def writable(self):
+ return True
+
+ def write(self, value):
+ required_bytes = (
+ self._output.tell() + len(value) + self._reserved_bytes)
+ if required_bytes > self._max_bytes:
+ self._raise_limit(required_bytes)
+ return self._output.write(value)
+
+ def _raise_limit(self, required_bytes):
+ raise ValueError(
+ "ROSBag staging exceeds configured limit of %d bytes "
+ "(requires at least %d bytes)."
+ % (self._max_bytes, required_bytes))
+
+ def tell(self):
+ return self._output.tell()
+
+ def flush(self):
+ if not self._output.closed:
+ self._output.flush()
+
+ def close(self):
+ if not self.closed:
+ try:
+ super().close()
+ finally:
+ self._output.close()
+
+
+@contextmanager
+def _staging_sink(path, max_bytes):
+ if max_bytes is None:
+ with pa.OSFile(str(path), "wb") as sink:
+ yield sink, None
+ return
+ with _BoundedStagingOutput(path, max_bytes) as output:
+ with pa.PythonFile(output, mode="w") as sink:
+ yield sink, output
+
+
+def _load_rosbag_manifests(
+ table,
+ manifests,
+ transform,
+ source_file_io,
+ AnyReader,
+ *,
+ default_typestore,
+ typestore_factory,
+ staging):
+ from pypaimon.multimodal.rosbag.api import RosbagLoadResult
+
+ target_schema = _target_schema(table.raw_table)
+ with TemporaryDirectory(
+ prefix="pypaimon_rosbag_", dir=staging.directory) as temp_dir:
+ _check_staging_free_space(temp_dir, staging)
+ spool_path = Path(temp_dir) / "validated.arrow"
+ batch_count = 0
+ row_count = 0
+ with _staging_sink(
+ spool_path, staging.max_bytes) as (sink, bounded_output):
+ with pa.ipc.new_file(sink, target_schema) as spool:
+ for source_index, manifest in enumerate(manifests):
+ source_staging_bytes = _source_staging_bytes(manifest)
+ _check_staging_bytes(
+ sink.tell() + source_staging_bytes, staging)
+ if bounded_output is not None:
+ bounded_output.set_reserved_bytes(
+ source_staging_bytes)
+ try:
+ for table_value in _transform_rosbag_manifest(
+ manifest,
+ transform,
+ source_file_io,
+ AnyReader,
+ target_schema,
+ default_typestore=default_typestore,
+ typestore_factory=typestore_factory,
+ staging=staging,
+ staging_root=Path(temp_dir) /
+ ("source-%06d" % source_index),
+ batch_index=batch_count,
+ base_staging_bytes=sink.tell()):
+ batch_count += 1
+ row_count += table_value.num_rows
+ for record_batch in table_value.to_batches():
+ spool.write_batch(record_batch)
+ _check_staging_bytes(
+ sink.tell() + source_staging_bytes,
+ staging)
+ finally:
+ if bounded_output is not None:
+ bounded_output.set_reserved_bytes(0)
+ _check_staging_bytes(sink.tell(), staging)
+
+ for manifest in manifests:
+ _verify_manifest_members(manifest, source_file_io)
+ snapshot_id = _write_spool(table, spool_path)
+ return RosbagLoadResult(
+ source_count=len(manifests),
+ batch_count=batch_count,
+ row_count=row_count,
+ snapshot_id=snapshot_id,
+ )
+
+
+def _source_staging_bytes(manifest):
+ if urlparse(manifest.uri).scheme.lower() == "file":
+ return 0
+ return sum(member.size for member in manifest.members)
+
+
+def _check_staging_bytes(actual_bytes, staging):
+ if staging.max_bytes is not None and actual_bytes > staging.max_bytes:
+ raise ValueError(
+ "ROSBag staging exceeds configured limit of %d bytes "
+ "(requires at least %d bytes)."
+ % (staging.max_bytes, actual_bytes))
+
+
+def _check_staging_free_space(directory, staging):
+ if shutil.disk_usage(str(directory)).free < staging.min_free_bytes:
+ raise ValueError(
+ "ROSBag staging has less than %d free bytes at %s."
+ % (staging.min_free_bytes, directory))
+
+
+def _transform_rosbag_manifest(
+ manifest,
+ transform,
+ source_file_io,
+ AnyReader,
+ target_schema,
+ *,
+ default_typestore,
+ typestore_factory,
+ staging,
+ staging_root,
+ batch_index=0,
+ base_staging_bytes=0):
+ """Yield validated Arrow tables for one fully preflighted source."""
+ produced_rows = 0
+ with _materialized_rosbag(
+ manifest,
+ source_file_io,
+ staging_root,
+ staging,
+ base_staging_bytes=base_staging_bytes) as local_path:
+ if manifest.format == "ros2_sqlite3_fragment":
+ _validate_sqlite_fragment(local_path, manifest.uri)
+ _preflight_reader(
+ manifest,
+ local_path,
+ AnyReader,
+ default_typestore,
+ typestore_factory,
+ )
+ source = RosbagSource(
+ uri=manifest.uri,
+ local_path=local_path,
+ format=manifest.format,
+ )
+ typestore = _new_typestore(default_typestore, typestore_factory)
+ with AnyReader(
+ [local_path], default_typestore=typestore) as reader:
+ transformed = transform(reader, source)
+ batches = None
+ try:
+ batches = _arrow_batches(transformed)
+ for index, value in enumerate(batches, start=batch_index):
+ table_value = strict_arrow_table(
+ value,
+ target_schema,
+ manifest.uri,
+ index,
+ "ROSBag",
+ )
+ produced_rows += table_value.num_rows
+ yield table_value
+ finally:
+ _close_transform_iterator(
+ batches if batches is not None else transformed)
+ if produced_rows == 0:
+ raise ValueError(
+ "ROSBag source %s produced no rows." % manifest.uri)
+
+
+def _validate_sqlite_fragment(local_path, source_uri):
+ import sqlite3
+
+ database_uri = "file:%s?mode=ro&immutable=1" % quote(
+ str(local_path), safe="/")
+ try:
+ connection = sqlite3.connect(database_uri, uri=True)
+ try:
+ results = [row[0] for row in connection.execute(
+ "PRAGMA quick_check")]
+ finally:
+ connection.close()
+ except sqlite3.DatabaseError as error:
+ raise ValueError(
+ "SQLite integrity check failed for ROSBag fragment %s: %s"
+ % (source_uri, error)) from error
+ if results != ["ok"]:
+ raise ValueError(
+ "SQLite integrity check failed for ROSBag fragment %s: %s"
+ % (source_uri, "; ".join(results)))
+
+
+def _preflight_reader(
+ manifest,
+ local_path,
+ AnyReader,
+ default_typestore,
+ typestore_factory):
+ typestore = _new_typestore(default_typestore, typestore_factory)
+ with AnyReader([local_path], default_typestore=typestore) as reader:
+ topic_counts = {}
+ actual_count = 0
+ for connection, _, _ in reader.messages():
+ actual_count += 1
+ topic_counts[connection.id] = (
+ topic_counts.get(connection.id, 0) + 1)
+ declared_count = reader.message_count
+ connections = list(reader.connections)
+ if actual_count != declared_count:
+ raise ValueError(
+ "ROSBag source %s declared %d messages but %d were readable."
+ % (manifest.uri, declared_count, actual_count))
+ if (
+ manifest.expected_message_count is not None
+ and actual_count != manifest.expected_message_count):
+ raise ValueError(
+ "ROSBag source %s metadata declares %d messages but %d were "
+ "readable."
+ % (
+ manifest.uri,
+ manifest.expected_message_count,
+ actual_count,
+ ))
+ for connection in connections:
+ actual_topic_count = topic_counts.get(connection.id, 0)
+ if actual_topic_count != connection.msgcount:
+ raise ValueError(
+ "ROSBag source %s topic %s declares %d messages but %d "
+ "were readable."
+ % (
+ manifest.uri,
+ connection.topic,
+ connection.msgcount,
+ actual_topic_count,
+ ))
+
+
+def _new_typestore(default_typestore, typestore_factory):
+ if typestore_factory is not None:
+ return typestore_factory()
+ return default_typestore
+
+
+def _arrow_batches(transformed):
+ if isinstance(transformed, (pa.Table, pa.RecordBatch)):
+ return iter([transformed])
+ if transformed is None or isinstance(transformed, (str, bytes, dict)):
+ raise ValueError(
+ "ROSBag transform must return Arrow data or an iterable of "
+ "Arrow data.")
+ try:
+ return iter(transformed)
+ except TypeError as error:
+ raise ValueError(
+ "ROSBag transform must return Arrow data or an iterable of "
+ "Arrow data.") from error
+
+
+def _close_transform_iterator(iterator):
+ close = getattr(iterator, "close", None)
+ if close is not None:
+ close()
+
+
+def _write_spool(table, spool_path):
+ write_builder = table.raw_table.new_batch_write_builder()
+ table_write = None
+ table_commit = None
+ commit_started = False
+ snapshot_recorder = _SnapshotRecorder()
+ try:
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_commit.add_commit_callback(snapshot_recorder)
+ with pa.memory_map(str(spool_path), "r") as source:
+ spool = pa.ipc.open_file(source)
+ for index in range(spool.num_record_batches):
+ table_write.write_arrow(pa.Table.from_batches([
+ spool.get_batch(index)]))
+ commit_messages = table_write.prepare_commit()
+ commit_started = True
+ table_commit.commit(commit_messages)
+ if snapshot_recorder.snapshot_id is None:
+ raise RuntimeError(
+ "ROSBag append committed without reporting a snapshot id.")
+ return snapshot_recorder.snapshot_id
+ except BaseException:
+ if table_write is not None and not commit_started:
+ table_write.abort()
+ raise
+ finally:
+ try:
+ if table_write is not None:
+ table_write.close()
+ finally:
+ if table_commit is not None:
+ table_commit.close()
diff --git a/paimon-python/pypaimon/multimodal/rosbag/source.py
b/paimon-python/pypaimon/multimodal/rosbag/source.py
new file mode 100644
index 0000000000..6d43c6d775
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/rosbag/source.py
@@ -0,0 +1,371 @@
+# 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.
+
+"""ROSBag source descriptions and discovery."""
+
+from contextlib import closing
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from typing import Optional, Tuple
+from urllib.parse import quote, unquote, urlparse
+
+import pyarrow.fs as pafs
+
+from pypaimon.multimodal.source_utils import (
+ _normalize_source_path,
+ _qualified_status_path,
+)
+
+
+@dataclass(frozen=True)
+class RosbagSourceMember:
+ """One physical file belonging to a logical ROSBag source."""
+
+ uri: str
+ relative_path: str
+ size: int
+ mtime_ns: Optional[int]
+
+
+@dataclass(frozen=True)
+class RosbagSourceManifest:
+ """Immutable physical-file manifest for one logical ROSBag source."""
+
+ uri: str
+ format: str
+ members: Tuple[RosbagSourceMember, ...]
+ expected_message_count: Optional[int] = None
+ directory_members: Optional[Tuple[str, ...]] = None
+
+
+@dataclass(frozen=True)
+class RosbagSource:
+ """Read context supplied to a ROSBag transform.
+
+ ``format`` is one of ``ros1``, ``ros2_mcap``, ``ros2_sqlite3``, or
+ ``ros2_sqlite3_fragment``.
+ """
+
+ uri: str
+ local_path: Path
+ format: str
+
+ @property
+ def name(self):
+ """Base name of the original source URI."""
+ return PurePosixPath(unquote(urlparse(self.uri).path)).name
+
+ @property
+ def stem(self):
+ """Base name without the final suffix."""
+ return PurePosixPath(self.name).stem
+
+ @property
+ def is_remote(self):
+ """Whether the original source requires FileIO materialization."""
+ return urlparse(self.uri).scheme.lower() != "file"
+
+
+def _discover_rosbag_sources(
+ paths, source_file_io, allow_storage_fragment=False):
+ """Discover normalized logical ROSBag sources."""
+ manifests = {}
+ visited_directories = set()
+ for value in paths:
+ path = _normalize_source_path(value)
+ try:
+ status = source_file_io.get_file_status(path)
+ except FileNotFoundError as not_found_error:
+ try:
+ children = source_file_io.list_status(path)
+ except (FileNotFoundError, KeyError):
+ raise ValueError(
+ "ROSBag source path does not exist: %s" % path
+ ) from not_found_error
+ if not children:
+ raise ValueError(
+ "ROSBag source path does not exist: %s" % path
+ ) from not_found_error
+ status = pafs.FileInfo(path, pafs.FileType.Directory)
+ _discover_status(
+ path,
+ status,
+ source_file_io,
+ manifests,
+ visited_directories,
+ allow_storage_fragment,
+ explicit=True,
+ )
+ return [manifests[key] for key in sorted(manifests)]
+
+
+def _discover_status(
+ parent_path,
+ status,
+ source_file_io,
+ manifests,
+ visited_directories,
+ allow_storage_fragment,
+ *,
+ explicit):
+ qualified = _qualified_status_path(parent_path, status)
+ if status.type == pafs.FileType.File:
+ manifest = _file_manifest(
+ qualified,
+ status,
+ allow_storage_fragment,
+ explicit=explicit,
+ )
+ if manifest is not None:
+ manifests[manifest.uri] = manifest
+ return
+ if status.type != pafs.FileType.Directory:
+ if explicit:
+ raise ValueError("Unsupported ROSBag source: %s" % qualified)
+ return
+ if qualified in visited_directories:
+ return
+ visited_directories.add(qualified)
+
+ metadata_uri = _join_source_path(qualified, "metadata.yaml")
+ try:
+ source_file_io.get_file_status(metadata_uri)
+ except FileNotFoundError:
+ children = source_file_io.list_status(qualified)
+ if explicit and any(
+ _storage_member_status(qualified, child)
+ for child in children):
+ raise ValueError(
+ "ROS2 recording directory is missing metadata.yaml: %s"
+ % qualified)
+ for child in children:
+ _discover_status(
+ qualified,
+ child,
+ source_file_io,
+ manifests,
+ visited_directories,
+ allow_storage_fragment,
+ explicit=False,
+ )
+ return
+
+ manifest = _ros2_directory_manifest(qualified, source_file_io)
+ manifests[manifest.uri] = manifest
+
+
+def _storage_member_status(parent, status):
+ if status.type != pafs.FileType.File:
+ return False
+ path = _qualified_status_path(parent, status)
+ suffix = PurePosixPath(unquote(urlparse(path).path)).suffix.lower()
+ return suffix in (".db3", ".mcap")
+
+
+def _file_manifest(path, status, allow_storage_fragment, *, explicit):
+ decoded_path = unquote(urlparse(path).path)
+ if decoded_path.lower().endswith(".bag.active"):
+ raise ValueError(
+ "ROS1 recording appears to be still being written: %s" % path)
+ suffix = PurePosixPath(decoded_path).suffix.lower()
+ formats = {".bag": "ros1", ".mcap": "ros2_mcap"}
+ if suffix == ".db3" and not allow_storage_fragment:
+ raise ValueError(
+ "Standalone ROS2 .db3 files may be incomplete recording "
+ "fragments; pass allow_storage_fragment=True to import the "
+ "file without recording completeness guarantees: %s" % path)
+ if suffix == ".db3" and explicit:
+ formats[suffix] = "ros2_sqlite3_fragment"
+ if suffix not in formats:
+ if explicit:
+ raise ValueError("Unsupported ROSBag source file: %s" % path)
+ return None
+ member = RosbagSourceMember(
+ uri=path,
+ relative_path=PurePosixPath(decoded_path).name,
+ size=status.size,
+ mtime_ns=_status_mtime_ns(status),
+ )
+ return RosbagSourceManifest(
+ uri=path,
+ format=formats[suffix],
+ members=(member,),
+ )
+
+
+def _ros2_directory_manifest(directory, source_file_io):
+ metadata_uri = _join_source_path(directory, "metadata.yaml")
+ try:
+ metadata_status = source_file_io.get_file_status(metadata_uri)
+ except FileNotFoundError as error:
+ raise ValueError(
+ "ROS2 recording is missing metadata.yaml: %s" % directory
+ ) from error
+ if metadata_status.type != pafs.FileType.File:
+ raise ValueError(
+ "ROS2 metadata.yaml is not a file: %s" % metadata_uri)
+
+ metadata = _read_ros2_metadata(metadata_uri, source_file_io)
+ storage_identifier = metadata.get("storage_identifier")
+ formats = {
+ "sqlite3": "ros2_sqlite3",
+ "mcap": "ros2_mcap",
+ }
+ if storage_identifier not in formats:
+ raise ValueError(
+ "Unsupported ROS2 storage identifier %r in %s."
+ % (storage_identifier, metadata_uri))
+
+ members = [
+ _source_member(
+ "metadata.yaml",
+ metadata_status,
+ directory,
+ )
+ ]
+ normalized_members = set()
+ for value in metadata.get("relative_file_paths", []):
+ relative_path = _safe_ros2_member_path(value)
+ collision_key = relative_path.casefold()
+ if collision_key in normalized_members:
+ raise ValueError(
+ "duplicate ROS2 metadata relative_file_paths entry: %r"
+ % value)
+ normalized_members.add(collision_key)
+ member_uri = _join_source_path(directory, relative_path)
+ if storage_identifier == "sqlite3":
+ _reject_sqlite_sidecars(member_uri, source_file_io)
+ try:
+ status = source_file_io.get_file_status(member_uri)
+ except FileNotFoundError as error:
+ raise ValueError(
+ "ROS2 recording member is missing: %s" % member_uri
+ ) from error
+ if status.type != pafs.FileType.File:
+ raise ValueError(
+ "ROS2 recording member is not a file: %s" % member_uri)
+ members.append(_source_member(
+ relative_path,
+ status,
+ directory,
+ ))
+
+ return RosbagSourceManifest(
+ uri=directory,
+ format=formats[storage_identifier],
+ members=tuple(members),
+ expected_message_count=int(metadata.get("message_count", 0)),
+ directory_members=_listed_directory_members(
+ directory, source_file_io),
+ )
+
+
+def _read_ros2_metadata(metadata_uri, source_file_io):
+ try:
+ from ruamel.yaml import YAML
+ from ruamel.yaml.error import YAMLError
+ except ImportError as error:
+ raise ImportError(
+ "ROSBag loading requires rosbags; install 'pypaimon[rosbag]'."
+ ) from error
+ with closing(source_file_io.new_input_stream(metadata_uri)) as stream:
+ content = stream.read()
+ try:
+ document = YAML(typ="safe").load(content)
+ return document["rosbag2_bagfile_information"]
+ except (KeyError, TypeError, ValueError, YAMLError) as error:
+ raise ValueError(
+ "Cannot read ROS2 metadata: %s" % metadata_uri) from error
+
+
+def _listed_directory_members(directory, source_file_io):
+ return tuple(sorted(
+ _qualified_status_path(directory, status)
+ for status in source_file_io.list_status(directory)
+ ))
+
+
+def _source_member(relative_path, status, parent):
+ return RosbagSourceMember(
+ uri=_qualified_status_path(parent, status),
+ relative_path=relative_path,
+ size=status.size,
+ mtime_ns=_status_mtime_ns(status),
+ )
+
+
+def _join_source_path(directory, relative_path):
+ return "%s/%s" % (
+ directory.rstrip("/"), quote(relative_path, safe="/:%"))
+
+
+def _reject_sqlite_sidecars(database_uri, source_file_io):
+ for suffix in ("-wal", "-shm", "-journal"):
+ sidecar = database_uri + suffix
+ try:
+ source_file_io.get_file_status(sidecar)
+ except FileNotFoundError:
+ continue
+ raise ValueError(
+ "ROS2 SQLite recording is not finalized or requires recovery; "
+ "found sidecar file: %s" % sidecar)
+
+
+def _safe_ros2_member_path(value):
+ if not isinstance(value, str) or not value:
+ raise ValueError(
+ "unsafe ROS2 metadata relative_file_paths entry: %r"
+ % value)
+ decoded = value
+ for _ in range(8):
+ next_value = unquote(decoded)
+ if next_value == decoded:
+ break
+ decoded = next_value
+ else:
+ raise ValueError(
+ "unsafe ROS2 metadata relative_file_paths entry: %r"
+ % value)
+ parsed = urlparse(decoded)
+ path = PurePosixPath(decoded)
+ if (
+ "\\" in decoded
+ or parsed.scheme
+ or parsed.netloc
+ or parsed.query
+ or parsed.fragment
+ or path.is_absolute()
+ or any(part in ("", ".", "..") for part in path.parts)):
+ raise ValueError(
+ "unsafe ROS2 metadata relative_file_paths entry: %r"
+ % value)
+ if len(path.parts) != 1:
+ raise ValueError(
+ "nested ROS2 metadata relative_file_paths entries are not "
+ "supported by rosbags: %r" % value)
+ return path.as_posix()
+
+
+def _status_mtime_ns(status):
+ value = getattr(status, "mtime_ns", None)
+ if value is not None:
+ return None if value < 0 else value
+ value = getattr(status, "mtime", None)
+ if value is None:
+ return None
+ if hasattr(value, "timestamp"):
+ value = value.timestamp()
+ return None if value < 0 else int(value * 1_000_000_000)
diff --git a/paimon-python/pypaimon/multimodal/rosbag/staging.py
b/paimon-python/pypaimon/multimodal/rosbag/staging.py
new file mode 100644
index 0000000000..77daf0b909
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/rosbag/staging.py
@@ -0,0 +1,158 @@
+# 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.
+
+"""Bounded local materialization for ROSBag FileIO sources."""
+
+import shutil
+from contextlib import closing, contextmanager
+from pathlib import Path
+from urllib.parse import urlparse
+
+import pyarrow.fs as pafs
+
+from pypaimon.multimodal.rosbag.source import _status_mtime_ns
+from pypaimon.multimodal.source_utils import _qualified_status_path
+
+
+@contextmanager
+def _materialized_rosbag(
+ manifest,
+ source_file_io,
+ staging_root,
+ staging,
+ base_staging_bytes=0):
+ """Yield a local file/directory and reject detected source mutations."""
+ if urlparse(manifest.uri).scheme.lower() == "file":
+ local_path = Path(source_file_io.to_filesystem_path(manifest.uri))
+ yield local_path
+ _verify_manifest_members(manifest, source_file_io)
+ return
+
+ if staging.copy_buffer_bytes <= 0:
+ raise ValueError("staging.copy_buffer_bytes must be positive.")
+ staging_root.mkdir(parents=True, exist_ok=False)
+ try:
+ local_path = _stage_remote_manifest(
+ manifest,
+ source_file_io,
+ staging_root,
+ staging,
+ base_staging_bytes=base_staging_bytes,
+ )
+ yield local_path
+ _verify_manifest_members(manifest, source_file_io)
+ finally:
+ shutil.rmtree(str(staging_root), ignore_errors=True)
+
+
+def _stage_remote_manifest(
+ manifest,
+ source_file_io,
+ staging_root,
+ staging,
+ base_staging_bytes=0):
+ if shutil.disk_usage(str(staging_root)).free < staging.min_free_bytes:
+ raise ValueError(
+ "ROSBag staging has less than %d free bytes at %s."
+ % (staging.min_free_bytes, staging_root))
+
+ expected_bytes = sum(member.size for member in manifest.members)
+ if (
+ staging.max_bytes is not None
+ and base_staging_bytes + expected_bytes > staging.max_bytes):
+ raise ValueError(
+ "ROSBag staging exceeds configured limit of %d bytes while "
+ "copying %s."
+ % (staging.max_bytes, manifest.uri))
+
+ total_copied = 0
+ for member in manifest.members:
+ destination = staging_root / member.relative_path
+ _require_contained_path(staging_root, destination)
+ destination.parent.mkdir(parents=True, exist_ok=True)
+ copied = 0
+ with closing(source_file_io.new_input_stream(member.uri)) as source:
+ with destination.open("wb") as output:
+ while True:
+ chunk = source.read(staging.copy_buffer_bytes)
+ if not chunk:
+ break
+ next_total = total_copied + len(chunk)
+ if (
+ staging.max_bytes is not None
+ and base_staging_bytes + next_total
+ > staging.max_bytes):
+ raise ValueError(
+ "ROSBag staging exceeds configured limit of %d "
+ "bytes while copying %s."
+ % (staging.max_bytes, member.uri))
+ output.write(chunk)
+ copied += len(chunk)
+ total_copied = next_total
+ if copied != member.size:
+ raise ValueError(
+ "ROSBag source %s changed or returned a short read: "
+ "expected %d bytes, copied %d."
+ % (member.uri, member.size, copied))
+
+ _verify_manifest_members(manifest, source_file_io)
+ local_path = (
+ staging_root
+ if any(member.relative_path == "metadata.yaml"
+ for member in manifest.members)
+ else staging_root / manifest.members[0].relative_path
+ )
+ return local_path
+
+
+def _verify_manifest_members(manifest, source_file_io):
+ if manifest.directory_members is not None:
+ actual_members = tuple(sorted(
+ _qualified_status_path(manifest.uri, status)
+ for status in source_file_io.list_status(manifest.uri)
+ ))
+ if actual_members != manifest.directory_members:
+ raise ValueError(
+ "ROSBag source directory members changed during ingestion: "
+ "%s" % manifest.uri)
+ for member in manifest.members:
+ try:
+ status = source_file_io.get_file_status(member.uri)
+ except FileNotFoundError as error:
+ raise ValueError(
+ "ROSBag source changed during ingestion; member disappeared: "
+ "%s" % member.uri) from error
+ if status.type != pafs.FileType.File or status.size != member.size:
+ raise ValueError(
+ "ROSBag source changed during ingestion: %s" % member.uri)
+ mtime_ns = _status_mtime_ns(status)
+ if (
+ member.mtime_ns is not None
+ and mtime_ns is not None
+ and mtime_ns != member.mtime_ns):
+ raise ValueError(
+ "ROSBag source changed during ingestion: %s" % member.uri)
+
+
+def _require_contained_path(root, destination):
+ resolved_root = root.resolve()
+ resolved_destination = destination.resolve()
+ try:
+ resolved_destination.relative_to(resolved_root)
+ except ValueError as error:
+ raise ValueError(
+ "ROSBag staging member escapes its temporary directory: %s"
+ % destination) from error
diff --git a/paimon-python/pypaimon/multimodal/source_utils.py
b/paimon-python/pypaimon/multimodal/source_utils.py
index 1e3c58014e..6eb0e27823 100644
--- a/paimon-python/pypaimon/multimodal/source_utils.py
+++ b/paimon-python/pypaimon/multimodal/source_utils.py
@@ -17,8 +17,44 @@
"""Shared validation for external multimodal sources."""
import os
+import re
+from pathlib import Path, PureWindowsPath
from typing import Mapping
-from urllib.parse import urlparse
+from urllib.parse import quote, unquote, urlparse, urlunparse
+
+from pypaimon.filesystem.resolving_file_io import ResolvingFileIO
+
+
+class _SourceFileIO:
+ """Resolve external source URIs without using target warehouse options."""
+
+ def __init__(self, options):
+ self._resolver = ResolvingFileIO(options)
+
+ def _resolve(self, path):
+ file_io = self._resolver._get_fileio(path)
+ native_path = file_io.to_filesystem_path(path)
+ if urlparse(path).scheme.lower() != "file":
+ native_path = unquote(native_path)
+ return file_io, native_path
+
+ def get_file_status(self, path):
+ file_io, native_path = self._resolve(path)
+ return file_io.get_file_status(native_path)
+
+ def list_status(self, path):
+ file_io, native_path = self._resolve(path)
+ return file_io.list_status(native_path)
+
+ def new_input_stream(self, path):
+ file_io, native_path = self._resolve(path)
+ return file_io.new_input_stream(native_path)
+
+ def to_filesystem_path(self, path):
+ return self._resolve(path)[1]
+
+ def close(self):
+ self._resolver.close()
def _source_path_text(value):
@@ -32,6 +68,61 @@ def _source_path_text(value):
return path
+def _normalize_source_path(value):
+ path = _source_path_text(value)
+ parsed = urlparse(path)
+ if _is_windows_drive_path(parsed):
+ windows_path = PureWindowsPath(path)
+ if not windows_path.is_absolute():
+ raise ValueError("Windows source paths must be absolute: %s" %
path)
+ return "file:///%s" % quote(windows_path.as_posix(), safe="/:")
+ if not parsed.scheme:
+ return Path(path).expanduser().resolve().as_uri()
+ return _quote_uri_path(path)
+
+
+def _quote_uri_path(uri):
+ match = re.match(r"^([A-Za-z][A-Za-z0-9+.-]*://[^/]*)(.*)$", uri)
+ if match is None:
+ return uri
+ return match.group(1) + quote(match.group(2), safe="/:%")
+
+
+def _qualified_status_path(parent_path, status):
+ status_path = str(status.path)
+ status_uri = urlparse(status_path)
+ if status_uri.scheme and not _is_windows_drive_path(status_uri):
+ return _quote_uri_path(status_path)
+
+ parent_uri = urlparse(parent_path)
+ scheme = parent_uri.scheme.lower()
+ if scheme == "file":
+ return _normalize_source_path(status_path)
+ if not scheme or _is_windows_drive_path(parent_uri):
+ return _normalize_source_path(status_path)
+
+ if scheme in ("hdfs", "viewfs"):
+ return urlunparse((
+ scheme,
+ parent_uri.netloc,
+ quote("/" + status_path.lstrip("/"), safe="/:"),
+ "",
+ "",
+ "",
+ ))
+
+ key = status_path.lstrip("/")
+ if parent_uri.netloc and not (
+ key == parent_uri.netloc
+ or key.startswith(parent_uri.netloc + "/")):
+ key = parent_uri.netloc + "/" + key
+ return "%s://%s" % (scheme, quote(key, safe="/:"))
+
+
+def _is_windows_drive_path(parsed):
+ return len(parsed.scheme) == 1 and not parsed.netloc
+
+
def _validated_source_options(source_options):
if source_options is None:
return {}
diff --git a/paimon-python/pypaimon/ray/__init__.py
b/paimon-python/pypaimon/ray/__init__.py
index e29d845061..fc0a8818fd 100644
--- a/paimon-python/pypaimon/ray/__init__.py
+++ b/paimon-python/pypaimon/ray/__init__.py
@@ -32,6 +32,7 @@ from pypaimon.ray.update_by_row_id import update_by_row_id
from pypaimon.ray.read_by_row_id import read_by_row_id
from pypaimon.ray.process_row_id_ranges import process_row_id_ranges
from pypaimon.ray.hdf5 import load_from_hdf5
+from pypaimon.ray.rosbag import load_from_rosbag
__all__ = [
"read_paimon",
@@ -44,6 +45,7 @@ __all__ = [
"read_by_row_id",
"process_row_id_ranges",
"load_from_hdf5",
+ "load_from_rosbag",
"WhenMatched",
"WhenNotMatched",
"source_col",
diff --git a/paimon-python/pypaimon/ray/rosbag.py
b/paimon-python/pypaimon/ray/rosbag.py
new file mode 100644
index 0000000000..8347dda7aa
--- /dev/null
+++ b/paimon-python/pypaimon/ray/rosbag.py
@@ -0,0 +1,212 @@
+# 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.
+
+"""Distributed ROSBag validation and ingestion with a materialize barrier."""
+
+import pickle
+from pathlib import Path
+import sys
+from tempfile import TemporaryDirectory
+from typing import Any, Dict, Mapping, Optional
+
+
+def load_from_rosbag(
+ table_identifier: str,
+ paths,
+ catalog_options: Dict[str, str],
+ *,
+ transform,
+ default_typestore=None,
+ typestore_factory=None,
+ source_options: Optional[Mapping[str, object]] = None,
+ staging=None,
+ allow_storage_fragment: bool = False,
+ concurrency: Optional[int] = None,
+ ray_remote_args: Optional[Dict[str, Any]] = None):
+ """Validate ROSBag sources on Ray before starting the Paimon sink.
+
+ ``transform(reader, source)`` returns Arrow tables or record batches.
+ Ray may retry it, so it must be deterministic and free of non-idempotent
+ external side effects. Validated output is materialized before the
+ append-only Paimon write starts; a commit exception may have an unknown
+ result.
+ """
+ if sys.version_info < (3, 10):
+ raise RuntimeError(
+ "Ray ROSBag loading requires Python 3.10 or newer; the rosbag "
+ "extra is not available on older Python versions.")
+ if not callable(transform):
+ raise ValueError("transform must be callable.")
+ if default_typestore is not None and typestore_factory is not None:
+ raise ValueError(
+ "default_typestore and typestore_factory are mutually exclusive.")
+ if typestore_factory is not None and not callable(typestore_factory):
+ raise ValueError("typestore_factory must be callable.")
+
+ from pypaimon.catalog.catalog_factory import CatalogFactory
+ from pypaimon.common.options import Options
+ from pypaimon.multimodal.rosbag.api import (
+ RosbagLoadResult,
+ RosbagStagingConfig,
+ _path_values,
+ _validate_staging_config,
+ )
+ from pypaimon.multimodal.rosbag.source import _discover_rosbag_sources
+ from pypaimon.multimodal.source_utils import (
+ _SourceFileIO,
+ _validated_source_options,
+ _validate_source_kerberos,
+ )
+ from pypaimon.multimodal.table import _target_schema
+ from pypaimon.ray.ray_paimon import _require_ray_data, write_paimon
+
+ if staging is None:
+ staging = RosbagStagingConfig()
+ if not isinstance(staging, RosbagStagingConfig):
+ raise ValueError("staging must be a RosbagStagingConfig.")
+ _validate_staging_config(staging)
+ options = _validated_source_options(source_options)
+ path_values = _path_values(paths)
+ _validate_source_kerberos(path_values, options, source_name="ROSBag")
+ source_file_io = _SourceFileIO(Options(options))
+ try:
+ manifests = _discover_rosbag_sources(
+ path_values,
+ source_file_io,
+ allow_storage_fragment=allow_storage_fragment,
+ )
+ finally:
+ source_file_io.close()
+ if not manifests:
+ return RosbagLoadResult(0, None, 0, None)
+
+ ray_data = _require_ray_data()
+ _require_ray_serializable("transform", transform)
+ if default_typestore is not None:
+ _require_ray_serializable("default_typestore", default_typestore)
+ if typestore_factory is not None:
+ _require_ray_serializable("typestore_factory", typestore_factory)
+ table = CatalogFactory.create(catalog_options).get_table(table_identifier)
+ target_schema = _target_schema(table)
+ inputs = ray_data.from_items(
+ [{"manifest": pickle.dumps(manifest)} for manifest in manifests],
+ override_num_blocks=len(manifests),
+ )
+ transformed = inputs.map_batches(
+ _TransformRosbagSource,
+ fn_constructor_kwargs={
+ "transform": transform,
+ "default_typestore": default_typestore,
+ "typestore_factory": typestore_factory,
+ "source_options": options,
+ "staging": staging,
+ "target_schema": target_schema,
+ },
+ batch_format="pyarrow",
+ batch_size=1,
+ concurrency=concurrency,
+ **dict(ray_remote_args or {}),
+ )
+ validated = transformed.materialize()
+ from pypaimon.multimodal.rosbag.staging import _verify_manifest_members
+ source_file_io = _SourceFileIO(Options(options))
+ try:
+ for manifest in manifests:
+ _verify_manifest_members(manifest, source_file_io)
+ finally:
+ source_file_io.close()
+ write_result = write_paimon(
+ validated,
+ table_identifier,
+ catalog_options,
+ concurrency=concurrency,
+ ray_remote_args=ray_remote_args,
+ )
+ return RosbagLoadResult(
+ source_count=len(manifests),
+ batch_count=None,
+ row_count=0 if write_result is None else write_result.row_count,
+ snapshot_id=(
+ None if write_result is None else write_result.snapshot_id),
+ )
+
+
+def _require_ray_serializable(name, value):
+ import ray.cloudpickle as cloudpickle
+
+ try:
+ cloudpickle.dumps(value)
+ except Exception as error:
+ raise ValueError(
+ "%s must be Ray-serializable: %s" % (name, error)) from error
+
+
+class _TransformRosbagSource:
+ """Ray callable that will validate one complete source per input batch."""
+
+ def __init__(
+ self,
+ *,
+ transform,
+ default_typestore,
+ typestore_factory,
+ source_options,
+ staging,
+ target_schema):
+ self.transform = transform
+ self.default_typestore = default_typestore
+ self.typestore_factory = typestore_factory
+ self.source_options = source_options
+ self.staging = staging
+ self.target_schema = target_schema
+
+ def __call__(self, batch):
+ if batch.num_rows != 1:
+ raise ValueError(
+ "Ray ROSBag transform requires one source per batch.")
+
+ from pypaimon.common.options import Options
+ from pypaimon.multimodal.rosbag.loader import (
+ _transform_rosbag_manifest,
+ )
+ from pypaimon.multimodal.source_utils import _SourceFileIO
+ try:
+ from rosbags.highlevel import AnyReader
+ except ImportError as error:
+ raise ImportError(
+ "Ray ROSBag loading requires rosbags on every worker; "
+ "install 'pypaimon[ray,rosbag]'.") from error
+
+ manifest = pickle.loads(batch["manifest"][0].as_py())
+ source_file_io = _SourceFileIO(Options(self.source_options))
+ try:
+ with TemporaryDirectory(
+ prefix="pypaimon_ray_rosbag_",
+ dir=self.staging.directory) as temp_dir:
+ for table in _transform_rosbag_manifest(
+ manifest,
+ self.transform,
+ source_file_io,
+ AnyReader,
+ self.target_schema,
+ default_typestore=self.default_typestore,
+ typestore_factory=self.typestore_factory,
+ staging=self.staging,
+ staging_root=Path(temp_dir) / "source"):
+ if table.num_rows:
+ yield table
+ finally:
+ source_file_io.close()
diff --git a/paimon-python/pypaimon/tests/multimodal_hdf5_test.py
b/paimon-python/pypaimon/tests/multimodal_hdf5_test.py
index 27063cefc2..7131060dbc 100644
--- a/paimon-python/pypaimon/tests/multimodal_hdf5_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_hdf5_test.py
@@ -71,7 +71,7 @@ class Hdf5RuntimeContractTest(unittest.TestCase):
resolver._get_fileio.return_value = backend
with patch(
- "pypaimon.multimodal.hdf5.ResolvingFileIO",
+ "pypaimon.multimodal.source_utils.ResolvingFileIO",
return_value=resolver):
source_io = _Hdf5SourceFileIO(Options({}))
self.assertEqual(
@@ -501,7 +501,7 @@ class MultimodalHdf5Test(unittest.TestCase):
yield from self._transform(h5, source)
with patch(
- "pypaimon.multimodal.hdf5.ResolvingFileIO",
+ "pypaimon.multimodal.source_utils.ResolvingFileIO",
return_value=source_file_io) as resolving_file_io:
result = self._load(
table,
@@ -541,7 +541,7 @@ class MultimodalHdf5Test(unittest.TestCase):
proxy, writer, committer = self._instrument_write(table)
with patch(
- "pypaimon.multimodal.hdf5.ResolvingFileIO",
+ "pypaimon.multimodal.source_utils.ResolvingFileIO",
return_value=source_file_io), patch.object(
table.raw_table,
"new_batch_write_builder",
@@ -566,7 +566,7 @@ class MultimodalHdf5Test(unittest.TestCase):
new_builder = Mock()
with patch(
- "pypaimon.multimodal.hdf5.ResolvingFileIO",
+ "pypaimon.multimodal.source_utils.ResolvingFileIO",
return_value=source_file_io), patch.object(
table.raw_table,
"new_batch_write_builder",
@@ -790,7 +790,7 @@ class MultimodalHdf5Test(unittest.TestCase):
with patch.object(
self.conn, "get_table", return_value=table), patch(
- "pypaimon.multimodal.hdf5.ResolvingFileIO") as resolving:
+ "pypaimon.multimodal.source_utils.ResolvingFileIO") as
resolving:
with self.assertRaisesRegex(
ValueError, "process-isolated"):
self.conn.load_from_hdf5(
diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
index 7f1f8d10a8..7362c91d23 100644
--- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
@@ -29,7 +29,7 @@ import pyarrow.fs as pafs
import pypaimon.multimodal as pmm
from pypaimon.common.options import Options
-from pypaimon.multimodal.hdf5 import _Hdf5SourceFileIO
+from pypaimon.multimodal.source_utils import _SourceFileIO
from pypaimon.multimodal.lerobot import load_from_lerobot
from pypaimon.multimodal.lerobot.loader import (
_image_bytes,
@@ -136,7 +136,7 @@ class LeRobotValidationTest(unittest.TestCase):
def test_double_encoded_file_uri_cannot_escape_source(self):
temp_dir = Path(tempfile.mkdtemp(prefix="pypaimon_lerobot_uri_"))
- source_file_io = _Hdf5SourceFileIO(Options({}))
+ source_file_io = _SourceFileIO(Options({}))
try:
root = temp_dir / "source"
root.mkdir()
@@ -163,7 +163,7 @@ class LeRobotValidationTest(unittest.TestCase):
def test_hdfs_source_rejects_explicit_keytab_before_resolution(self):
with patch(
- "pypaimon.multimodal.lerobot.source._Hdf5SourceFileIO"
+ "pypaimon.multimodal.lerobot.source._SourceFileIO"
) as source_file_io:
with self.assertRaisesRegex(ValueError, "process-isolated"):
load_from_lerobot(
@@ -685,7 +685,7 @@ class LeRobotImportTest(unittest.TestCase):
source_file_io = _RemoteLeRobotFileIO(self.image_source, source)
with patch(
- "pypaimon.multimodal.lerobot.source._Hdf5SourceFileIO",
+ "pypaimon.multimodal.lerobot.source._SourceFileIO",
return_value=source_file_io):
snapshot_id = self.connection.load_from_lerobot(
"oss_images",
diff --git a/paimon-python/pypaimon/tests/multimodal_rosbag_test.py
b/paimon-python/pypaimon/tests/multimodal_rosbag_test.py
new file mode 100644
index 0000000000..b56056ffda
--- /dev/null
+++ b/paimon-python/pypaimon/tests/multimodal_rosbag_test.py
@@ -0,0 +1,889 @@
+# 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 io
+import os
+from pathlib import Path
+import sys
+from unittest.mock import Mock, patch
+from urllib.parse import urlparse
+
+import pytest
+import pyarrow as pa
+import pyarrow.fs as pafs
+
+import pypaimon.multimodal as pmm
+from pypaimon.common.options import Options
+from pypaimon.multimodal.rosbag import RosbagSource
+from pypaimon.multimodal.rosbag.api import (
+ RosbagStagingConfig,
+ load_from_rosbag,
+)
+from pypaimon.multimodal.rosbag.loader import _BoundedStagingOutput
+from pypaimon.multimodal.rosbag.source import _discover_rosbag_sources
+from pypaimon.multimodal.rosbag.staging import (
+ _materialized_rosbag,
+ _stage_remote_manifest,
+ _verify_manifest_members,
+)
+from pypaimon.multimodal.source_utils import _SourceFileIO
+
+
+pytestmark = pytest.mark.skipif(
+ sys.version_info < (3, 10),
+ reason="rosbags 0.11 requires Python 3.10 or newer",
+)
+
+
+_MSGTYPE = "std_msgs/msg/String"
+_START_NS = 1_700_000_000_000_000_000
+
+
+def _write_ros2(path, storage_plugin, values=("first", "second")):
+ from rosbags.rosbag2 import Writer
+ from rosbags.typesys import Stores, get_typestore
+
+ typestore = get_typestore(Stores.ROS2_HUMBLE)
+ with Writer(
+ path,
+ version=Writer.VERSION_LATEST,
+ storage_plugin=storage_plugin) as writer:
+ connection = writer.add_connection(
+ "/text", _MSGTYPE, typestore=typestore)
+ for index, value in enumerate(values):
+ message = typestore.types[_MSGTYPE](data=value)
+ writer.write(
+ connection,
+ _START_NS + index,
+ typestore.serialize_cdr(message, _MSGTYPE),
+ )
+ return path
+
+
+def _write_ros1(path, values=("first", "second")):
+ from rosbags.rosbag1 import Writer
+ from rosbags.typesys import Stores, get_typestore
+
+ typestore = get_typestore(Stores.ROS1_NOETIC)
+ with Writer(path) as writer:
+ connection = writer.add_connection(
+ "/text", _MSGTYPE, typestore=typestore)
+ for index, value in enumerate(values):
+ message = typestore.types[_MSGTYPE](data=value)
+ writer.write(
+ connection,
+ _START_NS + index,
+ typestore.serialize_ros1(message, _MSGTYPE),
+ )
+ return path
+
+
+class _RemoteFileIO:
+
+ def __init__(self, objects, directories=None, virtual_directories=False):
+ self.objects = dict(objects)
+ self.directories = dict(directories or {})
+ self.virtual_directories = virtual_directories
+ self.close_count = 0
+
+ def _get_fileio(self, path):
+ return self
+
+ def to_filesystem_path(self, path):
+ parsed = urlparse(path)
+ return "%s%s" % (parsed.netloc, parsed.path)
+
+ def get_file_status(self, path):
+ if path in self.objects:
+ return pafs.FileInfo(
+ path, pafs.FileType.File, size=len(self.objects[path]))
+ if path in self.directories and not self.virtual_directories:
+ return pafs.FileInfo(path, pafs.FileType.Directory)
+ raise FileNotFoundError(path)
+
+ def list_status(self, path):
+ result = []
+ for child in self.directories[path]:
+ if child in self.objects:
+ result.append(pafs.FileInfo(
+ child,
+ pafs.FileType.File,
+ size=len(self.objects[child]),
+ ))
+ else:
+ result.append(pafs.FileInfo(child, pafs.FileType.Directory))
+ return result
+
+ def new_input_stream(self, path):
+ return io.BytesIO(self.objects[path])
+
+ def close(self):
+ self.close_count += 1
+
+
+def test_rosbag_source_exposes_original_and_local_paths(tmp_path):
+ local_path = tmp_path / "recording.bag"
+
+ source = RosbagSource(
+ uri="s3://bucket/robot/recording.bag",
+ local_path=local_path,
+ format="ros1",
+ )
+
+ assert source.uri == "s3://bucket/robot/recording.bag"
+ assert source.local_path == Path(local_path)
+ assert source.format == "ros1"
+ assert source.name == "recording.bag"
+ assert source.stem == "recording"
+ assert source.is_remote is True
+
+
+def test_rosbag_public_types_are_exported():
+ assert pmm.RosbagSource is RosbagSource
+ assert pmm.RosbagLoadResult.__name__ == "RosbagLoadResult"
+ assert pmm.RosbagStagingConfig.__name__ == "RosbagStagingConfig"
+
+
[email protected](
+ "staging,match",
+ [
+ (RosbagStagingConfig(max_bytes=-1), "max_bytes"),
+ (RosbagStagingConfig(min_free_bytes=-1), "min_free_bytes"),
+ (RosbagStagingConfig(copy_buffer_bytes=0), "copy_buffer_bytes"),
+ ],
+)
+def test_rejects_invalid_staging_config(staging, match):
+ with pytest.raises(ValueError, match=match):
+ load_from_rosbag(
+ object(), [], transform=lambda reader, source: (),
+ staging=staging)
+
+
+def test_discovers_explicit_local_ros1_bag(tmp_path):
+ bag = tmp_path / "recording.bag"
+ bag.write_bytes(b"rosbag")
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ manifests = _discover_rosbag_sources([bag], source_file_io)
+ finally:
+ source_file_io.close()
+
+ assert len(manifests) == 1
+ assert manifests[0].uri == bag.resolve().as_uri()
+ assert manifests[0].format == "ros1"
+ assert len(manifests[0].members) == 1
+ assert manifests[0].members[0].relative_path == "recording.bag"
+ assert manifests[0].members[0].size == 6
+
+
+def test_discovers_standalone_mcap_as_ros2(tmp_path):
+ mcap = tmp_path / "recording.mcap"
+ mcap.write_bytes(b"mcap")
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ manifests = _discover_rosbag_sources([mcap], source_file_io)
+ finally:
+ source_file_io.close()
+
+ assert len(manifests) == 1
+ assert manifests[0].format == "ros2_mcap"
+ assert manifests[0].members[0].relative_path == "recording.mcap"
+
+
+def test_standalone_db3_requires_explicit_fragment_mode(tmp_path):
+ database = tmp_path / "recording.db3"
+ database.write_bytes(b"sqlite")
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ with pytest.raises(ValueError, match="allow_storage_fragment=True"):
+ _discover_rosbag_sources([database], source_file_io)
+ finally:
+ source_file_io.close()
+
+
+def test_rejects_ros1_active_recording(tmp_path):
+ active = tmp_path / "recording.bag.active"
+ active.write_bytes(b"active")
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ with pytest.raises(ValueError, match="still being written"):
+ _discover_rosbag_sources([active], source_file_io)
+ finally:
+ source_file_io.close()
+
+
+def test_discovers_opted_in_standalone_db3_as_fragment(tmp_path):
+ database = tmp_path / "recording.db3"
+ database.write_bytes(b"sqlite")
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ manifests = _discover_rosbag_sources(
+ [database], source_file_io, allow_storage_fragment=True)
+ finally:
+ source_file_io.close()
+
+ assert len(manifests) == 1
+ assert manifests[0].format == "ros2_sqlite3_fragment"
+
+
+def test_rejects_corrupt_opted_in_sqlite_fragment_before_writer(tmp_path):
+ database = tmp_path / "recording.db3"
+ database.write_bytes(b"not-a-sqlite-database")
+ connection = pmm.connect(options={"warehouse": str(tmp_path /
"warehouse")})
+ table = connection.create_table(
+ "messages",
+ schema=pa.schema([pa.field("value", pa.string(), nullable=False)]),
+ options={"blob-as-descriptor": "false"},
+ )
+
+ with patch.object(
+ connection, "get_table", return_value=table), patch.object(
+ table.raw_table,
+ "new_batch_write_builder") as new_write_builder:
+ with pytest.raises(ValueError, match="SQLite integrity check failed"):
+ connection.load_from_rosbag(
+ table.identifier,
+ database,
+ transform=lambda reader, source: pa.table({"value": ["x"]}),
+ allow_storage_fragment=True,
+ )
+
+ new_write_builder.assert_not_called()
+
+
+def test_discovers_ros2_sqlite_directory_from_metadata(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+
+ recording = _write_ros2(
+ tmp_path / "recording", StoragePlugin.SQLITE3)
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ manifests = _discover_rosbag_sources([recording], source_file_io)
+ finally:
+ source_file_io.close()
+
+ assert len(manifests) == 1
+ assert manifests[0].uri == recording.resolve().as_uri()
+ assert manifests[0].format == "ros2_sqlite3"
+ assert manifests[0].expected_message_count == 2
+ assert [member.relative_path for member in manifests[0].members] == [
+ "metadata.yaml", "recording.db3"]
+
+
+def test_rejects_ros2_sqlite_sidecar_as_not_finalized(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+
+ recording = _write_ros2(
+ tmp_path / "recording", StoragePlugin.SQLITE3)
+ database = next(recording.glob("*.db3"))
+ Path("%s-wal" % database).touch()
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ with pytest.raises(ValueError, match="not finalized"):
+ _discover_rosbag_sources([recording], source_file_io)
+ finally:
+ source_file_io.close()
+
+
+def test_detects_ros2_directory_member_added_after_discovery(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+
+ recording = _write_ros2(
+ tmp_path / "recording", StoragePlugin.SQLITE3)
+ source_file_io = _SourceFileIO(Options({}))
+ try:
+ manifest = _discover_rosbag_sources(
+ [recording], source_file_io)[0]
+ (recording / "new-split.db3").write_bytes(b"new")
+
+ with pytest.raises(ValueError, match="members changed"):
+ _verify_manifest_members(manifest, source_file_io)
+ finally:
+ source_file_io.close()
+
+
+def test_detects_same_size_local_source_replacement(tmp_path):
+ bag = tmp_path / "recording.bag"
+ bag.write_bytes(b"old")
+ source_file_io = _SourceFileIO(Options({}))
+ try:
+ manifest = _discover_rosbag_sources([bag], source_file_io)[0]
+ original_mtime_ns = bag.stat().st_mtime_ns
+ bag.write_bytes(b"new")
+ os.utime(
+ bag,
+ ns=(original_mtime_ns + 1_000_000_000,) * 2,
+ )
+
+ with pytest.raises(ValueError, match="source changed"):
+ _verify_manifest_members(manifest, source_file_io)
+ finally:
+ source_file_io.close()
+
+
+def test_discovers_ros2_mcap_directory_from_metadata(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+
+ recording = _write_ros2(
+ tmp_path / "recording", StoragePlugin.MCAP)
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ manifests = _discover_rosbag_sources([recording], source_file_io)
+ finally:
+ source_file_io.close()
+
+ assert len(manifests) == 1
+ assert manifests[0].format == "ros2_mcap"
+ assert [member.relative_path for member in manifests[0].members] == [
+ "metadata.yaml", "recording.mcap"]
+
+
+def
test_explicit_ros2_storage_directory_without_metadata_is_incomplete(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+
+ recording = _write_ros2(
+ tmp_path / "recording", StoragePlugin.MCAP)
+ (recording / "metadata.yaml").unlink()
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ with pytest.raises(ValueError, match="missing metadata.yaml"):
+ _discover_rosbag_sources([recording], source_file_io)
+ finally:
+ source_file_io.close()
+
+
+def test_recursively_discovers_sorts_and_deduplicates_mixed_sources(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+
+ root = tmp_path / "sources"
+ root.mkdir()
+ ros1 = _write_ros1(root / "b.bag")
+ ros2 = _write_ros2(root / "nested" / "a", StoragePlugin.SQLITE3)
+ (root / "README.txt").write_text("ignored", encoding="utf-8")
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ manifests = _discover_rosbag_sources(
+ [root, ros1, ros2], source_file_io)
+ finally:
+ source_file_io.close()
+
+ assert [(item.uri, item.format) for item in manifests] == [
+ (ros1.resolve().as_uri(), "ros1"),
+ (ros2.resolve().as_uri(), "ros2_sqlite3"),
+ ]
+
+
+def test_rejects_encoded_ros2_member_path_traversal(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+ from ruamel.yaml import YAML
+
+ recording = _write_ros2(
+ tmp_path / "recording", StoragePlugin.SQLITE3)
+ metadata_path = recording / "metadata.yaml"
+ yaml = YAML()
+ metadata = yaml.load(metadata_path.read_text(encoding="utf-8"))
+ metadata["rosbag2_bagfile_information"]["relative_file_paths"] = [
+ "%252e%252e/escape.db3"]
+ with metadata_path.open("w", encoding="utf-8") as stream:
+ yaml.dump(metadata, stream)
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ with pytest.raises(ValueError, match="unsafe.*relative_file_paths"):
+ _discover_rosbag_sources([recording], source_file_io)
+ finally:
+ source_file_io.close()
+
+
+def test_rejects_duplicate_normalized_ros2_members(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+ from ruamel.yaml import YAML
+
+ recording = _write_ros2(
+ tmp_path / "recording", StoragePlugin.SQLITE3)
+ metadata_path = recording / "metadata.yaml"
+ yaml = YAML()
+ metadata = yaml.load(metadata_path.read_text(encoding="utf-8"))
+ information = metadata["rosbag2_bagfile_information"]
+ member = information["relative_file_paths"][0]
+ information["relative_file_paths"] = [member, member.upper()]
+ with metadata_path.open("w", encoding="utf-8") as stream:
+ yaml.dump(metadata, stream)
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ with pytest.raises(ValueError, match="duplicate.*relative_file_paths"):
+ _discover_rosbag_sources([recording], source_file_io)
+ finally:
+ source_file_io.close()
+
+
+def test_rejects_nested_ros2_member_paths_unsupported_by_rosbags(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+ from ruamel.yaml import YAML
+
+ recording = _write_ros2(
+ tmp_path / "recording", StoragePlugin.SQLITE3)
+ metadata_path = recording / "metadata.yaml"
+ yaml = YAML()
+ metadata = yaml.load(metadata_path.read_text(encoding="utf-8"))
+ information = metadata["rosbag2_bagfile_information"]
+ information["relative_file_paths"] = [
+ "nested/%s" % information["relative_file_paths"][0]]
+ with metadata_path.open("w", encoding="utf-8") as stream:
+ yaml.dump(metadata, stream)
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ with pytest.raises(ValueError, match="nested.*relative_file_paths"):
+ _discover_rosbag_sources([recording], source_file_io)
+ finally:
+ source_file_io.close()
+
+
+def test_rejects_malformed_ros2_metadata(tmp_path):
+ recording = tmp_path / "recording"
+ recording.mkdir()
+ (recording / "metadata.yaml").write_text("[", encoding="utf-8")
+ source_file_io = _SourceFileIO(Options({}))
+
+ try:
+ with pytest.raises(ValueError, match="Cannot read ROS2 metadata"):
+ _discover_rosbag_sources([recording], source_file_io)
+ finally:
+ source_file_io.close()
+
+
+def test_loads_ros1_after_preflight_into_one_snapshot(tmp_path):
+ bag = _write_ros1(tmp_path / "recording.bag")
+ connection = pmm.connect(options={"warehouse": str(tmp_path /
"warehouse")})
+ table = connection.create_table(
+ "messages",
+ schema=pa.schema([
+ pa.field("source", pa.string(), nullable=False),
+ pa.field("timestamp", pa.int64(), nullable=False),
+ pa.field("value", pa.string(), nullable=False),
+ ]),
+ options={"blob-as-descriptor": "false"},
+ )
+
+ def transform(reader, source):
+ rows = []
+ for ros_connection, timestamp, rawdata in reader.messages():
+ message = reader.deserialize(rawdata, ros_connection.msgtype)
+ rows.append({
+ "source": source.name,
+ "timestamp": timestamp,
+ "value": message.data,
+ })
+ return pa.Table.from_pylist(rows)
+
+ result = connection.load_from_rosbag(
+ table.identifier, bag, transform=transform)
+
+ assert result.source_count == 1
+ assert result.batch_count == 1
+ assert result.row_count == 2
+ assert result.snapshot_id is not None
+ assert table.scan().to_arrow().select([
+ "source", "timestamp", "value"]
+ ).to_pylist() == [
+ {
+ "source": "recording.bag",
+ "timestamp": _START_NS,
+ "value": "first",
+ },
+ {
+ "source": "recording.bag",
+ "timestamp": _START_NS + 1,
+ "value": "second",
+ },
+ ]
+
+
+def test_loads_remote_ros1_through_fileio_staging(tmp_path):
+ local_bag = _write_ros1(tmp_path / "source.bag")
+ remote = _RemoteFileIO({
+ "bucket/recording.bag": local_bag.read_bytes(),
+ })
+ connection = pmm.connect(options={"warehouse": str(tmp_path /
"warehouse")})
+ table = connection.create_table(
+ "messages",
+ schema=pa.schema([
+ pa.field("value", pa.string(), nullable=False),
+ ]),
+ options={"blob-as-descriptor": "false"},
+ )
+
+ def transform(reader, source):
+ values = [
+ reader.deserialize(rawdata, ros_connection.msgtype).data
+ for ros_connection, _, rawdata in reader.messages()
+ ]
+ return pa.table({"value": values})
+
+ with patch(
+ "pypaimon.multimodal.source_utils.ResolvingFileIO",
+ return_value=remote):
+ result = connection.load_from_rosbag(
+ table.identifier,
+ "s3://bucket/recording.bag",
+ transform=transform,
+ )
+
+ assert result.row_count == 2
+ assert table.scan().to_arrow().select(["value"]).to_pylist() == [
+ {"value": "first"}, {"value": "second"}]
+ assert remote.close_count == 1
+
+
+def test_loads_remote_ros2_directory_from_manifest_members(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+
+ local = _write_ros2(tmp_path / "local", StoragePlugin.SQLITE3)
+ metadata = local / "metadata.yaml"
+ database = next(local.glob("*.db3"))
+ remote = _RemoteFileIO(
+ objects={
+ "bucket/recording/metadata.yaml": metadata.read_bytes(),
+ "bucket/recording/%s" % database.name: database.read_bytes(),
+ },
+ directories={
+ "bucket/recording": [
+ "bucket/recording/metadata.yaml",
+ "bucket/recording/%s" % database.name,
+ ],
+ },
+ )
+ connection = pmm.connect(options={"warehouse": str(tmp_path /
"warehouse")})
+ table = connection.create_table(
+ "messages",
+ schema=pa.schema([pa.field("value", pa.string(), nullable=False)]),
+ options={"blob-as-descriptor": "false"},
+ )
+
+ def transform(reader, source):
+ return pa.table({"value": [
+ reader.deserialize(rawdata, ros_connection.msgtype).data
+ for ros_connection, _, rawdata in reader.messages()
+ ]})
+
+ with patch(
+ "pypaimon.multimodal.source_utils.ResolvingFileIO",
+ return_value=remote):
+ result = connection.load_from_rosbag(
+ table.identifier,
+ "s3://bucket/recording",
+ transform=transform,
+ )
+
+ assert result.source_count == 1
+ assert result.row_count == 2
+ assert table.scan().to_arrow().select(["value"]).to_pylist() == [
+ {"value": "first"}, {"value": "second"}]
+
+
+def test_discovers_remote_ros2_virtual_directory_prefix(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+
+ local = _write_ros2(tmp_path / "local", StoragePlugin.MCAP)
+ metadata = local / "metadata.yaml"
+ mcap = next(local.glob("*.mcap"))
+ remote = _RemoteFileIO(
+ objects={
+ "bucket/recording/metadata.yaml": metadata.read_bytes(),
+ "bucket/recording/%s" % mcap.name: mcap.read_bytes(),
+ },
+ directories={
+ "bucket/recording": [
+ "bucket/recording/metadata.yaml",
+ "bucket/recording/%s" % mcap.name,
+ ],
+ },
+ virtual_directories=True,
+ )
+
+ with patch(
+ "pypaimon.multimodal.source_utils.ResolvingFileIO",
+ return_value=remote):
+ source_file_io = _SourceFileIO(Options({}))
+ try:
+ manifests = _discover_rosbag_sources(
+ ["s3://bucket/recording"], source_file_io)
+ finally:
+ source_file_io.close()
+
+ assert len(manifests) == 1
+ assert manifests[0].format == "ros2_mcap"
+
+
+def test_remote_source_staging_is_removed_after_transform_scope(tmp_path):
+ local_bag = _write_ros1(tmp_path / "source.bag")
+ remote = _RemoteFileIO({
+ "bucket/recording.bag": local_bag.read_bytes(),
+ })
+ staging_root = tmp_path / "staging"
+
+ with patch(
+ "pypaimon.multimodal.source_utils.ResolvingFileIO",
+ return_value=remote):
+ source_file_io = _SourceFileIO(Options({}))
+ try:
+ manifest = _discover_rosbag_sources(
+ ["s3://bucket/recording.bag"], source_file_io)[0]
+ with _materialized_rosbag(
+ manifest,
+ source_file_io,
+ staging_root,
+ RosbagStagingConfig(min_free_bytes=0)) as local_path:
+ assert local_path.is_file()
+ assert not staging_root.exists()
+ finally:
+ source_file_io.close()
+
+
+def test_remote_staging_limit_stops_before_writing_excess_bytes(tmp_path):
+ class UnderreportedRemoteFileIO(_RemoteFileIO):
+
+ def get_file_status(self, path):
+ status = super().get_file_status(path)
+ return pafs.FileInfo(
+ status.path,
+ status.type,
+ size=1,
+ )
+
+ uri = "s3://bucket/recording.bag"
+ remote = UnderreportedRemoteFileIO({
+ "bucket/recording.bag": b"x" * 1024,
+ })
+ with patch(
+ "pypaimon.multimodal.source_utils.ResolvingFileIO",
+ return_value=remote):
+ source_file_io = _SourceFileIO(Options({}))
+ try:
+ manifest = _discover_rosbag_sources([uri], source_file_io)[0]
+ staging_root = tmp_path / "staging"
+ staging_root.mkdir()
+
+ with pytest.raises(ValueError, match="configured limit"):
+ _stage_remote_manifest(
+ manifest,
+ source_file_io,
+ staging_root,
+ RosbagStagingConfig(
+ max_bytes=1,
+ min_free_bytes=0,
+ copy_buffer_bytes=1024,
+ ),
+ )
+
+ staged = staging_root / "recording.bag"
+ assert staged.stat().st_size <= 1
+ finally:
+ source_file_io.close()
+
+
+def test_metadata_count_failure_happens_before_transform_and_writer(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+ from ruamel.yaml import YAML
+
+ recording = _write_ros2(
+ tmp_path / "recording", StoragePlugin.SQLITE3)
+ metadata_path = recording / "metadata.yaml"
+ yaml = YAML()
+ metadata = yaml.load(metadata_path.read_text(encoding="utf-8"))
+ metadata["rosbag2_bagfile_information"]["message_count"] += 1
+ with metadata_path.open("w", encoding="utf-8") as stream:
+ yaml.dump(metadata, stream)
+
+ connection = pmm.connect(options={"warehouse": str(tmp_path /
"warehouse")})
+ table = connection.create_table(
+ "messages",
+ schema=pa.schema([pa.field("value", pa.string(), nullable=False)]),
+ options={"blob-as-descriptor": "false"},
+ )
+ transform = Mock()
+
+ with patch.object(
+ connection, "get_table", return_value=table), patch.object(
+ table.raw_table,
+ "new_batch_write_builder") as new_write_builder:
+ with pytest.raises(ValueError, match="declared 3 messages.*2"):
+ connection.load_from_rosbag(
+ table.identifier, recording, transform=transform)
+
+ transform.assert_not_called()
+ new_write_builder.assert_not_called()
+
+
+def test_topic_count_failure_happens_before_transform_and_writer(tmp_path):
+ from rosbags.rosbag2 import StoragePlugin
+ from ruamel.yaml import YAML
+
+ recording = _write_ros2(
+ tmp_path / "recording", StoragePlugin.MCAP)
+ metadata_path = recording / "metadata.yaml"
+ yaml = YAML()
+ metadata = yaml.load(metadata_path.read_text(encoding="utf-8"))
+ topic = metadata["rosbag2_bagfile_information"][
+ "topics_with_message_count"][0]
+ topic["message_count"] += 1
+ with metadata_path.open("w", encoding="utf-8") as stream:
+ yaml.dump(metadata, stream)
+
+ connection = pmm.connect(options={"warehouse": str(tmp_path /
"warehouse")})
+ table = connection.create_table(
+ "messages",
+ schema=pa.schema([pa.field("value", pa.string(), nullable=False)]),
+ options={"blob-as-descriptor": "false"},
+ )
+ transform = Mock()
+
+ with patch.object(
+ connection, "get_table", return_value=table), patch.object(
+ table.raw_table,
+ "new_batch_write_builder") as new_write_builder:
+ with pytest.raises(ValueError, match="topic /text declares 3.*2"):
+ connection.load_from_rosbag(
+ table.identifier, recording, transform=transform)
+
+ transform.assert_not_called()
+ new_write_builder.assert_not_called()
+
+
+def test_rechecks_all_sources_after_last_transform_before_writer(tmp_path):
+ first = _write_ros1(tmp_path / "a.bag", values=("first",))
+ second = _write_ros1(tmp_path / "b.bag", values=("second",))
+ connection = pmm.connect(options={"warehouse": str(tmp_path /
"warehouse")})
+ table = connection.create_table(
+ "messages",
+ schema=pa.schema([pa.field("value", pa.string(), nullable=False)]),
+ options={"blob-as-descriptor": "false"},
+ )
+
+ def transform(reader, source):
+ values = [
+ reader.deserialize(rawdata, ros_connection.msgtype).data
+ for ros_connection, _, rawdata in reader.messages()
+ ]
+ if source.name == "b.bag":
+ with first.open("ab") as stream:
+ stream.write(b"changed")
+ return pa.table({"value": values})
+
+ with patch.object(
+ connection, "get_table", return_value=table), patch.object(
+ table.raw_table,
+ "new_batch_write_builder") as new_write_builder:
+ with pytest.raises(ValueError, match="source changed"):
+ connection.load_from_rosbag(
+ table.identifier,
+ [first, second],
+ transform=transform,
+ )
+
+ new_write_builder.assert_not_called()
+
+
+def test_arrow_ipc_staging_limit_fails_before_writer(tmp_path):
+ bag = _write_ros1(tmp_path / "recording.bag", values=("value",))
+ connection = pmm.connect(options={"warehouse": str(tmp_path /
"warehouse")})
+ table = connection.create_table(
+ "messages",
+ schema=pa.schema([pa.field("value", pa.string(), nullable=False)]),
+ options={"blob-as-descriptor": "false"},
+ )
+
+ def transform(reader, source):
+ return pa.table({"value": ["larger-than-one-byte"]})
+
+ with patch.object(
+ connection, "get_table", return_value=table), patch.object(
+ table.raw_table,
+ "new_batch_write_builder") as new_write_builder:
+ with pytest.raises(ValueError, match="staging exceeds.*1 bytes"):
+ connection.load_from_rosbag(
+ table.identifier,
+ bag,
+ transform=transform,
+ staging=RosbagStagingConfig(
+ max_bytes=1,
+ min_free_bytes=0,
+ ),
+ )
+
+ new_write_builder.assert_not_called()
+
+
+def test_bounded_staging_output_rejects_bytes_before_writing(tmp_path):
+ spool_path = tmp_path / "validated.arrow"
+
+ with _BoundedStagingOutput(spool_path, 1) as output:
+ with pytest.raises(ValueError, match="configured limit"):
+ output.write(b"too large")
+
+ assert spool_path.stat().st_size == 0
+
+
+def test_bounded_staging_output_reserves_remote_source_bytes(tmp_path):
+ spool_path = tmp_path / "validated.arrow"
+
+ with _BoundedStagingOutput(spool_path, 2) as output:
+ output.set_reserved_bytes(1)
+ output.write(b"x")
+ with pytest.raises(ValueError, match="configured limit"):
+ output.write(b"y")
+
+ assert spool_path.stat().st_size == 1
+
+
+def test_arrow_ipc_staging_requires_free_space_before_transform(tmp_path):
+ bag = _write_ros1(tmp_path / "recording.bag", values=("value",))
+ connection = pmm.connect(options={"warehouse": str(tmp_path /
"warehouse")})
+ table = connection.create_table(
+ "messages",
+ schema=pa.schema([pa.field("value", pa.string(), nullable=False)]),
+ options={"blob-as-descriptor": "false"},
+ )
+ transform = Mock(return_value=pa.table({"value": ["value"]}))
+
+ with patch.object(
+ connection, "get_table", return_value=table), patch.object(
+ table.raw_table,
+ "new_batch_write_builder") as new_write_builder, patch(
+ "pypaimon.multimodal.rosbag.loader.shutil.disk_usage",
+ return_value=Mock(free=0)):
+ with pytest.raises(ValueError, match="less than 1 free bytes"):
+ connection.load_from_rosbag(
+ table.identifier,
+ bag,
+ transform=transform,
+ staging=RosbagStagingConfig(min_free_bytes=1),
+ )
+
+ transform.assert_not_called()
+ new_write_builder.assert_not_called()
diff --git a/paimon-python/pypaimon/tests/ray_rosbag_test.py
b/paimon-python/pypaimon/tests/ray_rosbag_test.py
new file mode 100644
index 0000000000..db033321a6
--- /dev/null
+++ b/paimon-python/pypaimon/tests/ray_rosbag_test.py
@@ -0,0 +1,251 @@
+# 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 pickle
+import sys
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.common.options import Options
+from pypaimon.multimodal.rosbag.api import RosbagStagingConfig
+from pypaimon.multimodal.rosbag.source import _discover_rosbag_sources
+from pypaimon.multimodal.source_utils import _SourceFileIO
+from pypaimon.ray.rosbag import _TransformRosbagSource, load_from_rosbag
+
+
+pytestmark = pytest.mark.skipif(
+ sys.version_info < (3, 10),
+ reason="rosbags 0.11 requires Python 3.10 or newer",
+)
+
+
+_MSGTYPE = "std_msgs/msg/String"
+
+
+def _write_ros1(path):
+ from rosbags.rosbag1 import Writer
+ from rosbags.typesys import Stores, get_typestore
+
+ typestore = get_typestore(Stores.ROS1_NOETIC)
+ with Writer(path) as writer:
+ connection = writer.add_connection(
+ "/text", _MSGTYPE, typestore=typestore)
+ message = typestore.types[_MSGTYPE](data="worker")
+ writer.write(
+ connection,
+ 1_700_000_000_000_000_000,
+ typestore.serialize_ros1(message, _MSGTYPE),
+ )
+ return path
+
+
+def test_ray_rosbag_loader_is_exported():
+ from pypaimon.ray import load_from_rosbag as exported
+
+ assert exported is load_from_rosbag
+
+
+def test_ray_rosbag_rejects_invalid_staging_before_catalog_access():
+ with patch(
+ "pypaimon.catalog.catalog_factory.CatalogFactory.create"
+ ) as create_catalog:
+ with pytest.raises(ValueError, match="copy_buffer_bytes"):
+ load_from_rosbag(
+ "default.messages",
+ [],
+ {"warehouse": "/tmp/warehouse"},
+ transform=lambda reader, source: (),
+ staging=RosbagStagingConfig(copy_buffer_bytes=0),
+ )
+
+ create_catalog.assert_not_called()
+
+
+def test_ray_rosbag_materializes_validation_before_paimon_write(tmp_path):
+ bag = tmp_path / "recording.bag"
+ bag.write_bytes(b"bag")
+ lazy_dataset = Mock(name="lazy_dataset")
+ materialized_dataset = Mock(name="materialized_dataset")
+ lazy_dataset.materialize.return_value = materialized_dataset
+ inputs = Mock(name="inputs")
+ inputs.map_batches.return_value = lazy_dataset
+ ray_data = Mock(name="ray_data")
+ ray_data.from_items.return_value = inputs
+ table = Mock()
+ table.table_schema.to_arrow_schema.return_value = pa.schema([
+ pa.field("value", pa.int64(), nullable=False)])
+ catalog = Mock()
+ catalog.get_table.return_value = table
+ write_result = SimpleNamespace(row_count=3, snapshot_id=7)
+
+ with patch(
+ "pypaimon.ray.ray_paimon._require_ray_data",
+ return_value=ray_data), patch(
+ "pypaimon.catalog.catalog_factory.CatalogFactory.create",
+ return_value=catalog), patch(
+ "pypaimon.multimodal.table._target_schema",
+ return_value=pa.schema([
+ pa.field("value", pa.int64(), nullable=False)])),
patch(
+ "pypaimon.ray.ray_paimon.write_paimon",
+ return_value=write_result) as write_paimon:
+ result = load_from_rosbag(
+ "default.messages",
+ bag,
+ {"warehouse": str(tmp_path / "warehouse")},
+ transform=lambda reader, source: pa.table({"value": [1]}),
+ )
+
+ lazy_dataset.materialize.assert_called_once_with()
+ assert write_paimon.call_args.args[0] is materialized_dataset
+ assert result.source_count == 1
+ assert result.batch_count is None
+ assert result.row_count == 3
+ assert result.snapshot_id == 7
+
+
+def test_ray_rosbag_materialize_failure_never_starts_paimon_write(tmp_path):
+ bag = tmp_path / "recording.bag"
+ bag.write_bytes(b"bag")
+ lazy_dataset = Mock()
+ lazy_dataset.materialize.side_effect = ValueError("invalid source")
+ inputs = Mock()
+ inputs.map_batches.return_value = lazy_dataset
+ ray_data = Mock()
+ ray_data.from_items.return_value = inputs
+ catalog = Mock()
+ catalog.get_table.return_value = Mock()
+
+ with patch(
+ "pypaimon.ray.ray_paimon._require_ray_data",
+ return_value=ray_data), patch(
+ "pypaimon.catalog.catalog_factory.CatalogFactory.create",
+ return_value=catalog), patch(
+ "pypaimon.multimodal.table._target_schema",
+ return_value=pa.schema([
+ pa.field("value", pa.int64(), nullable=False)])),
patch(
+ "pypaimon.ray.ray_paimon.write_paimon") as write_paimon:
+ with pytest.raises(ValueError, match="invalid source"):
+ load_from_rosbag(
+ "default.messages",
+ bag,
+ {"warehouse": str(tmp_path / "warehouse")},
+ transform=lambda reader, source: pa.table({"value": [1]}),
+ )
+
+ write_paimon.assert_not_called()
+
+
+def test_ray_rosbag_rechecks_sources_after_materialize_before_write(tmp_path):
+ bag = tmp_path / "recording.bag"
+ bag.write_bytes(b"bag")
+ materialized_dataset = Mock()
+ lazy_dataset = Mock()
+
+ def materialize():
+ with bag.open("ab") as stream:
+ stream.write(b"changed")
+ return materialized_dataset
+
+ lazy_dataset.materialize.side_effect = materialize
+ inputs = Mock()
+ inputs.map_batches.return_value = lazy_dataset
+ ray_data = Mock()
+ ray_data.from_items.return_value = inputs
+ catalog = Mock()
+ catalog.get_table.return_value = Mock()
+
+ with patch(
+ "pypaimon.ray.ray_paimon._require_ray_data",
+ return_value=ray_data), patch(
+ "pypaimon.catalog.catalog_factory.CatalogFactory.create",
+ return_value=catalog), patch(
+ "pypaimon.multimodal.table._target_schema",
+ return_value=pa.schema([
+ pa.field("value", pa.int64(), nullable=False)])),
patch(
+ "pypaimon.ray.ray_paimon.write_paimon") as write_paimon:
+ with pytest.raises(ValueError, match="source changed"):
+ load_from_rosbag(
+ "default.messages",
+ bag,
+ {"warehouse": str(tmp_path / "warehouse")},
+ transform=lambda reader, source: pa.table({"value": [1]}),
+ )
+
+ write_paimon.assert_not_called()
+
+
+def test_ray_rosbag_worker_validates_and_transforms_one_source(tmp_path):
+ bag = _write_ros1(tmp_path / "recording.bag")
+ source_file_io = _SourceFileIO(Options({}))
+ try:
+ manifest = _discover_rosbag_sources([bag], source_file_io)[0]
+ finally:
+ source_file_io.close()
+ schema = pa.schema([pa.field("value", pa.string(), nullable=False)])
+
+ def transform(reader, source):
+ values = [
+ reader.deserialize(rawdata, connection.msgtype).data
+ for connection, _, rawdata in reader.messages()
+ ]
+ return pa.table({"value": values})
+
+ worker = _TransformRosbagSource(
+ transform=transform,
+ default_typestore=None,
+ typestore_factory=None,
+ source_options={},
+ staging=RosbagStagingConfig(min_free_bytes=0),
+ target_schema=schema,
+ )
+
+ output = list(worker(pa.table({
+ "manifest": [pickle.dumps(manifest)],
+ })))
+
+ assert len(output) == 1
+ assert output[0].schema == schema
+ assert output[0].to_pylist() == [{"value": "worker"}]
+
+
+def test_ray_rosbag_rejects_nonserializable_transform_on_driver(tmp_path):
+ bag = tmp_path / "recording.bag"
+ bag.write_bytes(b"bag")
+
+ class NonSerializableTransform:
+
+ def __init__(self):
+ self.values = (value for value in [1])
+
+ def __call__(self, reader, source):
+ return pa.table({"value": list(self.values)})
+
+ ray_data = Mock()
+ with patch(
+ "pypaimon.ray.ray_paimon._require_ray_data",
+ return_value=ray_data):
+ with pytest.raises(ValueError, match="transform must be
Ray-serializable"):
+ load_from_rosbag(
+ "default.messages",
+ bag,
+ {"warehouse": str(tmp_path / "warehouse")},
+ transform=NonSerializableTransform(),
+ )
+
+ ray_data.from_items.assert_not_called()
diff --git a/paimon-python/setup.py b/paimon-python/setup.py
index 8f0e57edc2..12e24632bc 100644
--- a/paimon-python/setup.py
+++ b/paimon-python/setup.py
@@ -237,6 +237,10 @@ setup(
# HDF5 loading is explicitly guarded and documented as Python 3.8+.
'h5py>=3,<4; python_version>="3.8"',
],
+ 'rosbag': [
+ # 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
