JingsongLi commented on code in PR #10351: URL: https://github.com/apache/paimon/pull/10351#discussion_r4245004592
########## paimon-python/pypaimon/filesystem/io_cache_file_io.py: ########## @@ -0,0 +1,304 @@ +# 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. + +"""FileIO that sends each request to an io-cache target or the origin endpoint.""" + +import threading +from typing import Callable, Dict, List, Optional + +import pyarrow.fs as pafs +from pyarrow._fs import FileSystemHandler + +from pypaimon.common.file_io import FileIO +from pypaimon.filesystem.io_cache_routing import IoCacheRouting, Op + + +class IoCacheRoutingFileIO(FileIO): + """Sends each call to origin or to a target FileIO, all built for the table path. + Calls and attributes that are not routed go to origin.""" + + def __init__(self, routing: IoCacheRouting, origin: FileIO, + create_target: Callable[[str], FileIO]): + self._routing = routing + self._origin = origin + self._create_target = create_target + self._targets: Dict[str, FileIO] = {} + self._targets_lock = threading.Lock() + self._filesystem = pafs.PyFileSystem(_IoCacheRoutingHandler(self)) + + @property + def properties(self): + return self._origin.properties + + @property + def filesystem(self): + """Filesystem for readers; it takes the URIs returned by to_filesystem_path.""" + return self._filesystem + + def _route(self, op: Op, path: str) -> Optional[str]: + return self._routing.route(op, path) + + def _file_io(self, name: Optional[str]) -> FileIO: + """Origin, or the FileIO of a target, created on first use.""" + if name is None: + return self._origin + file_io = self._targets.get(name) + if file_io is not None: + return file_io + with self._targets_lock: Review Comment: [P1] Recreate routed FileIO state after fork before acquiring inherited locks This lock remains held while a target FileIO is initialized. If a process forks during a cold target initialization in another thread, the child inherits the locked state and an empty target cache; the owning thread exists only in the parent. A child read then waits forever, even after the parent completes initialization. This is reachable with inherited REST table readers in Linux fork workers: streaming Torch datasets retain TableRead, and their PID handling only resets distributed rank. Pickle hooks are not invoked by fork. I reproduced the interleaving through production RESTTokenFileIO.new_input_stream with fs.oss.impl=legacy, with the REST cache lock explicitly free at fork. Only backend construction/read was replaced with a test double. The parent finished its read before the child attempted one; current code timed out on this inherited target lock, while the exact merge-base REST reader completed. Please detect PID changes/reinitialize the routed wrapper and target state before taking this lock, or register suitable after-fork recreation, and add a bounded fork regression. ########## paimon-python/pypaimon/read/reader/lance_utils.py: ########## @@ -78,7 +87,7 @@ def to_lance_specified(file_io: FileIO, file_path: str) -> Tuple[str, Optional[D if properties.contains(OssOptions.OSS_ENDPOINT): storage_options['oss_endpoint'] = properties.get(OssOptions.OSS_ENDPOINT) - storage_options['virtual_hosted_style_request'] = 'true' + storage_options['virtual_hosted_style_request'] = 'false' if path_style else 'true' Review Comment: [P1] Select a Lance backend that supports path-style cache targets The newly selected target's path-style flag does not work with the retained oss:// URI. Lance's OSS provider ignores the generic virtual_hosted_style_request=false option and prefixes the bucket onto the endpoint. With supported pylance 0.39.0, io-cache.endpoint=http://127.0.0.1:<port> and io-cache.target.default.path-style-access=true, an actual LanceFileReader over the returned URI/options attempts http://bucket.127.0.0.1:<port>/... and fails with invalid IPv4 address; the target receives no request. This prevents Lance reads through the IP/path-style target advertised by this PR. Using the same real Lance file, credentials and storage options with only the URI changed to s3://bucket/... successfully reads all three rows via target HEAD/GET. Please select the S3-compatible provider for path-style OSS targets (or another provider explicitly supporting this addressing), while preserving ordinary virtual-hosted OSS handling. Add an actual local HTTP Lance read regression: checking only the returned false flag misses the failure. -- 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]
