[Feature][Python][Ray] Support distributed vector search on primary-key tables #10772

Search before asking

• I searched in the issues and found nothing similar.

Motivation

I would like to query primary-key vector indexes from a Python/Ray application, where records and embeddings can be updated or deleted by key. For large tables spanning many buckets, a single vector query should be able to use multiple Ray workers.

PyPaimon currently supports Ray vector search for Data Evolution tables, while primary-key vector queries use local execution. Both the Python and Native adapters reject Ray execution for primary-key tables. This limits the resources available to a single query, even when the application already runs on a Ray cluster.

Solution

Extend the existing Ray search execution APIs to support primary-key vector indexes. For example, given a primary-key table with an existing vector index, the following should execute the search across Ray workers:

neighbors = (
    docs.search([0.1, 0.2, 0.3], column="embedding")
    .select(["id", "title"])
    .limit(10)
    .to_arrow(execution="ray", concurrency=4)
)

Expected behavior:

• Search different primary-key buckets in parallel and merge their candidates into a global Top-K.
• Preserve physical row positions, snapshot consistency and deletion-vector filtering so updates and deletes are handled correctly.
• Preserve existing filter, search-mode and refinement semantics.
• Reuse concurrency controls and propagate worker failures without returning partial results as successful queries.

The initial scope could focus on single-vector queries over existing indexes. Batch queries and distributed result lookup through to_ray() can be discussed separately.

Anything else?

Related work:

• [python] Support distributed vector search with Ray #9930 introduced Ray vector search for Data Evolution tables and explicitly excluded primary-key queries.
• [Python] Delegate Ray search through backend task execution #10724 and [core] Support scheduled execution of snapshot-scoped search plans paimon-rust#1191 provide the backend task execution framework, but do not enable distributed primary-key search.
• Execute primary-key vector search over Java-planned bucket splits (engine-distributed read) paimon-rust#755 covers executing Java-planned PK bucket splits in external engines. This request focuses on PyPaimon/Ray integration and could reuse that work.

Are you willing to submit a PR?

• I'm willing to submit a PR!