JingsongLi commented on code in PR #10426: URL: https://github.com/apache/paimon/pull/10426#discussion_r4216179368
########## paimon-python/pypaimon/ray/vector_index_build.py: ########## @@ -0,0 +1,81 @@ +# 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. + +"""Ray shard construction with one driver-side global-index commit.""" + +from pypaimon.globalindex.build_plan import split_by_global_index_shard +from pypaimon.globalindex.index_file_utils import new_global_index_file_name +from pypaimon.globalindex.vindex.vindex_vector_global_index_reader import VINDEX_IDENTIFIERS +from pypaimon.ray.vector_search import _execution_options + + +def validate_build_options(builder, concurrency, ray_remote_args): + if builder._index_type not in VINDEX_IDENTIFIERS: + raise ValueError("Ray index building supports only native vector indexes.") + concurrency, options = _execution_options(concurrency, ray_remote_args) + # Retrying a write task can race with a lost worker writing the same file. + # A failed build can instead be rerun after its uncommitted outputs close. + if options.get("max_retries", 0) != 0 or options.get("retry_exceptions", False): + raise ValueError("Ray index building requires max_retries=0 and retry_exceptions=False.") + options.update(max_retries=0, retry_exceptions=False) + if builder._core_options.global_index_row_count_per_shard() <= 0: + raise ValueError("Option 'global-index.row-count-per-shard' must be greater than 0.") + return concurrency, options + + +def build_vector_index(builder, splits, unindexed_ranges, index_field, table_read, + index_path, concurrency, remote_args): + import ray + + shards = split_by_global_index_shard( + splits, builder._core_options.global_index_row_count_per_shard(), unindexed_ranges) + # Allocate names before dispatch, so failed/lost result delivery cannot hide + # files from cleanup. Workers never publish index manifests or snapshots. + names = [new_global_index_file_name("vindex-" + builder._index_type) for _ in shards] + context = ray.put((builder, index_field, table_read, index_path)) Review Comment: [P2] Reconstruct Jindo FileIO before serializing the worker context For an OSS table with `pyjindosdk` installed, the default `fs.oss.impl=jindo` creates an `OssFileIO` whose Arrow `PyFileSystem` contains `JindoFileSystemHandler._known_file_sizes_lock`. `PyArrowFileIO.__getstate__` removes only its own `_legacy_bucket_lock` and retains that filesystem, while the Jindo handler has no serialization hook. Consequently, this new `ray.put` raises `TypeError: cannot pickle '_thread.lock' object` before dispatching any shard, making `execution="ray"` unusable for this supported storage backend. I reproduced this with actual `ray.put` on the complete builder/TableRead context, substituting only the unavailable Jindo SDK's Config/connect calls. The local FileIO context serialized successfully; the Jindo context failed on the lock. Please serialize the connection configuration and reconstruct the Jindo filesystem/client and lock on workers, or provide equivalent serialization support, and add a regression test for the Jindo build context. -- 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]
