feat(python): expose core vector search and scored results

bindings/python/README.md

@@ -56,6 +56,30 @@ ctx.sql("DROP TEMPORARY TABLE paimon.default.my_temp")
 
 For the full SQL reference, see the [SQL Integration docs](https://paimon.apache.org/docs/master/sql/).
 
+### Native Vector Search
+
+A table with an existing vector column can search through the Rust core without
+materializing user rows in Python:
+
+```python
+search = (table.new_vector_search_builder()
+          .with_vector_column("embedding")
+          .with_query_vector([1.0, 0.0])
+          .with_limit(10))
+plan = search.new_vector_search_scan().scan()
+result = search.new_vector_search_read().read_plan(plan)
+batches = result.new_read_builder().with_projection(["id"]).read()
+```
+
+The plan pins one snapshot and can be reused with different queries. Scalar and
+partition filters are applied before ranking. Batch builders return one result
+per input query, in input order, including empty results.
+
+Data Evolution results expose scored global row IDs through `row_ids()`. PK
+results expose physical positions and Java-compatible scored split bytes;
+`new_read_builder()` reads both result types from their source snapshot and
+returns projected columns plus `__paimon_search_score` in relevance order.
+
 ### Native Read / Write
 
 Beyond SQL, you can use the lower-level read and write APIs directly from Python.

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -221,6 +221,8 @@ class Table:
     def schema(self) -> TableSchema: ...
     def new_read_builder(self, options: Optional[Dict[str, str]] = None) -> ReadBuilder: ...
     def new_stream_scan(self) -> StreamTableScan: ...
+    def new_vector_search_builder(self) -> "VectorSearchBuilder": ...
+    def new_batch_vector_search_builder(self) -> "BatchVectorSearchBuilder": ...
     def new_batch_write_builder(self) -> "BatchWriteBuilder": ...
     def new_postpone_fixed_bucket_write_builder(self) -> "PostponeFixedBucketWriteBuilder": ...
     def new_stream_write_builder(self) -> "StreamWriteBuilder": ...
@@ -534,3 +536,64 @@ class SQLContext:
     def sql_with_options(
         self, sql: str, options: SQLOptions
     ) -> List[pyarrow.RecordBatch]: ...
+
+
+class VectorSearchBuilder:
+    def with_vector_column(self, name: str) -> "VectorSearchBuilder": ...
+    def with_limit(self, limit: int) -> "VectorSearchBuilder": ...
+    def with_query_vector(self, vector: Sequence[float]) -> "VectorSearchBuilder": ...
+    def with_option(self, key: str, value: str) -> "VectorSearchBuilder": ...
+    def with_options(self, options: Dict[str, str]) -> "VectorSearchBuilder": ...
+    def with_filter(self, predicate: Dict[str, Any]) -> "VectorSearchBuilder": ...
+    def with_partition_filter(self, predicate: Dict[str, Any]) -> "VectorSearchBuilder":
+        """Add a partition-only predicate, resolved by table field names."""
+        ...
+    def new_vector_search_scan(self) -> "VectorScan": ...
+    def new_vector_search_read(self) -> "VectorRead": ...
+    def execute_local(self) -> "SearchResult": ...
+
+class BatchVectorSearchBuilder:
+    def with_vector_column(self, name: str) -> "BatchVectorSearchBuilder": ...
+    def with_limit(self, limit: int) -> "BatchVectorSearchBuilder": ...
+    def with_query_vectors(self, vectors: Sequence[Sequence[float]]) -> "BatchVectorSearchBuilder": ...
+    def with_option(self, key: str, value: str) -> "BatchVectorSearchBuilder": ...
+    def with_options(self, options: Dict[str, str]) -> "BatchVectorSearchBuilder": ...
+    def with_filter(self, predicate: Dict[str, Any]) -> "BatchVectorSearchBuilder": ...
+    def with_partition_filter(self, predicate: Dict[str, Any]) -> "BatchVectorSearchBuilder": ...
+    def new_vector_search_scan(self) -> "VectorScan": ...
+    def new_batch_vector_search_read(self) -> "BatchVectorRead": ...
+    def execute_batch_local(self) -> List["SearchResult"]: ...
+
+class VectorScan:
+    def scan(self) -> "VectorScanPlan": ...
+
+class VectorScanPlan:
+    def snapshot_id(self) -> Optional[int]: ...
+
+class VectorRead:
+    def read_plan(self, plan: VectorScanPlan) -> "SearchResult": ...
+
+class BatchVectorRead:
+    def read_batch_plan(self, plan: VectorScanPlan) -> List["SearchResult"]: ...
+
+class SearchResult:
+    """Snapshot-scoped global row IDs or primary-key physical positions."""
+    def snapshot_id(self) -> Optional[int]: ...
+    def __len__(self) -> int: ...
+    def new_read_builder(self) -> "SearchResultReadBuilder": ...
+    def row_ids(self) -> Dict[int, float]:
+        """DE scores keyed by global row ID; rejects PK results, including empty ones."""
+        ...
+    def positions(self) -> List[Tuple[bytes, int, str, int, float]]:
+        """PK partition bytes, bucket, file, position and score in relevance order."""
+        ...
+    def splits(self) -> List[bytes]:
+        """PK physical selections in Java SplitSerializer v1 format, including scores."""
+        ...
+
+
+class SearchResultReadBuilder:
+    def with_projection(self, columns: List[str]) -> "SearchResultReadBuilder": ...
+    def read(self) -> List[pyarrow.RecordBatch]:
+        """Read user columns and scores in relevance order from the search snapshot."""
+        ...

bindings/python/src/context.rs

@@ -726,6 +726,14 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
     let this = PyModule::new(py, "datafusion")?;
     this.add_class::<PaimonCatalog>()?;
     this.add_class::<crate::table::PyTable>()?;
+    this.add_class::<crate::vector_search::PyVectorSearchBuilder>()?;
+    this.add_class::<crate::vector_search::PyBatchVectorSearchBuilder>()?;
+    this.add_class::<crate::vector_search::PyVectorScan>()?;
+    this.add_class::<crate::vector_search::PyVectorScanPlan>()?;
+    this.add_class::<crate::vector_search::PyVectorRead>()?;
+    this.add_class::<crate::vector_search::PyBatchVectorRead>()?;
+    this.add_class::<crate::vector_search::PySearchResult>()?;
+    this.add_class::<crate::vector_search::PySearchResultReadBuilder>()?;
     this.add_class::<crate::read::PyReadBuilder>()?;
     this.add_class::<crate::read::PyTableScan>()?;
     this.add_class::<crate::read::PyStreamTableScan>()?;

bindings/python/src/lib.rs

@@ -30,6 +30,7 @@ mod table;
 mod udf;
 mod update_assignment;
 mod variant;
+mod vector_search;
 mod write;
 // ---- #285: observability ----
 mod partition;

bindings/python/src/table.rs

@@ -198,6 +198,14 @@ impl PyTable {
         }
     }
 
+    fn new_vector_search_builder(&self) -> crate::vector_search::PyVectorSearchBuilder {
+        crate::vector_search::PyVectorSearchBuilder::new(Arc::clone(&self.inner))
+    }
+
+    fn new_batch_vector_search_builder(&self) -> crate::vector_search::PyBatchVectorSearchBuilder {
+        crate::vector_search::PyBatchVectorSearchBuilder::new(Arc::clone(&self.inner))
+    }
+
     fn new_batch_write_builder(&self) -> PyBatchWriteBuilder {
         PyBatchWriteBuilder::new(Arc::clone(&self.inner))
     }