leaves12138 commented on code in PR #1087: URL: https://github.com/apache/paimon-rust/pull/1087#discussion_r4231175155
########## bindings/python/src/vector_search.rs: ########## @@ -0,0 +1,430 @@ +// 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. + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow::pyarrow::ToPyArrow; +use futures::TryStreamExt; +use paimon::spec::Predicate; +use paimon::table::{BatchVectorRead, Table, VectorRead, VectorScan, VectorScanPlan}; +use paimon::vector_search::SearchResult; +use paimon_datafusion::runtime::runtime; +use pyo3::prelude::*; +use pyo3::types::{PyBytes, PyDict}; + +use crate::error::to_py_err; +use crate::predicate::dict_to_table_predicate; + +#[derive(Clone, Default)] +struct SearchConfig { + column: Option<String>, + limit: Option<usize>, + options: Vec<HashMap<String, String>>, + filters: Vec<Predicate>, + partition_filters: Vec<Predicate>, +} + +impl SearchConfig { + fn single<'a>( + &self, + table: &'a Table, + ) -> paimon::Result<paimon::table::VectorSearchBuilder<'a>> { + let mut builder = table.new_vector_search_builder(); + if let Some(column) = &self.column { + builder.with_vector_column(column); + } + if let Some(limit) = self.limit { + builder.with_limit(limit); + } + for options in &self.options { + builder.with_options(options.clone()); + } + for filter in &self.filters { + builder.with_filter(filter.clone()); + } + for filter in &self.partition_filters { + builder.with_partition_filter(filter.clone())?; + } + Ok(builder) + } + + fn batch<'a>( + &self, + table: &'a Table, + ) -> paimon::Result<paimon::table::BatchVectorSearchBuilder<'a>> { + let mut builder = table.new_batch_vector_search_builder(); + if let Some(column) = &self.column { + builder.with_vector_column(column); + } + if let Some(limit) = self.limit { + builder.with_limit(limit); + } + for options in &self.options { + builder.with_options(options.clone()); + } + for filter in &self.filters { + builder.with_filter(filter.clone()); + } + for filter in &self.partition_filters { + builder.with_partition_filter(filter.clone())?; + } + Ok(builder) + } +} + +#[pyclass(name = "VectorSearchBuilder", module = "pypaimon_rust.datafusion")] +pub struct PyVectorSearchBuilder { + table: Arc<Table>, + config: SearchConfig, + query: Option<Vec<f32>>, +} + +impl PyVectorSearchBuilder { + pub(crate) fn new(table: Arc<Table>) -> Self { + Self { + table, + config: SearchConfig::default(), + query: None, + } + } +} + +#[pymethods] +impl PyVectorSearchBuilder { + fn with_vector_column(mut slf: PyRefMut<'_, Self>, name: String) -> PyRefMut<'_, Self> { + slf.config.column = Some(name); + slf + } + fn with_limit(mut slf: PyRefMut<'_, Self>, limit: usize) -> PyRefMut<'_, Self> { + slf.config.limit = Some(limit); + slf + } + fn with_query_vector(mut slf: PyRefMut<'_, Self>, vectors: Vec<f32>) -> PyRefMut<'_, Self> { + slf.query = Some(vectors); + slf + } + fn with_option(mut slf: PyRefMut<'_, Self>, key: String, value: String) -> PyRefMut<'_, Self> { + slf.config.options.push(HashMap::from([(key, value)])); + slf + } + fn with_options( + mut slf: PyRefMut<'_, Self>, + options: HashMap<String, String>, + ) -> PyRefMut<'_, Self> { + slf.config.options.push(options); + slf + } + fn with_filter<'py>( + mut slf: PyRefMut<'py, Self>, + predicate: &Bound<'_, PyDict>, + ) -> PyResult<PyRefMut<'py, Self>> { + let predicate = dict_to_table_predicate(predicate, slf.table.schema(), true)?; + slf.config.filters.push(predicate); + Ok(slf) + } + fn with_partition_filter<'py>( + mut slf: PyRefMut<'py, Self>, + predicate: &Bound<'_, PyDict>, + ) -> PyResult<PyRefMut<'py, Self>> { + let predicate = dict_to_table_predicate(predicate, slf.table.schema(), true)?; + slf.config + .single(&slf.table) + .map_err(to_py_err)? + .with_partition_filter(predicate.clone()) + .map_err(to_py_err)?; + slf.config.partition_filters.push(predicate); + Ok(slf) + } + fn new_vector_search_scan(&self) -> PyResult<PyVectorScan> { + let inner = self + .config + .single(&self.table) + .map_err(to_py_err)? + .new_scan() + .map_err(to_py_err)?; + Ok(PyVectorScan { inner }) + } + fn new_vector_search_read(&self) -> PyResult<PyVectorRead> { + let mut builder = self.config.single(&self.table).map_err(to_py_err)?; + if let Some(query) = &self.query { + builder.with_query_vector(query.clone()); + } + let inner = builder.new_read().map_err(to_py_err)?; + Ok(PyVectorRead { inner }) + } + fn execute_local(&self, py: Python<'_>) -> PyResult<PySearchResult> { + let mut builder = self.config.single(&self.table).map_err(to_py_err)?; + if let Some(query) = &self.query { + builder.with_query_vector(query.clone()); + } + let results = py + .detach(|| runtime().block_on(builder.execute())) Review Comment: [P1] Preserve Java's raw-search routing for unevaluable scalar filters This exposes the core DE filtered-search path, but `DeVectorScan::plan` always resolves the predicate by reading matching row IDs and passes that allow-list to ANN. That is not Java's `AbstractDataEvolutionVectorRead.prepareSplits` behavior: when no scalar index can evaluate the predicate, Java moves the covered vector-index ranges to raw scoring and applies the residual there. The companion Python PR implements that routing in its classic reader, then bypasses it here through Native execution. I reproduced the difference on a real REST table using the companion `vector_table` fixture, 50 near vectors and 50 distant vectors in the same partition, an IVF-flat index with `nlist=4`, query `[0, 0]`, filter `id >= 100` with no scalar index on `id`, `ivf.nprobe=1`, and limit 3. The classic/reference path returns 3 matching hits; this binding and the new PyPaimon Native path both return 0. The difference occurs for single and batch searches in FAST, FULL, and DETAIL. Setting Native nprobe to 4 restores all 3 hits, confirming that filtered hits are lost through ANN candidate selection rather than being absent from the data. Please implement Java's scalar-evaluation/range-routing distinction in core before exposing this path, including both single and batch readers. A data-derived allow-list sent to ANN is not a substitute for the required raw fallback. ########## bindings/python/src/vector_search.rs: ########## @@ -0,0 +1,430 @@ +// 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. + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow::pyarrow::ToPyArrow; +use futures::TryStreamExt; +use paimon::spec::Predicate; +use paimon::table::{BatchVectorRead, Table, VectorRead, VectorScan, VectorScanPlan}; +use paimon::vector_search::SearchResult; +use paimon_datafusion::runtime::runtime; +use pyo3::prelude::*; +use pyo3::types::{PyBytes, PyDict}; + +use crate::error::to_py_err; +use crate::predicate::dict_to_table_predicate; + +#[derive(Clone, Default)] +struct SearchConfig { + column: Option<String>, + limit: Option<usize>, + options: Vec<HashMap<String, String>>, + filters: Vec<Predicate>, + partition_filters: Vec<Predicate>, +} + +impl SearchConfig { + fn single<'a>( + &self, + table: &'a Table, + ) -> paimon::Result<paimon::table::VectorSearchBuilder<'a>> { + let mut builder = table.new_vector_search_builder(); + if let Some(column) = &self.column { + builder.with_vector_column(column); + } + if let Some(limit) = self.limit { + builder.with_limit(limit); + } + for options in &self.options { + builder.with_options(options.clone()); + } + for filter in &self.filters { + builder.with_filter(filter.clone()); + } + for filter in &self.partition_filters { + builder.with_partition_filter(filter.clone())?; + } + Ok(builder) + } + + fn batch<'a>( + &self, + table: &'a Table, + ) -> paimon::Result<paimon::table::BatchVectorSearchBuilder<'a>> { + let mut builder = table.new_batch_vector_search_builder(); + if let Some(column) = &self.column { + builder.with_vector_column(column); + } + if let Some(limit) = self.limit { + builder.with_limit(limit); + } + for options in &self.options { + builder.with_options(options.clone()); + } + for filter in &self.filters { + builder.with_filter(filter.clone()); Review Comment: [P2] Honor the Java scalar-candidate refinement switch Both the single and batch configuration paths delegate all filters to core's unconditional `matching_row_ids_for_filter` data read. For an inexact scalar index answer this ignores `global-index.filter.refine-from-data=false`. Java's `AbstractDataEvolutionVectorRead.scalarMatchedRows` explicitly excludes those candidates when refinement is disabled; it does not read the data to turn them into exact matches. A real REST reproduction uses names `alpha`, `beta zeta`, and `gamma`, an IVF-flat index on the vector column, a BTREE index on `name`, filter `name LIKE '%zeta%'`, and limit 1. With refinement disabled the classic/Java-reference policy returns 0 hits, but both the Rust binding and Native PyPaimon return 1; batch returns `[1, 1]` instead of `[0, 0]`. All three search modes reproduce it. With refinement enabled the paths agree. Please preserve exact/unevaluable/inexact scalar evaluation state and apply the configured refinement policy before ANN, rather than always materializing an exact row-ID filter. ########## bindings/python/src/vector_search.rs: ########## @@ -0,0 +1,430 @@ +// 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. + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow::pyarrow::ToPyArrow; +use futures::TryStreamExt; +use paimon::spec::Predicate; +use paimon::table::{BatchVectorRead, Table, VectorRead, VectorScan, VectorScanPlan}; +use paimon::vector_search::SearchResult; +use paimon_datafusion::runtime::runtime; +use pyo3::prelude::*; +use pyo3::types::{PyBytes, PyDict}; + +use crate::error::to_py_err; +use crate::predicate::dict_to_table_predicate; + +#[derive(Clone, Default)] +struct SearchConfig { + column: Option<String>, + limit: Option<usize>, + options: Vec<HashMap<String, String>>, + filters: Vec<Predicate>, + partition_filters: Vec<Predicate>, +} + +impl SearchConfig { + fn single<'a>( + &self, + table: &'a Table, + ) -> paimon::Result<paimon::table::VectorSearchBuilder<'a>> { + let mut builder = table.new_vector_search_builder(); + if let Some(column) = &self.column { + builder.with_vector_column(column); + } + if let Some(limit) = self.limit { + builder.with_limit(limit); + } + for options in &self.options { + builder.with_options(options.clone()); + } + for filter in &self.filters { + builder.with_filter(filter.clone()); + } + for filter in &self.partition_filters { + builder.with_partition_filter(filter.clone())?; + } + Ok(builder) + } + + fn batch<'a>( + &self, + table: &'a Table, + ) -> paimon::Result<paimon::table::BatchVectorSearchBuilder<'a>> { + let mut builder = table.new_batch_vector_search_builder(); + if let Some(column) = &self.column { + builder.with_vector_column(column); + } + if let Some(limit) = self.limit { + builder.with_limit(limit); + } + for options in &self.options { + builder.with_options(options.clone()); + } + for filter in &self.filters { + builder.with_filter(filter.clone()); + } + for filter in &self.partition_filters { + builder.with_partition_filter(filter.clone())?; + } + Ok(builder) + } +} + +#[pyclass(name = "VectorSearchBuilder", module = "pypaimon_rust.datafusion")] +pub struct PyVectorSearchBuilder { + table: Arc<Table>, + config: SearchConfig, + query: Option<Vec<f32>>, +} + +impl PyVectorSearchBuilder { + pub(crate) fn new(table: Arc<Table>) -> Self { + Self { + table, + config: SearchConfig::default(), + query: None, + } + } +} + +#[pymethods] +impl PyVectorSearchBuilder { + fn with_vector_column(mut slf: PyRefMut<'_, Self>, name: String) -> PyRefMut<'_, Self> { + slf.config.column = Some(name); + slf + } + fn with_limit(mut slf: PyRefMut<'_, Self>, limit: usize) -> PyRefMut<'_, Self> { + slf.config.limit = Some(limit); + slf + } + fn with_query_vector(mut slf: PyRefMut<'_, Self>, vectors: Vec<f32>) -> PyRefMut<'_, Self> { Review Comment: [P2] Match the published single-vector keyword argument The new stub declares `with_query_vector(self, vector: Sequence[float])`, and the existing PyPaimon builder accepts that same `vector` keyword, but this PyO3 method names the argument `vectors`. Consequently `builder.with_query_vector(vector=[1.0, 0.0])` raises `TypeError: ... got an unexpected keyword argument 'vector'` despite matching the public type signature. Positional-only test calls miss this. Please rename the parameter to `vector` (or set the equivalent explicit Python signature) and add a keyword-call regression. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
