You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "TheR1sing3un (via GitHub)" <gi...@apache.org> on 2026/10/08 05:59:13 UTC
[PR] [python] Support distributed vector index construction with Ray [paimon]
TheR1sing3un opened a new pull request, #10426:
URL: https://github.com/apache/paimon/pull/10426
### Purpose
Allow `create_global_index` and multimodal `create_index` to build native vector-index shards with `execution="ray"`. The driver plans uncovered row ranges from one snapshot, workers construct shards with bounded concurrency, and the driver publishes all resulting entries in one commit. Local execution remains the default.
Allocate output names before dispatch and disable automatic write-task retries. If a task fails, drain outstanding tasks before removing uncommitted shard files, including files whose commit messages were never returned. Document shared storage, dependency, snapshot-retention and driver-failure requirements.
### Tests
50 targeted tests passed with Ray 2.54.0 and paimon-vindex 0.5.0, covering distributed construction, incremental coverage, deleted rows, one-commit publication, build/commit separation, cleanup after a worker writes then fails, option validation, and existing global-index/multimodal APIs. Flake8 and `git diff --check` passed. No benchmark code included.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support distributed vector index construction with Ray [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support distributed vector index construction with Ray [paimon]
Posted by "TheR1sing3un (via GitHub)" <gi...@apache.org>.
TheR1sing3un commented on code in PR #10426:
URL: https://github.com/apache/paimon/pull/10426#discussion_r4217857307
##########
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:
fixed
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support distributed vector index construction with Ray [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #10426:
URL: https://github.com/apache/paimon/pull/10426
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Support distributed vector index construction with Ray [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #10426:
URL: https://github.com/apache/paimon/pull/10426#issuecomment-6082411853
+1
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org