From fcc81d3fdef816f090d938cb31524a7380d345aa Mon Sep 17 00:00:00 2001 From: zhangstar333 Date: Fri, 28 Aug 2026 12:33:59 +0800 Subject: [PATCH 1/2] add lance_scanner_additional_sql_filter --- README.md | 2 +- include/lance/lance.h | 16 +++++++ include/lance/lance.hpp | 7 ++++ src/scanner.rs | 93 +++++++++++++++++++++++++++++++++++++---- tests/c_api_test.rs | 80 +++++++++++++++++++++++++++++++++-- 5 files changed, 187 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 436c2f1..d6ae6c8 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,7 @@ Based on the [liblance RFC](https://github.com/lance-format/lance/discussions/60 |--------|-----------|-------------| | [x] | Async scan | Callback-based `lance_scanner_scan_async()` for non-blocking scans | | [x] | Dataset metadata | `lance_dataset_version()`, `lance_dataset_count_rows()`, `lance_dataset_latest_version()` | -| [x] | Substrait filter pushdown | `lance_scanner_set_substrait_filter()` accepts a serialized Substrait `ExtendedExpression` (preferred over SQL strings for query engines) | +| [x] | Filter pushdown | `lance_scanner_set_substrait_filter()` accepts a serialized Substrait `ExtendedExpression`; `lance_scanner_additional_sql_filter()` adds SQL predicates with AND before scanning starts | ## Building diff --git a/include/lance/lance.h b/include/lance/lance.h index 55cfc71..ccd7e8c 100644 --- a/include/lance/lance.h +++ b/include/lance/lance.h @@ -906,6 +906,22 @@ int32_t lance_scanner_set_substrait_filter( size_t len ); +/** + * Add an SQL filter that is combined with the selected primary filter using + * AND. The primary filter is the Substrait filter when set, otherwise it is + * the SQL filter passed to `lance_scanner_new`. Multiple additional SQL + * filters are also combined using AND. + * + * Must be called before the scan starts. The filter string is copied. + * + * @param filter Non-NULL, non-empty SQL filter expression + * @return 0 on success, -1 on error + */ +int32_t lance_scanner_additional_sql_filter( + LanceScanner* scanner, + const char* filter +); + /** Type of a dynamically named scan metric. */ typedef enum { LANCE_SCAN_METRIC_COUNT = 0, diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp index b9ba7f9..330e419 100644 --- a/include/lance/lance.hpp +++ b/include/lance/lance.hpp @@ -1185,6 +1185,13 @@ class Scanner { return substrait_filter(bytes.data(), bytes.size()); } + /// Add an SQL filter that is combined with the selected primary filter using AND. + Scanner& additional_sql_filter(const std::string& filter) { + if (lance_scanner_additional_sql_filter(handle_.get(), filter.c_str()) != 0) + check_error(); + return *this; + } + /// Register a non-null callback for scan statistics after successful full exhaustion. /// The registration applies to every stream derived from this scanner, including /// concurrent streams and streams created after an earlier callback returns. The diff --git a/src/scanner.rs b/src/scanner.rs index 536d72e..7110111 100644 --- a/src/scanner.rs +++ b/src/scanner.rs @@ -11,7 +11,7 @@ use std::sync::{Arc, Condvar, Mutex, Weak}; use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker}; use arrow::ffi_stream::FFI_ArrowArrayStream; -use arrow_schema::SchemaRef; +use arrow_schema::{Schema as ArrowSchema, SchemaRef}; use datafusion::physical_plan::ExecutionPlan; use futures::{FutureExt, Stream, StreamExt}; use lance::Dataset; @@ -20,6 +20,9 @@ use lance::dataset::scanner::{ }; use lance::io::exec::fts::MatchQueryExec; use lance_core::Result; +use lance_datafusion::exec::{LanceExecutionOptions, get_session_context}; +use lance_datafusion::planner::Planner; +use lance_datafusion::substrait::parse_substrait; use lance_index::scalar::FullTextSearchQuery; use lance_io::stream::RecordBatchStream; use lance_table::format::IndexMetadata; @@ -56,6 +59,7 @@ pub struct LanceScanner { columns: Option>, filter: Option, substrait_filter: Option>, + additional_sql_filters: Vec, limit: Option, offset: Option, batch_size: Option, @@ -124,6 +128,7 @@ impl LanceScanner { columns: None, filter: None, substrait_filter: None, + additional_sql_filters: Vec::new(), limit: None, offset: None, batch_size: None, @@ -177,6 +182,41 @@ impl LanceScanner { Ok(()) } + fn apply_filter(&self, scanner: &mut lance::dataset::scanner::Scanner) -> Result<()> { + if self.additional_sql_filters.is_empty() { + if let Some(substrait) = &self.substrait_filter { + scanner.filter_substrait(substrait)?; + } else if let Some(sql) = &self.filter { + scanner.filter(sql)?; + } + return Ok(()); + } + + let schema = Arc::new(ArrowSchema::from(self.dataset.schema())); + let planner = Planner::new(Arc::clone(&schema)); + let mut combined = if let Some(substrait) = &self.substrait_filter { + let context = get_session_context(&LanceExecutionOptions::default()); + Some( + parse_substrait(substrait, schema, &context.state()) + .now_or_never() + .expect("Substrait filter parsing must complete synchronously")?, + ) + } else if let Some(sql) = &self.filter { + Some(planner.parse_filter(sql)?) + } else { + None + }; + for sql in &self.additional_sql_filters { + let sql = planner.parse_filter(sql)?; + combined = Some(match combined { + Some(existing) => existing.and(sql), + None => sql, + }); + } + scanner.filter_expr(planner.optimize_expr(combined.expect("additional filter exists"))?); + Ok(()) + } + /// Build the underlying Scanner and open a stream. fn materialize_stream(&mut self) -> Result<()> { let prepared_scanner = self.build_scanner()?; @@ -193,12 +233,7 @@ impl LanceScanner { if let Some(cols) = &self.columns { scanner.project(cols)?; } - // Substrait filter takes precedence over SQL filter when both are set. - if let Some(bytes) = &self.substrait_filter { - scanner.filter_substrait(bytes)?; - } else if let Some(filter) = &self.filter { - scanner.filter(filter)?; - } + self.apply_filter(&mut scanner)?; if self.limit.is_some() || self.offset.is_some() { scanner.limit(self.limit, self.offset)?; } @@ -782,6 +817,50 @@ unsafe fn scanner_set_substrait_filter_inner( Ok(0) } +/// Add an SQL filter that is combined with the scanner's selected primary filter using AND. +/// +/// The primary filter is the Substrait filter when one is set, otherwise it is the SQL filter +/// passed to `lance_scanner_new`. Multiple additional SQL filters are also combined using AND. +/// This must be called before the scan starts. The string is copied into the scanner. +/// +/// Returns 0 on success, -1 on error (check `lance_last_error_*`). +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_scanner_additional_sql_filter( + scanner: *mut LanceScanner, + filter: *const c_char, +) -> i32 { + scanner_poison_check!(scanner, -1); + scanner_ffi_try!(scanner, unsafe { + scanner_additional_sql_filter_inner(scanner, filter) + }) +} + +unsafe fn scanner_additional_sql_filter_inner( + scanner: *mut LanceScanner, + filter: *const c_char, +) -> Result { + if scanner.is_null() { + return Err(lance_core::Error::invalid_input_source( + "scanner is NULL".into(), + )); + } + let filter = unsafe { helpers::parse_c_string(filter)? } + .ok_or_else(|| lance_core::Error::invalid_input_source("filter must not be NULL".into()))?; + if filter.is_empty() { + return Err(lance_core::Error::invalid_input_source( + "additional SQL filter must be non-empty".into(), + )); + } + let scanner = unsafe { &mut *scanner }; + if scanner.scan_started.load(Ordering::Acquire) { + return Err(lance_core::Error::invalid_input_source( + "additional SQL filter must be set before the scan starts".into(), + )); + } + scanner.additional_sql_filters.push(filter.to_string()); + Ok(0) +} + /// Register a callback that receives execution statistics after the scan stream /// is fully consumed to EOF. /// diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs index 5949c85..17dc6c6 100644 --- a/tests/c_api_test.rs +++ b/tests/c_api_test.rs @@ -6900,18 +6900,17 @@ fn test_scanner_with_substrait_filter() { #[test] fn test_scanner_substrait_filter_overrides_sql_filter() { - // If both SQL and Substrait filters are set, Substrait wins (last write). + // If both primary filters are set, Substrait wins. let (_tmp, uri) = create_test_dataset(); let c_uri = c_str(&uri); let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; assert!(!ds.is_null()); - // Start with SQL filter "id < 0" (matches 0 rows). let sql = c_str("id < 0"); let scanner = unsafe { lance_scanner_new(ds, ptr::null(), sql.as_ptr()) }; assert!(!scanner.is_null()); - // Override with Substrait filter "id > 3" (matches 2 rows). + // Attach Substrait filter "id > 3" (matches id=4 and id=5). let bytes = substrait_id_gt_3(); let rc = unsafe { lance_scanner_set_substrait_filter(scanner, bytes.as_ptr(), bytes.len()) }; assert_eq!(rc, 0); @@ -6928,6 +6927,81 @@ fn test_scanner_substrait_filter_overrides_sql_filter() { unsafe { lance_dataset_close(ds) }; } +#[test] +fn test_scanner_additional_sql_filters_are_anded_with_substrait() { + let (_tmp, uri) = create_test_dataset(); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) }; + assert!(!scanner.is_null()); + + let bytes = substrait_id_gt_3(); + assert_eq!( + unsafe { lance_scanner_set_substrait_filter(scanner, bytes.as_ptr(), bytes.len()) }, + 0 + ); + for sql in [c_str("id < 6"), c_str("id < 5")] { + assert_eq!( + unsafe { lance_scanner_additional_sql_filter(scanner, sql.as_ptr()) }, + 0 + ); + } + + let mut ffi_stream = FFI_ArrowArrayStream::empty(); + assert_eq!( + unsafe { lance_scanner_to_arrow_stream(scanner, &mut ffi_stream) }, + 0 + ); + + let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut ffi_stream) }.unwrap(); + let total_rows: usize = reader.map(|r| r.unwrap().num_rows()).sum(); + assert_eq!(total_rows, 1, "id > 3 AND id < 6 AND id < 5 matches id=4"); + + unsafe { lance_scanner_close(scanner) }; + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_scanner_additional_sql_filter_rejects_invalid_inputs() { + let (_tmp, uri) = create_test_dataset(); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) }; + assert!(!scanner.is_null()); + + let filter = c_str("id > 3"); + assert_eq!( + unsafe { lance_scanner_additional_sql_filter(ptr::null_mut(), filter.as_ptr()) }, + -1 + ); + assert_eq!( + unsafe { lance_scanner_additional_sql_filter(scanner, ptr::null()) }, + -1 + ); + let empty = c_str(""); + assert_eq!( + unsafe { lance_scanner_additional_sql_filter(scanner, empty.as_ptr()) }, + -1 + ); + + let mut ffi_stream = FFI_ArrowArrayStream::empty(); + assert_eq!( + unsafe { lance_scanner_to_arrow_stream(scanner, &mut ffi_stream) }, + 0 + ); + assert_eq!( + unsafe { lance_scanner_additional_sql_filter(scanner, filter.as_ptr()) }, + -1, + "additional filters must be rejected after the scan starts" + ); + drop(unsafe { ArrowArrayStreamReader::from_raw(&mut ffi_stream) }.unwrap()); + + unsafe { lance_scanner_close(scanner) }; + unsafe { lance_dataset_close(ds) }; +} + #[test] fn test_scanner_set_substrait_filter_invalid_inputs() { let (_tmp, uri) = create_test_dataset(); From cc95d9f097f56b2a37b89d6c46a6096956b2e2f3 Mon Sep 17 00:00:00 2001 From: zhangstar333 Date: Fri, 28 Aug 2026 17:38:33 +0800 Subject: [PATCH 2/2] update --- src/scanner.rs | 48 +++++++---------- tests/c_api_test.rs | 129 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 149 insertions(+), 28 deletions(-) diff --git a/src/scanner.rs b/src/scanner.rs index 7110111..5b8c34e 100644 --- a/src/scanner.rs +++ b/src/scanner.rs @@ -11,7 +11,7 @@ use std::sync::{Arc, Condvar, Mutex, Weak}; use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker}; use arrow::ffi_stream::FFI_ArrowArrayStream; -use arrow_schema::{Schema as ArrowSchema, SchemaRef}; +use arrow_schema::SchemaRef; use datafusion::physical_plan::ExecutionPlan; use futures::{FutureExt, Stream, StreamExt}; use lance::Dataset; @@ -20,9 +20,6 @@ use lance::dataset::scanner::{ }; use lance::io::exec::fts::MatchQueryExec; use lance_core::Result; -use lance_datafusion::exec::{LanceExecutionOptions, get_session_context}; -use lance_datafusion::planner::Planner; -use lance_datafusion::substrait::parse_substrait; use lance_index::scalar::FullTextSearchQuery; use lance_io::stream::RecordBatchStream; use lance_table::format::IndexMetadata; @@ -183,37 +180,32 @@ impl LanceScanner { } fn apply_filter(&self, scanner: &mut lance::dataset::scanner::Scanner) -> Result<()> { + if let Some(substrait) = &self.substrait_filter { + scanner.filter_substrait(substrait)?; + } else if let Some(sql) = &self.filter { + scanner.filter(sql)?; + } + if self.additional_sql_filters.is_empty() { - if let Some(substrait) = &self.substrait_filter { - scanner.filter_substrait(substrait)?; - } else if let Some(sql) = &self.filter { - scanner.filter(sql)?; - } return Ok(()); } - let schema = Arc::new(ArrowSchema::from(self.dataset.schema())); - let planner = Planner::new(Arc::clone(&schema)); - let mut combined = if let Some(substrait) = &self.substrait_filter { - let context = get_session_context(&LanceExecutionOptions::default()); - Some( - parse_substrait(substrait, schema, &context.state()) - .now_or_never() - .expect("Substrait filter parsing must complete synchronously")?, - ) - } else if let Some(sql) = &self.filter { - Some(planner.parse_filter(sql)?) - } else { - None - }; + // Let Lance resolve every SQL expression against the scanner's full + // filterable schema. Besides stored columns, this includes metadata + // columns and query-generated columns such as _distance and _score. + let mut combined = scanner.get_expr_filter()?; for sql in &self.additional_sql_filters { - let sql = planner.parse_filter(sql)?; + let mut additional_scanner = scanner.clone(); + additional_scanner.filter(sql)?; + let additional = additional_scanner + .get_expr_filter()? + .expect("additional SQL filter exists"); combined = Some(match combined { - Some(existing) => existing.and(sql), - None => sql, + Some(existing) => existing.and(additional), + None => additional, }); } - scanner.filter_expr(planner.optimize_expr(combined.expect("additional filter exists"))?); + scanner.filter_expr(combined.expect("additional SQL filter exists")); Ok(()) } @@ -233,7 +225,6 @@ impl LanceScanner { if let Some(cols) = &self.columns { scanner.project(cols)?; } - self.apply_filter(&mut scanner)?; if self.limit.is_some() || self.offset.is_some() { scanner.limit(self.limit, self.offset)?; } @@ -305,6 +296,7 @@ impl LanceScanner { } else { None }; + self.apply_filter(&mut scanner)?; if let Some(callback) = &self.scan_statistics_callback { scanner.scan_stats_callback(callback.clone()); } diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs index 17dc6c6..5b0c013 100644 --- a/tests/c_api_test.rs +++ b/tests/c_api_test.rs @@ -6963,6 +6963,135 @@ fn test_scanner_additional_sql_filters_are_anded_with_substrait() { unsafe { lance_dataset_close(ds) }; } +#[test] +fn test_scanner_additional_sql_filter_preserves_metadata_primary_filter() { + let (_tmp, uri) = create_test_dataset(); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let primary = c_str( + "_rowid IS NOT NULL AND _rowaddr IS NOT NULL \ + AND _row_created_at_version IS NOT NULL \ + AND _row_last_updated_at_version IS NOT NULL", + ); + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), primary.as_ptr()) }; + assert!(!scanner.is_null()); + + let additional = c_str("id > 3"); + assert_eq!( + unsafe { lance_scanner_additional_sql_filter(scanner, additional.as_ptr()) }, + 0 + ); + + let mut ffi_stream = FFI_ArrowArrayStream::empty(); + assert_eq!( + unsafe { lance_scanner_to_arrow_stream(scanner, &mut ffi_stream) }, + 0 + ); + let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut ffi_stream) }.unwrap(); + let total_rows: usize = reader.map(|batch| batch.unwrap().num_rows()).sum(); + assert_eq!(total_rows, 2, "metadata predicate AND id > 3"); + + unsafe { lance_scanner_close(scanner) }; + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_scanner_additional_sql_filter_preserves_distance_primary_filter() { + let (_tmp, uri) = create_vector_dataset(16, 8); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let primary = c_str("_distance IS NOT NULL"); + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), primary.as_ptr()) }; + assert!(!scanner.is_null()); + let query = [0.0_f32; 8]; + let column = c_str("embedding"); + assert_eq!( + unsafe { + lance_scanner_nearest( + scanner, + column.as_ptr(), + query.as_ptr().cast(), + query.len(), + LanceDataType::Float32 as i32, + 16, + ) + }, + 0 + ); + let additional = c_str("id < 3"); + assert_eq!( + unsafe { lance_scanner_additional_sql_filter(scanner, additional.as_ptr()) }, + 0 + ); + + let mut ffi_stream = FFI_ArrowArrayStream::empty(); + assert_eq!( + unsafe { lance_scanner_to_arrow_stream(scanner, &mut ffi_stream) }, + 0 + ); + let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut ffi_stream) }.unwrap(); + let total_rows: usize = reader.map(|batch| batch.unwrap().num_rows()).sum(); + assert_eq!(total_rows, 3, "_distance predicate AND id < 3"); + + unsafe { lance_scanner_close(scanner) }; + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_scanner_additional_sql_filter_preserves_score_primary_filter() { + let (_tmp, uri) = create_test_dataset(); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let column = c_str("name"); + let inverted_params = c_str(r#"{"base_tokenizer":"simple","language":"English"}"#); + assert_eq!( + unsafe { + lance_dataset_create_scalar_index( + ds, + column.as_ptr(), + ptr::null(), + LanceScalarIndexType::Inverted as i32, + inverted_params.as_ptr(), + false, + ) + }, + 0 + ); + + let primary = c_str("_score IS NOT NULL"); + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), primary.as_ptr()) }; + assert!(!scanner.is_null()); + let query = c_str("alice"); + let columns = [column.as_ptr(), ptr::null()]; + assert_eq!( + unsafe { lance_scanner_full_text_search(scanner, query.as_ptr(), columns.as_ptr(), 0) }, + 0 + ); + let additional = c_str("id >= 1"); + assert_eq!( + unsafe { lance_scanner_additional_sql_filter(scanner, additional.as_ptr()) }, + 0 + ); + + let mut ffi_stream = FFI_ArrowArrayStream::empty(); + assert_eq!( + unsafe { lance_scanner_to_arrow_stream(scanner, &mut ffi_stream) }, + 0 + ); + let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut ffi_stream) }.unwrap(); + let total_rows: usize = reader.map(|batch| batch.unwrap().num_rows()).sum(); + assert_eq!(total_rows, 1, "_score predicate AND id >= 1"); + + unsafe { lance_scanner_close(scanner) }; + unsafe { lance_dataset_close(ds) }; +} + #[test] fn test_scanner_additional_sql_filter_rejects_invalid_inputs() { let (_tmp, uri) = create_test_dataset();