[JingsongLi]

Purpose

Depends on apache/paimon-rust#1191. Draft until its new Plan APIs reach Rust main. Current Python master also uses the writer API from apache/paimon-rust#1188.

Keep Ray as a scheduler while Python or Native execution owns vector, batch-vector, full-text and hybrid search semantics.

Changes

• Add the public SearchTaskExecutor contract and optional executor on read_plan / read_batch_plan, plus HybridSearchBuilder.execute(executor=...).
• Move Python shard algorithms from Ray into execution/python/read/search. Ray uses public scan/read/execute methods and a generic bounded scheduler without importing either backend.
• Use ordinary snapshot-and-split plans in both backends. Native translates Python-created, copied or pickled plans into typed SDK VectorScanPlan(splits, snapshot=None) / FullTextScanPlan(splits, snapshot=None). Snapshot and six concrete split types cross the public binding API as objects; metadata codecs remain private to adapters and pickle. Transported scans/readers recreate Native adapters without initializing Python execution.
• Treat supplied source subsets and empty plans as authoritative. Readers retain query configuration, filters and options; Rust Core coordinates global ANN selection/refinement, one raw BM25 corpus and hybrid fusion.
• Delegate all Native builder execution to the SDK's unified execute() method while preserving existing public PyPaimon search methods. Update direct SDK tests and old plan-context assertions to the typed source-plan contract.
• Keep opaque shard transport in private binding entry points. Preserve shared prerequisites, query concurrency budgets, cancellation isolation, original failures and iterator cleanup.
• Preserve existing public constructors and split-only methods. Primary-key plans can additionally carry their complete captured snapshot; existing explicit snapshot-ID plans resolve that ID without selecting latest.
• Preserve captured snapshot, bucket path, bucket count and aligned deletion metadata when aggregating primary-key sources. Normalize missing child snapshot IDs at the Native bridge without mutating the supplied plan.
• Keep scheduled full-text sources associated with their actual column, matching Java and local Python execution. Preserve original worker/cancellation failures when iterator cleanup also fails, while surfacing standalone cleanup failures.

Validation

• Public API, execution boundary, Ray scheduling, hybrid and cancellation group: 80 passed.
• Final review regression group: 110 passed, 88 Native integration cases skipped locally because the paired SDK is not installed. This covers primary-key source metadata, strict execute-only adapters, typed plan transport, source columns, scheduler failures, cancellation and cleanup identity.
• Adjacent primary-key/index/ranker regressions: 41 passed. Packaging tests from the initial implementation: 9 passed.
• Repository-configured Flake8 passed; Python 3.6 grammar and ASF headers passed for all 46 changed Python files. Full collection passed during implementation (22,400 tests collected).
• All 117 paired Rust Python binding unit tests passed, exercising actual registered SDK classes, typed Plan constructors, pickle, private codecs and scheduler callbacks.

The Native extension was not rebuilt or installed. Native CI continues to build Rust main; there are no SDK-version probes or Python execution fallbacks to bypass this dependency.