JingsongLi commented on code in PR #9411:
URL: https://github.com/apache/paimon/pull/9411#discussion_r3869755635
##########
paimon-python/pypaimon/filesystem/pyarrow_file_io.py:
##########
@@ -784,7 +788,8 @@ def to_filesystem_path(self, path: str) -> str:
from pyarrow.fs import S3FileSystem
parsed = urlparse(path)
- normalized_path = re.sub(r'/+', '/', parsed.path) if parsed.path else
''
+ normalized_path = (
+ unquote(re.sub(r'/+', '/', parsed.path)) if parsed.path else '')
Review Comment:
**[P1] Preserve Paimon-escaped filesystem paths**
This decodes values that are not merely URI escaping: Paimon deliberately
uses literal `%2F`, `%3A`, `%23`, and `%25` sequences in physical partition
directory names. I reproduced `s3://bucket/t/p=a%2Fb/...` becoming the native
key `bucket/t/p=a/b/...`, while
`native_plan_test.py::test_partition_path_keeps_existing_rust_path` explicitly
requires the `%2F` path when no legacy Python directory exists. Existing
Java/Rust-written tables can therefore become unreadable or address the wrong
object after upgrade. Please preserve the established literal-path behavior in
shared FileIO (including the analogous native-HDFS change) and handle HDF5
URI/display escaping without globally unquoting Paimon paths.
##########
paimon-python/pypaimon/multimodal/hdf5.py:
##########
@@ -0,0 +1,600 @@
+# 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
+
+
+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_kerberos_isolation(table, path_values, validated_options)
+ source_file_io = ResolvingFileIO(
+ 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_kerberos_isolation(table, 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
+
+ target_path = getattr(table.raw_table, "table_path", "")
+ if urlparse(target_path).scheme.lower() not in ("hdfs", "viewfs"):
+ return
Review Comment:
**[P1] Guard Kerberos state even for non-HDFS targets**
Returning here only proves that the current target is not HDFS; it does not
mean that no other HDFS client exists in the process. An explicit-keytab source
still runs `kinit` and overwrites process-global `KRB5CCNAME`. On this head I
reproduced a local target passing this check, the source changing the cache
from principal A's cache to principal B's cache, and `close()` leaving B's
cache active. Unrelated concurrent or later HDFS operations may then run under
the source identity. Please use a client-private credential cache, or
reject/require process isolation for explicit source keytabs whenever
process-wide isolation cannot be guaranteed.
##########
paimon-python/pypaimon/multimodal/hdf5.py:
##########
@@ -0,0 +1,600 @@
+# 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
+
+
+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_kerberos_isolation(table, path_values, validated_options)
+ source_file_io = ResolvingFileIO(
+ 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_kerberos_isolation(table, 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
+
+ target_path = getattr(table.raw_table, "table_path", "")
+ if urlparse(target_path).scheme.lower() not in ("hdfs", "viewfs"):
+ return
+ target_file_io = getattr(table.raw_table, "file_io", None)
+ target_properties = getattr(target_file_io, "properties", None)
+ target_options = (
+ target_properties.to_map()
+ if target_properties is not None
+ and callable(getattr(target_properties, "to_map", None))
+ else {}
+ )
+ target_principal = (
+ target_options.get("security.kerberos.login.principal")
+ or target_options.get("security.principal")
+ )
+ if target_principal != source_principal:
+ raise ValueError(
+ "HDF5 source and target use different Kerberos principals; "
+ "loading would overwrite process-global Kerberos credentials. "
+ "Use the same principal or isolate the load in another process.")
+
+
+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
+ _validate_array_nullability(
+ array.keys.slice(start, length), target_type.key_field,
+ "%s.%s" % (path, target_type.key_field.name))
+ _validate_array_nullability(
+ array.items.slice(start, length), target_type.item_field,
Review Comment:
**[P2] Ignore map entries hidden by null parents**
The `[start, stop)` slice includes physical child entries owned by null map
slots, although those entries are logically absent and do not need to satisfy
child nullability. I reproduced a valid nullable `MAP<STRING, INT NOT NULL>`
with logical rows `[None, [('visible', 1)]]`: `Table.validate(full=True)` and
both Parquet and ORC accept it, but this validator rejects a hidden null item
under the null parent. Please exclude offset ranges belonging to null parent
maps before recursively validating keys/items, and add a null-parent-map
regression test.
##########
paimon-python/pypaimon/multimodal/hdf5.py:
##########
@@ -0,0 +1,600 @@
+# 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
+
+
+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_kerberos_isolation(table, path_values, validated_options)
+ source_file_io = ResolvingFileIO(
Review Comment:
**[P2] Do not retain source-only AWS configuration globally**
When this temporary resolver first opens an S3/OSS source with explicit
access keys, `PyArrowFileIO` sets `AWS_EC2_METADATA_DISABLED=true`
process-wide. Closing the resolver does not restore it; I reproduced the
variable remaining set after `close()`. A later unrelated AWS client that
relies on EC2 instance-role credentials can then fail, contradicting the
documented promise that source options are not retained after the call. Please
use client-scoped credential-provider configuration; save/restore of a global
variable would still be unsafe for concurrent loads.
##########
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]
Review Comment:
**[P2] Quote the extras requirement for zsh**
The default zsh used on macOS interprets the brackets as a filename glob, so
copying this command fails with `zsh: no matches found: pypaimon[hdf5]` before
pip runs. Please use `pip install 'pypaimon[hdf5]'`, matching the quoted form
already used in the README.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
