JingsongLi commented on code in PR #9411: URL: https://github.com/apache/paimon/pull/9411#discussion_r3868584400
########## paimon-python/pypaimon/multimodal/hdf5.py: ########## @@ -0,0 +1,475 @@ +# 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 +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, urlparse, urlunparse +from urllib.request import url2pathname + +import pyarrow as pa +import pyarrow.fs as pafs + +from pypaimon.common.options import Options +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(url2pathname(parsed.path)) + + @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 = 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 Hdf5AppendResult: + """Counts and optional committed snapshot for one ``append_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 append_hdf5( + table, + paths, + *, + transform: Callable, + source_options: Optional[Mapping[str, object]] = None): + """Append HDF5 files to 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 not callable(transform): + raise ValueError("transform must be callable.") + source_file_io = ResolvingFileIO( Review Comment: **[P1] Keep source Kerberos credentials out of process-global state**\n\nConstructing the source resolver can create `HdfsNativeFileIO`, whose Kerberos setup runs plain `kinit` and updates process-global `KRB5CCNAME`; that backend explicitly documents that the last writer wins, and closing the temporary resolver does not restore the prior ticket. If the target uses principal A and the source uses principal B, the target writer can subsequently authenticate as B. Please bind a unique credential cache to the source client, or reject differing principals when the backend cannot isolate them. ########## paimon-python/pypaimon/multimodal/hdf5.py: ########## @@ -0,0 +1,475 @@ +# 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 +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, urlparse, urlunparse +from urllib.request import url2pathname + +import pyarrow as pa +import pyarrow.fs as pafs + +from pypaimon.common.options import Options +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(url2pathname(parsed.path)) + + @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 = 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 Hdf5AppendResult: + """Counts and optional committed snapshot for one ``append_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 append_hdf5( + table, + paths, + *, + transform: Callable, + source_options: Optional[Mapping[str, object]] = None): + """Append HDF5 files to 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 not callable(transform): + raise ValueError("transform must be callable.") + source_file_io = ResolvingFileIO( + Options(_validated_source_options(source_options))) + try: + files = _discover_hdf5_files(paths, source_file_io) + if not files: + return Hdf5AppendResult( + 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( + "append_hdf5 requires h5py; install pypaimon[hdf5]." + ) from error + return _append_hdf5_files( + table, files, transform, source_file_io, h5py) + finally: + source_file_io.close() + + +def _append_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: + 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 + if arrow_table.num_rows: + table_write.write_arrow(arrow_table) + finally: + _close_transform_iterator( + batches if batches is not None else transformed) + + if row_count == 0: + raise ValueError("HDF5 transform produced no rows.") + 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 Hdf5AppendResult( + 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 not parsed.scheme: + return Path(path).expanduser().resolve().as_uri() + 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="/:") + return path + + +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 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, + "/" + status_path.lstrip("/"), + "", + "", + "", + )) + + 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, key) + + +def _is_windows_drive_path(parsed): + return len(parsed.scheme) == 1 and not parsed.netloc + + +def _hdf5_suffix(path): + parsed = urlparse(path) + return PurePosixPath(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 _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: + return table.cast(target_schema, safe=True) Review Comment: **[P1] Validate nested nullability before writing**\n\n`Table.cast(target_schema, safe=True)` does not reject nulls inside list or fixed-list elements and map values; it can relabel existing child data as non-nullable. I reproduced an `ARRAY<INT NOT NULL>` value of `[1, None]` committing successfully through ORC and reading the null back. This violates the strict-schema contract. Please recursively validate null bitmaps for every non-nullable nested field and add ORC array/map regression coverage. ########## paimon-python/setup.py: ########## @@ -233,6 +233,9 @@ def read_requirements(): ], }, extras_require={ + 'hdf5': [ + 'h5py>=3,<4; python_version>="3.8"', Review Comment: **[P2] Avoid a no-op HDF5 extra on supported Python versions**\n\nThe package declares Python 3.6+ support and the documentation tells users to install `pypaimon[hdf5]`, but this marker installs no `h5py` on Python 3.6 or 3.7. The runtime error then repeats the same ineffective installation command. Please either provide a compatible dependency constraint or explicitly document and guard the Python 3.8+ requirement, including in the runtime error. ########## paimon-python/pypaimon/multimodal/hdf5.py: ########## @@ -0,0 +1,475 @@ +# 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 +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, urlparse, urlunparse +from urllib.request import url2pathname + +import pyarrow as pa +import pyarrow.fs as pafs + +from pypaimon.common.options import Options +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(url2pathname(parsed.path)) + + @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 = 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 Hdf5AppendResult: + """Counts and optional committed snapshot for one ``append_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 append_hdf5( + table, + paths, + *, + transform: Callable, + source_options: Optional[Mapping[str, object]] = None): + """Append HDF5 files to 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 not callable(transform): + raise ValueError("transform must be callable.") + source_file_io = ResolvingFileIO( + Options(_validated_source_options(source_options))) + try: + files = _discover_hdf5_files(paths, source_file_io) + if not files: + return Hdf5AppendResult( + 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( + "append_hdf5 requires h5py; install pypaimon[hdf5]." + ) from error + return _append_hdf5_files( + table, files, transform, source_file_io, h5py) + finally: + source_file_io.close() + + +def _append_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: + 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 + if arrow_table.num_rows: + table_write.write_arrow(arrow_table) + finally: + _close_transform_iterator( + batches if batches is not None else transformed) + + if row_count == 0: + raise ValueError("HDF5 transform produced no rows.") + 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 Hdf5AppendResult( + 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) Review Comment: **[P1] Do not treat an incomplete directory listing as empty**\n\n`LocalFileIO.list_status` catches `PermissionError` and returns an empty or partial list. I reproduced a recursive tree with one readable file and one mode-000 subdirectory committing only the readable sibling; an unreadable root is reported as an empty-directory no-op. Please propagate listing permission failures, or expose an explicit incomplete-listing result that this path converts into a pre-writer failure. ########## paimon-python/pypaimon/multimodal/hdf5.py: ########## @@ -0,0 +1,475 @@ +# 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 +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, urlparse, urlunparse +from urllib.request import url2pathname + +import pyarrow as pa +import pyarrow.fs as pafs + +from pypaimon.common.options import Options +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(url2pathname(parsed.path)) + + @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 = 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 Hdf5AppendResult: + """Counts and optional committed snapshot for one ``append_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 append_hdf5( + table, + paths, + *, + transform: Callable, + source_options: Optional[Mapping[str, object]] = None): + """Append HDF5 files to 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 not callable(transform): + raise ValueError("transform must be callable.") + source_file_io = ResolvingFileIO( + Options(_validated_source_options(source_options))) + try: + files = _discover_hdf5_files(paths, source_file_io) + if not files: + return Hdf5AppendResult( + 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( + "append_hdf5 requires h5py; install pypaimon[hdf5]." + ) from error + return _append_hdf5_files( + table, files, transform, source_file_io, h5py) + finally: + source_file_io.close() + + +def _append_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: + 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 + if arrow_table.num_rows: + table_write.write_arrow(arrow_table) + finally: + _close_transform_iterator( + batches if batches is not None else transformed) + + if row_count == 0: + raise ValueError("HDF5 transform produced no rows.") + 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 Hdf5AppendResult( + 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 not parsed.scheme: + return Path(path).expanduser().resolve().as_uri() + 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="/:") Review Comment: **[P2] Round-trip standard Windows file URIs**\n\nThis emits the standard `file:///C:/...` form, but `LocalFileIO._to_file` only handles a drive in the URI authority and converts the emitted URI to `Path("/C:/...")`, which has no drive under Windows semantics. UNC authorities are dropped as well. Please decode standard file URIs with Windows-aware `url2pathname` behavior, preserve UNC authorities, and add a normalization-to-open test on Windows. ########## paimon-python/pypaimon/multimodal/hdf5.py: ########## @@ -0,0 +1,475 @@ +# 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 +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, urlparse, urlunparse +from urllib.request import url2pathname + +import pyarrow as pa +import pyarrow.fs as pafs + +from pypaimon.common.options import Options +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(url2pathname(parsed.path)) + + @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 = 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 Hdf5AppendResult: + """Counts and optional committed snapshot for one ``append_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 append_hdf5( + table, + paths, + *, + transform: Callable, + source_options: Optional[Mapping[str, object]] = None): + """Append HDF5 files to 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 not callable(transform): + raise ValueError("transform must be callable.") + source_file_io = ResolvingFileIO( + Options(_validated_source_options(source_options))) + try: + files = _discover_hdf5_files(paths, source_file_io) + if not files: + return Hdf5AppendResult( + 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( + "append_hdf5 requires h5py; install pypaimon[hdf5]." + ) from error + return _append_hdf5_files( + table, files, transform, source_file_io, h5py) + finally: + source_file_io.close() + + +def _append_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: + 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 + if arrow_table.num_rows: + table_write.write_arrow(arrow_table) + finally: + _close_transform_iterator( + batches if batches is not None else transformed) + + if row_count == 0: Review Comment: **[P1] Enforce the no-row contract per source file**\n\nThis aggregate check only fails when every source is empty. With one valid file and one transform that yields nothing, I reproduced a successful commit returning `file_count=2`, `batch_count=1`, and `row_count=1`; the empty source is silently counted as ingested. Please track rows per `source` and fail before advancing to the next source or committing. ########## paimon-python/pypaimon/multimodal/hdf5.py: ########## @@ -0,0 +1,475 @@ +# 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 +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, urlparse, urlunparse +from urllib.request import url2pathname + +import pyarrow as pa +import pyarrow.fs as pafs + +from pypaimon.common.options import Options +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(url2pathname(parsed.path)) + + @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 = 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 Hdf5AppendResult: + """Counts and optional committed snapshot for one ``append_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 append_hdf5( + table, + paths, + *, + transform: Callable, + source_options: Optional[Mapping[str, object]] = None): + """Append HDF5 files to 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 not callable(transform): + raise ValueError("transform must be callable.") + source_file_io = ResolvingFileIO( + Options(_validated_source_options(source_options))) + try: + files = _discover_hdf5_files(paths, source_file_io) + if not files: + return Hdf5AppendResult( + 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( + "append_hdf5 requires h5py; install pypaimon[hdf5]." + ) from error + return _append_hdf5_files( + table, files, transform, source_file_io, h5py) + finally: + source_file_io.close() + + +def _append_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: + 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 + if arrow_table.num_rows: + table_write.write_arrow(arrow_table) + finally: + _close_transform_iterator( + batches if batches is not None else transformed) + + if row_count == 0: + raise ValueError("HDF5 transform produced no rows.") + 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 Hdf5AppendResult( + 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 not parsed.scheme: + return Path(path).expanduser().resolve().as_uri() + 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="/:") + return path + + +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 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, + "/" + status_path.lstrip("/"), + "", + "", + "", + )) + + 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, key) Review Comment: **[P2] Preserve filesystem-native paths when building URIs**\n\nRaw `FileInfo.path` values are interpolated into a URI without escaping reserved characters. For example, `episode.h5#backup` passes the suffix check but is later opened as `episode.h5`, while `episode#backup.h5` is silently skipped because the suffix lands in the fragment; `?` behaves similarly. Please percent-encode raw status paths consistently, or carry the native filesystem path separately from the display URI. -- 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]
