[Python] Delegate Ray search through backend task execution

.github/workflows/ci-python.yml

@@ -312,7 +312,8 @@ jobs:
   compatibility:
     name: Package and Ray compatibility
     timeout-minutes: 60
-    runs-on: ubuntu-latest
+    # setup-python provides Python 3.9 builds for Ubuntu 24.04, not 26.04.
+    runs-on: ubuntu-24.04
     steps:
       - name: Checkout code
         uses: actions/checkout@v6

docs/docs/pypaimon/multimodal-search.md

@@ -200,8 +200,8 @@ data. Historical snapshot and deletion semantics remain the same. Plain
 
 Use `execution="ray"` to execute vector queries across Ray workers and return
 Arrow results to the driver. This supports single and batch vector queries on
-data-evolution tables; primary-key tables and hybrid queries are not supported
-by this execution mode.
+data-evolution tables; primary-key tables are not supported by this execution
+mode. Full-text and hybrid queries are described below.
 
 Install the same PyPaimon and index dependencies on the driver and workers:
 
@@ -235,6 +235,12 @@ the number of in-flight Ray tasks, with a default of 4. `ray_remote_args` suppli
 Ray task options, including resources and retry settings; `num_returns` is
 managed by PyPaimon. These two arguments require `execution="ray"`.
 
+Ray schedules tasks from the selected search backend. With Native reads enabled
+on a supported REST table, Rust performs the query, filtering and ranking on the
+driver and workers. Install the same Native SDK on every node. Selecting Ray
+does not force Python search execution. A failure after execution starts is
+propagated without rerunning the query through another backend.
+
 For single and batch vector queries, `to_pandas()` and `to_list()` accept the
 same `execution`, `concurrency`, and `ray_remote_args` arguments as `to_arrow()`.
 For example, use `query.to_pandas(execution="ray", concurrency=4)` to return a
@@ -274,14 +280,17 @@ filtering and final row lookup.
 In particular, `pre_filter` filters candidates before ranking; `where()` filters
 the selected rows and can return fewer than the requested number of results.
 
-For indexed vector searches, workers evaluate scalar predicates and build
+For indexed vector searches using the Python backend, workers evaluate scalar predicates and build
 live-row filters at the query snapshot. Tasks using the same scalar index files
 share one Ray scalar-index evaluation; its result is passed directly to dependent
 tasks without materializing the bitmap on the driver. Candidate verification
 reads only filter columns and row IDs in the task's shard range, without scanning
 the scalar indexes again. The final shard filter is reused across a batch's query
 vectors. This lets expensive verification scale with search concurrency; partially
 overlapping scalar index inputs can still cause repeated index reads.
+The Native backend resolves shared snapshot filters in Rust before distributing
+index and data-reading tasks. Query matrices and other shared task context are
+transported once per task group.
 
 Index search, raw scans, refinement, lookup, and task retries use the same read
 snapshot. The existing `snapshot_id` and `tag_name` arguments to `search()` also
@@ -324,11 +333,55 @@ uncovered corpus, preserving existing BM25 statistics; that corpus must fit the
 worker's memory. Pre-filters and deletion vectors keep the local ranking semantics.
 
 Hybrid routes share one resolved snapshot, use Ray for their shard work, and
-reuse the existing RRF, MRR, or weighted-score fusion on the driver. Routes run
-successively so `concurrency` bounds the whole query. Final Arrow/Pandas row lookup
+reuse the existing RRF, MRR, or weighted-score fusion on the driver. Concurrent
+routes share the query's task budget, so `concurrency` bounds the whole query.
+Final Arrow/Pandas row lookup
 remains on the driver. Native resources close on workers; task errors propagate
 and pending read tasks are cancelled.
 
+### Scheduling a search plan
+
+Lower-level vector, batch-vector and full-text readers accept an optional
+`SearchTaskExecutor` in `read_plan()` or `read_batch_plan()`. The reader keeps
+ownership of the complete snapshot-scoped query, including global candidate
+selection and refinement. A scheduler only places the backend's serializable
+tasks and returns their results:
+
+```python
+from pypaimon.ray.search_execution import RaySearchTaskExecutor
+
+builder = (table.new_batch_vector_search_builder()
+           .with_vector_column("embedding")
+           .with_query_vectors([[0.1, 0.2, 0.3], [0.4, 0.5, 0.6]])
+           .with_limit(10))
+plan = builder.new_vector_search_scan().scan()
+results = builder.new_batch_vector_search_read().read_batch_plan(
+    plan, executor=RaySearchTaskExecutor(concurrency=4))
+```
+
+Plans contain ordinary snapshot and typed split metadata. A copied, pickled,
+or Python-constructed plan can be passed to a Native reader without choosing
+the Python backend. The Native SDK also accepts snapshot and split objects
+through its plan constructors. For example, to execute an explicitly selected
+set of sources:
+
+```python
+from pypaimon.table.source.search_types import VectorSearchScanPlan
+
+selected = VectorSearchScanPlan(plan.splits()[:1], plan.snapshot())
+results = builder.new_batch_vector_search_read().read_batch_plan(selected)
+```
+
+The supplied splits are authoritative: readers do not add uncovered raw ranges
+or rediscover other search indexes. A Native plan without a captured snapshot
+must have no sources and never selects the latest snapshot. Query vectors, limits and
+filters belong to the reader. Rust Core coordinates global candidate selection
+and refinement; its internal shard protocol is separate from the public source
+plan.
+
+Hybrid builders expose `execute(executor=...)` for the same scheduler.
+PyPaimon's existing `execute_local()` methods retain their local defaults.
+
 ## Search Hybrid
 
 Use `search_hybrid` to combine vector and full-text routes, then rerank the

docs/docs/pypaimon/writing.md

@@ -119,18 +119,25 @@ remain unsupported.
 
 Append `with_write_type()` preserves the selected column order. Both the full
 table schema and the selected schema are accepted as Arrow input; only selected
-columns are written. Full Arrow inputs and named Rows can carry partition and
-hash columns used only for routing, independently of the physical write type.
+columns are written. Selected Arrow inputs can also carry partition, hash and
+configured row-kind columns used for routing and event handling. The selected
+columns must retain their declared order. Named Rows bind fields by name.
 Reads restore omitted partition columns from split metadata. Selected-schema
 inputs must include the routing columns they need; `write_arrow(data, bucket=...)`
 also accepts a bucket assigned upstream.
 
-Changing `with_write_type()` or `with_blob_consumer()` affects writers created
-for new partition/bucket pairs. Existing bucket writers retain their original
-columns and consumer, including across stream `prepare_commit(identifier)`
-calls. Full-schema inputs are projected to each target's captured columns by
-name. Selected-schema inputs must match the target's captured layout; missing
-columns are rejected before any group from the batch is written.
+Configure `with_write_type()` before the first effective write. Empty batches
+and ignored events do not lock the selection. Row-to-Arrow encoding failures
+before execution starts also leave it configurable. After writing, the same ordered
+selection can be set again, but changing its columns or their order is rejected,
+including after `prepare_commit()` or idle writer cleanup. Create a new writer
+to use a different write type.
+
+Changing `with_blob_consumer()` affects newly created partition/bucket writers.
+Active bucket writers retain their consumer across stream
+`prepare_commit(identifier)` calls. Unaware bucket writers retire on an idle
+prepare; fixed bucket writers also require a newer committed identifier from
+the same commit user. A reopened bucket uses the current consumer.
 
 Batch `prepare_commit()` accepts no arguments and can be called once. Stream
 `prepare_commit(identifier)` requires an integer-like value in the signed

paimon-python/conftest.py

@@ -15,6 +15,7 @@
 # limitations under the License.
 
 import os
+from functools import wraps
 
 import pytest
 
@@ -139,6 +140,7 @@ def tracked_read(self, splits, *args, **kwargs):
         def track_write(method):
             original_write = getattr(NativeTableWrite, method)
 
+            @wraps(original_write)
             def tracked_write(self, data, *args, **kwargs):
                 global _native_write_count
                 native = self._native_writer is not None

paimon-python/dev/test_packaging.py

@@ -122,6 +122,12 @@ def test_distribution_contents(self):
         for name in ("pypaimon/sample/data/data.jsonl", "pypaimon/_version.py"):
             self.assertIn(name, names)
         execution_files = (
+            "pypaimon/table/source/search_execution.py",
+            "pypaimon/ray/search_execution.py",
+            "pypaimon/execution/native/read/search_execution.py",
+            "pypaimon/execution/python/read/search/distributed_search.py",
+            "pypaimon/execution/python/read/search/distributed_vector_search.py",
+            "pypaimon/execution/python/read/search/distributed_full_text_search.py",
             "pypaimon/execution/python/map_blob_key_serializer.py",
             "pypaimon/execution/python/write/external_path_provider.py",
             "pypaimon/execution/python/read/predicate_json_parser.py",

paimon-python/pypaimon/execution/README.md

@@ -144,6 +144,30 @@ Ray package. Public Ray helper entry points delegate to those same helpers.
 System-table output applies common Arrow filtering/projection to `SystemSplit`
 results from either metadata provider without creating a Python file reader.
 
+Search readers accept an optional `SearchTaskExecutor` when executing a complete
+plan. `SearchTask` and its dependency/context transport contract stay in
+`table/source/search_execution`; Ray only implements bounded scheduling,
+context sharing and cancellation. It does not inherit a Python search reader
+or select Python on behalf of a Native reader. Python's distributed adapters
+own their index/filter/raw/refinement operations. Native adapters transport
+opaque shared context and task bytes; Core owns both their interpretation and
+global query coordination. The global candidate limit is applied before
+refinement, and full-text raw fallback retains one corpus for BM25 statistics.
+
+Vector and full-text plans are snapshot-and-split data in both backends. The
+Rust SDK constructors accept typed `Snapshot` and search split objects; indexed,
+raw and primary-key splits keep their own fields and index metadata. The Native
+adapter translates ordinary Python plans into these values, including plans
+made by Python scans or sent to another process. Existing snapshot and index
+metadata codecs are private adapters, rather than a public JSON plan protocol.
+Readers own the table, schema, query and filter. Native selection is retained
+after transport, and source subsets never add missing indexes or raw ranges.
+For Native, a plan without a captured snapshot must have no sources and never
+selects latest. Shard callbacks use an internal binding bridge; they are not part of Rust's public
+Plan/Read API. SDK vector, batch-vector, full-text and hybrid builders share
+one `execute()` entry point. PyPaimon's existing public search methods delegate
+to it through the Native adapters.
+
 ## Corresponding execution APIs
 
 Compare the operation's inputs, results and lifecycle when changing a backend.
@@ -183,14 +207,20 @@ Invalid call arguments fail before either executor prepares files.
 Public stream write, update and commit entry points normalize integer-like
 commit identifiers to Java `long` / Rust `i64` before execution starts.
 Write-column selections reject empty selections and duplicate names before
-replacing the current configuration. Like Java, append writers capture their
-write columns and Blob consumer per partition/bucket. Reconfiguration affects
-future bucket writers; existing buckets retain their configuration across
-`prepare_commit` calls. Full-schema inputs project by column name, and batches
-are checked against every target's captured layout before staging any group.
-Native named-row conversion queries Rust for the target columns using only
-routing metadata, so Python does not duplicate bucket state or encode unused
-payloads. Index build/drop requests bind partition
+replacing the current configuration. Append writers keep one ordered write type
+after their first effective write; empty batches and ignored events do not lock
+it. Repeating the same selection is a no-op, while a different selection requires
+a new writer, including after prepares or idle cleanup. Full-schema inputs are
+projected to the selected columns by name. Selected Arrow inputs may also carry
+routing/event fields while retaining the selected column order. Named-row
+conversion encodes selected ordinary and inline reference fields before capturing
+a bucket writer; encoding failures leave the configuration open. Dedicated live
+Blob payloads are consumed only during execution, and unused payloads remain
+untouched.
+Blob consumers are captured per partition/bucket; changes apply to newly created
+writers, and active writers retain their consumer across prepares. Idle writers
+retire at the Java commit boundary, so reopening uses the current consumer.
+Index build/drop requests bind partition
 predicates to the partition layout and reject predicates containing
 non-partition fields.
 

paimon-python/pypaimon/execution/native/read/native_full_text_search.py

@@ -75,7 +75,7 @@ def try_native_full_text_search(builder):
     table = builder._table
     primary_key = builder._primary_key_full_text_definition() is not None
     # Once execution starts, propagate errors instead of running a second search.
-    result = native.execute_local()
+    result = native.execute()
     if primary_key:
         return PrimaryKeyScoredResult._from_native(table, result)
     return DictBasedScoredIndexResult(result.row_ids())

paimon-python/pypaimon/execution/native/read/native_hybrid_search.py

@@ -27,7 +27,7 @@
 from pypaimon.execution.native.write.native_commit import _rest_catalog_supported
 
 
-def try_native_hybrid_search(builder):
+def try_native_hybrid_search(builder, executor=None):
     """Qualify before execution; core owns one snapshot, routes and fusion."""
     table = builder._table
     if type(table) is not FileStoreTable:
@@ -41,6 +41,8 @@ def try_native_hybrid_search(builder):
             or not _rest_catalog_supported(table)):
         return None
     builder._validate_search()
+    if executor is not None and not table.options.data_evolution_enabled():
+        raise ValueError('Distributed search supports only data-evolution tables')
     if any(route.is_full_text() for route in builder._routes):
         # Preserve the same typed analyzer normalization used by native full-text.
         options = {'full-text.' + key: value for key, value in NativeFullTextIndexOptions.from_options(
@@ -63,4 +65,14 @@ def try_native_hybrid_search(builder):
         # Qualify strict literal support before executing any route or ranking.
         return None
     # Propagate execution failures: retrying Python can select a newer snapshot.
-    return DictBasedScoredIndexResult(native.execute_local().row_ids())
+    if executor is None:
+        result = native.execute()
+    else:
+        from pypaimon.execution.native.read.search_execution import native_task_executor
+        callback = native_task_executor(table, executor)
+        try:
+            result = native._execute_with_executor(callback)
+        finally:
+            callback.cancel()
+            callback.raise_if_failed()
+    return DictBasedScoredIndexResult(result.row_ids())