This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new cabdeb9f refactor(vector-search): remove extra prefilter guards (#769)
cabdeb9f is described below
commit cabdeb9f2f035d0c6c7eb45adda39746cdbf278a
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 1 17:07:30 2026 +0800
refactor(vector-search): remove extra prefilter guards (#769)
---
.../integrations/datafusion/src/filter_pushdown.rs | 35 ----------------------
.../datafusion/src/lateral_vector_search.rs | 6 ++--
.../integrations/datafusion/src/vector_search.rs | 13 ++------
crates/paimon/src/io/file_io.rs | 20 -------------
crates/paimon/src/table/vector_search_builder.rs | 31 +------------------
5 files changed, 6 insertions(+), 99 deletions(-)
diff --git a/crates/integrations/datafusion/src/filter_pushdown.rs
b/crates/integrations/datafusion/src/filter_pushdown.rs
index 6155062e..cdbe6460 100644
--- a/crates/integrations/datafusion/src/filter_pushdown.rs
+++ b/crates/integrations/datafusion/src/filter_pushdown.rs
@@ -89,20 +89,6 @@ pub(crate) fn analyze_filters(
}
}
-pub(crate) fn is_safe_vector_prefilter(predicate: &Predicate) -> bool {
- match predicate {
- Predicate::Leaf { literals, .. } => !literals.iter().any(|literal| {
- matches!(literal, Datum::Float(value) if value.is_nan())
- || matches!(literal, Datum::Double(value) if value.is_nan())
- }),
- Predicate::And(children) | Predicate::Or(children) => {
- children.iter().all(is_safe_vector_prefilter)
- }
- Predicate::Not(inner) => is_safe_vector_prefilter(inner),
- Predicate::AlwaysTrue | Predicate::AlwaysFalse => true,
- }
-}
-
#[cfg(test)]
pub(crate) fn build_pushed_predicate(filters: &[Expr], fields: &[DataField])
-> Option<Predicate> {
analyze_filters(filters, fields, true).pushed_predicate
@@ -926,27 +912,6 @@ mod tests {
);
}
- #[test]
- fn test_vector_prefilter_rejects_nan_literals() {
- let predicate = Predicate::Leaf {
- column: "score".to_string(),
- index: 0,
- data_type: DataType::Float(FloatType::new()),
- op: PredicateOperator::Eq,
- literals: vec![Datum::Float(f32::NAN)],
- };
- assert!(!is_safe_vector_prefilter(&predicate));
-
- let finite = Predicate::Leaf {
- column: "score".to_string(),
- index: 0,
- data_type: DataType::Float(FloatType::new()),
- op: PredicateOperator::Eq,
- literals: vec![Datum::Float(f32::INFINITY)],
- };
- assert!(is_safe_vector_prefilter(&finite));
- }
-
#[test]
fn test_negated_inexact_float_array_membership_falls_open() {
use datafusion::functions_nested::expr_fn::array_has;
diff --git a/crates/integrations/datafusion/src/lateral_vector_search.rs
b/crates/integrations/datafusion/src/lateral_vector_search.rs
index 674ef2af..8e88c538 100644
--- a/crates/integrations/datafusion/src/lateral_vector_search.rs
+++ b/crates/integrations/datafusion/src/lateral_vector_search.rs
@@ -62,7 +62,7 @@ use paimon::vector_search::SearchResult;
use tokio::sync::OnceCell;
use crate::error::to_datafusion_error;
-use crate::filter_pushdown::{analyze_filters, is_safe_vector_prefilter};
+use crate::filter_pushdown::analyze_filters;
use crate::vector_search::LateralVectorSearchTableProvider;
#[derive(Debug)]
@@ -163,9 +163,7 @@ impl OptimizerRule for RewriteLateralVectorSearch {
true,
);
match analysis.pushed_predicate {
- Some(predicate)
- if !analysis.requires_residual &&
is_safe_vector_prefilter(&predicate) =>
- {
+ Some(predicate) if !analysis.requires_residual => {
target_predicates.push(predicate);
}
_ => {
diff --git a/crates/integrations/datafusion/src/vector_search.rs
b/crates/integrations/datafusion/src/vector_search.rs
index f680800d..d37d16f2 100644
--- a/crates/integrations/datafusion/src/vector_search.rs
+++ b/crates/integrations/datafusion/src/vector_search.rs
@@ -49,7 +49,7 @@ use paimon::spec::{
use paimon::table::Table;
use crate::error::to_datafusion_error;
-use crate::filter_pushdown::{analyze_filters, is_safe_vector_prefilter};
+use crate::filter_pushdown::analyze_filters;
use crate::runtime::{await_with_runtime, block_on_with_runtime};
use crate::table::{datafusion_read_fields, PaimonTableProvider};
use crate::table_function_args::{
@@ -273,9 +273,7 @@ impl TableProvider for VectorSearchTableProvider {
"vector_search cannot apply a partially translated scalar
pre-filter".to_string(),
));
}
- let pushed_predicate = filter_analysis
- .pushed_predicate
- .filter(is_safe_vector_prefilter);
+ let pushed_predicate = filter_analysis.pushed_predicate;
// An outer `LIMIT 0` needs no rows.
if limit == Some(0) {
@@ -312,12 +310,7 @@ impl TableProvider for VectorSearchTableProvider {
.iter()
.map(|filter| {
let analysis = analyze_filters(std::slice::from_ref(*filter),
fields, true);
- if analysis
- .pushed_predicate
- .as_ref()
- .is_some_and(is_safe_vector_prefilter)
- && !analysis.requires_residual
- {
+ if analysis.pushed_predicate.is_some() &&
!analysis.requires_residual {
// Keep DataFusion's residual filter as a correctness
backstop
// while using the same predicate before vector Top-K.
TableProviderFilterPushDown::Inexact
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index b15c1e2d..1257f632 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -63,16 +63,11 @@ pub(crate) trait FileIOProvider: std::fmt::Debug + Send +
Sync {
#[derive(Clone)]
pub struct FileIO {
- /// Private identity shared by clones, but not by independently built
storage backends.
- storage_lineage: Arc<StorageLineage>,
storage: Arc<Storage>,
cache: Option<Arc<LocalCache>>,
provider: Option<Arc<dyn FileIOProvider>>,
}
-#[derive(Debug)]
-struct StorageLineage;
-
impl std::fmt::Debug for FileIO {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FileIO")
@@ -84,10 +79,6 @@ impl std::fmt::Debug for FileIO {
}
impl FileIO {
- pub(crate) fn shares_storage_lineage(&self, other: &Self) -> bool {
- Arc::ptr_eq(&self.storage_lineage, &other.storage_lineage)
- }
-
/// Attach an externally managed block cache.
///
/// `block_size` controls the aligned ranges presented to the cache.
@@ -542,7 +533,6 @@ impl FileIOBuilder {
let cache = self.cache.clone();
let storage = Storage::build(self)?;
Ok(FileIO {
- storage_lineage: Arc::new(StorageLineage),
storage: Arc::new(storage),
cache,
provider: None,
@@ -1264,16 +1254,6 @@ mod file_action_test {
assert!(!file_io_2.exists(path).await.unwrap());
}
- #[test]
- fn test_storage_lineage_is_shared_only_by_file_io_clones() {
- let file_io = setup_memory_file_io();
- let clone = file_io.clone();
- let independent = setup_memory_file_io();
-
- assert!(file_io.shares_storage_lineage(&clone));
- assert!(!file_io.shares_storage_lineage(&independent));
- }
-
#[tokio::test]
async fn test_get_status_fs() {
let file_io = setup_fs_file_io();
diff --git a/crates/paimon/src/table/vector_search_builder.rs
b/crates/paimon/src/table/vector_search_builder.rs
index 1fc662b9..3753a4f8 100644
--- a/crates/paimon/src/table/vector_search_builder.rs
+++ b/crates/paimon/src/table/vector_search_builder.rs
@@ -222,8 +222,7 @@ impl PreparedVectorSearchFilter {
}
fn same_vector_search_table(left: &Table, right: &Table) -> bool {
- left.file_io().shares_storage_lineage(right.file_io())
- && left.location().trim_end_matches('/') ==
right.location().trim_end_matches('/')
+ left.location().trim_end_matches('/') ==
right.location().trim_end_matches('/')
&& left.branch() == right.branch()
}
@@ -7537,34 +7536,6 @@ mod tests {
);
}
- #[tokio::test]
- async fn
prepared_filter_from_same_location_but_different_file_io_is_rejected() {
- let location = "memory:/prepared_filter_shared_location";
- let source =
-
vector_test_table_with_file_io(FileIOBuilder::new("memory").build().unwrap(),
location);
- let prepared = source
- .prepare_vector_search_filter(id_gt_filter(&source, 0))
- .await
- .unwrap();
- let target =
-
vector_test_table_with_file_io(FileIOBuilder::new("memory").build().unwrap(),
location);
-
- let error = target
- .new_batch_vector_search_builder()
- .with_vector_column("embedding")
- .with_query_vectors(vec![vec![1.0, 0.0]])
- .with_limit(1)
- .with_prepared_filter(prepared)
- .execute()
- .await
- .expect_err("a prepared filter must stay bound to its FileIO
lineage");
-
- assert!(
- error.to_string().contains("different table"),
- "unexpected error: {error}"
- );
- }
-
#[tokio::test]
async fn de_scalar_filter_applies_to_unindexed_raw_fallback() {
let table = de_vector_table().await;
