From 96ed0c438d962ef30b356368fe956752b4ed2edf Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 23 Oct 2026 13:44:33 +0800 Subject: [PATCH 1/3] [core] Expose phased hybrid search and pure score fusion --- bindings/python/project-description.md | 39 ++ .../python/pypaimon_rust/datafusion.pyi | 49 +- bindings/python/src/context.rs | 5 + bindings/python/src/full_text_search.rs | 20 +- bindings/python/src/hybrid_search.rs | 523 +++++++++++++++++- bindings/python/src/vector_search.rs | 21 +- .../paimon/src/table/hybrid_search_builder.rs | 396 ++++++++++--- .../hybrid_search_builder/parity_tests.rs | 8 +- .../hybrid_search_builder/phased_tests.rs | 358 ++++++++++++ crates/paimon/src/table/mod.rs | 3 +- 10 files changed, 1332 insertions(+), 90 deletions(-) create mode 100644 crates/paimon/src/table/hybrid_search_builder/phased_tests.rs diff --git a/bindings/python/project-description.md b/bindings/python/project-description.md index 9738ab17b..1b281db28 100644 --- a/bindings/python/project-description.md +++ b/bindings/python/project-description.md @@ -128,6 +128,45 @@ when_not_matched)` with `WhenNotMatched` objects. The factories compile conditio once and return an executor with `execute(Table | RecordBatch) -> Table`. The calling engine owns joins, authorization and table writes. +### Hybrid Search Phases + +Hybrid queries can execute routes separately and then fuse their results: + +```python +hybrid = (table.new_hybrid_search_builder() + .add_vector_route("embedding", [1.0, 0.0], 20) + .with_limit(10)) +routes = hybrid.route_builders() +results = [hybrid.to_route_result(route, route.search_builder().execute()) + for route in routes] +ranked = hybrid.rank(results) +``` + +`route_builders()` checks read authorization and captures one snapshot for all +routes, including an empty view. Returned search builders remain configurable +and retain that snapshot after later commits. `to_route_result()` accepts scored +global results and normalizes empty unscored selections; nonempty unscored or +physical primary-key selections cannot participate in global row-ID fusion. + +For externally evaluated scores, `HybridSearchWeightedResult` and +`HybridSearchRanker.RRF`, `.WEIGHTED_SCORE`, or `.MRR` provide the same fusion +without a table or query: + +```python +from pypaimon_rust.datafusion import ( + GlobalIndexResult, HybridSearchRanker, HybridSearchWeightedResult, +) + +hits = GlobalIndexResult.from_scored_row_ids({7: 0.9, 8: 0.5}) +ranked = HybridSearchRanker.MRR.rank([HybridSearchWeightedResult(hits, 2.0)], 1) +assert ranked.row_ids() == {7: 2.0} +``` + +Rankers use Java's float32 scores, weights and arithmetic. A shared captured read +context is retained only when every nonempty input belongs to the same table +view. Fusing external scores or different views returns scores without attaching +an unrelated table or snapshot. + ### Multimodal SQL Helpers `SQLContext` also registers Python scalar UDFs for common BLOB media and vector diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index 976c33090..221e455e1 100644 --- a/bindings/python/python/pypaimon_rust/datafusion.pyi +++ b/bindings/python/python/pypaimon_rust/datafusion.pyi @@ -18,7 +18,7 @@ from datetime import datetime from os import PathLike from typing import ( - Any, Callable, Dict, Iterable, List, Literal, Mapping, Optional, Sequence, Tuple, + Any, Callable, ClassVar, Dict, Iterable, List, Literal, Mapping, Optional, Sequence, Tuple, TypeAlias, Union, ) @@ -1140,6 +1140,47 @@ class GlobalIndexResultReadBuilder: """Read user columns and scores in relevance order from the search snapshot.""" ... +class HybridSearchWeightedResult: + def __init__(self, result: GlobalIndexResult, weight: float = 1.0) -> None: + """Capture a scored global result and a finite positive fusion weight.""" + ... + def result(self) -> GlobalIndexResult: ... + def weight(self) -> float: ... + +class HybridSearchRanker: + RRF: ClassVar[HybridSearchRanker] + WEIGHTED_SCORE: ClassVar[HybridSearchRanker] + MRR: ClassVar[HybridSearchRanker] + def rank(self, results: Sequence[HybridSearchWeightedResult], limit: int) -> GlobalIndexResult: + """Fuse evaluated results without table access or authorization calls.""" + ... + +class HybridSearchRoute: + @staticmethod + def vector(field_name: str, vector: Sequence[float], limit: int, + weight: float = 1.0, options: Optional[Dict[str, str]] = None) -> HybridSearchRoute: ... + @staticmethod + def full_text(field_name: str, query: str, limit: int, + weight: float = 1.0, options: Optional[Dict[str, str]] = None) -> HybridSearchRoute: ... + def kind(self) -> Literal["vector", "full_text"]: ... + def field_name(self) -> str: ... + def vector_value(self) -> Optional[List[float]]: ... + def full_text_query(self) -> Optional[str]: ... + def limit(self) -> int: ... + def weight(self) -> float: ... + def options(self) -> Dict[str, str]: ... + +class HybridSearchRouteBuilder: + def route(self) -> HybridSearchRoute: ... + def snapshot(self) -> Optional[Snapshot]: ... + def search_builder(self) -> Union[VectorSearchBuilder, FullTextSearchBuilder]: ... + def execute(self) -> GlobalIndexResult: ... + +class HybridSearchRouteResult: + def __init__(self, route: HybridSearchRoute, result: GlobalIndexResult) -> None: ... + def route(self) -> HybridSearchRoute: ... + def result(self) -> GlobalIndexResult: ... + class HybridSearchBuilder: def with_limit(self, limit: int) -> HybridSearchBuilder: ... def with_snapshot(self, snapshot: Optional[Snapshot]) -> HybridSearchBuilder: @@ -1154,6 +1195,12 @@ class HybridSearchBuilder: weight: float = 1.0, options: Optional[Dict[str, str]] = None) -> HybridSearchBuilder: ... def add_full_text_route(self, field_name: str, query: str, limit: int, weight: float = 1.0, options: Optional[Dict[str, str]] = None) -> HybridSearchBuilder: ... + def route_builders(self) -> List[HybridSearchRouteBuilder]: + """Capture one snapshot and return configured routes sharing that view.""" + ... + def to_route_result(self, route_builder: HybridSearchRouteBuilder, + result: GlobalIndexResult) -> HybridSearchRouteResult: ... + def rank(self, route_results: Sequence[HybridSearchRouteResult]) -> GlobalIndexResult: ... def execute(self) -> GlobalIndexResult: ... class FullTextSearchBuilder: diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs index 112a5014b..75af55b72 100644 --- a/bindings/python/src/context.rs +++ b/bindings/python/src/context.rs @@ -1001,6 +1001,11 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> this.add_class::()?; this.add_class::()?; this.add_class::()?; + this.add_class::()?; + this.add_class::()?; + this.add_class::()?; + this.add_class::()?; + this.add_class::()?; this.add_class::()?; this.add_class::()?; this.add_class::()?; diff --git a/bindings/python/src/full_text_search.rs b/bindings/python/src/full_text_search.rs index 7f3a441c7..7ce9d2994 100644 --- a/bindings/python/src/full_text_search.rs +++ b/bindings/python/src/full_text_search.rs @@ -23,7 +23,9 @@ use crate::predicate::dict_to_table_predicate; use crate::search_plan::{full_text_splits_from_python, full_text_splits_to_python}; use crate::snapshot::PySnapshot; use paimon::spec::Predicate; -use paimon::table::{FullTextRead, FullTextScan, FullTextScanPlan, Table}; +use paimon::table::{ + FullTextRead, FullTextScan, FullTextScanPlan, HybridSearchRouteBuilder, Table, +}; use paimon_datafusion::runtime::runtime; use pyo3::prelude::*; use pyo3::types::{PyDict, PyTuple}; @@ -68,6 +70,22 @@ impl PyFullTextSearchBuilder { config: Config::default(), } } + + pub(crate) fn from_hybrid_route(route: &HybridSearchRouteBuilder) -> PyResult { + // Core validates the route kind and owns the captured table view. + route.full_text_search_builder().map_err(to_py_err)?; + let config = route.route(); + Ok(Self { + table: Arc::clone(route.table()), + config: Config { + query: config + .full_text_query() + .map(|query| (config.field_name().to_string(), query.to_string())), + limit: Some(config.limit()), + filters: route.filter().cloned().into_iter().collect(), + }, + }) + } } #[pymethods] diff --git a/bindings/python/src/hybrid_search.rs b/bindings/python/src/hybrid_search.rs index 42ef92ee3..32457492c 100644 --- a/bindings/python/src/hybrid_search.rs +++ b/bindings/python/src/hybrid_search.rs @@ -18,11 +18,16 @@ //! Hybrid configuration forwarding; snapshot, routes and fusion belong to core. use crate::error::to_py_err; +use crate::full_text_search::PyFullTextSearchBuilder; use crate::global_index_result::PyGlobalIndexResult; use crate::predicate::dict_to_table_predicate; use crate::snapshot::PySnapshot; +use crate::vector_search::PyVectorSearchBuilder; use paimon::spec::{Predicate, Snapshot}; -use paimon::table::{HybridSearchRanker, HybridSearchRoute, Table}; +use paimon::table::{ + HybridSearchRanker, HybridSearchRoute, HybridSearchRouteBuilder, HybridSearchRouteKind, + HybridSearchRouteResult, HybridSearchWeightedResult, Table, +}; use paimon_datafusion::runtime::runtime; use pyo3::prelude::*; use pyo3::types::PyDict; @@ -30,6 +35,232 @@ use pyo3::types::PyDict; use std::collections::HashMap; use std::sync::Arc; +#[pyclass( + name = "HybridSearchWeightedResult", + module = "pypaimon_rust.datafusion", + frozen +)] +pub struct PyHybridSearchWeightedResult { + inner: HybridSearchWeightedResult, +} + +#[pymethods] +impl PyHybridSearchWeightedResult { + #[new] + #[pyo3(signature = (result, weight=1.0))] + fn new(result: &PyGlobalIndexResult, weight: f32) -> PyResult { + HybridSearchWeightedResult::new(result.inner.clone(), weight) + .map(|inner| Self { inner }) + .map_err(to_py_err) + } + + fn result(&self) -> PyGlobalIndexResult { + PyGlobalIndexResult::new(self.inner.result().clone()) + } + + fn weight(&self) -> f32 { + self.inner.weight() + } +} + +#[pyclass( + name = "HybridSearchRanker", + module = "pypaimon_rust.datafusion", + frozen +)] +pub struct PyHybridSearchRanker { + inner: HybridSearchRanker, +} + +#[pymethods] +impl PyHybridSearchRanker { + #[classattr] + const RRF: Self = Self { + inner: HybridSearchRanker::Rrf, + }; + + #[classattr] + const WEIGHTED_SCORE: Self = Self { + inner: HybridSearchRanker::WeightedScore, + }; + + #[classattr] + const MRR: Self = Self { + inner: HybridSearchRanker::Mrr, + }; + + fn rank( + &self, + py: Python<'_>, + results: Vec>, + limit: usize, + ) -> PyResult { + let results: Vec<_> = results.iter().map(|result| result.inner.clone()).collect(); + py.detach(|| self.inner.rank(&results, limit)) + .map(PyGlobalIndexResult::new) + .map_err(to_py_err) + } +} + +/// Immutable route configuration; constructors and validation belong to core. +#[pyclass( + name = "HybridSearchRoute", + module = "pypaimon_rust.datafusion", + frozen +)] +pub struct PyHybridSearchRoute { + inner: HybridSearchRoute, +} + +#[pymethods] +impl PyHybridSearchRoute { + #[staticmethod] + #[pyo3(signature = (field_name, vector, limit, weight=1.0, options=None))] + fn vector( + field_name: String, + vector: Vec, + limit: usize, + weight: f32, + options: Option>, + ) -> PyResult { + HybridSearchRoute::vector( + field_name, + vector, + limit, + weight, + options.unwrap_or_default(), + ) + .map(|inner| Self { inner }) + .map_err(to_py_err) + } + + #[staticmethod] + #[pyo3(signature = (field_name, query, limit, weight=1.0, options=None))] + fn full_text( + field_name: String, + query: String, + limit: usize, + weight: f32, + options: Option>, + ) -> PyResult { + HybridSearchRoute::full_text( + field_name, + query, + limit, + weight, + options.unwrap_or_default(), + ) + .map(|inner| Self { inner }) + .map_err(to_py_err) + } + + fn kind(&self) -> &'static str { + match self.inner.kind() { + HybridSearchRouteKind::Vector => "vector", + HybridSearchRouteKind::FullText => "full_text", + } + } + + fn field_name(&self) -> &str { + self.inner.field_name() + } + + fn vector_value(&self) -> Option> { + self.inner.vector_value().map(<[f32]>::to_vec) + } + + fn full_text_query(&self) -> Option<&str> { + self.inner.full_text_query() + } + + fn limit(&self) -> usize { + self.inner.limit() + } + + fn weight(&self) -> f32 { + self.inner.weight() + } + + fn options(&self) -> HashMap { + self.inner.options().clone() + } +} + +/// A core-owned route sharing the snapshot captured by route_builders(). +#[pyclass( + name = "HybridSearchRouteBuilder", + module = "pypaimon_rust.datafusion", + frozen +)] +pub struct PyHybridSearchRouteBuilder { + inner: HybridSearchRouteBuilder, +} + +#[pymethods] +impl PyHybridSearchRouteBuilder { + fn route(&self) -> PyHybridSearchRoute { + PyHybridSearchRoute { + inner: self.inner.route().clone(), + } + } + + fn snapshot(&self) -> Option { + self.inner + .table() + .travel_snapshot() + .cloned() + .map(PySnapshot::new) + } + + fn search_builder(&self, py: Python<'_>) -> PyResult> { + match self.inner.route().kind() { + HybridSearchRouteKind::Vector => { + Py::new(py, PyVectorSearchBuilder::from_hybrid_route(&self.inner)?) + .map(Py::into_any) + } + HybridSearchRouteKind::FullText => { + Py::new(py, PyFullTextSearchBuilder::from_hybrid_route(&self.inner)?) + .map(Py::into_any) + } + } + } + + fn execute(&self, py: Python<'_>) -> PyResult { + py.detach(|| runtime().block_on(self.inner.execute())) + .map(PyGlobalIndexResult::new) + .map_err(to_py_err) + } +} + +#[pyclass( + name = "HybridSearchRouteResult", + module = "pypaimon_rust.datafusion", + frozen +)] +pub struct PyHybridSearchRouteResult { + inner: HybridSearchRouteResult, +} + +#[pymethods] +impl PyHybridSearchRouteResult { + #[new] + fn new(route: &PyHybridSearchRoute, result: &PyGlobalIndexResult) -> PyResult { + HybridSearchRouteResult::new(route.inner.clone(), result.inner.clone()) + .map(|inner| Self { inner }) + .map_err(to_py_err) + } + + fn route(&self) -> PyHybridSearchRoute { + PyHybridSearchRoute { + inner: self.inner.route().clone(), + } + } + + fn result(&self) -> PyGlobalIndexResult { + PyGlobalIndexResult::new(self.inner.result().clone()) + } +} + #[pyclass(name = "HybridSearchBuilder", module = "pypaimon_rust.datafusion")] pub struct PyHybridSearchBuilder { table: Arc, @@ -179,6 +410,43 @@ impl PyHybridSearchBuilder { .map_err(to_py_err) } + fn route_builders(&self, py: Python<'_>) -> PyResult> { + let builder = self.builder(); + py.detach(|| runtime().block_on(builder.route_builders())) + .map(|routes| { + routes + .into_iter() + .map(|inner| PyHybridSearchRouteBuilder { inner }) + .collect() + }) + .map_err(to_py_err) + } + + fn to_route_result( + &self, + route_builder: &PyHybridSearchRouteBuilder, + result: &PyGlobalIndexResult, + ) -> PyResult { + self.builder() + .to_route_result(&route_builder.inner, result.inner.clone()) + .map(|inner| PyHybridSearchRouteResult { inner }) + .map_err(to_py_err) + } + + fn rank( + &self, + py: Python<'_>, + route_results: Vec>, + ) -> PyResult { + let route_results: Vec<_> = route_results + .iter() + .map(|result| result.inner.clone()) + .collect(); + py.detach(|| self.builder().rank(&route_results)) + .map(PyGlobalIndexResult::new) + .map_err(to_py_err) + } + fn _execute_with_executor( &self, py: Python<'_>, @@ -203,7 +471,258 @@ mod tests { use super::*; use paimon::catalog::Identifier; use paimon::io::FileIOBuilder; - use paimon::spec::{CommitKind, DataType, IntType, Schema, TableSchema}; + use paimon::spec::{ + ArrayType, CommitKind, DataType, FloatType, IntType, Schema, TableSchema, VarCharType, + }; + use pyo3::types::PyModule; + + fn module(py: Python<'_>) -> Bound<'_, PyModule> { + let package = PyModule::new(py, "pypaimon_rust").unwrap(); + crate::pypaimon_rust(py, &package).unwrap(); + py.import("sys") + .unwrap() + .getattr("modules") + .unwrap() + .set_item("pypaimon_rust", &package) + .unwrap(); + package.getattr("datafusion").unwrap().cast_into().unwrap() + } + + fn table() -> Arc
{ + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("text", DataType::VarChar(VarCharType::string_type())) + .column( + "embedding", + DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))), + ) + .options(HashMap::from([ + ("bucket".into(), "-1".into()), + ("row-tracking.enabled".into(), "true".into()), + ("data-evolution.enabled".into(), "true".into()), + ("global-index.enabled".into(), "true".into()), + ])) + .build() + .unwrap(); + Arc::new(Table::new( + FileIOBuilder::new("memory").build().unwrap(), + Identifier::new("default", "hybrid_phases"), + "memory:/hybrid_phases".into(), + TableSchema::new(0, &schema), + None, + )) + } + + #[test] + fn hybrid_rankers_fuse_typed_external_results_without_a_table() { + Python::attach(|py| { + let locals = PyDict::new(py); + locals.set_item("sdk", module(py)).unwrap(); + py.run( + cr#" +import math +first = sdk.GlobalIndexResult.from_scored_row_ids({10: 3.0, 20: 1.0}) +second = sdk.GlobalIndexResult.from_scored_row_ids({20: 5.0, 30: 1.0}) +results = (sdk.HybridSearchWeightedResult(first, 2.0), + sdk.HybridSearchWeightedResult(second)) +expected = ( + (sdk.HybridSearchRanker.RRF, {20: 2.0 / 62.0 + 1.0 / 61.0, 10: 2.0 / 61.0}), + (sdk.HybridSearchRanker.WEIGHTED_SCORE, {10: 2.0, 20: 1.0}), + (sdk.HybridSearchRanker.MRR, {10: 2.0, 20: 2.0}), +) +for ranker, scores in expected: + actual = ranker.rank(results, 2) + assert isinstance(actual, sdk.GlobalIndexResult) + assert actual.snapshot_id() is None + assert actual.row_ids().keys() == scores.keys() + assert all(math.isclose(actual.row_ids()[key], score, rel_tol=1e-6) + for key, score in scores.items()) + assert ranker.rank([], 2).row_ids() == {} + empty = sdk.HybridSearchWeightedResult(sdk.GlobalIndexResult.from_row_ranges([])) + assert empty.result().row_ids() == {} + assert ranker.rank([empty], 2).row_ids() == {} + try: + ranker.rank(results, 0) + except ValueError as error: + assert 'Limit must be positive' in str(error) + else: + raise AssertionError('zero fusion limit must fail') +for weight in (0.0, -1.0, float('inf')): + try: + sdk.HybridSearchWeightedResult(first, weight) + except ValueError as error: + assert 'Weight must be finite and positive' in str(error) + else: + raise AssertionError('invalid weight must fail') +for result, message in ( + (sdk.GlobalIndexResult.from_row_ranges([(1, 2)]), 'scored index results'), + (sdk.GlobalIndexResult.from_splits(None, []), 'primary-key physical results'), +): + try: + sdk.HybridSearchWeightedResult(result) + except NotImplementedError as error: + assert message in str(error) + else: + raise AssertionError('unsupported result family must fail') +try: + sdk.HybridSearchRanker.RRF.rank([first], 2) +except TypeError: + pass +else: + raise AssertionError('rank must accept typed weighted results') +"#, + Some(&locals), + Some(&locals), + ) + .unwrap(); + }); + } + + #[test] + fn hybrid_route_results_are_immutable_and_validate_result_families() { + Python::attach(|py| { + let locals = PyDict::new(py); + locals.set_item("sdk", module(py)).unwrap(); + py.run( + cr#" +route = sdk.HybridSearchRoute.vector('embedding', [1.0, 0.0], 3, 2.0, {'metric': 'l2'}) +vector = route.vector_value() +vector[0] = 99.0 +options = route.options() +options['metric'] = 'cosine' +assert route.vector_value() == [1.0, 0.0] +assert route.options() == {'metric': 'l2'} +assert route.kind() == 'vector' and route.full_text_query() is None +text = sdk.HybridSearchRoute.full_text('text', '{"match":{"query":"paimon"}}', 4) +assert text.kind() == 'full_text' and text.vector_value() is None +assert text.full_text_query() == '{"match":{"query":"paimon"}}' +for configuration in (route, text): + scored = sdk.GlobalIndexResult.from_scored_row_ids({7: 0.5}) + result = sdk.HybridSearchRouteResult(configuration, scored) + assert result.route().kind() == configuration.kind() + assert result.result().row_ids() == {7: 0.5} + empty = sdk.HybridSearchRouteResult(configuration, sdk.GlobalIndexResult.from_row_ranges([])) + assert empty.result().row_ids() == {} + try: + sdk.HybridSearchRouteResult(configuration, sdk.GlobalIndexResult.from_row_ranges([(1, 2)])) + except NotImplementedError as error: + assert 'scored index results' in str(error) + else: + raise AssertionError('unscored nonempty route result must fail') + try: + configuration.extra = True + except AttributeError: + pass + else: + raise AssertionError('route configuration must be immutable') +assert sdk.HybridSearchRoute.vector('embedding', [], 3).vector_value() == [] +assert sdk.HybridSearchRoute.full_text('text', '', 3).full_text_query() == '' +for create in ( + lambda: sdk.HybridSearchRoute.vector('embedding', [1.0], 0), + lambda: sdk.HybridSearchRoute.full_text('text', '{}', 3, options={'ignored': 'no'}), +): + try: + create() + except ValueError: + pass + else: + raise AssertionError('invalid route must fail core validation') +try: + sdk.HybridSearchRouteResult({}, sdk.GlobalIndexResult.from_scored_row_ids({})) +except TypeError: + pass +else: + raise AssertionError('route result must require typed route metadata') +"#, + Some(&locals), + Some(&locals), + ) + .unwrap(); + }); + } + + #[test] + fn hybrid_phases_return_configured_search_builders_on_one_empty_view() { + Python::attach(|py| { + let locals = PyDict::new(py); + locals.set_item("sdk", module(py)).unwrap(); + locals + .set_item("builder", PyHybridSearchBuilder::new(table())) + .unwrap(); + locals + .set_item("invalid", PyHybridSearchBuilder::new(table())) + .unwrap(); + locals + .set_item("incomplete", PyHybridSearchBuilder::new(table())) + .unwrap(); + let snapshot = Snapshot::builder() + .version(3) + .id(7) + .schema_id(0) + .base_manifest_list("base".into()) + .delta_manifest_list("delta".into()) + .commit_user("captured".into()) + .commit_identifier(1) + .commit_kind(CommitKind::APPEND) + .time_millis(0) + .build(); + locals + .set_item("snapshot", PySnapshot::new(snapshot)) + .unwrap(); + py.run( + cr#" +builder.add_vector_route('embedding', [1.0, 0.0], 3, 2.0) +builder.add_full_text_route('text', '{"match":{"query":"paimon"}}', 4) +builder.with_limit(2).with_snapshot(None) +routes = builder.route_builders() +assert len(routes) == 2 +assert all(isinstance(route, sdk.HybridSearchRouteBuilder) for route in routes) +assert all(route.snapshot() is None for route in routes) +vector, text = (route.search_builder() for route in routes) +assert isinstance(vector, sdk.VectorSearchBuilder) +assert isinstance(text, sdk.FullTextSearchBuilder) +for scan in (vector.new_vector_search_scan(), text.new_full_text_scan()): + plan = scan.scan() + assert plan.snapshot() is None and plan.splits() == [] +assert vector.execute().row_ids() == {} and text.execute().row_ids() == {} +results = [builder.to_route_result(route, route.execute()) for route in routes] +assert [result.route().weight() for result in results] == [2.0, 1.0] +fused = builder.rank(tuple(results)) +assert fused.row_ids() == {} and fused.snapshot_id() is None +# Reading an empty fused search retains its core table context. +assert fused.new_read_builder().with_projection(['id']).read() == [] +weighted = [sdk.HybridSearchWeightedResult(result.result(), result.route().weight()) + for result in results] +assert sdk.HybridSearchRanker.RRF.rank(weighted, 2).new_read_builder().read() == [] +builder.with_snapshot(snapshot) +captured = builder.route_builders() +assert all(isinstance(route.snapshot(), sdk.Snapshot) for route in captured) +assert all(route.snapshot().id() == 7 for route in captured) +# Reconfiguring the hybrid builder cannot change handles already returned. +builder.with_snapshot(None) +assert all(route.snapshot().id() == 7 for route in captured) +assert all(route.snapshot() is None for route in routes) +incomplete.add_vector_route('embedding', [], 3).add_full_text_route('text', '', 3) +incomplete.with_limit(2).with_snapshot(None) +unfinished = incomplete.route_builders() +vector, text = (route.search_builder() for route in unfinished) +vector.with_query_vector([1.0, 0.0]) +text.with_query('text', '{"match":{"query":"paimon"}}') +assert vector.execute().row_ids() == {} and text.execute().row_ids() == {} +invalid.add_vector_route('missing', [1.0], 3).with_limit(2).with_snapshot(None) +try: + invalid.route_builders() +except ValueError as error: + assert "Vector column 'missing' does not exist" in str(error) +else: + raise AssertionError('empty capture must still validate the route schema') +"#, + Some(&locals), + Some(&locals), + ) + .unwrap(); + }); + } #[test] fn hybrid_snapshot_uses_typed_metadata_and_distinguishes_empty_capture() { diff --git a/bindings/python/src/vector_search.rs b/bindings/python/src/vector_search.rs index 3bdfced80..fe6344c29 100644 --- a/bindings/python/src/vector_search.rs +++ b/bindings/python/src/vector_search.rs @@ -19,7 +19,9 @@ use std::collections::HashMap; use std::sync::Arc; use paimon::spec::Predicate; -use paimon::table::{BatchVectorRead, Table, VectorRead, VectorScan, VectorScanPlan}; +use paimon::table::{ + BatchVectorRead, HybridSearchRouteBuilder, Table, VectorRead, VectorScan, VectorScanPlan, +}; use paimon_datafusion::runtime::runtime; use pyo3::prelude::*; use pyo3::types::{PyDict, PyTuple}; @@ -102,6 +104,23 @@ impl PyVectorSearchBuilder { query: None, } } + + pub(crate) fn from_hybrid_route(route: &HybridSearchRouteBuilder) -> PyResult { + // Core validates the route kind and owns the captured table view. + route.vector_search_builder().map_err(to_py_err)?; + let config = route.route(); + Ok(Self { + table: Arc::clone(route.table()), + config: SearchConfig { + column: Some(config.field_name().to_string()), + limit: Some(config.limit()), + options: vec![config.options().clone()], + filters: route.filter().cloned().into_iter().collect(), + partition_filters: Vec::new(), + }, + query: config.vector_value().map(<[f32]>::to_vec), + }) + } } #[pymethods] diff --git a/crates/paimon/src/table/hybrid_search_builder.rs b/crates/paimon/src/table/hybrid_search_builder.rs index 6638e719e..8cb1f7e30 100644 --- a/crates/paimon/src/table/hybrid_search_builder.rs +++ b/crates/paimon/src/table/hybrid_search_builder.rs @@ -36,7 +36,7 @@ use crate::table::source::DataSplit; use crate::table::vector_search_common::{ collect_ranked_rows, ensure_no_reserved_read_columns, reorder_and_strip_position, RankedRow, }; -use crate::table::{ArrowRecordBatchStream, RowRange, Table}; +use crate::table::{ArrowRecordBatchStream, GlobalIndexResult, RowRange, Table}; use crate::vector_search::ScoredRowIds; #[cfg(feature = "fulltext")] @@ -54,6 +54,9 @@ const MAX_ROUTE_WORKERS: usize = 4; #[cfg(all(test, feature = "fulltext"))] mod parity_tests; +#[cfg(test)] +mod phased_tests; + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum HybridSearchRanker { Rrf, @@ -84,6 +87,35 @@ impl HybridSearchRanker { Self::Mrr => Self::MRR, } } + + /// Fuse scored global results without resolving a table or executing queries. + /// A shared captured table remains attached when every nonempty input uses + /// the same read view. External or conflicting views retain only scores. + pub fn rank( + &self, + results: &[HybridSearchWeightedResult], + limit: usize, + ) -> crate::Result { + if limit == 0 { + return Err(crate::Error::ConfigInvalid { + message: "Limit must be positive".into(), + }); + } + let weighted = results + .iter() + .map(|result| { + Ok(WeightedRouteResult { + result: result.result.row_ids()?.clone(), + weight: result.weight, + }) + }) + .collect::>>()?; + let hits = rank_results(*self, &weighted, limit); + Ok(match rank_read_table(results) { + Some(table) => GlobalIndexResult::from_global_search(Arc::new(table.clone()), hits), + None => GlobalIndexResult::from_scored_row_ids(hits), + }) + } } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -113,12 +145,6 @@ impl HybridSearchRoute { ) -> crate::Result { let field_name = field_name.into(); Self::validate_common(&field_name, limit, weight)?; - if vector.is_empty() { - return Err(crate::Error::DataInvalid { - message: "Search vector cannot be empty".to_string(), - source: None, - }); - } Ok(Self { kind: HybridSearchRouteKind::Vector, field_name, @@ -146,11 +172,6 @@ impl HybridSearchRoute { let field_name = field_name.into(); let query = query.into(); Self::validate_common(&field_name, limit, weight)?; - if query.is_empty() { - return Err(crate::Error::ConfigInvalid { - message: "Full-text route query cannot be empty".to_string(), - }); - } Ok(Self { kind: HybridSearchRouteKind::FullText, @@ -212,6 +233,151 @@ impl HybridSearchRoute { } } +/// An independently executable route retaining the hybrid search's captured view. +/// +/// The configured search builders borrow this owned handle, so routes can outlive +/// the hybrid builder without resolving latest again when they execute. +#[derive(Clone, Debug)] +pub struct HybridSearchRouteBuilder { + table: Arc
, + route: HybridSearchRoute, + filter: Option, +} + +impl HybridSearchRouteBuilder { + pub fn route(&self) -> &HybridSearchRoute { + &self.route + } + + pub fn table(&self) -> &Arc
{ + &self.table + } + + pub fn filter(&self) -> Option<&Predicate> { + self.filter.as_ref() + } + + pub fn vector_search_builder(&self) -> crate::Result> { + self.require_kind(HybridSearchRouteKind::Vector)?; + Ok(vector_route_builder( + &self.table, + &self.route, + self.filter.as_ref(), + )) + } + + #[cfg(feature = "fulltext")] + pub fn full_text_search_builder(&self) -> crate::Result> { + self.require_kind(HybridSearchRouteKind::FullText)?; + Ok(full_text_route_builder( + &self.table, + &self.route, + self.filter.as_ref(), + )) + } + + /// Execute this route against the same view as every sibling route. + pub async fn execute(&self) -> crate::Result { + self.execute_internal(None).await + } + + fn require_kind(&self, expected: HybridSearchRouteKind) -> crate::Result<()> { + if self.route.kind != expected { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Hybrid route is {:?}, expected {expected:?}", + self.route.kind + ), + }); + } + Ok(()) + } + + async fn execute_internal( + &self, + executor: Option<&dyn super::SearchTaskExecutor>, + ) -> crate::Result { + match self.route.kind { + HybridSearchRouteKind::Vector => { + let builder = self.vector_search_builder()?; + if let Some(executor) = executor { + let plan = builder.new_scan()?.plan().await?; + builder.new_read()?.read_with_executor(plan, executor).await + } else { + builder.execute().await + } + } + HybridSearchRouteKind::FullText => Ok(GlobalIndexResult::from_global_search( + self.table.clone(), + execute_full_text_route(&self.table, &self.route, self.filter.as_ref(), executor) + .await?, + )), + } + } +} + +/// A scored global selection and its positive fusion weight. +#[derive(Clone, Debug)] +pub struct HybridSearchWeightedResult { + result: GlobalIndexResult, + weight: f32, +} + +impl HybridSearchWeightedResult { + pub fn new(result: GlobalIndexResult, weight: f32) -> crate::Result { + if !weight.is_finite() || weight <= 0.0 { + return Err(crate::Error::ConfigInvalid { + message: format!("Weight must be finite and positive, got: {weight}"), + }); + } + if result.is_split_result() { + return Err(crate::Error::Unsupported { + message: "Hybrid route ranking requires global row-ID results; primary-key physical results are unsupported".into(), + }); + } + let result = if result.row_ids().is_ok() { + result + } else if result.is_empty() { + GlobalIndexResult::from_scored_row_ids(ScoredRowIds::empty()) + } else { + return Err(crate::Error::Unsupported { + message: "Hybrid search requires scored index results".into(), + }); + }; + Ok(Self { result, weight }) + } + + pub fn weight(&self) -> f32 { + self.weight + } + + pub fn result(&self) -> &GlobalIndexResult { + &self.result + } +} + +/// A checked scored global result with its route's configuration. +#[derive(Clone, Debug)] +pub struct HybridSearchRouteResult { + route: HybridSearchRoute, + weighted: HybridSearchWeightedResult, +} + +impl HybridSearchRouteResult { + pub fn new(route: HybridSearchRoute, result: GlobalIndexResult) -> crate::Result { + let weighted = HybridSearchWeightedResult::new(result, route.weight)?; + Ok(Self { route, weighted }) + } + + pub fn route(&self) -> &HybridSearchRoute { + &self.route + } + + pub fn result(&self) -> &GlobalIndexResult { + self.weighted.result() + } +} + pub struct HybridSearchBuilder<'a> { table: &'a Table, routes: Vec, @@ -311,6 +477,40 @@ impl<'a> HybridSearchBuilder<'a> { self } + /// Configure every global route against one captured snapshot, including an + /// explicitly empty view. Queries may be completed on the returned builders; + /// routes remain usable after this builder is dropped. + pub async fn route_builders(&self) -> crate::Result> { + self.validate_global_configuration()?; + let table = self + .resolve_pinned_route_table() + .await? + .unwrap_or_else(|| self.table.copy_with_pinned_snapshot(None)); + Ok(self.owned_route_builders(Arc::new(table))) + } + + pub fn to_route_result( + &self, + route: &HybridSearchRouteBuilder, + result: GlobalIndexResult, + ) -> crate::Result { + HybridSearchRouteResult::new(route.route.clone(), result) + } + + /// Fuse already evaluated global routes without reading table metadata or + /// re-running a search. Preserve the shared captured read context, if any. + pub fn rank( + &self, + route_results: &[HybridSearchRouteResult], + ) -> crate::Result { + let limit = self.validate_search()?; + let weighted = route_results + .iter() + .map(|route| route.weighted.clone()) + .collect::>(); + self.ranker.rank(&weighted, limit) + } + pub async fn execute(&self) -> crate::Result> { self.execute_scored().await?.row_ids()?.to_row_ranges() } @@ -340,31 +540,12 @@ impl<'a> HybridSearchBuilder<'a> { message: "Scheduled hybrid search requires a data-evolution table".into(), }); } - let limit = self.validate_search()?; - - // A primary-key hybrid fuses PHYSICAL positions, not global row ids, so a - // scored/row-range result is unsupported on it: fail loud and direct callers - // to the materialized `execute_read`. A mixed PK/global set of routes cannot - // be fused at all (different address spaces) and also fails loud. The - // append/data-evolution global path is unchanged (mirrors Java `rank`). - match self.classify_routes(&core)? { - HybridAddressSpace::PrimaryKey => { - return Err(crate::Error::DataInvalid { - message: "primary-key hybrid search does not produce global row ids; use the \ - materialized read (execute_read) instead" - .to_string(), - source: None, - }); - } - HybridAddressSpace::Global => {} - } - - self.validate_global_routes()?; + self.validate_global_search()?; // Resolve once even for an empty table or a retained tag. Re-resolving // selectors independently in the routes can observe different versions. let (table, hits) = match self.resolve_pinned_route_table().await? { Some(table) => { - let hits = self.execute_global_routes(&table, limit, executor).await?; + let hits = self.execute_global_routes(&table, executor).await?; (table, hits) } None => ( @@ -391,38 +572,70 @@ impl<'a> HybridSearchBuilder<'a> { }) } + fn validate_global_configuration(&self) -> crate::Result<()> { + let core = CoreOptions::new(self.table.schema().options()); + core.ensure_read_authorized()?; + self.validate_search()?; + if self.classify_routes(&core)? == HybridAddressSpace::PrimaryKey { + return Err(crate::Error::DataInvalid { + message: "primary-key hybrid search does not produce global row ids; use the \ + materialized read (execute_read) instead" + .to_string(), + source: None, + }); + } + for route in &self.routes { + if route.kind == HybridSearchRouteKind::Vector + && !self + .table + .schema() + .fields() + .iter() + .any(|field| field.name() == route.field_name) + { + return Err(crate::Error::ConfigInvalid { + message: format!("Vector column '{}' does not exist", route.field_name), + }); + } + #[cfg(not(feature = "fulltext"))] + if route.kind == HybridSearchRouteKind::FullText { + return Err(crate::Error::ConfigInvalid { + message: "Full-text hybrid routes require the fulltext feature".into(), + }); + } + } + Ok(()) + } + + fn validate_global_search(&self) -> crate::Result<()> { + self.validate_global_configuration()?; + self.validate_global_routes()?; + Ok(()) + } + + fn owned_route_builders(&self, table: Arc
) -> Vec { + self.routes + .iter() + .map(|route| HybridSearchRouteBuilder { + table: table.clone(), + route: route.clone(), + filter: self.filter.clone(), + }) + .collect() + } + fn vector_route_builder<'t>( &self, table: &'t Table, route: &HybridSearchRoute, ) -> super::VectorSearchBuilder<'t> { - let mut builder = table.new_vector_search_builder(); - builder - .with_vector_column(&route.field_name) - .with_query_vector(route.vector.clone().expect("validated vector route")) - .with_limit(route.limit) - .with_options(route.options.clone()); - if let Some(filter) = &self.filter { - builder.with_filter(filter.clone()); - } - builder + vector_route_builder(table, route, self.filter.as_ref()) } fn validate_global_routes(&self) -> crate::Result<()> { for route in &self.routes { match route.kind { HybridSearchRouteKind::Vector => { - if !self - .table - .schema() - .fields() - .iter() - .any(|field| field.name() == route.field_name) - { - return Err(crate::Error::ConfigInvalid { - message: format!("Vector column '{}' does not exist", route.field_name), - }); - } let builder = self.vector_route_builder(self.table, route); builder.new_scan()?; builder.new_read()?; @@ -448,41 +661,19 @@ impl<'a> HybridSearchBuilder<'a> { async fn execute_global_routes( &self, table: &Table, - limit: usize, executor: Option<&dyn super::SearchTaskExecutor>, ) -> crate::Result { // Bound concurrent route work while retaining input order for float // accumulation and keeping each route's weight and limit attached to it. - let route_results = stream::iter(self.routes.clone()) + let route_results = stream::iter(self.owned_route_builders(Arc::new(table.clone()))) .map(|route| async move { - let result = match route.kind { - HybridSearchRouteKind::Vector => { - let builder = self.vector_route_builder(table, &route); - let result = if let Some(executor) = executor { - let plan = builder.new_scan()?.plan().await?; - builder - .new_read()? - .read_with_executor(plan, executor) - .await? - } else { - builder.execute().await? - }; - result.into_row_ids()? - } - HybridSearchRouteKind::FullText => { - execute_full_text_route(table, &route, self.filter.as_ref(), executor) - .await? - } - }; - Ok::<_, crate::Error>(WeightedRouteResult { - result, - weight: route.weight, - }) + let result = route.execute_internal(executor).await?; + self.to_route_result(&route, result) }) .buffered(MAX_ROUTE_WORKERS) .try_collect::>() .await?; - Ok(rank_results(self.ranker, &route_results, limit)) + self.rank(&route_results)?.into_row_ids() } /// Materialize the fused hybrid search hits into Arrow rows, best-fused-score @@ -1054,6 +1245,23 @@ fn build_hybrid_indexed_splits( Ok(out) } +fn vector_route_builder<'t>( + table: &'t Table, + route: &HybridSearchRoute, + filter: Option<&Predicate>, +) -> super::VectorSearchBuilder<'t> { + let mut builder = table.new_vector_search_builder(); + builder + .with_vector_column(&route.field_name) + .with_query_vector(route.vector.clone().expect("validated vector route")) + .with_limit(route.limit) + .with_options(route.options.clone()); + if let Some(filter) = filter { + builder.with_filter(filter.clone()); + } + builder +} + #[cfg(feature = "fulltext")] fn full_text_route_builder<'t>( table: &'t Table, @@ -1111,6 +1319,34 @@ struct WeightedRouteResult { weight: f32, } +fn rank_read_table(results: &[HybridSearchWeightedResult]) -> Option<&Table> { + let mut nonempty = results.iter().filter(|result| !result.result.is_empty()); + let Some(first) = nonempty.next() else { + return results + .iter() + .filter_map(|result| result.result.table()) + .find(|table| table.is_time_traveled()); + }; + let table = first.result.table()?; + if !table.is_time_traveled() { + return None; + } + for result in nonempty { + let other = result.result.table()?; + if table.identifier() != other.identifier() + || table.location() != other.location() + || table.branch() != other.branch() + || table.schema() != other.schema() + || first.result.snapshot_id() != result.result.snapshot_id() + || table.travel_snapshot() != other.travel_snapshot() + || !other.is_time_traveled() + { + return None; + } + } + Some(table) +} + fn rank_results( ranker: HybridSearchRanker, route_results: &[WeightedRouteResult], diff --git a/crates/paimon/src/table/hybrid_search_builder/parity_tests.rs b/crates/paimon/src/table/hybrid_search_builder/parity_tests.rs index 1d542a95d..eea5408f7 100644 --- a/crates/paimon/src/table/hybrid_search_builder/parity_tests.rs +++ b/crates/paimon/src/table/hybrid_search_builder/parity_tests.rs @@ -30,7 +30,7 @@ use arrow_array::{ArrayRef, Int32Array, ListArray, StringArray}; use arrow_schema::{DataType as ArrowType, Field, Schema as ArrowSchema}; use std::sync::Arc; -fn table() -> Table { +pub(super) fn table() -> Table { let schema = Schema::builder() .column("id", DataType::Int(IntType::new())) .column("text", DataType::VarChar(VarCharType::string_type())) @@ -61,7 +61,7 @@ fn table() -> Table { ) } -async fn append(table: &Table, start: i32) { +pub(super) async fn append(table: &Table, start: i32) { let columns: Vec = vec![ Arc::new(Int32Array::from(vec![ start, @@ -104,7 +104,7 @@ async fn append(table: &Table, start: i32) { .unwrap(); } -async fn index(table: &Table, kind: &str, column: &str) { +pub(super) async fn index(table: &Table, kind: &str, column: &str) { let mut builder = table.new_global_index_build_builder(); builder.with_index_type(kind).with_index_column(column); if kind == "ivf-flat" { @@ -158,7 +158,7 @@ async fn full_text_route_rejects_plain_text_as_java_does() { assert!(builder.execute_scored().await.is_err()); } -fn mixed_query(table: &Table, route_limit: usize) -> HybridSearchBuilder<'_> { +pub(super) fn mixed_query(table: &Table, route_limit: usize) -> HybridSearchBuilder<'_> { let mut builder = table.new_hybrid_search_builder(); builder .add_vector_route( diff --git a/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs b/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs new file mode 100644 index 000000000..113f31f0a --- /dev/null +++ b/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs @@ -0,0 +1,358 @@ +// 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 super::*; +use crate::catalog::Identifier; +use crate::io::FileIOBuilder; +use crate::spec::{DataType, IntType, Schema, TableSchema}; + +fn scored(ids: &[u64], scores: &[f32], weight: f32) -> HybridSearchWeightedResult { + HybridSearchWeightedResult::new( + GlobalIndexResult::from_scored_row_ids(ScoredRowIds::new(ids.to_vec(), scores.to_vec())), + weight, + ) + .unwrap() +} + +#[test] +fn ranker_fuses_external_results_with_weights_and_deterministic_ties() { + let results = [ + scored(&[1, 2, 3], &[10.0, 5.0, 0.0], 2.0), + scored(&[2, 3, 4], &[100.0, 50.0, 0.0], 1.0), + ]; + for (ranker, ids, expected) in [ + ( + HybridSearchRanker::Rrf, + vec![2, 3, 1], + vec![2.0 / 62.0 + 1.0 / 61.0, 2.0 / 63.0 + 1.0 / 62.0, 2.0 / 61.0], + ), + ( + HybridSearchRanker::WeightedScore, + vec![1, 2, 3], + vec![2.0, 2.0, 0.5], + ), + ( + HybridSearchRanker::Mrr, + vec![1, 2, 3], + vec![2.0, 2.0, 2.0 / 3.0 + 0.5], + ), + ] { + let ranked = ranker.rank(&results, 3).unwrap(); + let hits = ranked.row_ids().unwrap(); + assert_eq!(hits.row_ids, ids, "ranker {ranker:?}"); + for (actual, expected) in hits.scores.iter().zip(expected) { + assert!((actual - expected).abs() < 1e-6, "ranker {ranker:?}"); + } + assert!(ranked.table().is_none()); + } + + // Callers may rank an arbitrary subset or repeat a result. Only each + // supplied result's weight participates, independently of a route list. + let repeated = HybridSearchRanker::Mrr + .rank(&[results[0].clone(), results[0].clone()], 1) + .unwrap(); + assert_eq!(repeated.row_ids().unwrap().row_ids, vec![1]); + assert_eq!(repeated.row_ids().unwrap().scores, vec![4.0]); +} + +#[test] +fn weighted_results_normalize_empty_selections_and_reject_invalid_inputs() { + let empty = + HybridSearchWeightedResult::new(GlobalIndexResult::from_row_ranges(vec![]), 2.0).unwrap(); + assert_eq!(empty.weight(), 2.0); + assert!(empty.result().row_ids().unwrap().is_empty()); + assert!(HybridSearchRanker::Rrf + .rank(&[empty], 2) + .unwrap() + .is_empty()); + assert!(HybridSearchRanker::Mrr.rank(&[], 2).unwrap().is_empty()); + assert!(HybridSearchRanker::Rrf.rank(&[], 0).is_err()); + assert!(HybridSearchWeightedResult::new( + GlobalIndexResult::from_row_ranges(vec![RowRange::new(1, 1)]), + 1.0, + ) + .is_err()); + assert!( + HybridSearchWeightedResult::new(GlobalIndexResult::from_splits(Some(1), vec![]), 1.0,) + .is_err() + ); + for weight in [0.0, -1.0, f32::INFINITY] { + assert!(HybridSearchWeightedResult::new( + GlobalIndexResult::from_scored_row_ids(ScoredRowIds::empty()), + weight, + ) + .is_err()); + } +} + +#[test] +fn builder_ranks_supplied_results_without_authentication_or_query_validation() { + // No metadata exists and this query could never execute: rank consumes + // evaluated scores only, just as Java HybridSearchBuilder.rank does. + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option("query-auth.enabled", "true") + .build() + .unwrap(); + let table = Table::new( + FileIOBuilder::new("memory").build().unwrap(), + Identifier::new("default", "rank-only"), + "memory:/rank-only-no-metadata".into(), + TableSchema::new(0, &schema), + None, + ); + let route = HybridSearchRoute::vector("missing", vec![0.0], 1, 2.0, HashMap::new()).unwrap(); + let result = HybridSearchRouteResult::new( + route.clone(), + GlobalIndexResult::from_scored_row_ids(ScoredRowIds::new(vec![7], vec![0.5])), + ) + .unwrap(); + assert_eq!(result.route().field_name(), "missing"); + let mut builder = table.new_hybrid_search_builder(); + builder.add_route(route).with_limit(1).with_mrr_ranker(); + let ranked = builder.rank(&[result]).unwrap(); + assert_eq!(ranked.row_ids().unwrap().row_ids, vec![7]); + assert_eq!(ranked.row_ids().unwrap().scores, vec![2.0]); + builder.with_limit(0); + assert!(builder.rank(&[]).is_err()); + assert!(table.new_hybrid_search_builder().rank(&[]).is_err()); +} + +#[cfg(feature = "fulltext")] +mod captured_routes { + use super::*; + use crate::spec::{Datum, PredicateBuilder}; + + use super::super::parity_tests::{append, index, mixed_query, table}; + + #[tokio::test] + async fn routes_share_one_snapshot_and_remain_owned_after_later_commits() { + let table = table(); + append(&table, 0).await; + index(&table, "ivf-flat", "embedding").await; + index(&table, "full-text", "text").await; + let predicates = PredicateBuilder::new(table.schema().fields()); + let keep = predicates + .equal("label", Datum::String("keep".into())) + .unwrap(); + let (routes, expected) = { + let mut builder = mixed_query(&table, 4); + builder.with_filter(keep); + let expected = builder.execute_scored().await.unwrap(); + (builder.route_builders().await.unwrap(), expected) + }; + assert_eq!(routes.len(), 2); + assert!(Arc::ptr_eq(routes[0].table(), routes[1].table())); + assert!(routes[0].filter().is_some()); + assert_eq!( + routes[0].table().travel_snapshot().map(|s| s.id()), + expected.snapshot_id() + ); + assert!(routes[0].vector_search_builder().is_ok()); + assert!(routes[0].full_text_search_builder().is_err()); + assert!(routes[1].full_text_search_builder().is_ok()); + assert!(routes[1].vector_search_builder().is_err()); + + append(&table, 4).await; + let mut results = vec![]; + for route in &routes { + let result = route.execute().await.unwrap(); + assert_eq!(result.snapshot_id(), expected.snapshot_id()); + results.push(HybridSearchRouteResult::new(route.route().clone(), result).unwrap()); + } + let builder = mixed_query(&table, 4); + let ranked = builder.rank(&results).unwrap(); + assert_eq!(ranked.snapshot_id(), expected.snapshot_id()); + assert_eq!( + ranked.row_ids().unwrap().row_ids, + expected.row_ids().unwrap().row_ids + ); + assert_eq!( + ranked.row_ids().unwrap().scores, + expected.row_ids().unwrap().scores + ); + assert!(ranked + .row_ids() + .unwrap() + .row_ids + .iter() + .all(|id| [1, 2].contains(id))); + // Repeating a route uses its retained metadata rather than latest. + assert_eq!( + routes[0].execute().await.unwrap().snapshot_id(), + expected.snapshot_id() + ); + assert_ne!( + builder.execute_scored().await.unwrap().snapshot_id(), + expected.snapshot_id() + ); + } + + #[tokio::test] + async fn empty_route_view_stays_empty_after_a_first_commit() { + let table = table(); + let builder = mixed_query(&table, 4); + let routes = builder.route_builders().await.unwrap(); + assert!(Arc::ptr_eq(routes[0].table(), routes[1].table())); + assert!(routes[0].table().is_time_traveled()); + assert!(routes[0].table().travel_snapshot().is_none()); + append(&table, 0).await; + let mut results = vec![]; + for route in &routes { + results.push( + builder + .to_route_result(route, route.execute().await.unwrap()) + .unwrap(), + ); + } + let ranked = builder.rank(&results).unwrap(); + assert!(ranked.is_empty()); + assert!(ranked.table().unwrap().is_time_traveled()); + assert!(ranked.snapshot_id().is_none()); + } + + #[tokio::test] + async fn incomplete_queries_can_be_finished_on_captured_route_builders() { + let table = table(); + append(&table, 0).await; + index(&table, "ivf-flat", "embedding").await; + index(&table, "full-text", "text").await; + let snapshot = table.resolve_read_snapshot().await.unwrap().unwrap(); + let mut builder = table.new_hybrid_search_builder(); + builder + .add_vector_route("embedding", vec![], 4, 1.0, HashMap::new()) + .unwrap() + .add_full_text_route("text", "", 4, 1.0, HashMap::new()) + .unwrap() + .with_limit(16); + let routes = builder.route_builders().await.unwrap(); + assert!(routes[0].route().vector_value().unwrap().is_empty()); + assert_eq!(routes[1].route().full_text_query(), Some("")); + // Preparing routes does not execute their incomplete queries. + assert!(builder.execute_scored().await.is_err()); + let mut empty_text = table.new_hybrid_search_builder(); + empty_text + .add_full_text_route("text", "", 4, 1.0, HashMap::new()) + .unwrap() + .with_limit(4); + assert!(empty_text.execute_scored().await.is_err()); + + append(&table, 4).await; + let mut vector = routes[0].vector_search_builder().unwrap(); + vector.with_query_vector(vec![0.0, 0.0]); + let mut text = routes[1].full_text_search_builder().unwrap(); + text.with_query("text", r#"{"match":{"query":"paimon"}}"#); + let results = [ + builder + .to_route_result(&routes[0], vector.execute().await.unwrap()) + .unwrap(), + builder + .to_route_result(&routes[1], text.execute_scored().await.unwrap()) + .unwrap(), + ]; + let actual = builder.rank(&results).unwrap(); + let expected = mixed_query(&table, 4) + .with_snapshot(Some(&snapshot)) + .execute_scored() + .await + .unwrap(); + assert_eq!(actual.snapshot_id(), Some(snapshot.id())); + assert_eq!( + actual.row_ids().unwrap().row_ids, + expected.row_ids().unwrap().row_ids + ); + assert_eq!( + actual.row_ids().unwrap().scores, + expected.row_ids().unwrap().scores + ); + } + + #[tokio::test] + async fn rank_drops_read_context_for_different_snapshots_or_external_hits() { + let table = table(); + append(&table, 0).await; + let first = table.resolve_read_snapshot().await.unwrap().unwrap(); + let captured = table.copy_with_pinned_snapshot(Some(&first)); + let left = HybridSearchWeightedResult::new( + GlobalIndexResult::from_global_search( + Arc::new(captured.clone()), + ScoredRowIds::new(vec![1], vec![0.9]), + ), + 2.0, + ) + .unwrap(); + let sibling = HybridSearchWeightedResult::new( + GlobalIndexResult::from_global_search( + Arc::new(captured), + ScoredRowIds::new(vec![2], vec![0.8]), + ), + 1.0, + ) + .unwrap(); + let compatible = HybridSearchRanker::Mrr + .rank(&[left.clone(), sibling], 2) + .unwrap(); + assert_eq!(compatible.snapshot_id(), Some(first.id())); + assert!(compatible.table().is_some()); + + append(&table, 4).await; + let later = table.resolve_read_snapshot().await.unwrap().unwrap(); + let right = HybridSearchWeightedResult::new( + GlobalIndexResult::from_global_search( + Arc::new(table.copy_with_pinned_snapshot(Some(&later))), + ScoredRowIds::new(vec![2], vec![0.8]), + ), + 1.0, + ) + .unwrap(); + for right in [right, scored(&[2], &[0.8], 1.0)] { + let ranked = HybridSearchRanker::Mrr + .rank(&[left.clone(), right], 2) + .unwrap(); + assert_eq!(ranked.row_ids().unwrap().row_ids, vec![1, 2]); + assert_eq!(ranked.row_ids().unwrap().scores, vec![2.0, 1.0]); + assert!(ranked.table().is_none()); + assert!(ranked.snapshot_id().is_none()); + assert!(ranked.new_read_builder().read().await.is_err()); + } + } + + #[tokio::test] + async fn route_preparation_validates_before_resolving_snapshots() { + let table = table(); + let guarded = table.copy_with_options(HashMap::from([( + "query-auth.enabled".into(), + "true".into(), + )])); + let error = mixed_query(&guarded, 4).route_builders().await.unwrap_err(); + assert!(format!("{error:?}").contains("query-auth.enabled")); + assert!(table + .new_hybrid_search_builder() + .route_builders() + .await + .is_err()); + let mut missing = table.new_hybrid_search_builder(); + missing + .add_vector_route("missing", vec![0.0], 4, 1.0, HashMap::new()) + .unwrap() + .with_limit(4); + assert!(missing.route_builders().await.is_err()); + let mut zero = mixed_query(&table, 4); + zero.with_limit(0); + assert!(zero.route_builders().await.is_err()); + } +} diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index dc463d171..0e1193fa2 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -230,7 +230,8 @@ pub use global_index_result::{ }; pub use global_index_types::{normalize_global_index_type, SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP}; pub use hybrid_search_builder::{ - HybridSearchBuilder, HybridSearchRanker, HybridSearchRoute, HybridSearchRouteKind, + HybridSearchBuilder, HybridSearchRanker, HybridSearchRoute, HybridSearchRouteBuilder, + HybridSearchRouteKind, HybridSearchRouteResult, HybridSearchWeightedResult, }; pub use incremental_scan::{ IncrementalPlan, IncrementalScan, IncrementalScanMode, IncrementalSplit, From 2cf5601c7d74d59fdd0865dc2640f4489fe25dea Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 23 Oct 2026 14:24:44 +0800 Subject: [PATCH 2/3] Unify search results and retain mutable route builders --- bindings/python/README.md | 19 ++ bindings/python/project-description.md | 6 +- .../python/pypaimon_rust/datafusion.pyi | 9 +- bindings/python/src/full_text_search.rs | 79 ++----- bindings/python/src/hybrid_search.rs | 63 ++++-- bindings/python/src/vector_search.rs | 85 ++------ .../datafusion/src/full_text_search.rs | 6 +- .../datafusion/src/hybrid_search.rs | 2 +- .../full_text_index_build_builder/tests.rs | 4 + .../src/table/full_text_search_builder.rs | 127 +++++++----- .../full_text_search_builder/append_tests.rs | 28 +-- .../full_text_search_builder/parity_tests.rs | 75 ++++--- .../table/full_text_search_builder/read.rs | 31 +-- .../table/full_text_search_builder/scan.rs | 29 ++- .../parallel_tests.rs | 5 +- .../paimon/src/table/hybrid_search_builder.rs | 192 ++++++++++-------- .../hybrid_search_builder/parity_tests.rs | 52 ++--- .../hybrid_search_builder/phased_tests.rs | 178 +++++++++++++++- .../paimon/src/table/vector_search_builder.rs | 118 +++++++---- .../data_evolution_search_sidecar_test.rs | 4 +- docs/src/sql.md | 3 +- 21 files changed, 670 insertions(+), 445 deletions(-) diff --git a/bindings/python/README.md b/bindings/python/README.md index 4e5ae99b6..c12c8f3f6 100644 --- a/bindings/python/README.md +++ b/bindings/python/README.md @@ -106,6 +106,25 @@ 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. +Hybrid searches return the same `GlobalIndexResult` from `execute()`. To run +routes separately, call `route_builders()` once and configure each route through +its `search_builder()`: + +```python +hybrid = (table.new_hybrid_search_builder() + .add_vector_route("embedding", [1.0, 0.0], 10) + .with_limit(5)) +route = hybrid.route_builders()[0] +route.search_builder().with_query_vector([0.0, 1.0]).with_limit(3) +result = route.execute() +fused = hybrid.rank([hybrid.to_route_result(route, result)]) +``` + +Each route returns the same live search builder on repeated calls. Query, +limit, option and filter changes affect later `route.execute()` calls. All +returned routes keep the snapshot captured by `route_builders()`, including +a captured empty table, even if later writes publish a newer snapshot. + ### Native Read / Write Beyond SQL, you can use the lower-level read and write APIs directly from Python. diff --git a/bindings/python/project-description.md b/bindings/python/project-description.md index 1b281db28..870373f34 100644 --- a/bindings/python/project-description.md +++ b/bindings/python/project-description.md @@ -143,8 +143,10 @@ ranked = hybrid.rank(results) ``` `route_builders()` checks read authorization and captures one snapshot for all -routes, including an empty view. Returned search builders remain configurable -and retain that snapshot after later commits. `to_route_result()` accepts scored +routes, including an empty view. Each route retains its search builder: repeated +`search_builder()` calls return the same object, and configuring it changes later +`route.execute()` calls. Returned builders retain their snapshot after later +commits. `to_route_result()` accepts scored global results and normalizes empty unscored selections; nonempty unscored or physical primary-key selections cannot participate in global row-ID fusion. diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index 221e455e1..fb50bd68d 100644 --- a/bindings/python/python/pypaimon_rust/datafusion.pyi +++ b/bindings/python/python/pypaimon_rust/datafusion.pyi @@ -1173,7 +1173,12 @@ class HybridSearchRoute: class HybridSearchRouteBuilder: def route(self) -> HybridSearchRoute: ... def snapshot(self) -> Optional[Snapshot]: ... - def search_builder(self) -> Union[VectorSearchBuilder, FullTextSearchBuilder]: ... + def search_builder(self) -> Union[VectorSearchBuilder, FullTextSearchBuilder]: + """Return this route's live builder; mutations affect subsequent route execution. + + Repeated calls return the same object on the captured snapshot. + """ + ... def execute(self) -> GlobalIndexResult: ... class HybridSearchRouteResult: @@ -1196,7 +1201,7 @@ class HybridSearchBuilder: def add_full_text_route(self, field_name: str, query: str, limit: int, weight: float = 1.0, options: Optional[Dict[str, str]] = None) -> HybridSearchBuilder: ... def route_builders(self) -> List[HybridSearchRouteBuilder]: - """Capture one snapshot and return configured routes sharing that view.""" + """Capture one snapshot and return configured routes with live search builders.""" ... def to_route_result(self, route_builder: HybridSearchRouteBuilder, result: GlobalIndexResult) -> HybridSearchRouteResult: ... diff --git a/bindings/python/src/full_text_search.rs b/bindings/python/src/full_text_search.rs index 7ce9d2994..c0e5aa7bb 100644 --- a/bindings/python/src/full_text_search.rs +++ b/bindings/python/src/full_text_search.rs @@ -22,68 +22,31 @@ use crate::global_index_result::PyGlobalIndexResult; use crate::predicate::dict_to_table_predicate; use crate::search_plan::{full_text_splits_from_python, full_text_splits_to_python}; use crate::snapshot::PySnapshot; -use paimon::spec::Predicate; use paimon::table::{ - FullTextRead, FullTextScan, FullTextScanPlan, HybridSearchRouteBuilder, Table, + FullTextRead, FullTextScan, FullTextScanPlan, FullTextSearchBuilder, HybridSearchRouteBuilder, + Table, }; use paimon_datafusion::runtime::runtime; use pyo3::prelude::*; use pyo3::types::{PyDict, PyTuple}; use std::sync::Arc; -#[derive(Default)] -struct Config { - query: Option<(String, String)>, - limit: Option, - filters: Vec, -} - -impl Config { - fn builder<'a>( - &self, - table: &'a Table, - ) -> paimon::Result> { - let mut builder = table.new_full_text_search_builder(); - if let Some((column, query)) = &self.query { - builder.with_query(column, query); - } - if let Some(limit) = self.limit { - builder.with_limit(limit); - } - for filter in &self.filters { - builder.with_filter(filter.clone()); - } - Ok(builder) - } -} - #[pyclass(name = "FullTextSearchBuilder", module = "pypaimon_rust.datafusion")] pub struct PyFullTextSearchBuilder { table: Arc
, - config: Config, + inner: FullTextSearchBuilder<'static>, } impl PyFullTextSearchBuilder { pub(crate) fn new(table: Arc
) -> Self { - Self { - table, - config: Config::default(), - } + let inner = table.new_full_text_search_builder().into_owned(); + Self { table, inner } } pub(crate) fn from_hybrid_route(route: &HybridSearchRouteBuilder) -> PyResult { - // Core validates the route kind and owns the captured table view. - route.full_text_search_builder().map_err(to_py_err)?; - let config = route.route(); Ok(Self { table: Arc::clone(route.table()), - config: Config { - query: config - .full_text_query() - .map(|query| (config.field_name().to_string(), query.to_string())), - limit: Some(config.limit()), - filters: route.filter().cloned().into_iter().collect(), - }, + inner: route.full_text_search_builder().map_err(to_py_err)?, }) } } @@ -95,11 +58,11 @@ impl PyFullTextSearchBuilder { field_name: String, query: String, ) -> PyRefMut<'_, Self> { - slf.config.query = Some((field_name, query)); + slf.inner.with_query(&field_name, &query); slf } fn with_limit(mut slf: PyRefMut<'_, Self>, limit: usize) -> PyRefMut<'_, Self> { - slf.config.limit = Some(limit); + slf.inner.with_limit(limit); slf } fn with_filter<'py>( @@ -107,7 +70,7 @@ impl PyFullTextSearchBuilder { predicate: &Bound<'_, PyDict>, ) -> PyResult> { let filter = dict_to_table_predicate(predicate, slf.table.schema(), true)?; - slf.config.filters.push(filter); + slf.inner.with_filter(filter); Ok(slf) } fn with_partition_filter<'py>( @@ -115,37 +78,21 @@ impl PyFullTextSearchBuilder { predicate: &Bound<'_, PyDict>, ) -> PyResult> { let filter = dict_to_table_predicate(predicate, slf.table.schema(), true)?; - slf.config - .builder(&slf.table) - .map_err(to_py_err)? - .with_partition_filter(filter.clone()) - .map_err(to_py_err)?; - slf.config.filters.push(filter); + slf.inner.with_partition_filter(filter).map_err(to_py_err)?; Ok(slf) } fn new_full_text_scan(&self) -> PyResult { Ok(PyFullTextScan { - inner: self - .config - .builder(&self.table) - .map_err(to_py_err)? - .new_scan() - .map_err(to_py_err)?, + inner: self.inner.new_scan().map_err(to_py_err)?, }) } fn new_full_text_read(&self) -> PyResult { Ok(PyFullTextRead { - inner: self - .config - .builder(&self.table) - .map_err(to_py_err)? - .new_read() - .map_err(to_py_err)?, + inner: self.inner.new_read().map_err(to_py_err)?, }) } fn execute(&self, py: Python<'_>) -> PyResult { - let builder = self.config.builder(&self.table).map_err(to_py_err)?; - py.detach(|| runtime().block_on(builder.execute_scored())) + py.detach(|| runtime().block_on(self.inner.execute())) .map(PyGlobalIndexResult::new) .map_err(to_py_err) } diff --git a/bindings/python/src/hybrid_search.rs b/bindings/python/src/hybrid_search.rs index 32457492c..6888f21cc 100644 --- a/bindings/python/src/hybrid_search.rs +++ b/bindings/python/src/hybrid_search.rs @@ -194,6 +194,24 @@ impl PyHybridSearchRoute { )] pub struct PyHybridSearchRouteBuilder { inner: HybridSearchRouteBuilder, + search_builder: Py, +} + +impl PyHybridSearchRouteBuilder { + fn new(py: Python<'_>, inner: HybridSearchRouteBuilder) -> PyResult { + let search_builder = match inner.route().kind() { + HybridSearchRouteKind::Vector => { + Py::new(py, PyVectorSearchBuilder::from_hybrid_route(&inner)?)?.into_any() + } + HybridSearchRouteKind::FullText => { + Py::new(py, PyFullTextSearchBuilder::from_hybrid_route(&inner)?)?.into_any() + } + }; + Ok(Self { + inner, + search_builder, + }) + } } #[pymethods] @@ -212,17 +230,8 @@ impl PyHybridSearchRouteBuilder { .map(PySnapshot::new) } - fn search_builder(&self, py: Python<'_>) -> PyResult> { - match self.inner.route().kind() { - HybridSearchRouteKind::Vector => { - Py::new(py, PyVectorSearchBuilder::from_hybrid_route(&self.inner)?) - .map(Py::into_any) - } - HybridSearchRouteKind::FullText => { - Py::new(py, PyFullTextSearchBuilder::from_hybrid_route(&self.inner)?) - .map(Py::into_any) - } - } + fn search_builder(&self, py: Python<'_>) -> Py { + self.search_builder.clone_ref(py) } fn execute(&self, py: Python<'_>) -> PyResult { @@ -405,21 +414,20 @@ impl PyHybridSearchBuilder { fn execute(&self, py: Python<'_>) -> PyResult { let builder = self.builder(); - py.detach(|| runtime().block_on(builder.execute_scored())) + py.detach(|| runtime().block_on(builder.execute())) .map(PyGlobalIndexResult::new) .map_err(to_py_err) } fn route_builders(&self, py: Python<'_>) -> PyResult> { let builder = self.builder(); - py.detach(|| runtime().block_on(builder.route_builders())) - .map(|routes| { - routes - .into_iter() - .map(|inner| PyHybridSearchRouteBuilder { inner }) - .collect() - }) - .map_err(to_py_err) + let routes = py + .detach(|| runtime().block_on(builder.route_builders())) + .map_err(to_py_err)?; + routes + .into_iter() + .map(|inner| PyHybridSearchRouteBuilder::new(py, inner)) + .collect() } fn to_route_result( @@ -681,6 +689,18 @@ assert all(route.snapshot() is None for route in routes) vector, text = (route.search_builder() for route in routes) assert isinstance(vector, sdk.VectorSearchBuilder) assert isinstance(text, sdk.FullTextSearchBuilder) +assert routes[0].search_builder() is vector +assert routes[1].search_builder() is text +for route, search in zip(routes, (vector, text)): + assert search.with_limit(0) is search + try: + route.execute() + except ValueError as error: + assert 'Limit must be' in str(error), str(error) + else: + raise AssertionError('route.execute must observe search builder mutations') + search.with_limit(1) + assert route.execute().row_ids() == {} for scan in (vector.new_vector_search_scan(), text.new_full_text_scan()): plan = scan.scan() assert plan.snapshot() is None and plan.splits() == [] @@ -709,6 +729,9 @@ vector, text = (route.search_builder() for route in unfinished) vector.with_query_vector([1.0, 0.0]) text.with_query('text', '{"match":{"query":"paimon"}}') assert vector.execute().row_ids() == {} and text.execute().row_ids() == {} +assert unfinished[0].search_builder() is vector +assert unfinished[1].search_builder() is text +assert all(route.execute().row_ids() == {} for route in unfinished) invalid.add_vector_route('missing', [1.0], 3).with_limit(2).with_snapshot(None) try: invalid.route_builders() diff --git a/bindings/python/src/vector_search.rs b/bindings/python/src/vector_search.rs index fe6344c29..302b2e16d 100644 --- a/bindings/python/src/vector_search.rs +++ b/bindings/python/src/vector_search.rs @@ -21,6 +21,7 @@ use std::sync::Arc; use paimon::spec::Predicate; use paimon::table::{ BatchVectorRead, HybridSearchRouteBuilder, Table, VectorRead, VectorScan, VectorScanPlan, + VectorSearchBuilder, }; use paimon_datafusion::runtime::runtime; use pyo3::prelude::*; @@ -42,29 +43,6 @@ struct SearchConfig { } impl SearchConfig { - fn single<'a>( - &self, - table: &'a Table, - ) -> paimon::Result> { - 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, @@ -92,33 +70,19 @@ impl SearchConfig { #[pyclass(name = "VectorSearchBuilder", module = "pypaimon_rust.datafusion")] pub struct PyVectorSearchBuilder { table: Arc
, - config: SearchConfig, - query: Option>, + inner: VectorSearchBuilder<'static>, } impl PyVectorSearchBuilder { pub(crate) fn new(table: Arc
) -> Self { - Self { - table, - config: SearchConfig::default(), - query: None, - } + let inner = table.new_vector_search_builder().into_owned(); + Self { table, inner } } pub(crate) fn from_hybrid_route(route: &HybridSearchRouteBuilder) -> PyResult { - // Core validates the route kind and owns the captured table view. - route.vector_search_builder().map_err(to_py_err)?; - let config = route.route(); Ok(Self { table: Arc::clone(route.table()), - config: SearchConfig { - column: Some(config.field_name().to_string()), - limit: Some(config.limit()), - options: vec![config.options().clone()], - filters: route.filter().cloned().into_iter().collect(), - partition_filters: Vec::new(), - }, - query: config.vector_value().map(<[f32]>::to_vec), + inner: route.vector_search_builder().map_err(to_py_err)?, }) } } @@ -126,26 +90,26 @@ impl PyVectorSearchBuilder { #[pymethods] impl PyVectorSearchBuilder { fn with_vector_column(mut slf: PyRefMut<'_, Self>, name: String) -> PyRefMut<'_, Self> { - slf.config.column = Some(name); + slf.inner.with_vector_column(&name); slf } fn with_limit(mut slf: PyRefMut<'_, Self>, limit: usize) -> PyRefMut<'_, Self> { - slf.config.limit = Some(limit); + slf.inner.with_limit(limit); slf } fn with_query_vector(mut slf: PyRefMut<'_, Self>, vector: Vec) -> PyRefMut<'_, Self> { - slf.query = Some(vector); + slf.inner.with_query_vector(vector); slf } fn with_option(mut slf: PyRefMut<'_, Self>, key: String, value: String) -> PyRefMut<'_, Self> { - slf.config.options.push(HashMap::from([(key, value)])); + slf.inner.with_options(HashMap::from([(key, value)])); slf } fn with_options( mut slf: PyRefMut<'_, Self>, options: HashMap, ) -> PyRefMut<'_, Self> { - slf.config.options.push(options); + slf.inner.with_options(options); slf } fn with_filter<'py>( @@ -153,7 +117,7 @@ impl PyVectorSearchBuilder { predicate: &Bound<'_, PyDict>, ) -> PyResult> { let predicate = dict_to_table_predicate(predicate, slf.table.schema(), true)?; - slf.config.filters.push(predicate); + slf.inner.with_filter(predicate); Ok(slf) } fn with_partition_filter<'py>( @@ -161,38 +125,22 @@ impl PyVectorSearchBuilder { predicate: &Bound<'_, PyDict>, ) -> PyResult> { 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()) + slf.inner + .with_partition_filter(predicate) .map_err(to_py_err)?; - slf.config.partition_filters.push(predicate); Ok(slf) } fn new_vector_search_scan(&self) -> PyResult { - let inner = self - .config - .single(&self.table) - .map_err(to_py_err)? - .new_scan() - .map_err(to_py_err)?; + let inner = self.inner.new_scan().map_err(to_py_err)?; Ok(PyVectorScan { inner }) } fn new_vector_search_read(&self) -> PyResult { - 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)?; + let inner = self.inner.new_read().map_err(to_py_err)?; Ok(PyVectorRead { inner }) } fn execute(&self, py: Python<'_>) -> PyResult { - 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())) + .detach(|| runtime().block_on(self.inner.execute())) .map_err(to_py_err)?; Ok(PyGlobalIndexResult::new(results)) } @@ -703,6 +651,7 @@ for factory, expected in ( ): builder = factory() assert callable(builder.execute) + assert not hasattr(builder, 'execute_scored') assert not hasattr(builder, 'execute_local') assert not hasattr(builder, 'execute_batch_local') try: diff --git a/crates/integrations/datafusion/src/full_text_search.rs b/crates/integrations/datafusion/src/full_text_search.rs index 4a4b4c577..a237d5998 100644 --- a/crates/integrations/datafusion/src/full_text_search.rs +++ b/crates/integrations/datafusion/src/full_text_search.rs @@ -162,7 +162,11 @@ impl TableProvider for FullTextSearchTableProvider { .with_text_column(&self.column_name) .with_query_text(&self.query_text) .with_limit(self.limit); - builder.execute().await.map_err(to_datafusion_error) + builder + .execute() + .await + .and_then(|result| result.row_ids()?.to_row_ranges()) + .map_err(to_datafusion_error) }) .await?; diff --git a/crates/integrations/datafusion/src/hybrid_search.rs b/crates/integrations/datafusion/src/hybrid_search.rs index 314796031..cc1898a27 100644 --- a/crates/integrations/datafusion/src/hybrid_search.rs +++ b/crates/integrations/datafusion/src/hybrid_search.rs @@ -227,7 +227,7 @@ impl TableProvider for HybridSearchTableProvider { .with_ranker(&self.ranker) .map_err(to_datafusion_error)?; builder - .execute_scored() + .execute() .await .and_then(|result| result.into_row_ids()) .map_err(to_datafusion_error) diff --git a/crates/paimon/src/table/full_text_index_build_builder/tests.rs b/crates/paimon/src/table/full_text_index_build_builder/tests.rs index 3dc165314..3e57a9f18 100644 --- a/crates/paimon/src/table/full_text_index_build_builder/tests.rs +++ b/crates/paimon/src/table/full_text_index_build_builder/tests.rs @@ -133,6 +133,10 @@ async fn search(table: &Table, query: &str) -> Vec { .execute() .await .unwrap() + .row_ids() + .unwrap() + .to_row_ranges() + .unwrap() } fn text_batch(row_ids: Vec, names: Vec>) -> RecordBatch { diff --git a/crates/paimon/src/table/full_text_search_builder.rs b/crates/paimon/src/table/full_text_search_builder.rs index e5e81d6ac..dfebafaba 100644 --- a/crates/paimon/src/table/full_text_search_builder.rs +++ b/crates/paimon/src/table/full_text_search_builder.rs @@ -43,7 +43,9 @@ use paimon_ftindex_core::io::PosWriter; use paimon_ftindex_core::{FullTextIndexConfig, FullTextIndexWriter}; use roaring::RoaringTreemap; use serde_json::json; +use std::borrow::Cow; use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, RwLock}; const FULL_TEXT_INDEX_TYPE: &str = "full-text"; mod execution; @@ -57,6 +59,9 @@ pub use scan::{FullTextScan, FullTextScanPlan}; /// Builder for executing full-text search on a Paimon table. /// +/// Clones share configuration. Setters update every handle, and each execution +/// captures the current configuration before planning its sources. +/// /// Usage: /// ```ignore /// let result = table.new_full_text_search_builder() @@ -68,8 +73,14 @@ pub use scan::{FullTextScan, FullTextScanPlan}; /// ``` /// /// Reference: `org.apache.paimon.table.source.FullTextSearchBuilder` +#[derive(Clone)] pub struct FullTextSearchBuilder<'a> { - table: &'a Table, + table: Cow<'a, Table>, + config: Arc>, +} + +#[derive(Clone, Default)] +struct FullTextSearchConfig { text_column: Option, query_text: Option, query_is_dsl: bool, @@ -81,85 +92,102 @@ pub struct FullTextSearchBuilder<'a> { impl<'a> FullTextSearchBuilder<'a> { pub(crate) fn new(table: &'a Table) -> Self { Self { - table, - text_column: None, - query_text: None, - query_is_dsl: false, - limit: None, - include_row_ids: None, - filter: None, + table: Cow::Borrowed(table), + config: Arc::new(RwLock::new(FullTextSearchConfig::default())), + } + } + + pub(crate) fn from_owned(table: Arc
) -> FullTextSearchBuilder<'static> { + FullTextSearchBuilder { + table: Cow::Owned(table.as_ref().clone()), + config: Arc::new(RwLock::new(FullTextSearchConfig::default())), } } + /// Own the table handle while preserving shared configuration with clones. + pub fn into_owned(self) -> FullTextSearchBuilder<'static> { + FullTextSearchBuilder { + table: Cow::Owned(self.table.into_owned()), + config: self.config, + } + } + + fn config_snapshot(&self) -> FullTextSearchConfig { + self.config + .read() + .expect("full-text search configuration lock poisoned") + .clone() + } + /// Set the text column to search. pub fn with_text_column(&mut self, name: &str) -> &mut Self { - self.text_column = Some(name.to_string()); + self.config.write().unwrap().text_column = Some(name.to_string()); self } /// Set the query text to search for. pub fn with_query_text(&mut self, query: &str) -> &mut Self { - self.query_text = Some(query.to_string()); - self.query_is_dsl = false; + let mut config = self.config.write().unwrap(); + config.query_text = Some(query.to_string()); + config.query_is_dsl = false; + drop(config); self } /// Set candidate row IDs to include in the search. pub fn with_include_row_ids(&mut self, include_row_ids: RoaringTreemap) -> &mut Self { - self.include_row_ids = Some(include_row_ids); + self.config.write().unwrap().include_row_ids = Some(include_row_ids); self } /// Set the top-k limit for results. pub fn with_limit(&mut self, limit: usize) -> &mut Self { - self.limit = Some(limit); + self.config.write().unwrap().limit = Some(limit); self } - /// Execute the full-text search and return row ranges. - /// - /// This reads the latest snapshot, loads the index manifest, and evaluates - /// the search against full-text indexes. - /// - /// Reference: `FullTextSearchBuilder.executeLocal()` - /// For primary-key file positions, use [`Self::execute_scored`]. - pub async fn execute(&self) -> crate::Result> { - self.execute_scored().await?.row_ids()?.to_row_ranges() - } - /// Search one snapshot without materializing user columns. Results carry /// global row IDs or primary-key file positions and can be read later with /// [`crate::table::GlobalIndexResult::new_read_builder`]. - pub async fn execute_scored(&self) -> crate::Result { - let read = self.new_read()?; - read.read(self.new_scan()?.scan().await?).await + pub async fn execute(&self) -> crate::Result { + let (scan, read) = self.new_scan_and_read()?; + read.read(scan.scan().await?).await } /// Configure the queried field and query, matching Java withQuery. pub fn with_query(&mut self, field_name: &str, query: &str) -> &mut Self { - self.with_text_column(field_name); - self.query_text = Some(query.to_string()); - self.query_is_dsl = true; + let mut config = self.config.write().unwrap(); + config.text_column = Some(field_name.to_string()); + config.query_text = Some(query.to_string()); + config.query_is_dsl = true; + drop(config); self } /// Exclude rows before Top-K; partition conjuncts prune the manifest. pub fn with_filter(&mut self, filter: Predicate) -> &mut Self { - super::vector_search_common::add_filter(&mut self.filter, filter); + super::vector_search_common::add_filter(&mut self.config.write().unwrap().filter, filter); self } pub fn with_partition_filter(&mut self, filter: Predicate) -> crate::Result<&mut Self> { - super::partition_filter::validate_partition_filter(self.table, &filter)?; + super::partition_filter::validate_partition_filter(&self.table, &filter)?; Ok(self.with_filter(filter)) } pub fn new_scan(&self) -> crate::Result { - FullTextScan::new(self) + FullTextScan::new(&self.table, &self.config_snapshot()) } pub fn new_read(&self) -> crate::Result { - FullTextRead::new(self) + FullTextRead::new(&self.table, &self.config_snapshot()) + } + + pub(crate) fn new_scan_and_read(&self) -> crate::Result<(FullTextScan, FullTextRead)> { + let config = self.config_snapshot(); + let read = FullTextRead::new(&self.table, &config)?; + let scan = FullTextScan::new(&self.table, &config)?; + Ok((scan, read)) } /// Run the full-text search and materialize the matching rows as Arrow batches, @@ -179,8 +207,10 @@ impl<'a> FullTextSearchBuilder<'a> { // Fail closed: returns data outside `TableScan`/`TableRead`. let core = CoreOptions::new(self.table.schema().options()); core.ensure_read_authorized()?; + let config = self.config_snapshot(); let text_column = - self.text_column + config + .text_column .as_deref() .ok_or_else(|| crate::Error::ConfigInvalid { message: "Text column must be set via with_text_column()".to_string(), @@ -195,7 +225,13 @@ impl<'a> FullTextSearchBuilder<'a> { }); } - self.execute_scored().await?.new_read_builder().read().await + let read = FullTextRead::new(&self.table, &config)?; + let scan = FullTextScan::new(&self.table, &config)?; + read.read(scan.scan().await?) + .await? + .new_read_builder() + .read() + .await } } @@ -1666,8 +1702,6 @@ mod tests { assert!(positions[0].score > 0.0); } - // (d) The PK full-text path produces physical positions, not global row-ids: - // Only the ranges-only `execute` API rejects the physical-position result. #[tokio::test] async fn scheduled_full_text_rejects_primary_key_plans() { struct NoScheduling; @@ -1700,31 +1734,32 @@ mod tests { } #[tokio::test] - async fn pk_full_text_ranges_reject_physical_positions() { + async fn pk_full_text_result_rejects_global_row_id_conversion() { let table = pk_full_text_table("pk_ft_execute_loud", &[]); - let err = table + let result = table .new_full_text_search_builder() .with_text_column("body") .with_query_text(r#"{"match":{"query":"alpha"}}"#) .with_limit(10) .execute() .await - .unwrap_err(); + .unwrap(); + let err = result.row_ids().unwrap_err(); assert!( matches!(&err, crate::Error::Unsupported { message } if message.contains("physical file positions")), - "the ranges-only API must reject primary-key file positions, got: {err}" + "global row ID conversion must reject primary-key file positions, got: {err}" ); } #[tokio::test] - async fn pk_full_text_scored_empty_result_keeps_its_address_space() { + async fn pk_full_text_empty_result_keeps_its_address_space() { let table = pk_full_text_table("pk_ft_scored_loud", &[]); let result = table .new_full_text_search_builder() .with_text_column("body") .with_query_text(r#"{"match":{"query":"alpha"}}"#) .with_limit(10) - .execute_scored() + .execute() .await .unwrap(); assert_eq!(result.snapshot_id(), None); @@ -1779,7 +1814,7 @@ mod tests { } // (e) A non-PK (append) table: execute_read is unsupported and must fail loud - // rather than silently return nothing; the append execute/execute_scored path + // rather than silently return nothing; the append execute path // stays unaffected (exercised by the raw-search tests above). #[tokio::test] async fn append_execute_read_fails_loud_unsupported() { @@ -1804,7 +1839,7 @@ mod tests { // Dispatch: with data-evolution ENABLED, a configured pk-full-text column must // NOT resolve to the PK path (mirrors Java `primaryKeyFullTextDefinition`), so - // execute_scored returns global row IDs rather than physical PK positions. + // execute returns global row IDs rather than physical PK positions. #[tokio::test] async fn data_evolution_enabled_does_not_take_pk_path() { let de_on = HashMap::from([ diff --git a/crates/paimon/src/table/full_text_search_builder/append_tests.rs b/crates/paimon/src/table/full_text_search_builder/append_tests.rs index cd810e20f..f4205204b 100644 --- a/crates/paimon/src/table/full_text_search_builder/append_tests.rs +++ b/crates/paimon/src/table/full_text_search_builder/append_tests.rs @@ -52,7 +52,7 @@ async fn scalar_modes_preserve_missing_and_partial_coverage() { assert_eq!( ids(query(&table) .with_filter(filter.clone()) - .execute_scored() + .execute() .await .unwrap()), expected, @@ -76,7 +76,7 @@ async fn scalar_modes_preserve_missing_and_partial_coverage() { assert_eq!( ids(query(&table) .with_filter(filter.clone()) - .execute_scored() + .execute() .await .unwrap()), expected, @@ -115,7 +115,7 @@ async fn candidate_refinement_keeps_rust_defaults_and_original_scores() { pb.greater_or_equal("id", Datum::Int(5)).unwrap(), ]); let all = query(&table) - .execute_scored() + .execute() .await .unwrap() .into_row_ids() @@ -123,7 +123,7 @@ async fn candidate_refinement_keeps_rust_defaults_and_original_scores() { let filtered = query(&table) .with_filter(filter.clone()) .with_limit(2) - .execute_scored() + .execute() .await .unwrap(); assert_eq!(ids(filtered.clone()), vec![5, 7]); @@ -139,13 +139,13 @@ async fn candidate_refinement_keeps_rust_defaults_and_original_scores() { )])); assert!(query(&no_refine) .with_filter(filter) - .execute_scored() + .execute() .await .unwrap() .is_empty()); assert!(query(&table) .with_filter(pb.equal("label", Datum::String("missing".into())).unwrap()) - .execute_scored() + .execute() .await .unwrap() .is_empty()); @@ -166,17 +166,17 @@ async fn partition_scalar_and_include_filters_precede_top_k() { ]); let mut builder = query(&table); builder.with_filter(filter).with_limit(2); - let result = builder.execute_scored().await.unwrap(); + let result = builder.execute().await.unwrap(); assert_eq!(user_ids(&table, result.clone()).await, vec![5, 7]); // Partition commit order assigns global IDs independently of user IDs. let last = *result.row_ids().unwrap().row_ids.iter().max().unwrap(); builder.with_include_row_ids([last, 9].into_iter().collect()); assert_eq!( - user_ids(&table, builder.execute_scored().await.unwrap()).await, + user_ids(&table, builder.execute().await.unwrap()).await, vec![7], ); builder.with_include_row_ids(RoaringTreemap::new()); - assert!(builder.execute_scored().await.unwrap().is_empty()); + assert!(builder.execute().await.unwrap().is_empty()); } } @@ -251,7 +251,7 @@ async fn deletion_vectors_filter_exact_index_hits_before_top_k_and_pin_snapshot( delete_row(&table, 5).await; assert_eq!(ids(read.read(plan).await.unwrap()), vec![1, 5]); - let result = builder.execute_scored().await.unwrap(); + let result = builder.execute().await.unwrap(); assert_eq!(ids(result.clone()), vec![1, 7]); let batches: Vec = result .new_read_builder() @@ -317,7 +317,7 @@ async fn removed_files_do_not_supply_exact_index_hits() { .await .unwrap(); assert_eq!(ids(read.read(plan).await.unwrap()), vec![1]); - assert_eq!(ids(builder.execute_scored().await.unwrap()), vec![5]); + assert_eq!(ids(builder.execute().await.unwrap()), vec![5]); } #[tokio::test] @@ -332,7 +332,7 @@ async fn empty_filtered_plans_stay_empty_after_index_build() { index(&table, "full-text", "text").await; index(&table, "btree", "label").await; assert!(read.read(plan).await.unwrap().is_empty()); - assert_eq!(ids(builder.execute_scored().await.unwrap()), vec![1]); + assert_eq!(ids(builder.execute().await.unwrap()), vec![1]); } #[tokio::test] @@ -353,7 +353,7 @@ async fn unsupported_append_filter_combinations_fail_before_empty_snapshot() { let table = table.copy_with_options(HashMap::from([(option.into(), value.into())])); let error = query(&table) .with_filter(keep_filter(&table)) - .execute_scored() + .execute() .await .unwrap_err(); assert!( @@ -377,7 +377,7 @@ async fn unsupported_append_filter_combinations_fail_before_empty_snapshot() { ); assert!(query(&pk) .with_filter(keep_filter(&pk)) - .execute_scored() + .execute() .await .unwrap_err() .to_string() diff --git a/crates/paimon/src/table/full_text_search_builder/parity_tests.rs b/crates/paimon/src/table/full_text_search_builder/parity_tests.rs index 90b365d6f..f2f37752a 100644 --- a/crates/paimon/src/table/full_text_search_builder/parity_tests.rs +++ b/crates/paimon/src/table/full_text_search_builder/parity_tests.rs @@ -144,12 +144,49 @@ fn ids(result: crate::table::GlobalIndexResult) -> Vec { ids } +#[tokio::test] +async fn cloned_full_text_builders_share_configuration_and_capture_executions() { + let table = new_table(false); + append(&table, 0, 8).await; + index(&table, "full-text", "text").await; + let builder = query(&table); + let mut shared = builder.clone().into_owned(); + assert_eq!( + ids(builder.execute().await.unwrap()), + vec![0, 1, 2, 4, 5, 6, 7] + ); + + shared.with_query_text("absent"); + assert!(builder.execute().await.unwrap().is_empty()); + shared + .with_query("text", r#"{"match":{"query":"paimon"}}"#) + .with_limit(1); + assert_eq!(builder.execute().await.unwrap().len(), 1); + + let filter = PredicateBuilder::new(table.schema().fields()) + .equal("id", Datum::Int(5)) + .unwrap(); + shared.with_filter(filter); + assert_eq!(ids(builder.execute().await.unwrap()), vec![5]); + shared.with_include_row_ids([0].into_iter().collect()); + assert!(builder.execute().await.unwrap().is_empty()); + shared.with_include_row_ids([5].into_iter().collect()); + + let (scan, read) = builder.new_scan_and_read().unwrap(); + shared.with_query_text("absent"); + assert_eq!( + ids(read.read(scan.scan().await.unwrap()).await.unwrap()), + vec![5] + ); + assert!(builder.execute().await.unwrap().is_empty()); +} + #[tokio::test] async fn full_text_result_reads_projection_and_scores_from_the_captured_snapshot() { let table = new_table(false); append(&table, 0, 8).await; index(&table, "full-text", "text").await; - let result = query(&table).execute_scored().await.unwrap(); + let result = query(&table).execute().await.unwrap(); let snapshot = result.snapshot_id().unwrap(); let expected = result.row_ids().unwrap(); assert!(result.positions().is_err()); @@ -252,7 +289,7 @@ async fn scalar_modes_and_refinement_follow_java_before_top_k() { assert_eq!( ids(query(&table) .with_filter(keep.clone()) - .execute_scored() + .execute() .await .unwrap()), expected @@ -264,18 +301,14 @@ async fn scalar_modes_and_refinement_follow_java_before_top_k() { "fast".into(), )])); assert_eq!( - ids(query(&fast) - .with_filter(keep) - .execute_scored() - .await - .unwrap()), + ids(query(&fast).with_filter(keep).execute().await.unwrap()), vec![1, 5, 7] ); let candidate = pb.contains("label", Datum::String("eep".into())).unwrap(); assert_eq!( ids(query(&fast) .with_filter(candidate.clone()) - .execute_scored() + .execute() .await .unwrap()), vec![1, 5, 7] @@ -286,14 +319,14 @@ async fn scalar_modes_and_refinement_follow_java_before_top_k() { )])); assert!(query(&no_refine) .with_filter(candidate) - .execute_scored() + .execute() .await .unwrap() .is_empty()); let high = pb.greater_or_equal("id", Datum::Int(5)).unwrap(); let mut search = query(&table); search.with_limit(1).with_filter(high); - assert_eq!(ids(search.execute_scored().await.unwrap()), vec![5]); + assert_eq!(ids(search.execute().await.unwrap()), vec![5]); } #[tokio::test] @@ -373,7 +406,7 @@ async fn empty_string_filters_preserve_java_search_refinement() { assert_eq!( ids(query(&table) .with_filter(filter.clone()) - .execute_scored() + .execute() .await .unwrap()), expected, @@ -443,7 +476,7 @@ async fn partition_filters_apply_to_indexed_and_raw_corpora() { query(&table) .with_partition_filter(filter.clone()) .unwrap() - .execute_scored() + .execute() .await .unwrap() ) @@ -457,11 +490,7 @@ async fn partition_filters_apply_to_indexed_and_raw_corpora() { assert_eq!( user_ids( &table, - query(&table) - .with_filter(filter) - .execute_scored() - .await - .unwrap() + query(&table).with_filter(filter).execute().await.unwrap() ) .await, vec![5, 7, 9, 11] @@ -485,7 +514,7 @@ async fn scan_read_pins_snapshot_and_keeps_query_configuration_on_reader() { vec![0, 1, 2, 4, 5, 6, 7] ); assert_eq!( - ids(query(&table).execute_scored().await.unwrap()), + ids(query(&table).execute().await.unwrap()), vec![0, 1, 2, 4, 5, 6, 7, 8, 9, 10, 11] ); let pb = PredicateBuilder::new(table.schema().fields()); @@ -522,7 +551,7 @@ async fn empty_plans_stay_empty_after_append_and_validate_query_first() { let empty = new_table(false); assert!(query(&empty) .with_limit(0) - .execute_scored() + .execute() .await .unwrap_err() .to_string() @@ -543,11 +572,11 @@ async fn raw_filter_keeps_bm25_statistics_of_the_whole_corpus() { append(&table, 0, 4).await; index(&table, "full-text", "text").await; append(&table, 4, 8).await; - let unfiltered = query(&table).execute_scored().await.unwrap(); + let unfiltered = query(&table).execute().await.unwrap(); let pb = PredicateBuilder::new(table.schema().fields()); let filtered = query(&table) .with_filter(pb.equal("label", Datum::String("keep".into())).unwrap()) - .execute_scored() + .execute() .await .unwrap(); assert_eq!(ids(filtered.clone()), vec![1, 5, 7, 9, 11]); @@ -606,7 +635,7 @@ async fn snapshot_pinning_preserves_schema_only_renames() { .with_query("body", r#"{"match":{"query":"paimon"}}"#) .with_limit(64); assert_eq!( - ids(builder.execute_scored().await.unwrap()), + ids(builder.execute().await.unwrap()), vec![0, 1, 2, 4, 5, 6, 7] ); let plan = builder.new_scan().unwrap().scan().await.unwrap(); @@ -648,7 +677,7 @@ async fn scalar_search_pre_filters_exclude_composite_indexes() { assert_eq!( ids(query(&table) .with_filter(predicate.clone()) - .execute_scored() + .execute() .await .unwrap()), expected diff --git a/crates/paimon/src/table/full_text_search_builder/read.rs b/crates/paimon/src/table/full_text_search_builder/read.rs index 2cd518134..81502982f 100644 --- a/crates/paimon/src/table/full_text_search_builder/read.rs +++ b/crates/paimon/src/table/full_text_search_builder/read.rs @@ -29,23 +29,24 @@ pub struct FullTextRead { } impl FullTextRead { - pub(super) fn new(builder: &FullTextSearchBuilder<'_>) -> crate::Result { - CoreOptions::new(builder.table.schema().options()).ensure_read_authorized()?; - let limit = builder.limit.filter(|limit| *limit > 0).ok_or_else(|| { - crate::Error::ConfigInvalid { - message: "Limit must be positive, set via with_limit()".into(), - } - })?; - let context = PlanContext::new(builder)?; - let query = builder + pub(super) fn new(table: &Table, config: &FullTextSearchConfig) -> crate::Result { + CoreOptions::new(table.schema().options()).ensure_read_authorized()?; + let limit = + config + .limit + .filter(|limit| *limit > 0) + .ok_or_else(|| crate::Error::ConfigInvalid { + message: "Limit must be positive, set via with_limit()".into(), + })?; + let context = PlanContext::new(table, config)?; + let query = config .query_text .as_deref() .ok_or_else(|| crate::Error::ConfigInvalid { message: "Query text must be set via with_query() or with_query_text()".into(), })?; if context.primary_key_field_id.is_some() { - if builder - .table + if table .schema() .core_options() .full_text_index_search_mode()? @@ -56,8 +57,8 @@ impl FullTextRead { if let Some(filter) = &context.filter { let (_, data) = super::super::bucket_filter::split_partition_and_data_predicates( filter.clone(), - builder.table.schema().fields(), - builder.table.schema().partition_keys(), + table.schema().fields(), + table.schema().partition_keys(), ); if !data.is_empty() { return Err(crate::Error::Unsupported { @@ -68,14 +69,14 @@ impl FullTextRead { } } } - let query = if builder.query_is_dsl || context.primary_key_field_id.is_some() { + let query = if config.query_is_dsl || context.primary_key_field_id.is_some() { query.to_string() } else { normalize_query_text(query, &context.column)? }; let search = FullTextSearch::new(query, limit, context.column.clone())?; Ok(Self { - table: builder.table.clone(), + table: table.clone(), context, search, }) diff --git a/crates/paimon/src/table/full_text_search_builder/scan.rs b/crates/paimon/src/table/full_text_search_builder/scan.rs index 1971bc330..a03f86a7b 100644 --- a/crates/paimon/src/table/full_text_search_builder/scan.rs +++ b/crates/paimon/src/table/full_text_search_builder/scan.rs @@ -31,17 +31,16 @@ pub(super) struct PlanContext { } impl PlanContext { - pub(super) fn new(builder: &FullTextSearchBuilder<'_>) -> crate::Result { - let core = builder.table.schema().core_options(); + pub(super) fn new(table: &Table, config: &FullTextSearchConfig) -> crate::Result { + let core = table.schema().core_options(); core.ensure_read_authorized()?; - let column = builder + let column = config .text_column .as_deref() .ok_or_else(|| crate::Error::ConfigInvalid { message: "Text column must be set via with_query() or with_text_column()".into(), })?; - let field = builder - .table + let field = table .schema() .fields() .iter() @@ -57,16 +56,16 @@ impl PlanContext { message: format!("Full-text column '{column}' must be a character string"), }); } - let has_row_filter = builder.filter.as_ref().is_some_and(|filter| { + let has_row_filter = config.filter.as_ref().is_some_and(|filter| { let (_, data) = split_partition_and_data_predicates( filter.clone(), - builder.table.schema().fields(), - builder.table.schema().partition_keys(), + table.schema().fields(), + table.schema().partition_keys(), ); !data.is_empty() }); if has_row_filter && !core.data_evolution_enabled() { - if !builder.table.schema().primary_keys().is_empty() + if !table.schema().primary_keys().is_empty() || !core.row_tracking_enabled() || core.file_format() != "parquet" { @@ -82,7 +81,7 @@ impl PlanContext { } let primary_key_field_id = resolves_to_pk_full_text_path(&core, column).then_some(field.id()); - if primary_key_field_id.is_some() && builder.include_row_ids.is_some() { + if primary_key_field_id.is_some() && config.include_row_ids.is_some() { return Err(crate::Error::Unsupported { message: "primary-key full-text search uses physical file positions, not global row IDs" @@ -91,8 +90,8 @@ impl PlanContext { } Ok(Self { column: column.into(), - filter: builder.filter.clone(), - include_row_ids: builder.include_row_ids.clone(), + filter: config.filter.clone(), + include_row_ids: config.include_row_ids.clone(), primary_key_field_id, }) } @@ -188,10 +187,10 @@ pub struct FullTextScan { } impl FullTextScan { - pub(super) fn new(builder: &FullTextSearchBuilder<'_>) -> crate::Result { + pub(super) fn new(table: &Table, config: &FullTextSearchConfig) -> crate::Result { Ok(Self { - table: builder.table.clone(), - context: PlanContext::new(builder)?, + table: table.clone(), + context: PlanContext::new(table, config)?, }) } diff --git a/crates/paimon/src/table/global_index_build_builder/parallel_tests.rs b/crates/paimon/src/table/global_index_build_builder/parallel_tests.rs index e87376e56..3045ffcf7 100644 --- a/crates/paimon/src/table/global_index_build_builder/parallel_tests.rs +++ b/crates/paimon/src/table/global_index_build_builder/parallel_tests.rs @@ -253,10 +253,7 @@ async fn parallel_native_writers_match_serial_shards_and_incremental_coverage() .execute() .await .unwrap(); - let mut ids = result - .into_iter() - .map(|range| range.from()) - .collect::>(); + let mut ids = result.into_row_ids().unwrap().row_ids; ids.sort_unstable(); assert_eq!(ids, (1..24).step_by(2).collect::>()); } diff --git a/crates/paimon/src/table/hybrid_search_builder.rs b/crates/paimon/src/table/hybrid_search_builder.rs index 8cb1f7e30..c21a227f9 100644 --- a/crates/paimon/src/table/hybrid_search_builder.rs +++ b/crates/paimon/src/table/hybrid_search_builder.rs @@ -20,6 +20,7 @@ //! Reference: `org.apache.paimon.table.source.HybridSearchBuilder`. use std::collections::{BTreeMap, HashMap}; +use std::fmt; use std::sync::Arc; use arrow_array::RecordBatch; @@ -235,13 +236,29 @@ impl HybridSearchRoute { /// An independently executable route retaining the hybrid search's captured view. /// -/// The configured search builders borrow this owned handle, so routes can outlive -/// the hybrid builder without resolving latest again when they execute. -#[derive(Clone, Debug)] +/// Typed builder handles share each route's live configuration and captured +/// table, so edits through a getter affect later executions of the route. +#[derive(Clone)] pub struct HybridSearchRouteBuilder { table: Arc
, route: HybridSearchRoute, - filter: Option, + builder: RouteSearchBuilder, +} + +#[derive(Clone)] +enum RouteSearchBuilder { + Vector(super::VectorSearchBuilder<'static>), + #[cfg(feature = "fulltext")] + FullText(super::FullTextSearchBuilder<'static>), +} + +impl fmt::Debug for HybridSearchRouteBuilder { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("HybridSearchRouteBuilder") + .field("table", &self.table) + .field("route", &self.route) + .finish() + } } impl HybridSearchRouteBuilder { @@ -253,27 +270,22 @@ impl HybridSearchRouteBuilder { &self.table } - pub fn filter(&self) -> Option<&Predicate> { - self.filter.as_ref() - } - - pub fn vector_search_builder(&self) -> crate::Result> { + pub fn vector_search_builder(&self) -> crate::Result> { self.require_kind(HybridSearchRouteKind::Vector)?; - Ok(vector_route_builder( - &self.table, - &self.route, - self.filter.as_ref(), - )) + match &self.builder { + RouteSearchBuilder::Vector(builder) => Ok(builder.clone()), + #[cfg(feature = "fulltext")] + _ => unreachable!("validated vector route"), + } } #[cfg(feature = "fulltext")] - pub fn full_text_search_builder(&self) -> crate::Result> { + pub fn full_text_search_builder(&self) -> crate::Result> { self.require_kind(HybridSearchRouteKind::FullText)?; - Ok(full_text_route_builder( - &self.table, - &self.route, - self.filter.as_ref(), - )) + match &self.builder { + RouteSearchBuilder::FullText(builder) => Ok(builder.clone()), + _ => unreachable!("validated full-text route"), + } } /// Execute this route against the same view as every sibling route. @@ -297,21 +309,26 @@ impl HybridSearchRouteBuilder { &self, executor: Option<&dyn super::SearchTaskExecutor>, ) -> crate::Result { - match self.route.kind { - HybridSearchRouteKind::Vector => { - let builder = self.vector_search_builder()?; + match &self.builder { + RouteSearchBuilder::Vector(builder) => { + if let Some(executor) = executor { + let (scan, read) = builder.new_scan_and_read()?; + let plan = scan.plan().await?; + read.read_with_executor(plan, executor).await + } else { + builder.execute().await + } + } + #[cfg(feature = "fulltext")] + RouteSearchBuilder::FullText(builder) => { if let Some(executor) = executor { - let plan = builder.new_scan()?.plan().await?; - builder.new_read()?.read_with_executor(plan, executor).await + let (scan, read) = builder.new_scan_and_read()?; + let plan = scan.scan().await?; + read.read_with_executor(plan, executor).await } else { builder.execute().await } } - HybridSearchRouteKind::FullText => Ok(GlobalIndexResult::from_global_search( - self.table.clone(), - execute_full_text_route(&self.table, &self.route, self.filter.as_ref(), executor) - .await?, - )), } } } @@ -511,12 +528,9 @@ impl<'a> HybridSearchBuilder<'a> { self.ranker.rank(&weighted, limit) } - pub async fn execute(&self) -> crate::Result> { - self.execute_scored().await?.row_ids()?.to_row_ranges() - } - - pub async fn execute_scored(&self) -> crate::Result { - self.execute_scored_internal(None).await + /// Execute and fuse global routes, retaining scores and the captured read view. + pub async fn execute(&self) -> crate::Result { + self.execute_internal(None).await } /// Schedule each route's shard work while preserving one shared snapshot @@ -526,10 +540,10 @@ impl<'a> HybridSearchBuilder<'a> { &self, executor: &dyn super::SearchTaskExecutor, ) -> crate::Result { - self.execute_scored_internal(Some(executor)).await + self.execute_internal(Some(executor)).await } - async fn execute_scored_internal( + async fn execute_internal( &self, executor: Option<&dyn super::SearchTaskExecutor>, ) -> crate::Result { @@ -616,10 +630,33 @@ impl<'a> HybridSearchBuilder<'a> { fn owned_route_builders(&self, table: Arc
) -> Vec { self.routes .iter() - .map(|route| HybridSearchRouteBuilder { - table: table.clone(), - route: route.clone(), - filter: self.filter.clone(), + .map(|route| { + let builder = match route.kind { + HybridSearchRouteKind::Vector => { + RouteSearchBuilder::Vector(configure_vector_route_builder( + super::VectorSearchBuilder::from_owned(table.clone()), + route, + self.filter.as_ref(), + )) + } + #[cfg(feature = "fulltext")] + HybridSearchRouteKind::FullText => { + RouteSearchBuilder::FullText(configure_full_text_route_builder( + super::FullTextSearchBuilder::from_owned(table.clone()), + route, + self.filter.as_ref(), + )) + } + #[cfg(not(feature = "fulltext"))] + HybridSearchRouteKind::FullText => { + unreachable!("full-text feature validated before preparing routes") + } + }; + HybridSearchRouteBuilder { + table: table.clone(), + route: route.clone(), + builder, + } }) .collect() } @@ -683,7 +720,7 @@ impl<'a> HybridSearchBuilder<'a> { /// full-text column in `pk-full-text.index.columns` with data-evolution /// disabled). A mixed PK/global set of routes fails loud, and an all-global /// (append/data-evolution) hybrid is unsupported here — those use - /// `execute`/`execute_scored`. Mirrors Java `HybridSearchBuilderImpl` PK path. + /// `execute`. Mirrors Java `HybridSearchBuilderImpl` PK path. pub async fn execute_read(&self) -> crate::Result { let core = CoreOptions::new(self.table.schema().options()); core.ensure_read_authorized()?; @@ -695,7 +732,7 @@ impl<'a> HybridSearchBuilder<'a> { } HybridAddressSpace::Global => Err(crate::Error::Unsupported { message: "materialized hybrid read (execute_read) is only supported on the \ - primary-key hybrid path; use execute/execute_scored for the \ + primary-key hybrid path; use execute for the \ append/data-evolution path" .to_string(), }), @@ -1250,7 +1287,14 @@ fn vector_route_builder<'t>( route: &HybridSearchRoute, filter: Option<&Predicate>, ) -> super::VectorSearchBuilder<'t> { - let mut builder = table.new_vector_search_builder(); + configure_vector_route_builder(table.new_vector_search_builder(), route, filter) +} + +fn configure_vector_route_builder<'t>( + mut builder: super::VectorSearchBuilder<'t>, + route: &HybridSearchRoute, + filter: Option<&Predicate>, +) -> super::VectorSearchBuilder<'t> { builder .with_vector_column(&route.field_name) .with_query_vector(route.vector.clone().expect("validated vector route")) @@ -1268,7 +1312,15 @@ fn full_text_route_builder<'t>( route: &HybridSearchRoute, filter: Option<&Predicate>, ) -> super::FullTextSearchBuilder<'t> { - let mut builder = table.new_full_text_search_builder(); + configure_full_text_route_builder(table.new_full_text_search_builder(), route, filter) +} + +#[cfg(feature = "fulltext")] +fn configure_full_text_route_builder<'t>( + mut builder: super::FullTextSearchBuilder<'t>, + route: &HybridSearchRoute, + filter: Option<&Predicate>, +) -> super::FullTextSearchBuilder<'t> { builder .with_query( &route.field_name, @@ -1284,36 +1336,6 @@ fn full_text_route_builder<'t>( builder } -#[cfg(feature = "fulltext")] -async fn execute_full_text_route( - table: &Table, - route: &HybridSearchRoute, - filter: Option<&Predicate>, - executor: Option<&dyn super::SearchTaskExecutor>, -) -> crate::Result { - let builder = full_text_route_builder(table, route, filter); - let result = if let Some(executor) = executor { - let read = builder.new_read()?; - let plan = builder.new_scan()?.scan().await?; - read.read_with_executor(plan, executor).await? - } else { - builder.execute_scored().await? - }; - result.into_row_ids() -} - -#[cfg(not(feature = "fulltext"))] -async fn execute_full_text_route( - _table: &Table, - _route: &HybridSearchRoute, - _filter: Option<&Predicate>, - _executor: Option<&dyn super::SearchTaskExecutor>, -) -> crate::Result { - Err(crate::Error::ConfigInvalid { - message: "Full-text hybrid routes require the fulltext feature".to_string(), - }) -} - struct WeightedRouteResult { result: ScoredRowIds, weight: f32, @@ -2070,9 +2092,9 @@ mod pk_hybrid_tests { } // (d) The PK hybrid path produces physical positions, not global row ids: - // execute / execute_scored must fail loud and point at execute_read. + // execute must fail loud and point at execute_read. #[tokio::test] - async fn pk_hybrid_execute_and_scored_fail_loud() { + async fn pk_hybrid_execute_fails_loud() { let table = build_hybrid_table( "memory:/pk_hybrid_guard", &[100, 101], @@ -2102,11 +2124,6 @@ mod pk_hybrid_tests { .unwrap(); builder.with_limit(2); - let scored_err = builder.execute_scored().await.unwrap_err(); - assert!( - format!("{scored_err:?}").contains("execute_read"), - "PK hybrid execute_scored must point at execute_read, got: {scored_err:?}" - ); let execute_err = builder.execute().await.unwrap_err(); assert!( format!("{execute_err:?}").contains("execute_read"), @@ -2186,10 +2203,10 @@ mod pk_hybrid_tests { } // (e) DE regression: a hybrid over non-PK routes (no PK configs) still takes the - // global-row-id execute_scored path unchanged. An empty table yields an empty + // global-row-id execute path unchanged. An empty table yields an empty // result without hitting the PK guard. #[tokio::test] - async fn global_hybrid_execute_scored_unchanged() { + async fn global_hybrid_execute_retains_global_result() { // A table without any PK index configuration: both routes resolve to the // append/data-evolution global path. let file_io = FileIOBuilder::new("memory").build().unwrap(); @@ -2238,10 +2255,7 @@ mod pk_hybrid_tests { // No snapshot -> both routes empty -> fused empty; the DE path must NOT hit // the PK fail-loud guard. - let result = builder - .execute_scored() - .await - .expect("DE hybrid path failed"); + let result = builder.execute().await.expect("DE hybrid path failed"); assert!( result.is_empty(), "empty table yields empty DE hybrid result" diff --git a/crates/paimon/src/table/hybrid_search_builder/parity_tests.rs b/crates/paimon/src/table/hybrid_search_builder/parity_tests.rs index eea5408f7..4aec727a8 100644 --- a/crates/paimon/src/table/hybrid_search_builder/parity_tests.rs +++ b/crates/paimon/src/table/hybrid_search_builder/parity_tests.rs @@ -135,7 +135,7 @@ async fn full_text_route_uses_java_structured_query() { builder.with_limit(4); assert_eq!( builder - .execute_scored() + .execute() .await .unwrap() .into_row_ids() @@ -155,7 +155,7 @@ async fn full_text_route_rejects_plain_text_as_java_does() { .add_full_text_route("text", "paimon", 4, 1.0, HashMap::new()) .unwrap(); builder.with_limit(4); - assert!(builder.execute_scored().await.is_err()); + assert!(builder.execute().await.is_err()); } pub(super) fn mixed_query(table: &Table, route_limit: usize) -> HybridSearchBuilder<'_> { @@ -206,21 +206,22 @@ async fn both_routes_filter_before_top_k_and_accumulate_predicates() { vec![6], "the vector route must filter before its top-k" ); - let text = execute_full_text_route(&pinned, &builder.routes[1], builder.filter.as_ref(), None) + let text = full_text_route_builder(&pinned, &builder.routes[1], builder.filter.as_ref()) + .execute() .await .unwrap(); assert_eq!( - text.row_ids, + text.row_ids().unwrap().row_ids, vec![6], "the text route must filter before its top-k" ); - let result = builder.execute_scored().await.unwrap(); + let result = builder.execute().await.unwrap(); assert_eq!(result.row_ids().unwrap().row_ids, vec![6]); assert!((result.row_ids().unwrap().scores[0] - 2.0 / 61.0).abs() < 1e-6); builder .with_partition_filter(pb.equal("pt", Datum::Int(0)).unwrap()) .unwrap(); - assert!(builder.execute_scored().await.unwrap().is_empty()); + assert!(builder.execute().await.unwrap().is_empty()); } #[tokio::test] @@ -238,7 +239,7 @@ async fn scalar_coverage_and_refinement_are_shared_by_both_routes() { )])); let mut ids = mixed_query(&table, 4) .with_filter(keep.clone()) - .execute_scored() + .execute() .await .unwrap() .into_row_ids() @@ -253,7 +254,7 @@ async fn scalar_coverage_and_refinement_are_shared_by_both_routes() { .with_filter(keep.clone()) .with_limit(4); assert_eq!( - text.execute_scored().await.unwrap().len(), + text.execute().await.unwrap().len(), if mode == "fast" { 0 } else { 2 } ); } @@ -269,7 +270,7 @@ async fn scalar_coverage_and_refinement_are_shared_by_both_routes() { ])); let mut ids = mixed_query(&table, 4) .with_filter(candidate.clone()) - .execute_scored() + .execute() .await .unwrap() .into_row_ids() @@ -303,7 +304,7 @@ async fn concurrent_routes_preserve_weights_limits_and_java_fusion() { .with_limit(4) .with_ranker(ranker) .unwrap(); - let result = builder.execute_scored().await.unwrap(); + let result = builder.execute().await.unwrap(); assert_eq!(result.row_ids().unwrap().row_ids, ids, "ranker {ranker}"); for (actual, expected) in result.row_ids().unwrap().scores.iter().zip(scores) { assert!( @@ -314,7 +315,7 @@ async fn concurrent_routes_preserve_weights_limits_and_java_fusion() { builder.with_limit(1); assert_eq!( builder - .execute_scored() + .execute() .await .unwrap() .into_row_ids() @@ -353,7 +354,7 @@ async fn retained_tag_keeps_both_routes_on_the_expired_snapshot() { "hybrid-before".into(), )])); let mut old = mixed_query(&tagged, 16) - .execute_scored() + .execute() .await .unwrap() .into_row_ids() @@ -361,14 +362,7 @@ async fn retained_tag_keeps_both_routes_on_the_expired_snapshot() { .row_ids; old.sort_unstable(); assert_eq!(old, vec![0, 1, 2, 3]); - assert_eq!( - mixed_query(&table, 16) - .execute_scored() - .await - .unwrap() - .len(), - 8 - ); + assert_eq!(mixed_query(&table, 16).execute().await.unwrap().len(), 8); } #[tokio::test] @@ -386,7 +380,7 @@ async fn caller_pinned_empty_and_snapshot_views_do_not_read_latest() { append(&table, 4).await; assert!(mixed_query(&table, 16) .with_snapshot(None) - .execute_scored() + .execute() .await .unwrap() .is_empty()); @@ -397,7 +391,7 @@ async fn caller_pinned_empty_and_snapshot_views_do_not_read_latest() { .unwrap(); let mut old = mixed_query(&table, 16) .with_snapshot(Some(&snapshot)) - .execute_scored() + .execute() .await .unwrap() .into_row_ids() @@ -410,11 +404,7 @@ async fn caller_pinned_empty_and_snapshot_views_do_not_read_latest() { #[tokio::test] async fn empty_tables_still_validate_route_and_partition_configuration() { let table = table(); - assert!(mixed_query(&table, 4) - .execute_scored() - .await - .unwrap() - .is_empty()); + assert!(mixed_query(&table, 4).execute().await.unwrap().is_empty()); let pb = PredicateBuilder::new(table.schema().fields()); assert!(mixed_query(&table, 4) .with_partition_filter(pb.equal("id", Datum::Int(1)).unwrap()) @@ -425,16 +415,16 @@ async fn empty_tables_still_validate_route_and_partition_configuration() { .unwrap() .with_limit(4); // Java does not parse the DSL until a route actually evaluates a corpus. - assert!(invalid.execute_scored().await.unwrap().is_empty()); + assert!(invalid.execute().await.unwrap().is_empty()); append(&table, 0).await; index(&table, "full-text", "text").await; - assert!(invalid.execute_scored().await.is_err()); + assert!(invalid.execute().await.is_err()); let mut missing = table.new_hybrid_search_builder(); missing .add_vector_route("missing", vec![0.0, 0.0], 4, 1.0, HashMap::new()) .unwrap() .with_limit(4); - assert!(missing.execute_scored().await.is_err()); + assert!(missing.execute().await.is_err()); } struct WireExecutor { @@ -495,7 +485,7 @@ async fn scheduled_hybrid_shares_snapshot_and_preserves_all_core_rankers() { .with_filter(keep.clone()) .with_ranker(ranker.as_str()) .unwrap(); - let expected = builder.execute_scored().await.unwrap(); + let expected = builder.execute().await.unwrap(); let executor = WireExecutor { table: table.clone(), kinds: Default::default(), diff --git a/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs b/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs index 113f31f0a..93945ae39 100644 --- a/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs +++ b/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs @@ -152,12 +152,11 @@ mod captured_routes { let (routes, expected) = { let mut builder = mixed_query(&table, 4); builder.with_filter(keep); - let expected = builder.execute_scored().await.unwrap(); + let expected = builder.execute().await.unwrap(); (builder.route_builders().await.unwrap(), expected) }; assert_eq!(routes.len(), 2); assert!(Arc::ptr_eq(routes[0].table(), routes[1].table())); - assert!(routes[0].filter().is_some()); assert_eq!( routes[0].table().travel_snapshot().map(|s| s.id()), expected.snapshot_id() @@ -197,7 +196,7 @@ mod captured_routes { expected.snapshot_id() ); assert_ne!( - builder.execute_scored().await.unwrap().snapshot_id(), + builder.execute().await.unwrap().snapshot_id(), expected.snapshot_id() ); } @@ -225,6 +224,169 @@ mod captured_routes { assert!(ranked.snapshot_id().is_none()); } + #[tokio::test] + async fn vector_route_getters_share_query_limit_and_filter_updates() { + let table = table(); + append(&table, 0).await; + index(&table, "ivf-flat", "embedding").await; + let routes = mixed_query(&table, 4).route_builders().await.unwrap(); + let route = routes[0].clone(); + let snapshot_id = route.table().travel_snapshot().unwrap().id(); + let mut vector = route.vector_search_builder().unwrap(); + vector.with_query_vector(vec![3.0, 0.0]).with_limit(2); + assert_eq!( + route.execute().await.unwrap().row_ids().unwrap().row_ids, + vec![3, 2] + ); + + let predicates = PredicateBuilder::new(table.schema().fields()); + route.vector_search_builder().unwrap().with_filter( + predicates + .equal("label", Datum::String("keep".into())) + .unwrap(), + ); + assert_eq!( + vector.execute().await.unwrap().row_ids().unwrap().row_ids, + vec![2, 1] + ); + let mut repeated = route.vector_search_builder().unwrap(); + repeated.with_query_vector(vec![0.0, 0.0]).with_limit(1); + // Handles and cloned routes keep the shared query state and owned view. + drop(routes); + append(&table, 4).await; + for result in [ + route.execute().await.unwrap(), + vector.execute().await.unwrap(), + ] { + assert_eq!(result.row_ids().unwrap().row_ids, vec![1]); + assert_eq!(result.snapshot_id(), Some(snapshot_id)); + } + } + + #[tokio::test] + async fn full_text_route_getters_share_query_limit_and_filter_updates() { + let table = table(); + append(&table, 0).await; + index(&table, "full-text", "text").await; + let routes = mixed_query(&table, 4).route_builders().await.unwrap(); + let route = routes[1].clone(); + let snapshot_id = route.table().travel_snapshot().unwrap().id(); + let mut text = route.full_text_search_builder().unwrap(); + text.with_limit(1); + assert_eq!(route.execute().await.unwrap().len(), 1); + + route + .full_text_search_builder() + .unwrap() + .with_query("text", r#"{"match":{"query":"lake"}}"#); + assert_eq!( + text.execute().await.unwrap().row_ids().unwrap().row_ids, + vec![2] + ); + let predicates = PredicateBuilder::new(table.schema().fields()); + let mut repeated = route.full_text_search_builder().unwrap(); + repeated + .with_query("text", r#"{"match":{"query":"paimon"}}"#) + .with_filter(predicates.equal("id", Datum::Int(1)).unwrap()); + drop(routes); + append(&table, 4).await; + for result in [ + route.execute().await.unwrap(), + text.execute().await.unwrap(), + ] { + assert_eq!(result.row_ids().unwrap().row_ids, vec![1]); + assert_eq!(result.snapshot_id(), Some(snapshot_id)); + } + } + + struct UpdatingExecutor { + table: Table, + update: std::sync::Mutex>>, + } + + #[async_trait::async_trait] + impl crate::table::SearchTaskExecutor for UpdatingExecutor { + async fn execute( + &self, + tasks: Vec, + ) -> crate::Result> { + let update = self.update.lock().unwrap().take(); + if let Some(update) = update { + update(); + } + let (context, frames) = crate::table::SearchTask::serialize_batch(&tasks)?; + let mut results = Vec::new(); + for frame in frames { + results.push(crate::table::SearchTaskResult::deserialize( + &self.table.execute_search_task(&frame, &context).await?, + )?); + } + Ok(results) + } + } + + #[tokio::test] + async fn active_route_execution_keeps_its_configuration_while_getters_are_mutated() { + fn require_send(value: T) -> T { + value + } + + let table = table(); + append(&table, 0).await; + index(&table, "ivf-flat", "embedding").await; + index(&table, "full-text", "text").await; + let routes = mixed_query(&table, 4).route_builders().await.unwrap(); + let predicates = PredicateBuilder::new(table.schema().fields()); + for route in routes { + let (update, expected, later): (Box, _, _) = + match route.route().kind() { + HybridSearchRouteKind::Vector => { + let mut vector = route.vector_search_builder().unwrap(); + let filter = predicates.equal("id", Datum::Int(3)).unwrap(); + ( + Box::new(move || { + vector + .with_query_vector(vec![3.0, 0.0]) + .with_limit(1) + .with_filter(filter); + }), + vec![0, 1, 2, 3], + vec![3], + ) + } + HybridSearchRouteKind::FullText => { + let mut text = route.full_text_search_builder().unwrap(); + let filter = predicates.equal("id", Datum::Int(2)).unwrap(); + ( + Box::new(move || { + text.with_query("text", r#"{"match":{"query":"lake"}}"#) + .with_limit(1) + .with_filter(filter); + }), + vec![0, 1, 2], + vec![2], + ) + } + }; + let executor = UpdatingExecutor { + table: table.clone(), + update: std::sync::Mutex::new(Some(update)), + }; + // Mutating through a worker callback must succeed while execution is + // awaiting I/O. The in-flight request keeps its original settings. + let result = require_send(route.execute_internal(Some(&executor))) + .await + .unwrap(); + let mut ids = result.into_row_ids().unwrap().row_ids; + ids.sort_unstable(); + assert_eq!(ids, expected); + assert_eq!( + route.execute().await.unwrap().row_ids().unwrap().row_ids, + later + ); + } + } + #[tokio::test] async fn incomplete_queries_can_be_finished_on_captured_route_builders() { let table = table(); @@ -243,13 +405,13 @@ mod captured_routes { assert!(routes[0].route().vector_value().unwrap().is_empty()); assert_eq!(routes[1].route().full_text_query(), Some("")); // Preparing routes does not execute their incomplete queries. - assert!(builder.execute_scored().await.is_err()); + assert!(builder.execute().await.is_err()); let mut empty_text = table.new_hybrid_search_builder(); empty_text .add_full_text_route("text", "", 4, 1.0, HashMap::new()) .unwrap() .with_limit(4); - assert!(empty_text.execute_scored().await.is_err()); + assert!(empty_text.execute().await.is_err()); append(&table, 4).await; let mut vector = routes[0].vector_search_builder().unwrap(); @@ -258,16 +420,16 @@ mod captured_routes { text.with_query("text", r#"{"match":{"query":"paimon"}}"#); let results = [ builder - .to_route_result(&routes[0], vector.execute().await.unwrap()) + .to_route_result(&routes[0], routes[0].execute().await.unwrap()) .unwrap(), builder - .to_route_result(&routes[1], text.execute_scored().await.unwrap()) + .to_route_result(&routes[1], routes[1].execute().await.unwrap()) .unwrap(), ]; let actual = builder.rank(&results).unwrap(); let expected = mixed_query(&table, 4) .with_snapshot(Some(&snapshot)) - .execute_scored() + .execute() .await .unwrap(); assert_eq!(actual.snapshot_id(), Some(snapshot.id())); diff --git a/crates/paimon/src/table/vector_search_builder.rs b/crates/paimon/src/table/vector_search_builder.rs index eef281f29..6b119964f 100644 --- a/crates/paimon/src/table/vector_search_builder.rs +++ b/crates/paimon/src/table/vector_search_builder.rs @@ -23,10 +23,22 @@ use crate::table::vector_scan::{PlanContext, VectorScan}; use crate::table::vector_search_common::{add_filter, validate_partition_filter}; use crate::table::Table; use crate::vector_search::GlobalIndexResult; +use std::borrow::Cow; use std::collections::HashMap; +use std::sync::{Arc, RwLock}; +/// Configures a vector query. Clones share the same live configuration. +/// +/// Each scan, reader, or execution captures its settings before performing I/O; +/// later edits affect future requests without changing work already created. +#[derive(Clone)] pub struct VectorSearchBuilder<'a> { - table: &'a Table, + table: Cow<'a, Table>, + config: Arc>, +} + +#[derive(Clone, Default)] +struct VectorSearchConfig { vector_column: Option, query_vector: Option>, limit: Option, @@ -37,32 +49,43 @@ pub struct VectorSearchBuilder<'a> { impl<'a> VectorSearchBuilder<'a> { pub(crate) fn new(table: &'a Table) -> Self { Self { - table, - vector_column: None, - query_vector: None, - limit: None, - options: HashMap::new(), - filter: None, + table: Cow::Borrowed(table), + config: Arc::new(RwLock::new(VectorSearchConfig::default())), + } + } + + pub(crate) fn from_owned(table: Arc
) -> VectorSearchBuilder<'static> { + VectorSearchBuilder { + table: Cow::Owned(table.as_ref().clone()), + config: Arc::new(RwLock::new(VectorSearchConfig::default())), + } + } + + /// Retain this builder's table and shared configuration without a borrow. + pub fn into_owned(self) -> VectorSearchBuilder<'static> { + VectorSearchBuilder { + table: Cow::Owned(self.table.into_owned()), + config: self.config, } } pub fn with_vector_column(&mut self, name: &str) -> &mut Self { - self.vector_column = Some(name.to_string()); + self.config.write().unwrap().vector_column = Some(name.to_string()); self } pub fn with_query_vector(&mut self, vector: Vec) -> &mut Self { - self.query_vector = Some(vector); + self.config.write().unwrap().query_vector = Some(vector); self } pub fn with_limit(&mut self, limit: usize) -> &mut Self { - self.limit = Some(limit); + self.config.write().unwrap().limit = Some(limit); self } pub fn with_options(&mut self, options: HashMap) -> &mut Self { - self.options.extend(options); + self.config.write().unwrap().options.extend(options); self } @@ -78,68 +101,89 @@ impl<'a> VectorSearchBuilder<'a> { /// ranges are scored from raw data even in FAST mode. Raw searches always /// apply the row predicate before Top-K, matching Java. pub fn with_filter(&mut self, filter: Predicate) -> &mut Self { - add_filter(&mut self.filter, filter); + add_filter(&mut self.config.write().unwrap().filter, filter); self } /// Add a partition-only predicate expressed against the table schema. pub fn with_partition_filter(&mut self, filter: Predicate) -> crate::Result<&mut Self> { - validate_partition_filter(self.table, &filter)?; + validate_partition_filter(&self.table, &filter)?; Ok(self.with_filter(filter)) } /// Create a query-independent scan. Only the vector column must be configured. pub fn new_scan(&self) -> crate::Result { + let config = self.config.read().unwrap().clone(); + self.scan_with_config(&config) + } + + fn scan_with_config(&self, config: &VectorSearchConfig) -> crate::Result { CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?; - let column = self - .vector_column - .as_deref() - .ok_or_else(|| crate::Error::ConfigInvalid { - message: "Vector column must be set via with_vector_column()".to_string(), - })?; - VectorScan::new(self.table, column, self.filter.as_ref(), None, None) - .map(|scan| scan.with_options(&self.options)) + let column = + config + .vector_column + .as_deref() + .ok_or_else(|| crate::Error::ConfigInvalid { + message: "Vector column must be set via with_vector_column()".to_string(), + })?; + VectorScan::new(&self.table, column, config.filter.as_ref(), None, None) + .map(|scan| scan.with_options(&config.options)) } /// Create an owned reader; query errors are reported before planning. pub fn new_read(&self) -> crate::Result { - let (column, query, limit) = self.query()?; - let context = PlanContext::new(self.table, column, self.filter.as_ref(), None, None)?; + let config = self.config.read().unwrap().clone(); + self.read_with_config(&config) + } + + fn read_with_config(&self, config: &VectorSearchConfig) -> crate::Result { + let (column, query, limit) = self.query(config)?; + let context = PlanContext::new(&self.table, column, config.filter.as_ref(), None, None)?; Ok(VectorRead { batch: BatchVectorRead::new( - self.table, + &self.table, column, &[query], limit, - &self.options, - self.filter.as_ref(), + &config.options, + config.filter.as_ref(), context, )?, }) } + pub(crate) fn new_scan_and_read(&self) -> crate::Result<(VectorScan, VectorRead)> { + let config = self.config.read().unwrap().clone(); + let read = self.read_with_config(&config)?; + Ok((self.scan_with_config(&config)?, read)) + } + /// Search locally using the same Scan -> Plan -> Read API exposed to engines. /// Use the result's `new_read_builder()` to materialize projected columns. pub async fn execute(&self) -> crate::Result { - let read = self.new_read()?; - read.read(self.new_scan()?.plan().await?).await + let (scan, read) = self.new_scan_and_read()?; + read.read(scan.plan().await?).await } - fn query(&self) -> crate::Result<(&str, &[f32], usize)> { + fn query<'c>( + &self, + config: &'c VectorSearchConfig, + ) -> crate::Result<(&'c str, &'c [f32], usize)> { CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?; - let column = self - .vector_column - .as_deref() - .ok_or_else(|| crate::Error::ConfigInvalid { - message: "Vector column must be set via with_vector_column()".to_string(), - })?; - let vector = self + let column = + config + .vector_column + .as_deref() + .ok_or_else(|| crate::Error::ConfigInvalid { + message: "Vector column must be set via with_vector_column()".to_string(), + })?; + let vector = config .query_vector .as_deref() .ok_or_else(|| crate::Error::ConfigInvalid { message: "Query vector must be set via with_query_vector()".to_string(), })?; - let limit = self.limit.ok_or_else(|| crate::Error::ConfigInvalid { + let limit = config.limit.ok_or_else(|| crate::Error::ConfigInvalid { message: "Limit must be set via with_limit()".to_string(), })?; Ok((column, vector, limit)) diff --git a/crates/paimon/tests/data_evolution_search_sidecar_test.rs b/crates/paimon/tests/data_evolution_search_sidecar_test.rs index 58fde33e4..01d65784f 100644 --- a/crates/paimon/tests/data_evolution_search_sidecar_test.rs +++ b/crates/paimon/tests/data_evolution_search_sidecar_test.rs @@ -375,7 +375,7 @@ async fn full_text_and_hybrid_hits_read_latest_variant_sidecars_with_ranked_scor .with_text_column("text") .with_query_text("needle") .with_limit(3) - .execute_scored() + .execute() .await .unwrap(); let text_hits = text.row_ids().unwrap(); @@ -427,7 +427,7 @@ async fn full_text_and_hybrid_hits_read_latest_variant_sidecars_with_ranked_scor ) .unwrap() .with_limit(3); - let fused = hybrid.execute_scored().await.unwrap(); + let fused = hybrid.execute().await.unwrap(); let hits = fused.row_ids().unwrap(); assert_eq!( hits.row_ids, diff --git a/docs/src/sql.md b/docs/src/sql.md index 0e2d1e2ab..13f79fb90 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -2228,7 +2228,8 @@ let result = builder.new_read()?.read(plan).await?; `FullTextScanPlan` owns a pinned snapshot and can outlive its scan. Readers for the same table, branch, column and filter can reuse the plan with different queries or limits. Pinning preserves the query's bound schema, including -schema-only column renames. `execute_scored()` performs both stages directly. +schema-only column renames. `execute()` performs both stages directly and returns +`GlobalIndexResult`, including scores and the captured read context. For Data Evolution, full-text and scalar `fast`, `full`, and `detail` modes follow Java's coverage rules. Inexact scalar candidates are read back only From fba0781f5d9393d07f39df3a47e56d388e4094b3 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 23 Oct 2026 14:43:23 +0800 Subject: [PATCH 3/3] Preserve storage scope when retaining hybrid result read contexts --- crates/paimon/src/io/file_io.rs | 13 +++ .../paimon/src/table/hybrid_search_builder.rs | 1 + .../hybrid_search_builder/phased_tests.rs | 91 +++++++++++++++++++ 3 files changed, 105 insertions(+) diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index ab0934272..9369cb74b 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -291,6 +291,19 @@ impl FileIO { self } + /// Whether both handles retain the same storage or provider instance. + pub(crate) fn shares_backend(&self, other: &Self) -> bool { + match (&self.backend, &other.backend) { + (FileIOBackend::Storage(left), FileIOBackend::Storage(right)) => { + Arc::ptr_eq(left, right) + } + (FileIOBackend::Provider(left), FileIOBackend::Provider(right)) => { + Arc::ptr_eq(left, right) + } + _ => false, + } + } + pub(crate) fn file_format_metadata_cache(&self) -> Arc { Arc::clone(&self.file_format_metadata_cache) } diff --git a/crates/paimon/src/table/hybrid_search_builder.rs b/crates/paimon/src/table/hybrid_search_builder.rs index c21a227f9..b6b5e83b8 100644 --- a/crates/paimon/src/table/hybrid_search_builder.rs +++ b/crates/paimon/src/table/hybrid_search_builder.rs @@ -1357,6 +1357,7 @@ fn rank_read_table(results: &[HybridSearchWeightedResult]) -> Option<&Table> { let other = result.result.table()?; if table.identifier() != other.identifier() || table.location() != other.location() + || !table.file_io().shares_backend(other.file_io()) || table.branch() != other.branch() || table.schema() != other.schema() || first.result.snapshot_id() != result.result.snapshot_id() diff --git a/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs b/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs index 93945ae39..5aab29c5d 100644 --- a/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs +++ b/crates/paimon/src/table/hybrid_search_builder/phased_tests.rs @@ -493,6 +493,97 @@ mod captured_routes { } } + #[derive(Debug)] + struct PartitionScopedFileIO { + delegate: crate::io::FileIO, + denied_partition: &'static str, + } + + #[async_trait::async_trait] + impl crate::io::FileIOProvider for PartitionScopedFileIO { + async fn create(&self, path: &str) -> crate::Result<(opendal::Operator, String)> { + if path.contains(self.denied_partition) { + return Err(crate::Error::IoUnsupported { + message: "partition is outside this FileIO credential scope".into(), + }); + } + self.delegate.create_static(path) + } + } + + #[tokio::test] + async fn rank_drops_read_context_across_partition_scoped_file_io() { + let table = table(); + append(&table, 0).await; + append(&table, 4).await; + let snapshot = table.resolve_read_snapshot().await.unwrap().unwrap(); + let predicates = PredicateBuilder::new(table.schema().fields()); + let mut results = Vec::new(); + for (partition, denied_partition) in [(0, "/pt=1/"), (1, "/pt=0/")] { + // Both handles read exactly the same immutable dataset. Only their + // credentials differ: each can read one partition's data files. + let file_io = table + .file_io() + .clone() + .with_provider(Arc::new(PartitionScopedFileIO { + delegate: table.file_io().clone(), + denied_partition, + })); + let scoped_table = Table::new( + file_io, + table.identifier().clone(), + table.location().to_string(), + table.schema().clone(), + None, + ) + .copy_with_pinned_snapshot(Some(&snapshot)); + let result = scoped_table + .new_vector_search_builder() + .with_vector_column("embedding") + .with_query_vector(vec![0.0, 0.0]) + .with_limit(4) + .with_partition_filter(predicates.equal("pt", Datum::Int(partition)).unwrap()) + .unwrap() + .execute() + .await + .unwrap(); + assert_eq!(result.len(), 4); + // Each route can materialize its own hits using its own credentials. + let batches = result + .new_read_builder() + .with_projection(&["id"]) + .read() + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::(), 4); + let weighted = HybridSearchWeightedResult::new(result, 1.0).unwrap(); + let same_scope = HybridSearchRanker::Mrr + .rank(&[weighted.clone(), weighted.clone()], 4) + .unwrap(); + assert!(same_scope.table().is_some()); + assert_eq!(same_scope.snapshot_id(), Some(snapshot.id())); + results.push(weighted); + } + + let ranked = HybridSearchRanker::Mrr.rank(&results, 8).unwrap(); + let mut row_ids = ranked.row_ids().unwrap().row_ids.clone(); + row_ids.sort_unstable(); + assert_eq!(row_ids, (0..8).collect::>()); + assert!(ranked.table().is_none()); + assert!(ranked.snapshot_id().is_none()); + match ranked.new_read_builder().read().await { + Err(crate::Error::Unsupported { message }) => { + assert!(message + .contains("Direct result reads require a result returned by a table search")); + } + Err(error) => panic!("expected missing read context, got: {error}"), + Ok(_) => panic!("mixed credential scopes must not retain one route's read context"), + } + } + #[tokio::test] async fn route_preparation_validates_before_resolving_snapshots() { let table = table();