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 ce35a70072 [python] Add HDF5 DataSource ingestion (#9411)
ce35a70072 is described below
commit ce35a70072317191fd3473f0fc7913e8850dd3f4
Author: Yann Byron <[email protected]>
AuthorDate: Fri Aug 28 13:52:09 2026 +0800
[python] Add HDF5 DataSource ingestion (#9411)
---
.github/workflows/paimon-python-checks.yml | 2 +
docs/docs/pypaimon/multimodal-api.mdx | 126 +++
paimon-python/README.md | 78 +-
paimon-python/dev/requirements-dev.txt | 2 +
paimon-python/pypaimon/filesystem/local_file_io.py | 47 +-
.../pypaimon/filesystem/pyarrow_file_io.py | 6 +-
paimon-python/pypaimon/multimodal/__init__.py | 6 +
paimon-python/pypaimon/multimodal/connection.py | 21 +
paimon-python/pypaimon/multimodal/hdf5.py | 626 ++++++++++++++
paimon-python/pypaimon/tests/file_io_test.py | 43 +-
paimon-python/pypaimon/tests/hdfs_native_test.py | 8 +
.../pypaimon/tests/multimodal_hdf5_test.py | 949 +++++++++++++++++++++
.../pypaimon/tests/oss_legacy_mode_test.py | 7 +-
paimon-python/setup.py | 4 +
14 files changed, 1905 insertions(+), 20 deletions(-)
diff --git a/.github/workflows/paimon-python-checks.yml
b/.github/workflows/paimon-python-checks.yml
index c66101d8e6..5b35fff4f8 100755
--- a/.github/workflows/paimon-python-checks.yml
+++ b/.github/workflows/paimon-python-checks.yml
@@ -137,6 +137,8 @@ jobs:
else
python -m pip install pyarrow==16.0.0 numpy==1.24.3
pandas==2.0.3 flake8==4.0.1
fi
+ python -m pip install 'h5py>=3,<4'
+ python -c "import h5py; print('h5py', h5py.__version__)"
if [[ "${{ matrix.python-version }}" == "3.11" ]]; then
# Exercise the 0.4 API in one lane until its wheel is published.
diff --git a/docs/docs/pypaimon/multimodal-api.mdx
b/docs/docs/pypaimon/multimodal-api.mdx
index 2bb3d2562c..995966a77c 100644
--- a/docs/docs/pypaimon/multimodal-api.mdx
+++ b/docs/docs/pypaimon/multimodal-api.mdx
@@ -190,6 +190,132 @@ docs.add([
])
```
+## Load HDF5
+
+`MultimodalConnection.load_from_hdf5` streams one or more local or remote HDF5
+files into an existing multimodal table. HDF5 loading requires Python 3.8 or
+newer. Install the optional dependency first:
+
+```shell
+pip install 'pypaimon[hdf5]'
+```
+
+The transform receives an open `h5py.File` and an `Hdf5File`. Its `path` is the
+resolved local `file://` URI or remote URI; `name` and `stem` provide
convenient
+path components. For local sources, `local_path` returns a decoded `Path` that
+can be used to read sibling files, including paths containing spaces or
Unicode;
+it is `None` for remote sources. The transform can return one Arrow table or
+record batch, or an iterable that yields multiple tables or batches:
+
+```python
+import pyarrow as pa
+
+EMBEDDING_VECTOR_TYPE = pa.list_(pa.float32(), 3)
+IMAGE_BLOB_TYPE = pa.large_binary()
+
+schema = pa.schema([
+ pa.field("episode_id", pa.string(), nullable=False),
+ pa.field("frame_index", pa.int32(), nullable=False),
+ # Arrow fixed-size lists map to Paimon VECTOR columns.
+ pa.field("embedding", EMBEDDING_VECTOR_TYPE, nullable=False),
+ # Arrow binary and large-binary values map to Paimon BLOB columns.
+ pa.field("image", IMAGE_BLOB_TYPE),
+])
+
+frames = conn.create_table("frames", schema=schema)
+
+
+def transform(h5, source):
+ episode_id = source.stem
+ for begin in range(0, len(h5["embedding"]), 128):
+ end = min(begin + 128, len(h5["embedding"]))
+ yield pa.RecordBatch.from_pydict({
+ "episode_id": [episode_id] * (end - begin),
+ "frame_index": list(range(begin, end)),
+ "embedding": h5["embedding"][begin:end].tolist(),
+ "image": [bytes(value) for value in h5["image"][begin:end]],
+ }, schema=schema)
+
+
+result = conn.load_from_hdf5(
+ "frames",
+ "/data/episodes",
+ transform=transform,
+)
+print(result.file_count)
+print(result.batch_count)
+print(result.row_count)
+print(result.snapshot_id)
+```
+
+A path can be one `.h5` or `.hdf5` file, an iterable of paths, or a directory.
+Directories are searched recursively. Before opening files, `load_from_hdf5`
+resolves every discovered path, removes duplicate files within that call, and
+sorts by the resolved path. Overlapping directory and file arguments therefore
+read a physical file once per call. Local paths and `file://`, `hdfs://`,
+`viewfs://`, `oss://`, `s3://`, and `gs://` URIs use PyPaimon's existing
+FileIO implementations.
+
+Remote credentials and endpoints belong to the HDF5 source, not necessarily
+the target warehouse. Pass standard FileIO settings through `source_options`;
+they are not inherited from the target table, written to table options, or
+retained after the call. This also applies to HDFS client settings: provide
+source-specific settings explicitly, while configuration discoverable by the
+underlying filesystem client can still come from its normal environment:
+
+```python
+remote_result = conn.load_from_hdf5(
+ "frames",
+ "oss://source-bucket/episodes",
+ transform=transform,
+ source_options={
+ "fs.oss.endpoint": "oss-cn-hangzhou.aliyuncs.com",
+ "fs.oss.accessKeyId": "SOURCE_ACCESS_KEY_ID",
+ "fs.oss.accessKeySecret": "SOURCE_ACCESS_KEY_SECRET",
+ },
+)
+```
+
+Because native HDFS Kerberos credentials use process-global ticket state, HDF5
+sources cannot run an explicit-keytab login in a shared process. Acquire the
+ticket in a process-isolated worker, omit the source principal/keytab options,
+and then call `load_from_hdf5` in that worker.
+
+HDF5 requires random access. `load_from_hdf5` passes the seekable stream
returned
+by `FileIO.new_input_stream` directly to h5py and never downloads it to a local
+temporary file. A non-seekable stream fails before commit. Recursive directory
+discovery requires backend listing support; legacy OSS on PyArrow before 16
+must use explicit file paths, Jindo, or a newer PyArrow version. HDF5 performs
+many small seeks, so benchmark remote-object latency for large production
+inputs even though no full-file download is required.
+
+An empty path iterable or existing directory with no HDF5 files is a no-op. It
+returns zero file, batch, and row counts with `snapshot_id=None`, without
+creating a writer or snapshot. A nonexistent path or unsupported file suffix
+is still an error, as is a discovered file whose transform produces no rows.
+
+The target schema is strict. Each batch must contain exactly the target columns
+in target order and must be safely convertible to the target Arrow schema.
+Missing columns are not filled with null, extra columns are not discarded, and
+invalid nullability, incompatible types, or fixed-size vector lengths fail the
+whole call.
+
+One call uses one writer and one commit for all discovered files, and a
+successful non-empty call creates exactly one snapshot. Any failure before the
+commit aborts the writer and leaves no partial snapshot. An exception after the
+commit starts has Paimon's unknown-commit-result semantics: `load_from_hdf5`
does
+not retry and does not abort files that a snapshot may already reference.
+
+For a future Ray integration, split HDF5 file descriptions across worker tasks,
+apply the same transform and strict batch contract in each task, and collect
+writer commit messages for one coordinator commit. Do not call `load_from_hdf5`
+independently in every task, because that would create one commit per task. The
+HDF5 core itself has no Ray dependency.
+
+`load_from_hdf5` is append-only and is **not retry-safe**. It does not add
source
+provenance columns, maintain a source ledger, skip prior inputs, or detect
+source drift. Calling it again with the same input appends the rows again.
+
## Overwrite
`overwrite` accepts the same input formats as `add` and replaces existing data
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 5f716c8828..0e213985af 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -31,6 +31,83 @@ pip3 install dist/*.tar.gz
The command will install the package and core dependencies to your local
Python environment.
+# HDF5 to multimodal tables
+
+HDF5 loading requires Python 3.8 or newer. Install the optional dependency and
+create the target multimodal table before loading local or remote HDF5 files as
+one or more Arrow batches:
+
+```commandline
+pip install 'pypaimon[hdf5,vortex]'
+```
+
+```python
+import pyarrow as pa
+import pypaimon.multimodal as pmm
+
+EMBEDDING_VECTOR_TYPE = pa.list_(pa.float32(), 3)
+IMAGE_BLOB_TYPE = pa.large_binary()
+
+schema = pa.schema([
+ pa.field("episode_id", pa.string(), nullable=False),
+ pa.field("frame_index", pa.int32(), nullable=False),
+ # Arrow fixed-size lists map to Paimon VECTOR columns.
+ pa.field("embedding", EMBEDDING_VECTOR_TYPE, nullable=False),
+ # Arrow binary and large-binary values map to Paimon BLOB columns.
+ pa.field("image", IMAGE_BLOB_TYPE),
+])
+
+
+def transform(h5, source):
+ episode_id = source.stem
+ for begin in range(0, len(h5["embedding"]), 128):
+ end = min(begin + 128, len(h5["embedding"]))
+ yield pa.RecordBatch.from_pydict({
+ "episode_id": [episode_id] * (end - begin),
+ "frame_index": list(range(begin, end)),
+ "embedding": h5["embedding"][begin:end].tolist(),
+ "image": [bytes(value) for value in h5["image"][begin:end]],
+ }, schema=schema)
+
+connection = pmm.connect(options={"warehouse": "/tmp/warehouse"})
+frames = connection.create_table(
+ "frames",
+ schema=schema,
+)
+result = connection.load_from_hdf5(
+ "frames", "/data/episodes", transform=transform)
+print(result.file_count, result.batch_count, result.row_count,
result.snapshot_id)
+```
+
+`load_from_hdf5` accepts one `.h5`/`.hdf5` file, an iterable of paths, or
+directories that are searched recursively. Paths are resolved, duplicate
+files within the call are removed, and the remaining files are processed in
+sorted order. Every yielded batch must have exactly the target columns and be
+safely convertible to the table schema; missing or extra columns, nulls for
+non-nullable fields, incompatible types, and invalid fixed-size vector lengths
+fail the call.
+
+Remote `hdfs://`, `viewfs://`, `oss://`, `s3://`, and `gs://` sources use
+PyPaimon's FileIO abstraction. Pass source-only credentials and endpoints via
+`source_options={"fs.oss.endpoint": "...", ...}`; target warehouse FileIO
+settings are deliberately not reused. h5py reads the seekable FileIO stream
+directly without a local temporary download. Legacy OSS with PyArrow before 16
+supports explicit files but requires Jindo or a newer PyArrow for recursive
+directory discovery. In transforms, `source.local_path` returns a decoded
+`Path` for local sources (including spaces and Unicode) and `None` for remote
+sources.
+
+An empty path iterable or an existing directory without HDF5 files returns
+zero counts and `snapshot_id=None` without creating a writer or snapshot.
+Nonexistent paths, unsupported file suffixes, and discovered files whose
+transform produces no rows remain errors.
+
+All files in one call use one writer and one commit, so success creates one
+snapshot. The API is append-only: it does not add provenance columns, keep a
+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.
+
# HDFS without a local Hadoop install
`pypaimon` supports HDFS through a pure-protocol client based on
@@ -109,4 +186,3 @@ unsupported platform such as Windows), `pypaimon`
automatically falls
back to the `pyarrow` (`libhdfs`/JVM) path and logs a warning. Disable
the fallback with `hdfs.client.fallback-to-pyarrow=false` if you want
hard failures instead.
-
diff --git a/paimon-python/dev/requirements-dev.txt
b/paimon-python/dev/requirements-dev.txt
index 28da953b3b..a1b143383c 100644
--- a/paimon-python/dev/requirements-dev.txt
+++ b/paimon-python/dev/requirements-dev.txt
@@ -21,6 +21,8 @@
duckdb==1.3.2
flake8==4.0.1
pytest~=7.0
+# HDF5 ingestion tests run in the supported h5py wheel lanes (Python 3.8+).
+h5py>=3,<4; python_version >= "3.8"
# merge_into needs Dataset.join (added in Ray 2.50). Python 3.8 has no 2.50
wheel.
ray>=2.10.0; python_version < "3.9"
ray>=2.50.0; python_version >= "3.9"
diff --git a/paimon-python/pypaimon/filesystem/local_file_io.py
b/paimon-python/pypaimon/filesystem/local_file_io.py
index b315226181..e3530c1951 100644
--- a/paimon-python/pypaimon/filesystem/local_file_io.py
+++ b/paimon-python/pypaimon/filesystem/local_file_io.py
@@ -22,7 +22,7 @@ import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Optional
-from urllib.parse import urlparse
+from urllib.parse import unquote, urlparse
import pyarrow
import pyarrow.fs as pafs
@@ -35,6 +35,24 @@ from pypaimon.schema.data_types import DataField,
AtomicType, PyarrowFieldParser
from pypaimon.write.blob_format_writer import BlobFormatWriter
+def _file_uri_path(parsed, windows=None) -> str:
+ """Decode a file URI path, preserving Windows drives and UNC hosts."""
+ if windows is None:
+ windows = os.name == "nt"
+ if windows:
+ from nturl2path import url2pathname as windows_url2pathname
+ if parsed.netloc and not parsed.netloc.endswith(":"):
+ value = "//%s%s" % (parsed.netloc, parsed.path)
+ elif parsed.netloc:
+ value = "/%s%s" % (parsed.netloc, parsed.path)
+ else:
+ value = parsed.path
+ return windows_url2pathname(value)
+ if parsed.netloc and parsed.netloc != "localhost":
+ return "//%s%s" % (parsed.netloc, unquote(parsed.path))
+ return unquote(parsed.path)
+
+
class LocalFileIO(FileIO):
"""
Local file system implementation of FileIO.
@@ -61,13 +79,14 @@ class LocalFileIO(FileIO):
if parsed.scheme == 'file' and parsed.netloc and
parsed.netloc.endswith(':'):
drive_letter = parsed.netloc.rstrip(':')
- path_part = parsed.path.lstrip('/') if parsed.path else ''
+ path_part = unquote(parsed.path).lstrip('/') if parsed.path else ''
if path_part:
return Path(f"{drive_letter}:/{path_part}")
else:
return Path(f"{drive_letter}:")
- local_path = parsed.path if parsed.scheme else path
+ local_path = (
+ _file_uri_path(parsed) if parsed.scheme == 'file' else path)
if not local_path:
return Path(".")
@@ -138,18 +157,16 @@ class LocalFileIO(FileIO):
if file_path.is_file():
results.append(self.get_file_status(path))
elif file_path.is_dir():
- try:
- for item in file_path.iterdir():
- try:
- if path.startswith('file://'):
- item_path = f"file://{item}"
- else:
- item_path = str(item)
- results.append(self.get_file_status(item_path))
- except FileNotFoundError:
- pass
- except PermissionError:
- pass
+ for item in file_path.iterdir():
+ try:
+ if path.startswith('file://'):
+ item_path = item.absolute().as_uri()
+ else:
+ item_path = str(item)
+ results.append(self.get_file_status(item_path))
+ except FileNotFoundError:
+ # A child may disappear between listing and stat.
+ pass
return results
diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
index 4423c9559e..362994fb65 100644
--- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
+++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
@@ -45,6 +45,10 @@ def _pyarrow_lt_7():
return parse(pyarrow.__version__) < parse("7.0.0")
+class LegacyOssDirectoryListingError(RuntimeError):
+ """Raised when legacy PyArrow OSS cannot enumerate a directory."""
+
+
class PyArrowFileIO(FileIO):
def __init__(self, path: str, catalog_options: Options):
self.properties = catalog_options
@@ -402,7 +406,7 @@ class PyArrowFileIO(FileIO):
def list_status(self, path: str):
if self._legacy_oss_mode():
- raise RuntimeError(
+ raise LegacyOssDirectoryListingError(
"Listing OSS directories is not supported with PyArrow < 16 "
"(it parses the first key segment as a bucket). Upgrade to "
"pyarrow >= 16, or install pyjindosdk and set
fs.oss.impl=jindo.")
diff --git a/paimon-python/pypaimon/multimodal/__init__.py
b/paimon-python/pypaimon/multimodal/__init__.py
index 1b5060ec8f..96afcbbea3 100644
--- a/paimon-python/pypaimon/multimodal/__init__.py
+++ b/paimon-python/pypaimon/multimodal/__init__.py
@@ -25,6 +25,10 @@ from pypaimon.multimodal.blob_store import (
PutObjectResult,
)
from pypaimon.multimodal.connection import MultimodalConnection, connect
+from pypaimon.multimodal.hdf5 import (
+ Hdf5File,
+ Hdf5LoadResult,
+)
from pypaimon.multimodal.table import (
MultimodalTable,
TextRoute,
@@ -41,6 +45,8 @@ from pypaimon.table.data_evolution_merge_into import (
__all__ = [
"BlobObject",
"BlobStore",
+ "Hdf5File",
+ "Hdf5LoadResult",
"MultimodalConnection",
"MultimodalTable",
"NoSuchKey",
diff --git a/paimon-python/pypaimon/multimodal/connection.py
b/paimon-python/pypaimon/multimodal/connection.py
index 8bd1c90777..ff90820daf 100644
--- a/paimon-python/pypaimon/multimodal/connection.py
+++ b/paimon-python/pypaimon/multimodal/connection.py
@@ -96,6 +96,27 @@ class MultimodalConnection:
raw_table,
)
+ def load_from_hdf5(
+ self,
+ table_name: str,
+ paths,
+ *,
+ transform,
+ source_options=None):
+ """Load HDF5 transforms into an existing table as one append commit.
+
+ Repeating a call appends the rows again. A commit exception has an
+ unknown result and is not safe to retry without checking table state.
+ Source filesystem options are isolated from the target warehouse.
+ """
+ from pypaimon.multimodal.hdf5 import load_from_hdf5
+ return load_from_hdf5(
+ self.get_table(table_name),
+ paths,
+ transform=transform,
+ source_options=source_options,
+ )
+
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
new file mode 100644
index 0000000000..68389cf4e2
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/hdf5.py
@@ -0,0 +1,626 @@
+# 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.
+
+"""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 typing import Callable, Mapping, Optional
+from urllib.parse import quote, unquote, urlparse, urlunparse
+
+import pyarrow as pa
+import pyarrow.compute as pc
+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.table import _target_schema
+from pypaimon.write.commit_callback import CommitCallback
+
+
+_HDF5_SUFFIXES = (".h5", ".hdf5")
+
+
+@dataclass(frozen=True)
+class Hdf5File:
+ """Read context supplied to an HDF5 transform."""
+
+ path: str
+
+ @property
+ def local_path(self) -> Optional[Path]:
+ """Decoded local path, or ``None`` for a remote source."""
+ parsed = urlparse(self.path)
+ if parsed.scheme.lower() != "file":
+ return None
+ return Path(_file_uri_path(parsed))
+
+ @property
+ def name(self) -> str:
+ """Base name of the local path or remote URI."""
+ local_path = self.local_path
+ if local_path is not None:
+ return local_path.name
+ parsed = urlparse(self.path)
+ path = unquote(parsed.path) if parsed.scheme else self.path
+ return PurePosixPath(path).name
+
+ @property
+ def stem(self) -> str:
+ """Base name without the final HDF5 suffix."""
+ return PurePosixPath(self.name).stem
+
+
+@dataclass(frozen=True)
+class Hdf5LoadResult:
+ """Counts and optional snapshot for one ``load_from_hdf5`` call."""
+
+ file_count: int
+ batch_count: int
+ row_count: int
+ snapshot_id: Optional[int]
+
+
+class _SnapshotRecorder(CommitCallback):
+
+ def __init__(self):
+ self.snapshot_id = None
+
+ def call(self, context):
+ 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 close(self):
+ self._resolver.close()
+
+
+def load_from_hdf5(
+ table,
+ paths,
+ *,
+ transform: Callable,
+ source_options: Optional[Mapping[str, object]] = None):
+ """Load HDF5 files into an existing multimodal table.
+
+ ``transform`` receives an open ``h5py.File`` and :class:`Hdf5File`, and
+ must return one Arrow table/batch or an iterable of Arrow tables/batches.
+ All unique files and batches in one call share one writer and one commit.
+ Local paths and FileIO-supported URIs are accepted. ``source_options`` are
+ used only for source FileIO resolution and are never inherited from the
+ target table's warehouse.
+
+ This API is strictly append-only. It does not track sources or detect
+ duplicates between calls, so calling it again writes the rows again. It is
+ not retry-safe: a commit exception may have happened after the snapshot
+ became visible and is returned without retrying or aborting written files.
+ Empty discovery is a no-op and returns zero counts with no snapshot.
+ """
+ if sys.version_info < (3, 8):
+ raise RuntimeError(
+ "load_from_hdf5 requires Python 3.8 or newer; the hdf5 extra "
+ "is not available on older Python versions.")
+ if not callable(transform):
+ raise ValueError("transform must be callable.")
+ validated_options = _validated_source_options(source_options)
+ path_values = _path_values(paths)
+ _validate_source_kerberos(path_values, validated_options)
+ source_file_io = _Hdf5SourceFileIO(Options(validated_options))
+ try:
+ files = _discover_hdf5_files(path_values, source_file_io)
+ if not files:
+ return Hdf5LoadResult(
+ file_count=0,
+ batch_count=0,
+ row_count=0,
+ snapshot_id=None,
+ )
+ # Preserve the dependency-free no-op for empty discovery; only require
+ # h5py once at least one HDF5 source will actually be opened.
+ try:
+ import h5py
+ except ImportError as error:
+ raise ImportError(
+ "load_from_hdf5 requires h5py; install 'pypaimon[hdf5]'."
+ ) from error
+ return _load_hdf5_files(
+ table, files, transform, source_file_io, h5py)
+ finally:
+ source_file_io.close()
+
+
+def _load_hdf5_files(table, files, transform, source_file_io, h5py):
+ target_schema = _target_schema(table.raw_table)
+ write_builder = table.raw_table.new_batch_write_builder()
+ table_write = None
+ table_commit = None
+ commit_started = False
+ batch_count = 0
+ row_count = 0
+ snapshot_recorder = _SnapshotRecorder()
+
+ try:
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_commit.add_commit_callback(snapshot_recorder)
+
+ for source in files:
+ source_row_count = 0
+ with closing(source_file_io.new_input_stream(source.path)) as
stream:
+ _require_seekable(stream, source)
+ with h5py.File(stream, "r") as h5:
+ transformed = transform(h5, source)
+ batches = None
+ try:
+ batches = _arrow_batches(transformed)
+ for value in batches:
+ arrow_table = _strict_arrow_table(
+ value,
+ target_schema,
+ source,
+ batch_count,
+ )
+ batch_count += 1
+ row_count += arrow_table.num_rows
+ source_row_count += arrow_table.num_rows
+ if arrow_table.num_rows:
+ table_write.write_arrow(arrow_table)
+ finally:
+ _close_transform_iterator(
+ batches if batches is not None else transformed)
+
+ if source_row_count == 0:
+ raise ValueError(
+ "HDF5 source %s produced no rows." % source.path)
+
+ commit_messages = table_write.prepare_commit()
+ commit_started = True
+ table_commit.commit(commit_messages)
+ if snapshot_recorder.snapshot_id is None:
+ raise RuntimeError(
+ "HDF5 append committed without reporting a snapshot id.")
+ return Hdf5LoadResult(
+ file_count=len(files),
+ batch_count=batch_count,
+ row_count=row_count,
+ snapshot_id=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()
+
+
+def _discover_hdf5_files(paths, source_file_io):
+ values = _path_values(paths)
+ normalized = {}
+ visited_directories = set()
+ for value in values:
+ input_path = _source_path_text(value)
+ path = _normalize_source_path(input_path)
+ try:
+ status = source_file_io.get_file_status(path)
+ except FileNotFoundError as error:
+ _discover_missing_path(
+ source_file_io,
+ path,
+ normalized,
+ visited_directories,
+ error,
+ input_path,
+ )
+ continue
+ _discover_status(
+ source_file_io,
+ path,
+ status,
+ normalized,
+ visited_directories,
+ explicit_path=input_path,
+ )
+ return [normalized[key] for key in sorted(normalized)]
+
+
+def _discover_missing_path(
+ source_file_io,
+ path,
+ normalized,
+ visited_directories,
+ not_found_error,
+ input_path):
+ if _hdf5_suffix(path):
+ raise ValueError(
+ "HDF5 path does not exist: %s" % input_path
+ ) from not_found_error
+ try:
+ children = source_file_io.list_status(path)
+ except LegacyOssDirectoryListingError as error:
+ _raise_legacy_directory_listing_error(path, error)
+ if not children:
+ raise ValueError(
+ "HDF5 path does not exist: %s" % input_path
+ ) from not_found_error
+ visited_directories.add(path)
+ _discover_directory_children(
+ source_file_io,
+ path,
+ children,
+ normalized,
+ visited_directories,
+ )
+
+
+def _discover_status(
+ source_file_io,
+ parent_path,
+ status,
+ normalized,
+ visited_directories,
+ explicit_path=None):
+ path = _qualified_status_path(parent_path, status)
+ if status.type == pafs.FileType.File:
+ if not _hdf5_suffix(path):
+ if explicit_path is not None:
+ raise ValueError(
+ "HDF5 file has unsupported suffix: %s; expected .h5 or "
+ ".hdf5." % explicit_path)
+ return
+ normalized[path] = Hdf5File(path=path)
+ return
+ if status.type != pafs.FileType.Directory:
+ raise ValueError("Unsupported HDF5 source status for path: %s" % path)
+ if path in visited_directories:
+ return
+ visited_directories.add(path)
+ try:
+ children = source_file_io.list_status(path)
+ except LegacyOssDirectoryListingError as error:
+ _raise_legacy_directory_listing_error(path, error)
+ _discover_directory_children(
+ source_file_io,
+ path,
+ children,
+ normalized,
+ visited_directories,
+ )
+
+
+def _discover_directory_children(
+ source_file_io,
+ directory,
+ children,
+ normalized,
+ visited_directories):
+ for child in children:
+ _discover_status(
+ source_file_io,
+ directory,
+ child,
+ normalized,
+ visited_directories,
+ explicit_path=None,
+ )
+
+
+def _raise_legacy_directory_listing_error(path, error):
+ raise ValueError(
+ "Recursive HDF5 discovery is unavailable for legacy OSS at %s; "
+ "pass explicit HDF5 file paths, use Jindo, or upgrade PyArrow."
+ % path) from error
+
+
+def _source_path_text(value):
+ try:
+ path = os.fspath(value)
+ except TypeError as error:
+ raise ValueError(
+ "paths must contain only filesystem paths or URIs.") from error
+ if isinstance(path, bytes):
+ raise ValueError("paths must contain only filesystem paths or URIs.")
+ 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 _hdf5_suffix(path):
+ parsed = urlparse(path)
+ return PurePosixPath(unquote(parsed.path)).suffix.lower() in _HDF5_SUFFIXES
+
+
+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 _validated_source_options(source_options):
+ if source_options is None:
+ return {}
+ if not isinstance(source_options, Mapping):
+ raise ValueError("source_options must be a mapping.")
+ return dict(source_options)
+
+
+def _validate_source_kerberos(paths, source_options):
+ source_principal = (
+ source_options.get("security.kerberos.login.principal")
+ or source_options.get("security.principal")
+ )
+ source_keytab = (
+ source_options.get("security.kerberos.login.keytab")
+ or source_options.get("security.keytab")
+ )
+ if not source_principal and not source_keytab:
+ return
+ if bool(source_principal) != bool(source_keytab):
+ raise ValueError(
+ "Source Kerberos principal and keytab must be both set or both "
+ "unset.")
+ if not any(
+ urlparse(_source_path_text(path)).scheme.lower()
+ in ("hdfs", "viewfs") for path in paths):
+ return
+ raise ValueError(
+ "HDF5 sources cannot use an explicit Kerberos keytab in a shared "
+ "process because kinit overwrites process-global credentials. "
+ "Run the load in a process-isolated worker with a pre-acquired "
+ "ticket cache and omit the source principal and keytab options.")
+
+
+def _require_seekable(stream, source):
+ required = ("read", "seek", "tell")
+ if any(not callable(getattr(stream, method, None)) for method in required):
+ raise ValueError(
+ "HDF5 source stream must be seekable: %s" % source.path)
+ seekable = getattr(stream, "seekable", None)
+ if callable(seekable) and not seekable():
+ raise ValueError(
+ "HDF5 source stream must be seekable: %s" % source.path)
+ try:
+ stream.seek(stream.tell())
+ except (OSError, TypeError, ValueError) as error:
+ raise ValueError(
+ "HDF5 source stream must be seekable: %s" % source.path
+ ) from error
+
+
+def _strict_arrow_table(data, target_schema, source, batch_index):
+ if isinstance(data, pa.RecordBatch):
+ table = pa.Table.from_batches([data])
+ elif isinstance(data, pa.Table):
+ table = data
+ else:
+ raise ValueError(
+ "HDF5 transform must return Arrow data or an iterable of Arrow
data.")
+
+ missing = [
+ name for name in target_schema.names if name not in table.column_names
+ ]
+ if missing:
+ raise ValueError(
+ "HDF5 batch %d from %s is missing columns: %s"
+ % (batch_index, source.path, missing))
+ extra = [
+ name for name in table.column_names if name not in target_schema.names
+ ]
+ if extra:
+ raise ValueError(
+ "HDF5 batch %d from %s has unexpected columns: %s"
+ % (batch_index, source.path, extra))
+ if table.column_names != target_schema.names:
+ raise ValueError(
+ "HDF5 batch %d from %s has columns in the wrong order: %s; "
+ "expected %s."
+ % (batch_index, source.path, table.column_names,
+ target_schema.names))
+ try:
+ _validate_nested_nullability(table, target_schema)
+ if table.schema.equals(target_schema, check_metadata=False):
+ return table
+ casted = table.cast(target_schema, safe=True)
+ _validate_nested_nullability(casted, target_schema)
+ return casted
+ except (ValueError, TypeError, NotImplementedError) as error:
+ raise ValueError(
+ "HDF5 batch %d from %s cannot be converted to the table schema: %s"
+ % (batch_index, source.path, error)) from error
+
+
+def _validate_nested_nullability(table, schema):
+ for field, column in zip(schema, table.columns):
+ for chunk in column.chunks:
+ _validate_array_nullability(chunk, field, field.name)
+
+
+def _validate_array_nullability(array, field, path):
+ if not field.nullable and array.null_count:
+ raise ValueError(
+ "non-nullable field %s contains %d null value(s)"
+ % (path, array.null_count))
+
+ target_type = field.type
+ source_type = array.type
+ if (pa.types.is_list(target_type)
+ or pa.types.is_large_list(target_type)
+ or pa.types.is_fixed_size_list(target_type)):
+ if not (pa.types.is_list(source_type)
+ or pa.types.is_large_list(source_type)
+ or pa.types.is_fixed_size_list(source_type)):
+ return
+ _validate_array_nullability(
+ pc.list_flatten(array),
+ target_type.value_field,
+ "%s.%s" % (path, target_type.value_field.name),
+ )
+ return
+
+ if pa.types.is_map(target_type):
+ if not pa.types.is_map(source_type):
+ return
+ start = array.offsets[0].as_py()
+ stop = array.offsets[-1].as_py()
+ length = stop - start
+ offsets = pc.subtract(
+ array.offsets,
+ pa.scalar(start, type=array.offsets.type),
+ )
+ entries = pa.StructArray.from_arrays(
+ [array.keys.slice(start, length),
+ array.items.slice(start, length)],
+ fields=[source_type.key_field, source_type.item_field],
+ )
+ logical_entries = pc.list_flatten(pa.ListArray.from_arrays(
+ offsets,
+ entries,
+ mask=pc.is_null(array),
+ ))
+ _validate_array_nullability(
+ logical_entries.field(0), target_type.key_field,
+ "%s.%s" % (path, target_type.key_field.name))
+ _validate_array_nullability(
+ logical_entries.field(1), target_type.item_field,
+ "%s.%s" % (path, target_type.item_field.name))
+ return
+
+ if pa.types.is_struct(target_type):
+ if not pa.types.is_struct(source_type):
+ return
+ parent_valid = pc.is_valid(array) if array.null_count else None
+ for index, child_field in enumerate(target_type):
+ child = array.field(index)
+ if parent_valid is not None:
+ child = pc.filter(child, parent_valid)
+ _validate_array_nullability(
+ child,
+ child_field,
+ "%s.%s" % (path, child_field.name),
+ )
+
+
+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(
+ "HDF5 transform must return Arrow data or an iterable of Arrow
data.")
+ try:
+ return iter(transformed)
+ except TypeError as error:
+ raise ValueError(
+ "HDF5 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()
diff --git a/paimon-python/pypaimon/tests/file_io_test.py
b/paimon-python/pypaimon/tests/file_io_test.py
index f261b0a73b..9803a53790 100644
--- a/paimon-python/pypaimon/tests/file_io_test.py
+++ b/paimon-python/pypaimon/tests/file_io_test.py
@@ -21,13 +21,14 @@ import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
+from urllib.parse import urlparse
import pyarrow.fs as pafs
from pypaimon.common.file_io import create_temp_path
from pypaimon.common.options import Options
from pypaimon.common.options.config import OssOptions
-from pypaimon.filesystem.local_file_io import LocalFileIO
+from pypaimon.filesystem.local_file_io import LocalFileIO, _file_uri_path
from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO, _pyarrow_lt_7
@@ -54,6 +55,22 @@ class FileIOTest(unittest.TestCase):
# Test bucket and path
self.assertEqual(file_io.to_filesystem_path("s3://my-bucket/path/to/file.txt"),
"my-bucket/path/to/file.txt")
+ self.assertEqual(
+ file_io.to_filesystem_path(
+ "s3://my-bucket/path/episode%23copy%3F1.h5"),
+ "my-bucket/path/episode%23copy%3F1.h5",
+ )
+ self.assertEqual(
+ file_io.to_filesystem_path(
+ "s3://my-bucket/path/episode%2523copy.h5"),
+ "my-bucket/path/episode%2523copy.h5",
+ )
+ self.assertEqual(
+ file_io.to_filesystem_path(
+ "s3://my-bucket/t/p=a%2Fb/data.parquet"),
+ "my-bucket/t/p=a%2Fb/data.parquet",
+ )
+
self.assertEqual(file_io.to_filesystem_path("oss://my-bucket/path/to/file.txt"),
"my-bucket/path/to/file.txt")
@@ -167,6 +184,11 @@ class FileIOTest(unittest.TestCase):
"/tmp/path/to/file.txt")
self.assertEqual(file_io.to_filesystem_path("file:///path/to/file.txt"),
"/path/to/file.txt")
+ self.assertEqual(
+ file_io.to_filesystem_path(
+ "file:///tmp/episode%20%E4%B8%AD%E6%96%87.h5"),
+ "/tmp/episode 中文.h5",
+ )
# Test empty paths
self.assertEqual(file_io.to_filesystem_path("file://"), ".")
@@ -209,6 +231,25 @@ class FileIOTest(unittest.TestCase):
self.assertEqual(s3_file_io.to_filesystem_path("C:\\path\\to\\file.txt"),
"C:\\path\\to\\file.txt")
+ def test_standard_windows_and_unc_file_uris_round_trip(self):
+ self.assertEqual(
+ r"C:\data set\episode.h5",
+ _file_uri_path(
+ urlparse("file:///C:/data%20set/episode.h5"), windows=True),
+ )
+ self.assertEqual(
+ r"\\server\share\episode.h5",
+ _file_uri_path(
+ urlparse("file://server/share/episode.h5"), windows=True),
+ )
+
+ def test_local_directory_listing_propagates_permission_errors(self):
+ with tempfile.TemporaryDirectory() as directory:
+ file_io = LocalFileIO(directory, Options({}))
+ with patch.object(Path, "iterdir", side_effect=PermissionError):
+ with self.assertRaises(PermissionError):
+ file_io.list_status(directory)
+
def test_path_normalization(self):
"""Test path normalization (multiple slashes)."""
file_io = LocalFileIO("file:///tmp/warehouse", Options({}))
diff --git a/paimon-python/pypaimon/tests/hdfs_native_test.py
b/paimon-python/pypaimon/tests/hdfs_native_test.py
index 957ee8f123..ae8d7d673a 100644
--- a/paimon-python/pypaimon/tests/hdfs_native_test.py
+++ b/paimon-python/pypaimon/tests/hdfs_native_test.py
@@ -648,6 +648,14 @@ class ToFilesystemPathTest(unittest.TestCase):
"/foo/bar",
)
+ def test_hdfs_uri_preserves_paimon_escaped_partition_path(self):
+ fio = self._make("hdfs://ns1/")
+ self.assertEqual(
+ fio.to_filesystem_path(
+ "hdfs://ns1/warehouse/t/p=a%2Fb/data.parquet"),
+ "/warehouse/t/p=a%2Fb/data.parquet",
+ )
+
def test_hdfs_uri_different_ns_unchanged(self):
fio = self._make("hdfs://ns1/")
self.assertEqual(
diff --git a/paimon-python/pypaimon/tests/multimodal_hdf5_test.py
b/paimon-python/pypaimon/tests/multimodal_hdf5_test.py
new file mode 100644
index 0000000000..27063cefc2
--- /dev/null
+++ b/paimon-python/pypaimon/tests/multimodal_hdf5_test.py
@@ -0,0 +1,949 @@
+# 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
+import shutil
+import sys
+import tempfile
+import unittest
+import warnings
+from dataclasses import fields
+from functools import partial
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import numpy as np
+import pyarrow as pa
+import pyarrow.fs as pafs
+
+import pypaimon.multimodal as pmm
+from pypaimon.common.options import Options
+from pypaimon.filesystem.pyarrow_file_io import LegacyOssDirectoryListingError
+from pypaimon.multimodal.hdf5 import (
+ _Hdf5SourceFileIO,
+ _normalize_source_path,
+ _qualified_status_path,
+ _strict_arrow_table,
+ load_from_hdf5,
+)
+
+try:
+ import h5py
+except ImportError:
+ h5py = None
+
+
+_OPTIONS = {
+ "file.format": "parquet",
+ "vector.file.format": "parquet",
+ "blob-as-descriptor": "false",
+}
+
+
+class Hdf5RuntimeContractTest(unittest.TestCase):
+
+ def test_hdf5_load_requires_python_38_or_newer(self):
+ with patch(
+ "pypaimon.multimodal.hdf5.sys.version_info", (3, 7)):
+ with self.assertRaisesRegex(RuntimeError, "Python 3.8 or newer"):
+ load_from_hdf5(Mock(), [], transform=lambda h5, source: ())
+
+ def test_hdf5_source_io_decodes_only_its_external_uri(self):
+ backend = Mock()
+ backend.to_filesystem_path.return_value = (
+ "bucket/episode%23copy%3F1.h5")
+ backend.get_file_status.return_value = "status"
+ resolver = Mock()
+ resolver._get_fileio.return_value = backend
+
+ with patch(
+ "pypaimon.multimodal.hdf5.ResolvingFileIO",
+ return_value=resolver):
+ source_io = _Hdf5SourceFileIO(Options({}))
+ self.assertEqual(
+ "status",
+ source_io.get_file_status(
+ "s3://bucket/episode%23copy%3F1.h5"),
+ )
+
+ backend.get_file_status.assert_called_once_with(
+ "bucket/episode#copy?1.h5")
+
+
+class _WriterProxy:
+
+ def __init__(self, delegate, write_error=None):
+ self.delegate = delegate
+ self.write_error = write_error
+ self.abort_count = 0
+ self.close_count = 0
+
+ def write_arrow(self, table):
+ if self.write_error is not None:
+ raise self.write_error
+ return self.delegate.write_arrow(table)
+
+ def prepare_commit(self):
+ return self.delegate.prepare_commit()
+
+ def abort(self):
+ self.abort_count += 1
+ return self.delegate.abort()
+
+ def close(self):
+ self.close_count += 1
+ return self.delegate.close()
+
+
+class _CommitterProxy:
+
+ def __init__(self, delegate, commit_error=None, raise_after_commit=False):
+ self.delegate = delegate
+ self.commit_error = commit_error
+ self.raise_after_commit = raise_after_commit
+ self.commit_count = 0
+ self.close_count = 0
+
+ def add_commit_callback(self, callback):
+ return self.delegate.add_commit_callback(callback)
+
+ def commit(self, messages):
+ self.commit_count += 1
+ if self.commit_error is not None and not self.raise_after_commit:
+ raise self.commit_error
+ result = self.delegate.commit(messages)
+ if self.commit_error is not None:
+ raise self.commit_error
+ return result
+
+ def close(self):
+ self.close_count += 1
+ return self.delegate.close()
+
+
+class _WriteBuilderProxy:
+
+ def __init__(self, writer, committer):
+ self.writer = writer
+ self.committer = committer
+
+ def new_write(self):
+ return self.writer
+
+ def new_commit(self):
+ return self.committer
+
+
+class _TrackedSourceStream(io.BytesIO):
+
+ def __init__(self, data, source_file_io):
+ super().__init__(data)
+ self.source_file_io = source_file_io
+
+ def seekable(self):
+ return self.source_file_io.seekable
+
+ def close(self):
+ if not self.closed:
+ self.source_file_io.closed_stream_count += 1
+ super().close()
+
+
+class _RemoteSourceFileIO:
+
+ def __init__(
+ self,
+ objects=None,
+ directories=None,
+ seekable=True,
+ native_stream=False):
+ self.objects = {
+ self._filesystem_path(path): value
+ for path, value in dict(objects or {}).items()
+ }
+ self.directories = {
+ self._filesystem_path(path): [
+ self._filesystem_path(child) for child in children]
+ for path, children in dict(directories or {}).items()
+ }
+ self.seekable = seekable
+ self.native_stream = native_stream
+ self.opened_paths = []
+ self.streams = []
+ self.closed_stream_count = 0
+ self.close_count = 0
+ self.list_error = None
+
+ @staticmethod
+ def _filesystem_path(path):
+ from urllib.parse import urlparse
+ parsed = urlparse(path)
+ return "%s%s" % (parsed.netloc, parsed.path)
+
+ def _status(self, path):
+ if path in self.objects:
+ file_type = pafs.FileType.File
+ elif path in self.directories:
+ file_type = pafs.FileType.Directory
+ else:
+ raise FileNotFoundError(path)
+ return pafs.FileInfo(path, file_type)
+
+ def _get_fileio(self, path):
+ return self
+
+ def to_filesystem_path(self, path):
+ return self._filesystem_path(path)
+
+ def get_file_status(self, path):
+ return self._status(path)
+
+ def list_status(self, path):
+ if self.list_error is not None:
+ raise self.list_error
+ return [self._status(child) for child in self.directories[path]]
+
+ def new_input_stream(self, path):
+ self.opened_paths.append(path)
+ if self.native_stream:
+ stream = pa.BufferReader(self.objects[path])
+ else:
+ stream = _TrackedSourceStream(self.objects[path], self)
+ self.streams.append(stream)
+ return stream
+
+ def close(self):
+ self.close_count += 1
+
+
[email protected](h5py is None, "h5py is not installed")
+class MultimodalHdf5Test(unittest.TestCase):
+
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp(prefix="pypaimon_hdf5_")
+ self.source_dir = Path(self.temp_dir) / "source"
+ self.warehouse = os.path.join(self.temp_dir, "warehouse")
+ self.source_dir.mkdir()
+ self.conn = pmm.connect(options={"warehouse": self.warehouse})
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ def _write_source(self, relative_path, offset=0):
+ path = self.source_dir / relative_path
+ path.parent.mkdir(parents=True, exist_ok=True)
+ with h5py.File(path, "w") as h5:
+ h5.create_dataset(
+ "values",
+ data=np.arange(
+ offset, offset + 12, dtype=np.float32).reshape(4, 3),
+ )
+ images = h5.create_dataset(
+ "images", (4,), dtype=h5py.vlen_dtype(np.dtype("uint8")))
+ for index in range(4):
+ jpeg = (b"\xff\xd8\xff" +
+ ("%s-%d" % (path.name, index)).encode("utf-8") +
+ b"\xff\xd9")
+ images[index] = np.frombuffer(
+ jpeg, dtype=np.uint8)
+ return path
+
+ @staticmethod
+ def _schema():
+ return pa.schema([
+ pa.field("source", pa.string(), nullable=False),
+ pa.field("frame_index", pa.int32(), nullable=False),
+ pa.field("score", pa.float64()),
+ pa.field("value", pa.list_(pa.float32(), 3), nullable=False),
+ pa.field("image", pa.large_binary()),
+ ])
+
+ def _create_table(self, name="frames"):
+ return self.conn.create_table(
+ name, schema=self._schema(), options=_OPTIONS)
+
+ def _load(self, table, paths, *, transform, source_options=None):
+ with patch.object(self.conn, "get_table", return_value=table):
+ return self.conn.load_from_hdf5(
+ table.identifier,
+ paths,
+ transform=transform,
+ source_options=source_options,
+ )
+
+ @staticmethod
+ def _transform(h5, source):
+ values = np.asarray(h5["values"][:], dtype=np.float32)
+ images = [
+ np.asarray(value, dtype=np.uint8).tobytes()
+ for value in h5["images"][:]
+ ]
+ for begin in range(0, len(values), 2):
+ end = min(begin + 2, len(values))
+ # Deliberately let Arrow infer int64, list<double>, and binary.
+ # HDF5 loading may safely cast after validating exact columns.
+ yield pa.Table.from_pydict({
+ "source": [source.name] * (end - begin),
+ "frame_index": list(range(begin, end)),
+ "score": [None if index == 0 else float(index)
+ for index in range(begin, end)],
+ "value": values[begin:end].tolist(),
+ "image": images[begin:end],
+ })
+
+ def _instrument_write(
+ self,
+ table,
+ *,
+ write_error=None,
+ commit_error=None,
+ raise_after_commit=False):
+ builder = table.raw_table.new_batch_write_builder()
+ writer = _WriterProxy(builder.new_write(), write_error=write_error)
+ committer = _CommitterProxy(
+ builder.new_commit(),
+ commit_error=commit_error,
+ raise_after_commit=raise_after_commit,
+ )
+ proxy = _WriteBuilderProxy(writer, committer)
+ return proxy, writer, committer
+
+ def test_load_from_hdf5_streams_multiple_files_in_one_snapshot(self):
+ first = self._write_source("a.hdf5", 0)
+ self._write_source("nested/b.h5", 100)
+ table = self._create_table()
+ table.add(pa.Table.from_pydict({
+ "source": ["seed"],
+ "frame_index": [-1],
+ "score": [None],
+ "value": [[-1.0, -1.0, -1.0]],
+ "image": [b"seed"],
+ }, schema=self._schema()))
+ before_snapshot =
table.raw_table.snapshot_manager().get_latest_snapshot()
+ seen = []
+ seen_paths = []
+
+ def transform(h5, source):
+ seen.append(source.name)
+ seen_paths.append(source.path)
+ yield from self._transform(h5, source)
+
+ result = self._load(
+ table,
+ [self.source_dir, first, first.resolve()],
+ transform=transform,
+ )
+
+ self.assertEqual(2, result.file_count)
+ self.assertEqual(4, result.batch_count)
+ self.assertEqual(8, result.row_count)
+ self.assertEqual(before_snapshot.id + 1, result.snapshot_id)
+ self.assertEqual(
+ result.snapshot_id,
+ table.raw_table.snapshot_manager().get_latest_snapshot().id,
+ )
+ self.assertEqual(["a.hdf5", "b.h5"], seen)
+ self.assertTrue(all(path.startswith("file://") for path in seen_paths))
+ rows = table.scan().select([
+ "source", "frame_index", "score", "value", "image",
+ ]).to_arrow().to_pylist()
+ self.assertEqual(9, len(rows))
+ self.assertEqual({"seed", "a.hdf5", "b.h5"}, {
+ row["source"] for row in rows
+ })
+ self.assertTrue(any(row["score"] is None for row in rows))
+ self.assertTrue(all(
+ row["image"] is None or isinstance(row["image"], bytes)
+ for row in rows
+ ))
+ self.assertTrue(all(
+ row["image"].startswith(b"\xff\xd8\xff")
+ and row["image"].endswith(b"\xff\xd9")
+ for row in rows if row["source"] != "seed"
+ ))
+ self.assertEqual(3, len(rows[-1]["value"]))
+
+ def test_load_from_hdf5_accepts_paths_and_reappends(self):
+ first = self._write_source("a.hdf5", 0)
+ self._write_source("nested/b.h5", 100)
+ table = self._create_table()
+
+ single = self._load(table, first, transform=self._transform)
+ duplicate_list = self._load(
+ table,
+ [first, first.resolve(), first.resolve().as_uri()],
+ transform=self._transform,
+ )
+ directory = self._load(
+ table, self.source_dir, transform=self._transform)
+
+ self.assertEqual((1, 2, 4), (
+ single.file_count, single.batch_count, single.row_count))
+ self.assertEqual((1, 2, 4), (
+ duplicate_list.file_count,
+ duplicate_list.batch_count,
+ duplicate_list.row_count,
+ ))
+ self.assertEqual((2, 4, 8), (
+ directory.file_count, directory.batch_count, directory.row_count))
+ self.assertEqual(16, table.scan().to_arrow().num_rows)
+ self.assertEqual(
+ ["file_count", "batch_count", "row_count", "snapshot_id"],
+ [field.name for field in fields(type(directory))],
+ )
+ self.assertEqual(["path"], [
+ field.name for field in fields(pmm.Hdf5File)
+ ])
+
+ def test_hdf5_file_exposes_decoded_local_path(self):
+ source = self._write_source("episode 中文.h5")
+ table = self._create_table("local_context")
+ seen = []
+
+ def transform(h5, context):
+ seen.append(context)
+ yield from self._transform(h5, context)
+
+ self._load(table, source, transform=transform)
+
+ self.assertEqual(source.resolve(), seen[0].local_path)
+ self.assertEqual("episode 中文.h5", seen[0].name)
+ self.assertEqual("episode 中文", seen[0].stem)
+ self.assertIsNone(
+ pmm.Hdf5File("s3://source-bucket/episode.h5").local_path)
+
+ expected = source.resolve()
+ context = pmm.Hdf5File(expected.as_uri())
+ with patch.object(Path, "resolve", side_effect=AssertionError):
+ self.assertEqual(expected, context.local_path)
+
+ def test_windows_drive_paths_are_normalized_as_file_uris(self):
+ path = r"C:\data set\episode.h5"
+ with warnings.catch_warnings():
+ warnings.simplefilter("error", DeprecationWarning)
+ self.assertEqual(
+ "file:///C:/data%20set/episode.h5",
+ _normalize_source_path(path),
+ )
+
+ def test_qualified_status_path_covers_fileio_path_shapes(self):
+ cases = (
+ ("s3://bucket/root", "s3://bucket/root/a.h5",
+ "s3://bucket/root/a.h5"),
+ ("file:///tmp/root", "/tmp/root/a.h5",
+ Path("/tmp/root/a.h5").resolve().as_uri()),
+ ("file:///C:/root", r"C:\root\a.h5",
+ "file:///C:/root/a.h5"),
+ ("hdfs://namenode:8020/root", "/root/a.h5",
+ "hdfs://namenode:8020/root/a.h5"),
+ ("viewfs://cluster/root", "root/a.h5",
+ "viewfs://cluster/root/a.h5"),
+ ("s3://bucket/root", "bucket/root/a.h5",
+ "s3://bucket/root/a.h5"),
+ ("s3://bucket/root", "root/a.h5",
+ "s3://bucket/root/a.h5"),
+ ("oss://bucket/root", "bucket/root/a.h5",
+ "oss://bucket/root/a.h5"),
+ ("gs://bucket/root", "root/a.h5",
+ "gs://bucket/root/a.h5"),
+ ("s3://bucket/root", "bucket/root/episode#backup.h5",
+ "s3://bucket/root/episode%23backup.h5"),
+ ("hdfs://namenode/root", "/root/episode?copy.hdf5",
+ "hdfs://namenode/root/episode%3Fcopy.hdf5"),
+ ("s3://bucket/root", "bucket/root/episode%23copy.h5",
+ "s3://bucket/root/episode%2523copy.h5"),
+ )
+ for parent, status_path, expected in cases:
+ with self.subTest(parent=parent, status_path=status_path):
+ status = pafs.FileInfo(status_path, pafs.FileType.File)
+ self.assertEqual(
+ expected, _qualified_status_path(parent, status))
+
+ def test_load_from_hdf5_reads_recursive_remote_sources(self):
+ root = "s3://source-bucket/episodes"
+ first = root + "/a.hdf5"
+ nested = root + "/nested"
+ second = nested + "/b.h5"
+ source_file_io = _RemoteSourceFileIO(
+ objects={
+ first: self._write_source("remote-a.hdf5", 0).read_bytes(),
+ second: self._write_source("remote-b.h5", 100).read_bytes(),
+ },
+ directories={
+ root: [nested, first],
+ nested: [second],
+ },
+ native_stream=True,
+ )
+ source_options = {
+ "fs.s3.endpoint": "https://source.example.com";,
+ "fs.s3.access-key": "source-key",
+ }
+ table = self._create_table("remote_frames")
+ seen = []
+
+ def transform(h5, source):
+ seen.append((source.path, source.name, source.stem))
+ yield from self._transform(h5, source)
+
+ with patch(
+ "pypaimon.multimodal.hdf5.ResolvingFileIO",
+ return_value=source_file_io) as resolving_file_io:
+ result = self._load(
+ table,
+ [root, first],
+ transform=transform,
+ source_options=source_options,
+ )
+
+ self.assertEqual((2, 4, 8), (
+ result.file_count, result.batch_count, result.row_count))
+ self.assertEqual([
+ "source-bucket/episodes/a.hdf5",
+ "source-bucket/episodes/nested/b.h5",
+ ], source_file_io.opened_paths)
+ self.assertEqual([
+ (first, "a.hdf5", "a"),
+ (second, "b.h5", "b"),
+ ], seen)
+ self.assertTrue(all(stream.closed for stream in
source_file_io.streams))
+ self.assertEqual(1, source_file_io.close_count)
+ actual_options = resolving_file_io.call_args.args[0].to_map()
+ self.assertEqual(source_options, actual_options)
+ self.assertEqual(
+ result.snapshot_id,
+ table.raw_table.snapshot_manager().get_latest_snapshot().id,
+ )
+
+ def test_non_seekable_remote_stream_aborts_and_closes_resources(self):
+ source = "s3://source-bucket/episode.h5"
+ source_file_io = _RemoteSourceFileIO(
+ objects={
+ source: self._write_source("remote.h5").read_bytes(),
+ },
+ seekable=False,
+ )
+ table = self._create_table("non_seekable")
+ proxy, writer, committer = self._instrument_write(table)
+
+ with patch(
+ "pypaimon.multimodal.hdf5.ResolvingFileIO",
+ return_value=source_file_io), patch.object(
+ table.raw_table,
+ "new_batch_write_builder",
+ return_value=proxy):
+ with self.assertRaisesRegex(ValueError, "seekable"):
+ self._load(table, source, transform=self._transform)
+
+ self.assertEqual(1, writer.abort_count)
+ self.assertEqual(1, writer.close_count)
+ self.assertEqual(1, committer.close_count)
+ self.assertEqual(1, source_file_io.closed_stream_count)
+ self.assertEqual(1, source_file_io.close_count)
+ self.assertIsNone(
+ table.raw_table.snapshot_manager().get_latest_snapshot())
+
+ def test_legacy_oss_directory_error_recommends_explicit_files(self):
+ source = "oss://source-bucket/episodes"
+ source_file_io = _RemoteSourceFileIO()
+ source_file_io.list_error = LegacyOssDirectoryListingError(
+ "Listing OSS directories is not supported with PyArrow < 16")
+ table = self._create_table("legacy_oss")
+ new_builder = Mock()
+
+ with patch(
+ "pypaimon.multimodal.hdf5.ResolvingFileIO",
+ return_value=source_file_io), patch.object(
+ table.raw_table,
+ "new_batch_write_builder",
+ new_builder):
+ with self.assertRaisesRegex(
+ ValueError, "pass explicit HDF5 file paths"):
+ self._load(table, source, transform=self._transform)
+
+ new_builder.assert_not_called()
+ self.assertEqual(1, source_file_io.close_count)
+
+ def test_load_from_hdf5_rejects_non_arrow_and_empty_output(self):
+ source = self._write_source("episode.h5")
+ cases = (
+ (lambda h5, info: {"frame_index": [0]}, "Arrow data"),
+ (lambda h5, info: iter(()), "produced no rows"),
+ )
+ for index, (transform, message) in enumerate(cases):
+ table = self._create_table("invalid_output_%d" % index)
+ with self.subTest(message=message):
+ with self.assertRaisesRegex(ValueError, message):
+ self._load(table, source, transform=transform)
+ self.assertIsNone(
+ table.raw_table.snapshot_manager().get_latest_snapshot())
+
+ def test_each_discovered_file_must_produce_rows(self):
+ valid = self._write_source("valid.h5")
+ empty = self._write_source("empty.h5")
+ table = self._create_table("per_source_rows")
+ proxy, writer, committer = self._instrument_write(table)
+
+ def transform(h5, source):
+ if source.name == "empty.h5":
+ return iter(())
+ return self._transform(h5, source)
+
+ with patch.object(
+ table.raw_table,
+ "new_batch_write_builder",
+ return_value=proxy):
+ with self.assertRaisesRegex(
+ ValueError, "empty.h5.*produced no rows"):
+ self._load(table, [valid, empty], transform=transform)
+
+ self.assertEqual(1, writer.abort_count)
+ self.assertEqual(0, committer.commit_count)
+ self.assertIsNone(
+ table.raw_table.snapshot_manager().get_latest_snapshot())
+
+ def test_strict_schema_rejects_nested_non_nullable_values(self):
+ source = pmm.Hdf5File("file:///tmp/episode.h5")
+ cases = (
+ (
+ pa.field(
+ "items",
+ pa.list_(pa.field("item", pa.int32(), nullable=False)),
+ ),
+ [[1, None]],
+ "items.item",
+ ),
+ (
+ pa.field(
+ "vector",
+ pa.list_(
+ pa.field("item", pa.float32(), nullable=False), 2),
+ ),
+ [[1.0, None]],
+ "vector.item",
+ ),
+ (
+ pa.field(
+ "attributes",
+ pa.map_(
+ pa.string(),
+ pa.field("value", pa.int32(), nullable=False),
+ ),
+ ),
+ [[("key", None)]],
+ "attributes.value",
+ ),
+ )
+
+ for field, values, path in cases:
+ schema = pa.schema([field])
+ batch = pa.Table.from_arrays(
+ [pa.array(values, type=field.type)], schema=schema)
+ with self.subTest(path=path):
+ with self.assertRaisesRegex(ValueError, path):
+ _strict_arrow_table(batch, schema, source, 0)
+
+ def test_orc_load_rejects_null_array_elements_and_map_values(self):
+ source = self._write_source("nested_nulls.h5")
+ cases = (
+ (
+ "orc_array_null",
+ pa.field(
+ "nested",
+ pa.list_(pa.field("item", pa.int32(), nullable=False)),
+ ),
+ [[1, None]],
+ ),
+ (
+ "orc_map_null",
+ pa.field(
+ "nested",
+ pa.map_(
+ pa.string(),
+ pa.field("value", pa.int32(), nullable=False),
+ ),
+ ),
+ [[("key", None)]],
+ ),
+ )
+ for name, nested_field, values in cases:
+ schema = pa.schema([
+ pa.field("id", pa.int32(), nullable=False),
+ nested_field,
+ ])
+ table = self.conn.create_table(
+ name,
+ schema=schema,
+ options={
+ "file.format": "orc",
+ "blob-as-descriptor": "false",
+ },
+ )
+
+ def transform(h5, context, field=nested_field, data=values):
+ return pa.Table.from_arrays(
+ [pa.array([1], type=pa.int32()),
+ pa.array(data, type=field.type)],
+ names=["id", "nested"],
+ )
+
+ with self.subTest(name=name):
+ with self.assertRaisesRegex(ValueError, "nested"):
+ self._load(table, source, transform=transform)
+ self.assertIsNone(
+ table.raw_table.snapshot_manager().get_latest_snapshot())
+
+ def test_nested_nullability_checks_only_logical_sliced_map_values(self):
+ item_field = pa.field("value", pa.int32(), nullable=False)
+ map_type = pa.map_(pa.string(), item_field)
+ schema = pa.schema([pa.field("attributes", map_type)])
+ values = pa.array(
+ [[("outside", None)], [("inside", 1)]], type=map_type)
+ sliced = pa.Table.from_arrays(
+ [values.slice(1, 1)], schema=schema)
+
+ result = _strict_arrow_table(
+ sliced, schema, pmm.Hdf5File("file:///tmp/episode.h5"), 0)
+
+ self.assertEqual(
+ [[("inside", 1)]], result.column("attributes").to_pylist())
+
+ def test_nested_nullability_ignores_entries_hidden_by_null_maps(self):
+ item_field = pa.field("value", pa.int32(), nullable=False)
+ map_type = pa.map_(pa.string(), item_field)
+ schema = pa.schema([pa.field("attributes", map_type)])
+ offsets = pa.array([0, 1, 2], type=pa.int32())
+ entries = pa.StructArray.from_arrays(
+ [
+ pa.array(["hidden", "visible"]),
+ pa.array([None, 1], type=pa.int32()),
+ ],
+ fields=[map_type.key_field, map_type.item_field],
+ )
+ validity = pa.array([False, True], type=pa.bool_()).buffers()[1]
+ values = pa.MapArray.from_buffers(
+ map_type,
+ 2,
+ [validity, offsets.buffers()[1]],
+ children=[entries],
+ )
+ table = pa.Table.from_arrays([values], schema=schema)
+
+ result = _strict_arrow_table(
+ table, schema, pmm.Hdf5File("file:///tmp/episode.h5"), 0)
+
+ self.assertEqual(
+ [None, [("visible", 1)]],
+ result.column("attributes").to_pylist(),
+ )
+
+ def test_load_from_hdf5_rejects_invalid_schemas(self):
+ source = self._write_source("episode.hdf5")
+ good = {
+ "source": ["episode.hdf5"],
+ "frame_index": [0],
+ "score": [None],
+ "value": [[1.0, 2.0, 3.0]],
+ "image": [b"jpeg"],
+ }
+ cases = []
+ missing = dict(good)
+ missing.pop("score")
+ cases.append((missing, "missing columns"))
+ extra = dict(good, unexpected=[1])
+ cases.append((extra, "unexpected columns"))
+ incompatible = dict(good, frame_index=["not-an-int"])
+ cases.append((incompatible, "cannot be converted"))
+ invalid_vector = dict(good, value=[[1.0, 2.0]])
+ cases.append((invalid_vector, "cannot be converted"))
+ null_non_nullable = dict(good, source=[None])
+ cases.append((null_non_nullable, "cannot be converted"))
+
+ for index, (data, message) in enumerate(cases):
+ table = self._create_table("strict_%d" % index)
+
+ def transform(h5, info, batch=data):
+ yield pa.Table.from_pydict(batch)
+
+ with self.subTest(message=message):
+ with self.assertRaisesRegex(ValueError, message):
+ self._load(table, source, transform=transform)
+ self.assertIsNone(
+ table.raw_table.snapshot_manager().get_latest_snapshot())
+
+ def test_hdfs_source_rejects_explicit_keytab_for_any_target(self):
+ table = self._create_table("kerberos_isolation")
+
+ with patch.object(
+ self.conn, "get_table", return_value=table), patch(
+ "pypaimon.multimodal.hdf5.ResolvingFileIO") as resolving:
+ with self.assertRaisesRegex(
+ ValueError, "process-isolated"):
+ self.conn.load_from_hdf5(
+ table.identifier,
+ "hdfs://source-ns/episode.h5",
+ transform=self._transform,
+ source_options={
+ "security.kerberos.login.principal": "source@REALM",
+ "security.kerberos.login.keytab": "/source.keytab",
+ },
+ )
+
+ resolving.assert_not_called()
+
+ def test_precommit_failures_abort_and_close_all_open_resources(self):
+ valid_source = self._write_source("valid.h5")
+ invalid_source = self.source_dir / "invalid.h5"
+ invalid_source.write_bytes(b"not hdf5")
+
+ def transform_failure(h5, source):
+ yield next(self._transform(h5, source))
+ raise RuntimeError("transform failed")
+
+ def schema_failure(h5, source, resource_state):
+ try:
+ yield pa.Table.from_pydict({"missing": [1]})
+ raise AssertionError("generator should have been closed")
+ finally:
+ resource_state["generator_closed"] = True
+
+ cases = (
+ (invalid_source, self._transform, None, OSError),
+ (valid_source, transform_failure, None, RuntimeError),
+ (valid_source, schema_failure, None, ValueError),
+ (valid_source, self._transform, RuntimeError("write failed"),
RuntimeError),
+ )
+ for index, (source, transform, write_error, error_type) in
enumerate(cases):
+ table = self._create_table("failure_%d" % index)
+ proxy, writer, committer = self._instrument_write(
+ table, write_error=write_error)
+ resource_state = {}
+ expects_generator_close = transform is schema_failure
+ if expects_generator_close:
+ transform = partial(
+ schema_failure, resource_state=resource_state)
+
+ def tracked_transform(h5, info, delegate=transform):
+ resource_state["h5"] = h5
+ return delegate(h5, info)
+
+ with self.subTest(error=error_type.__name__):
+ with patch.object(
+ table.raw_table,
+ "new_batch_write_builder",
+ return_value=proxy):
+ with self.assertRaises(error_type):
+ self._load(
+ table, source, transform=tracked_transform)
+ self.assertEqual(1, writer.abort_count)
+ self.assertEqual(1, writer.close_count)
+ self.assertEqual(1, committer.close_count)
+ self.assertEqual(0, committer.commit_count)
+ self.assertIsNone(
+ table.raw_table.snapshot_manager().get_latest_snapshot())
+ if "h5" in resource_state:
+ self.assertFalse(resource_state["h5"].id.valid)
+ if expects_generator_close:
+ self.assertTrue(resource_state.get("generator_closed"))
+
+ def test_empty_discovery_is_a_noop_without_creating_a_writer(self):
+ table = self._create_table()
+ new_builder = Mock()
+
+ with patch.object(
+ table.raw_table,
+ "new_batch_write_builder",
+ new_builder), patch.dict(sys.modules, {"h5py": None}):
+ for paths in ([], self.source_dir):
+ with self.subTest(paths=paths):
+ result = self._load(
+ table, paths, transform=self._transform)
+ self.assertEqual(
+ (0, 0, 0, None),
+ (
+ result.file_count,
+ result.batch_count,
+ result.row_count,
+ result.snapshot_id,
+ ),
+ )
+
+ new_builder.assert_not_called()
+ self.assertIsNone(
+ table.raw_table.snapshot_manager().get_latest_snapshot())
+
+ def test_discovery_failure_happens_before_writer_creation(self):
+ table = self._create_table()
+ new_builder = Mock()
+ missing = self.source_dir / "missing.h5"
+ with patch.object(
+ table.raw_table,
+ "new_batch_write_builder",
+ new_builder):
+ with self.assertRaisesRegex(
+ ValueError,
+ "HDF5 path does not exist: %s" % missing):
+ self._load(table, missing, transform=self._transform)
+ new_builder.assert_not_called()
+
+ def test_existing_file_with_unsupported_suffix_reports_suffix_only(self):
+ source = self.source_dir / "episode.txt"
+ source.write_text("not hdf5", encoding="utf-8")
+ table = self._create_table("unsupported_suffix")
+
+ with self.assertRaisesRegex(
+ ValueError,
+ "HDF5 file has unsupported suffix: %s" % source):
+ self._load(table, source, transform=self._transform)
+
+ def test_commit_exception_is_not_retried_or_aborted(self):
+ source = self._write_source("episode.h5")
+ table = self._create_table()
+ commit_error = RuntimeError("unknown commit result")
+ proxy, writer, committer = self._instrument_write(
+ table,
+ commit_error=commit_error,
+ raise_after_commit=True,
+ )
+
+ with patch.object(
+ table.raw_table,
+ "new_batch_write_builder",
+ return_value=proxy):
+ with self.assertRaisesRegex(RuntimeError, "unknown commit result"):
+ self._load(table, source, transform=self._transform)
+
+ self.assertEqual(1, committer.commit_count)
+ self.assertEqual(0, writer.abort_count)
+ self.assertEqual(1, writer.close_count)
+ self.assertEqual(1, committer.close_count)
+ self.assertEqual(4, table.scan().to_arrow().num_rows)
+ self.assertEqual(
+ 1, table.raw_table.snapshot_manager().get_latest_snapshot().id)
+
+ def test_hdf5_api_is_connection_only_and_has_no_managed_provenance(self):
+ table = self._create_table()
+ self.assertTrue(hasattr(self.conn, "load_from_hdf5"))
+ self.assertFalse(hasattr(table, "append_hdf5"))
+ self.assertFalse(hasattr(table, "from_hdf5"))
+ self.assertFalse(hasattr(pmm, "HDF5_SOURCE_PATH_COLUMN"))
+ self.assertFalse(hasattr(pmm, "HDF5_SOURCE_SHA256_COLUMN"))
+ self.assertFalse(hasattr(pmm, "Hdf5SourceDriftError"))
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
b/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
index 02f71f1af8..43c4972f0c 100644
--- a/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
+++ b/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
@@ -28,7 +28,10 @@ import pyarrow.fs as pafs
from pypaimon.common.options import Options
from pypaimon.common.options.config import OssOptions
-from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO
+from pypaimon.filesystem.pyarrow_file_io import (
+ LegacyOssDirectoryListingError,
+ PyArrowFileIO,
+)
TABLE_PATH = "oss://test-bucket/db-uuid.db/tbl-uuid"
@@ -235,7 +238,7 @@ class OssLegacyModeTest(unittest.TestCase):
"""Fail fast instead of the misleading raw NoSuchKey selector error."""
file_io = self._new_file_io(legacy=True)
- with self.assertRaises(RuntimeError) as ctx:
+ with self.assertRaises(LegacyOssDirectoryListingError) as ctx:
file_io.list_status(TABLE_PATH)
self.assertIn("pyarrow >= 16", str(ctx.exception))
file_io.filesystem.get_file_info.assert_not_called()
diff --git a/paimon-python/setup.py b/paimon-python/setup.py
index d864d5ce44..bf83fe4386 100644
--- a/paimon-python/setup.py
+++ b/paimon-python/setup.py
@@ -233,6 +233,10 @@ setup(
],
},
extras_require={
+ 'hdf5': [
+ # HDF5 loading is explicitly guarded and documented as Python 3.8+.
+ 'h5py>=3,<4; python_version>="3.8"',
+ ],
'ray': [
'ray>=2.10,<3; python_version>="3.8"',
],
