diff --git a/gensrc/proto/internal_service.proto b/gensrc/proto/internal_service.proto index 00cd171dcc03d5..55c64d302a03a5 100644 --- a/gensrc/proto/internal_service.proto +++ b/gensrc/proto/internal_service.proto @@ -831,6 +831,9 @@ message PRequestBlockDesc { repeated uint32 row_id = 5; optional PTupleDescriptor desc = 6; repeated uint32 column_idxs = 7; + // Numeric value of ROW_VERSION. Keep this as uint32 so a receiver can detect and reject a + // future version that is unknown to its current binary instead of treating it as version 0. + optional uint32 row_location_version = 8 [default = 0]; } message PTopNLazyMaterializationFileCacheStats { diff --git a/thirdparty/download-thirdparty.sh b/thirdparty/download-thirdparty.sh index 6b8d90c726f431..eb4aed22823e63 100755 --- a/thirdparty/download-thirdparty.sh +++ b/thirdparty/download-thirdparty.sh @@ -765,10 +765,13 @@ fi # Apply Doris lance-c patches. if [[ " ${TP_ARCHIVES[*]} " =~ " LANCE_C " ]]; then - if [[ "${LANCE_C_SOURCE}" == "lance-c-0.1.8" ]]; then + if [[ "${LANCE_C_SOURCE}" == "lance-c-0.1.9" ]]; then cd "${TP_SOURCE_DIR}/${LANCE_C_SOURCE}" if [[ ! -f "${PATCHED_MARK}" ]]; then - patch -p1 <"${TP_PATCH_DIR}/lance-c-0.1.8-pr-69.patch" + patch --batch --forward --reject-file=- --fuzz=0 --no-backup-if-mismatch -s \ + -p1 <"${TP_PATCH_DIR}/lance-c-0.1.9-pr-73.patch" + patch --batch --forward --reject-file=- --fuzz=0 --no-backup-if-mismatch -s \ + -p1 <"${TP_PATCH_DIR}/lance-c-0.1.9-pr-74.patch" touch "${PATCHED_MARK}" fi cd - diff --git a/thirdparty/patches/lance-c-0.1.8-pr-69.patch b/thirdparty/patches/lance-c-0.1.8-pr-69.patch deleted file mode 100644 index ee38eb33bcb2fd..00000000000000 --- a/thirdparty/patches/lance-c-0.1.8-pr-69.patch +++ /dev/null @@ -1,653 +0,0 @@ -From a4d6e489c627fe4b0e49d9a2991c9436313debc0 Mon Sep 17 00:00:00 2001 -From: zhangstar333 -Date: Tue, 1 Sep 2026 11:15:00 +0800 -Subject: [PATCH 1/2] update - ---- - src/fts_query.rs | 2 ++ - src/scanner.rs | 44 ++++++++++++++++++++++--- - tests/c_api_test.rs | 78 +++++++++++++++++++++++++++++++++++++++++++++ - 3 files changed, 120 insertions(+), 4 deletions(-) - -diff --git a/src/fts_query.rs b/src/fts_query.rs -index cd194c7..e85ed99 100644 ---- a/src/fts_query.rs -+++ b/src/fts_query.rs -@@ -54,6 +54,7 @@ pub(crate) struct FtsQueryContextInner { - pub(crate) query: FullTextSearchQuery, - pub(crate) segments: Vec, - pub(crate) scorer: Arc, -+ pub(crate) has_unindexed_fragments: bool, - } - - impl FtsQueryContextInner { -@@ -216,6 +217,7 @@ async fn prepare_fts_query_context( - query, - segments, - scorer, -+ has_unindexed_fragments: !unindexed_fragment_ids.is_empty(), - }) - } - -diff --git a/src/scanner.rs b/src/scanner.rs -index 5b8c34e..f60f0c5 100644 ---- a/src/scanner.rs -+++ b/src/scanner.rs -@@ -179,6 +179,45 @@ impl LanceScanner { - Ok(()) - } - -+ /// Restrict an INDEX_ONLY prepared FTS scan to fragments covered by the -+ /// selected committed segments. This is deliberately separate from -+ /// `fast_search`: that option is scanner-wide, changes unrelated scalar -+ /// index fallback behavior, and also forces `_rowid` into the output. -+ fn apply_prepared_fts_fragment_filter( -+ &self, -+ scanner: &mut lance::dataset::scanner::Scanner, -+ context: &FtsQueryContextInner, -+ segments: &[IndexMetadata], -+ ) -> Result<()> { -+ if !context.has_unindexed_fragments { -+ return Ok(()); -+ } -+ -+ let mut selected_fragment_ids = std::collections::HashSet::new(); -+ for segment in segments { -+ let fragment_bitmap = segment.fragment_bitmap.as_ref().ok_or_else(|| { -+ lance_core::Error::internal(format!( -+ "prepared FTS segment {} lost its validated fragment coverage", -+ segment.uuid -+ )) -+ })?; -+ selected_fragment_ids.extend(fragment_bitmap.iter()); -+ } -+ -+ let selected_fragments = self -+ .dataset -+ .get_fragments() -+ .into_iter() -+ .filter(|fragment| { -+ u32::try_from(fragment.id()) -+ .is_ok_and(|fragment_id| selected_fragment_ids.contains(&fragment_id)) -+ }) -+ .map(|fragment| fragment.metadata().clone()) -+ .collect(); -+ scanner.with_fragments(selected_fragments); -+ Ok(()) -+ } -+ - fn apply_filter(&self, scanner: &mut lance::dataset::scanner::Scanner) -> Result<()> { - if let Some(substrait) = &self.substrait_filter { - scanner.filter_substrait(substrait)?; -@@ -282,11 +321,8 @@ impl LanceScanner { - let distributed_fts = if let Some(context) = &self.fts_context { - context.validate_dataset_identity(&self.dataset)?; - let segments = select_fts_segments(context, self.fts_index_segments.as_deref())?; -+ self.apply_prepared_fts_fragment_filter(&mut scanner, context, &segments)?; - scanner.full_text_search(context.query.clone())?; -- // Both STRICT and INDEX_ONLY context scans must use only the -- // committed segments pinned in the context. In STRICT mode all -- // current fragments were already proven covered during prepare. -- scanner.fast_search(); - Some(PreparedFtsExecution { - context: Arc::clone(context), - segments, -diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs -index 74b9f85..3627c4e 100644 ---- a/tests/c_api_test.rs -+++ b/tests/c_api_test.rs -@@ -5762,6 +5762,84 @@ fn load_fts_segment_uuids(uri: &str, column: &str) -> Vec<[u8; 16]> { - }) - } - -+#[test] -+fn test_prepared_fts_row_id_output_is_explicit() { -+ let (_tmp, uri) = create_test_dataset(); -+ let uri_c = c_str(&uri); -+ let column = c_str("name"); -+ let query = c_str("alice"); -+ let inverted_params = c_str(r#"{"base_tokenizer":"simple","language":"English"}"#); -+ -+ let dataset = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; -+ assert_eq!( -+ unsafe { -+ lance_dataset_create_scalar_index( -+ dataset, -+ column.as_ptr(), -+ ptr::null(), -+ LanceScalarIndexType::Inverted as i32, -+ inverted_params.as_ptr(), -+ false, -+ ) -+ }, -+ 0 -+ ); -+ let context = unsafe { -+ lance_dataset_prepare_fts_query( -+ dataset, -+ column.as_ptr(), -+ query.as_ptr(), -+ 0, -+ LanceFtsCoverageMode::Strict as i32, -+ ) -+ }; -+ assert!(!context.is_null(), "{}", unsafe { -+ std::ffi::CStr::from_ptr(lance_last_error_message()).to_string_lossy() -+ }); -+ -+ let id = c_str("id"); -+ let columns = [id.as_ptr(), ptr::null()]; -+ let scan_schema = |with_row_id: bool| { -+ let scanner = unsafe { lance_scanner_new(dataset, columns.as_ptr(), ptr::null()) }; -+ assert!(!scanner.is_null()); -+ if with_row_id { -+ assert_eq!(unsafe { lance_scanner_with_row_id(scanner, true) }, 0); -+ } -+ assert_eq!( -+ unsafe { lance_scanner_set_fts_query_context(scanner, context) }, -+ 0 -+ ); -+ let mut stream = FFI_ArrowArrayStream::empty(); -+ assert_eq!( -+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) }, -+ 0, -+ "{}", -+ unsafe { std::ffi::CStr::from_ptr(lance_last_error_message()).to_string_lossy() } -+ ); -+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream).unwrap() }; -+ let schema = reader.schema(); -+ let rows: usize = reader.map(|batch| batch.unwrap().num_rows()).sum(); -+ assert!(rows > 0); -+ unsafe { lance_scanner_close(scanner) }; -+ schema -+ }; -+ -+ let without_row_id = scan_schema(false); -+ assert_eq!(without_row_id.fields().len(), 2); -+ assert!(without_row_id.field_with_name("id").is_ok()); -+ assert!(without_row_id.field_with_name("_score").is_ok()); -+ assert!(without_row_id.field_with_name("_rowid").is_err()); -+ -+ let with_row_id = scan_schema(true); -+ assert_eq!(with_row_id.fields().len(), 3); -+ assert!(with_row_id.field_with_name("id").is_ok()); -+ assert!(with_row_id.field_with_name("_score").is_ok()); -+ assert!(with_row_id.field_with_name("_rowid").is_ok()); -+ -+ unsafe { lance_fts_query_context_close(context) }; -+ unsafe { lance_dataset_close(dataset) }; -+} -+ - #[test] - fn test_prepare_fts_query_index_only_allows_unindexed_fragment() { - let (_tmp, uri) = create_test_dataset(); - -From 6f0fae4cc51bf144685564b1d2e9f6f7afc71f8a Mon Sep 17 00:00:00 2001 -From: zhangstar333 -Date: Tue, 1 Sep 2026 12:34:05 +0800 -Subject: [PATCH 2/2] update - ---- - src/fts_query.rs | 2 - - src/scanner.rs | 253 ++++++++++++++++++++++++++++++++++---------- - tests/c_api_test.rs | 79 ++++++++++++++ - 3 files changed, 279 insertions(+), 55 deletions(-) - -diff --git a/src/fts_query.rs b/src/fts_query.rs -index e85ed99..cd194c7 100644 ---- a/src/fts_query.rs -+++ b/src/fts_query.rs -@@ -54,7 +54,6 @@ pub(crate) struct FtsQueryContextInner { - pub(crate) query: FullTextSearchQuery, - pub(crate) segments: Vec, - pub(crate) scorer: Arc, -- pub(crate) has_unindexed_fragments: bool, - } - - impl FtsQueryContextInner { -@@ -217,7 +216,6 @@ async fn prepare_fts_query_context( - query, - segments, - scorer, -- has_unindexed_fragments: !unindexed_fragment_ids.is_empty(), - }) - } - -diff --git a/src/scanner.rs b/src/scanner.rs -index f60f0c5..0c29b17 100644 ---- a/src/scanner.rs -+++ b/src/scanner.rs -@@ -12,13 +12,13 @@ use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker}; - - use arrow::ffi_stream::FFI_ArrowArrayStream; - use arrow_schema::SchemaRef; --use datafusion::physical_plan::ExecutionPlan; -+use datafusion::physical_plan::{ExecutionPlan, empty::EmptyExec}; - use futures::{FutureExt, Stream, StreamExt}; - use lance::Dataset; - use lance::dataset::scanner::{ - DatasetRecordBatchStream, ExecutionStatsCallback, ExecutionSummaryCounts, - }; --use lance::io::exec::fts::MatchQueryExec; -+use lance::io::exec::fts::{FlatMatchQueryExec, MatchQueryExec}; - use lance_core::Result; - use lance_index::scalar::FullTextSearchQuery; - use lance_io::stream::RecordBatchStream; -@@ -179,45 +179,6 @@ impl LanceScanner { - Ok(()) - } - -- /// Restrict an INDEX_ONLY prepared FTS scan to fragments covered by the -- /// selected committed segments. This is deliberately separate from -- /// `fast_search`: that option is scanner-wide, changes unrelated scalar -- /// index fallback behavior, and also forces `_rowid` into the output. -- fn apply_prepared_fts_fragment_filter( -- &self, -- scanner: &mut lance::dataset::scanner::Scanner, -- context: &FtsQueryContextInner, -- segments: &[IndexMetadata], -- ) -> Result<()> { -- if !context.has_unindexed_fragments { -- return Ok(()); -- } -- -- let mut selected_fragment_ids = std::collections::HashSet::new(); -- for segment in segments { -- let fragment_bitmap = segment.fragment_bitmap.as_ref().ok_or_else(|| { -- lance_core::Error::internal(format!( -- "prepared FTS segment {} lost its validated fragment coverage", -- segment.uuid -- )) -- })?; -- selected_fragment_ids.extend(fragment_bitmap.iter()); -- } -- -- let selected_fragments = self -- .dataset -- .get_fragments() -- .into_iter() -- .filter(|fragment| { -- u32::try_from(fragment.id()) -- .is_ok_and(|fragment_id| selected_fragment_ids.contains(&fragment_id)) -- }) -- .map(|fragment| fragment.metadata().clone()) -- .collect(); -- scanner.with_fragments(selected_fragments); -- Ok(()) -- } -- - fn apply_filter(&self, scanner: &mut lance::dataset::scanner::Scanner) -> Result<()> { - if let Some(substrait) = &self.substrait_filter { - scanner.filter_substrait(substrait)?; -@@ -321,7 +282,6 @@ impl LanceScanner { - let distributed_fts = if let Some(context) = &self.fts_context { - context.validate_dataset_identity(&self.dataset)?; - let segments = select_fts_segments(context, self.fts_index_segments.as_deref())?; -- self.apply_prepared_fts_fragment_filter(&mut scanner, context, &segments)?; - scanner.full_text_search(context.query.clone())?; - Some(PreparedFtsExecution { - context: Arc::clone(context), -@@ -361,14 +321,24 @@ impl PreparedScanner { - return self.scanner.try_into_stream().await; - }; - let plan = self.scanner.create_plan().await?; -- let (plan, replaced) = replace_match_query_exec( -+ let selected_segments_have_current_fragments = segments_have_current_fragments( -+ &distributed_fts.context.dataset, -+ &distributed_fts.segments, -+ )?; -+ let (plan, rewritten) = rewrite_prepared_fts_plan( - plan, - &distributed_fts.segments, - &distributed_fts.context.scorer, -+ selected_segments_have_current_fragments, - )?; -- if replaced != 1 { -+ if rewritten.match_query_execs > 1 -+ || rewritten.flat_match_query_execs > 1 -+ || rewritten.match_query_execs + rewritten.flat_match_query_execs == 0 -+ || (selected_segments_have_current_fragments && rewritten.match_query_execs != 1) -+ { - return Err(lance_core::Error::internal(format!( -- "expected exactly one MatchQueryExec in prepared FTS plan, replaced {replaced}" -+ "unexpected prepared FTS plan for selected segments with current fragment coverage {selected_segments_have_current_fragments}: rewrote {} MatchQueryExec node(s) and removed {} FlatMatchQueryExec node(s)", -+ rewritten.match_query_execs, rewritten.flat_match_query_execs - ))); - } - let stream = lance_datafusion::exec::execute_plan( -@@ -415,22 +385,81 @@ fn select_fts_segments( - Ok(selected) - } - --fn replace_match_query_exec( -+fn segments_have_current_fragments( -+ dataset: &lance::Dataset, -+ segments: &[IndexMetadata], -+) -> Result { -+ let current_fragment_ids = dataset -+ .get_fragments() -+ .into_iter() -+ .map(|fragment| { -+ u32::try_from(fragment.id()).map_err(|_| { -+ lance_core::Error::internal(format!( -+ "current fragment id {} exceeds the validated u32 FTS coverage range", -+ fragment.id() -+ )) -+ }) -+ }) -+ .collect::>>()?; -+ for segment in segments { -+ let fragment_bitmap = segment.fragment_bitmap.as_ref().ok_or_else(|| { -+ lance_core::Error::internal(format!( -+ "prepared FTS segment {} lost its validated fragment coverage", -+ segment.uuid -+ )) -+ })?; -+ if fragment_bitmap -+ .iter() -+ .any(|fragment_id| current_fragment_ids.contains(&fragment_id)) -+ { -+ return Ok(true); -+ } -+ } -+ Ok(false) -+} -+ -+#[derive(Default)] -+struct PreparedFtsPlanRewriteCounts { -+ match_query_execs: usize, -+ flat_match_query_execs: usize, -+} -+ -+fn rewrite_prepared_fts_plan( - plan: Arc, - segments: &[IndexMetadata], - scorer: &Arc, --) -> Result<(Arc, usize)> { -+ selected_segments_have_current_fragments: bool, -+) -> Result<(Arc, PreparedFtsPlanRewriteCounts)> { -+ // Lance's ordinary FTS planner adds a flat-search branch for fragments not -+ // covered by the logical index. A prepared INDEX_ONLY scan must omit that -+ // branch, but using Scanner::with_fragments to do so would turn an -+ // otherwise unfiltered index search into a full row-id prefilter scan. -+ if plan.downcast_ref::().is_some() { -+ return Ok(( -+ Arc::new(EmptyExec::new(plan.schema())), -+ PreparedFtsPlanRewriteCounts { -+ match_query_execs: 0, -+ flat_match_query_execs: 1, -+ }, -+ )); -+ } -+ - let children = plan.children(); -- let mut replaced = 0; -+ let mut rewritten = PreparedFtsPlanRewriteCounts::default(); - let rebuilt = if children.is_empty() { - plan - } else { - let mut new_children = Vec::with_capacity(children.len()); - for child in children { -- let (new_child, child_replaced) = -- replace_match_query_exec(Arc::clone(child), segments, scorer)?; -+ let (new_child, child_rewritten) = rewrite_prepared_fts_plan( -+ Arc::clone(child), -+ segments, -+ scorer, -+ selected_segments_have_current_fragments, -+ )?; - new_children.push(new_child); -- replaced += child_replaced; -+ rewritten.match_query_execs += child_rewritten.match_query_execs; -+ rewritten.flat_match_query_execs += child_rewritten.flat_match_query_execs; - } - plan.with_new_children(new_children).map_err(|error| { - lance_core::Error::internal(format!( -@@ -440,6 +469,10 @@ fn replace_match_query_exec( - }; - - if let Some(exec) = rebuilt.downcast_ref::() { -+ rewritten.match_query_execs += 1; -+ if !selected_segments_have_current_fragments { -+ return Ok((Arc::new(EmptyExec::new(rebuilt.schema())), rewritten)); -+ } - let replacement = MatchQueryExec::new_with_segments( - Arc::clone(exec.dataset()), - exec.query().clone(), -@@ -448,9 +481,9 @@ fn replace_match_query_exec( - segments.to_vec(), - ) - .with_base_scorer(Arc::clone(scorer)); -- return Ok((Arc::new(replacement), replaced + 1)); -+ return Ok((Arc::new(replacement), rewritten)); - } -- Ok((rebuilt, replaced)) -+ Ok((rebuilt, rewritten)) - } - - /// Type of a dynamically named scan metric. -@@ -2118,6 +2151,9 @@ mod tests { - use super::*; - use crate::dataset::{lance_dataset_close, lance_dataset_open}; - use crate::error::{lance_last_error_code, lance_last_error_message}; -+ use crate::fts_query::{ -+ LanceFtsCoverageMode, lance_dataset_prepare_fts_query, lance_fts_query_context_close, -+ }; - use std::ffi::{CStr, CString}; - use std::sync::atomic::{AtomicI32, AtomicUsize}; - use std::sync::{Barrier, mpsc}; -@@ -2125,6 +2161,9 @@ mod tests { - - use arrow_array::{Int32Array, RecordBatch, StringArray}; - use arrow_schema::{DataType, Field, Schema}; -+ use lance::index::DatasetIndexExt; -+ use lance::io::exec::PreFilterSource; -+ use lance_index::{IndexType, scalar::InvertedIndexParams}; - - /// Write a 3-row dataset to a tempdir, returning (tempdir, uri). - fn create_test_dataset() -> (tempfile::TempDir, String) { -@@ -2169,6 +2208,114 @@ mod tests { - .store(true, Ordering::SeqCst); - } - -+ fn prepared_fts_plan_shape(plan: &Arc) -> (usize, usize, usize) { -+ let mut match_query_execs = 0; -+ let mut flat_match_query_execs = 0; -+ let mut filtered_row_id_prefilters = 0; -+ if let Some(exec) = plan.downcast_ref::() { -+ match_query_execs += 1; -+ if matches!(exec.prefilter_source(), PreFilterSource::FilteredRowIds(_)) { -+ filtered_row_id_prefilters += 1; -+ } -+ } -+ if plan.downcast_ref::().is_some() { -+ flat_match_query_execs += 1; -+ } -+ for child in plan.children() { -+ let (child_match, child_flat, child_filtered) = -+ prepared_fts_plan_shape(&Arc::clone(child)); -+ match_query_execs += child_match; -+ flat_match_query_execs += child_flat; -+ filtered_row_id_prefilters += child_filtered; -+ } -+ ( -+ match_query_execs, -+ flat_match_query_execs, -+ filtered_row_id_prefilters, -+ ) -+ } -+ -+ #[test] -+ fn prepared_fts_index_only_plan_does_not_scan_indexed_fragment_row_ids() { -+ let (_tmp, uri) = create_test_dataset(); -+ block_on(async { -+ let mut dataset = Dataset::open(&uri).await.unwrap(); -+ dataset -+ .create_index( -+ &["name"], -+ IndexType::Inverted, -+ None, -+ &InvertedIndexParams::default(), -+ false, -+ ) -+ .await -+ .unwrap(); -+ -+ let schema = Arc::new(Schema::new(vec![ -+ Field::new("id", DataType::Int32, false), -+ Field::new("name", DataType::Utf8, true), -+ ])); -+ let batch = RecordBatch::try_new( -+ schema.clone(), -+ vec![ -+ Arc::new(Int32Array::from(vec![4])), -+ Arc::new(StringArray::from(vec!["a"])), -+ ], -+ ) -+ .unwrap(); -+ dataset -+ .append( -+ arrow::record_batch::RecordBatchIterator::new(vec![Ok(batch)], schema), -+ None, -+ ) -+ .await -+ .unwrap(); -+ }); -+ -+ let (dataset, scanner) = open_dataset_and_scanner(&uri); -+ let column = CString::new("name").unwrap(); -+ let query = CString::new("a").unwrap(); -+ let context = unsafe { -+ lance_dataset_prepare_fts_query( -+ dataset, -+ column.as_ptr(), -+ query.as_ptr(), -+ 0, -+ LanceFtsCoverageMode::IndexOnly as i32, -+ ) -+ }; -+ assert!(!context.is_null()); -+ assert_eq!( -+ unsafe { lance_scanner_set_fts_query_context(scanner, context) }, -+ 0 -+ ); -+ -+ let prepared = unsafe { &*scanner }.build_scanner().unwrap(); -+ let distributed = prepared.distributed_fts.as_ref().unwrap(); -+ let segments = distributed.segments.clone(); -+ let scorer = Arc::clone(&distributed.context.scorer); -+ let plan = block_on(prepared.scanner.create_plan()).unwrap(); -+ assert_eq!( -+ prepared_fts_plan_shape(&plan), -+ (1, 1, 0), -+ "an unfiltered prepared FTS plan must not materialize selected fragment row IDs" -+ ); -+ -+ let has_current_fragments = -+ segments_have_current_fragments(&distributed.context.dataset, &segments).unwrap(); -+ let (rewritten, counts) = -+ rewrite_prepared_fts_plan(plan, &segments, &scorer, has_current_fragments).unwrap(); -+ assert_eq!(counts.match_query_execs, 1); -+ assert_eq!(counts.flat_match_query_execs, 1); -+ assert_eq!(prepared_fts_plan_shape(&rewritten), (1, 0, 0)); -+ -+ unsafe { -+ lance_scanner_close(scanner); -+ lance_fts_query_context_close(context); -+ lance_dataset_close(dataset); -+ } -+ } -+ - /// Assert the pending thread-local error is `Panic` carrying the poison - /// message; consumes it so the next assertion starts from a clean slate. - fn assert_poison_error_pending() { -diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs -index 3627c4e..8805764 100644 ---- a/tests/c_api_test.rs -+++ b/tests/c_api_test.rs -@@ -5943,6 +5943,85 @@ fn test_prepare_fts_query_index_only_allows_unindexed_fragment() { - unsafe { lance_dataset_close(dataset) }; - } - -+#[test] -+fn test_prepared_fts_index_only_empty_segment_returns_empty_shard() { -+ use lance::index::DatasetIndexExt; -+ use lance_index::{IndexType, scalar::InvertedIndexParams}; -+ -+ let (_tmp, uri) = create_test_dataset(); -+ lance_c::runtime::block_on(async { -+ let mut dataset = Dataset::open(&uri).await.unwrap(); -+ let params = InvertedIndexParams::default(); -+ dataset -+ .create_index_builder(&["name"], IndexType::Inverted, ¶ms) -+ .name("empty_name_fts".to_string()) -+ .train(false) -+ .await -+ .unwrap(); -+ let segments = dataset -+ .load_indices_by_name("empty_name_fts") -+ .await -+ .unwrap(); -+ assert_eq!(segments.len(), 1); -+ assert!( -+ segments[0] -+ .fragment_bitmap -+ .as_ref() -+ .is_some_and(|fragment_bitmap| fragment_bitmap.is_empty()) -+ ); -+ }); -+ -+ let uri_c = c_str(&uri); -+ let column = c_str("name"); -+ let query = c_str("alice"); -+ let dataset = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; -+ let context = unsafe { -+ lance_dataset_prepare_fts_query( -+ dataset, -+ column.as_ptr(), -+ query.as_ptr(), -+ 0, -+ LanceFtsCoverageMode::IndexOnly as i32, -+ ) -+ }; -+ assert!(!context.is_null(), "{}", unsafe { -+ std::ffi::CStr::from_ptr(lance_last_error_message()).to_string_lossy() -+ }); -+ let segment_uuids = load_fts_segment_uuids(&uri, "name"); -+ assert_eq!(segment_uuids.len(), 1); -+ -+ let scanner = unsafe { lance_scanner_new(dataset, ptr::null(), ptr::null()) }; -+ assert_eq!( -+ unsafe { lance_scanner_set_fts_query_context(scanner, context) }, -+ 0 -+ ); -+ assert_eq!( -+ unsafe { -+ lance_scanner_set_fts_index_segments( -+ scanner, -+ segment_uuids.as_ptr().cast::(), -+ segment_uuids.len(), -+ ) -+ }, -+ 0 -+ ); -+ -+ let mut stream = FFI_ArrowArrayStream::empty(); -+ assert_eq!( -+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) }, -+ 0, -+ "{}", -+ unsafe { std::ffi::CStr::from_ptr(lance_last_error_message()).to_string_lossy() } -+ ); -+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream).unwrap() }; -+ let total_rows: usize = reader.map(|batch| batch.unwrap().num_rows()).sum(); -+ assert_eq!(total_rows, 0); -+ -+ unsafe { lance_scanner_close(scanner) }; -+ unsafe { lance_fts_query_context_close(context) }; -+ unsafe { lance_dataset_close(dataset) }; -+} -+ - #[test] - fn test_prepared_fts_global_scorer_is_shared_across_segment_splits() { - use lance::index::DatasetIndexExt; diff --git a/thirdparty/patches/lance-c-0.1.9-pr-73.patch b/thirdparty/patches/lance-c-0.1.9-pr-73.patch new file mode 100644 index 00000000000000..7d5eb45ad6ce7d --- /dev/null +++ b/thirdparty/patches/lance-c-0.1.9-pr-73.patch @@ -0,0 +1,2317 @@ +From a4c71309ddb76ad79808e3e8f7797bcd9bfc174a Mon Sep 17 00:00:00 2001 +From: zhangstar333 +Date: Thu, 3 Sep 2026 13:00:43 +0800 +Subject: [PATCH] foyer + +--- + Cargo.lock | 205 ++++++ + Cargo.toml | 4 + + README.md | 22 + + include/lance/lance.h | 63 ++ + include/lance/lance.hpp | 29 + + src/data_cache.rs | 68 ++ + src/dataset.rs | 11 + + src/foyer_data_cache.rs | 1215 ++++++++++++++++++++++++++++++++++++ + src/lib.rs | 4 + + src/restore.rs | 8 + + src/session.rs | 13 +- + src/writer.rs | 1 + + tests/c_api_test.rs | 221 +++++++ + tests/cpp/test_c_api.c | 32 + + tests/cpp/test_cpp_api.cpp | 23 + + 15 files changed, 1918 insertions(+), 1 deletion(-) + create mode 100644 src/data_cache.rs + create mode 100644 src/foyer_data_cache.rs + +diff --git a/Cargo.lock b/Cargo.lock +index 60c1caf..85536f5 100644 +--- a/Cargo.lock ++++ b/Cargo.lock +@@ -434,6 +434,16 @@ dependencies = [ + "loom", + ] + ++[[package]] ++name = "asyncband" ++version = "0.6.7" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "94a214ba60d6231afd0e805e3c27c45a1626d9debaa5a5061c45a1ea1b2f1ed0" ++dependencies = [ ++ "hashbrown 0.17.1", ++ "slab", ++] ++ + [[package]] + name = "atoi" + version = "2.0.0" +@@ -1267,6 +1277,17 @@ version = "0.8.7" + source = "registry+https://github.com/rust-lang/crates.io-index" + checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + ++[[package]] ++name = "core_affinity" ++version = "0.8.3" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "a034b3a7b624016c6e13f5df875747cc25f884156aad2abd12b6c46797971342" ++dependencies = [ ++ "libc", ++ "num_cpus", ++ "winapi", ++] ++ + [[package]] + name = "countio" + version = "0.3.0" +@@ -2112,6 +2133,12 @@ dependencies = [ + "url", + ] + ++[[package]] ++name = "datasketches" ++version = "0.3.0" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "46c4cf71a36b46dcfc00e5014c0c20ccad2b1b6a008304d7d57d2749b2d41b3d" ++ + [[package]] + name = "der" + version = "0.7.10" +@@ -2298,6 +2325,16 @@ version = "0.2.3" + source = "registry+https://github.com/rust-lang/crates.io-index" + checksum = "f8eb564c5c7423d25c886fb561d1e4ee69f72354d16918afa32c08811f6b6a55" + ++[[package]] ++name = "fastant" ++version = "0.1.11" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "2e825441bfb2d831c47c97d05821552db8832479f44c571b97fededbf0099c07" ++dependencies = [ ++ "small_ctor", ++ "web-time", ++] ++ + [[package]] + name = "fastrand" + version = "2.3.0" +@@ -2369,6 +2406,127 @@ dependencies = [ + "percent-encoding", + ] + ++[[package]] ++name = "foyer" ++version = "0.22.4" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "cab5c4bac30455a0dbc4c858436fb440d51bc6543daafbdf35c5e6ca94c0afd7" ++dependencies = [ ++ "anyhow", ++ "asyncband", ++ "equivalent", ++ "foyer-common", ++ "foyer-memory", ++ "foyer-storage", ++ "foyer-tokio", ++ "futures-util", ++ "mixtrics", ++ "pin-project", ++ "serde", ++ "tracing", ++] ++ ++[[package]] ++name = "foyer-common" ++version = "0.22.4" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "e3634c6f3da978b6cae98d54367a2342041d3a4786b20c0384164cc7fce94787" ++dependencies = [ ++ "anyhow", ++ "bytes", ++ "cfg-if 1.0.4", ++ "foyer-tokio", ++ "mixtrics", ++ "parking_lot", ++ "pin-project", ++ "twox-hash", ++] ++ ++[[package]] ++name = "foyer-intrusive-collections" ++version = "0.10.0-dev" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "6e4fee46bea69e0596130e3210e65d3424e0ac1e6df3bde6636304bdf1ca4a3b" ++dependencies = [ ++ "memoffset", ++] ++ ++[[package]] ++name = "foyer-memory" ++version = "0.22.4" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "5349a61af676b3275bfef7a5fceceeadf6b0a7dd9ec14ba0edd78e1ff771d101" ++dependencies = [ ++ "anyhow", ++ "asyncband", ++ "bitflags", ++ "datasketches", ++ "equivalent", ++ "foyer-common", ++ "foyer-intrusive-collections", ++ "foyer-tokio", ++ "futures-util", ++ "hashbrown 0.17.1", ++ "itertools 0.15.0", ++ "mixtrics", ++ "parking_lot", ++ "paste", ++ "pin-project", ++ "serde", ++ "tracing", ++] ++ ++[[package]] ++name = "foyer-storage" ++version = "0.22.4" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "49cfcce3c10a1f2ac65bfcf0bd85501462d454c594c8a16a4a27cce773599381" ++dependencies = [ ++ "allocator-api2", ++ "anyhow", ++ "asyncband", ++ "bytes", ++ "core_affinity", ++ "equivalent", ++ "fastant", ++ "foyer-common", ++ "foyer-memory", ++ "foyer-tokio", ++ "fs4", ++ "futures-core", ++ "futures-util", ++ "hashbrown 0.17.1", ++ "io-uring", ++ "itertools 0.15.0", ++ "libc", ++ "lz4", ++ "parking_lot", ++ "pin-project", ++ "rand 0.10.1", ++ "tracing", ++ "twox-hash", ++ "zstd", ++] ++ ++[[package]] ++name = "foyer-tokio" ++version = "0.22.4" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "6b7315103199f3415a010befd6dd5a1c8e7082fbc49c0194931e65629995d9d1" ++dependencies = [ ++ "tokio", ++] ++ ++[[package]] ++name = "fs4" ++version = "0.13.1" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "8640e34b88f7652208ce9e88b1a37a2ae95227d84abec377ccd3c5cfeb141ed4" ++dependencies = [ ++ "rustix", ++ "windows-sys 0.59.0", ++] ++ + [[package]] + name = "fs_extra" + version = "1.3.0" +@@ -3392,6 +3550,15 @@ dependencies = [ + "either", + ] + ++[[package]] ++name = "itertools" ++version = "0.15.0" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "8b4baf93f58d4425749ca49a51c50ebab072c5df6994d08fed93541c331481dc" ++dependencies = [ ++ "either", ++] ++ + [[package]] + name = "itoa" + version = "1.0.18" +@@ -3703,8 +3870,11 @@ dependencies = [ + "arrow", + "arrow-array", + "arrow-schema", ++ "async-trait", ++ "bytes", + "chrono", + "datafusion", ++ "foyer", + "futures", + "half", + "lance", +@@ -3718,6 +3888,7 @@ dependencies = [ + "lance-table", + "libc", + "log", ++ "object_store", + "pin-project", + "prost", + "snafu", +@@ -4399,6 +4570,15 @@ version = "2.8.0" + source = "registry+https://github.com/rust-lang/crates.io-index" + checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" + ++[[package]] ++name = "memoffset" ++version = "0.9.1" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" ++dependencies = [ ++ "autocfg", ++] ++ + [[package]] + name = "mime" + version = "0.3.17" +@@ -4436,6 +4616,16 @@ dependencies = [ + "windows-sys 0.61.2", + ] + ++[[package]] ++name = "mixtrics" ++version = "0.2.5" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "2c46b5adfb7a3ae4996d327a5bdc90e78fec025806dd312bdbe6f07a755e0ec9" ++dependencies = [ ++ "itertools 0.15.0", ++ "parking_lot", ++] ++ + [[package]] + name = "moka" + version = "0.12.15" +@@ -6547,6 +6737,12 @@ version = "0.4.12" + source = "registry+https://github.com/rust-lang/crates.io-index" + checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + ++[[package]] ++name = "small_ctor" ++version = "0.1.2" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "88414a5ca1f85d82cc34471e975f0f74f6aa54c40f062efa42c0080e7f763f81" ++ + [[package]] + name = "smallvec" + version = "1.15.1" +@@ -7831,6 +8027,15 @@ dependencies = [ + "windows-targets 0.52.6", + ] + ++[[package]] ++name = "windows-sys" ++version = "0.59.0" ++source = "registry+https://github.com/rust-lang/crates.io-index" ++checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" ++dependencies = [ ++ "windows-targets 0.52.6", ++] ++ + [[package]] + name = "windows-sys" + version = "0.60.2" +diff --git a/Cargo.toml b/Cargo.toml +index d072a5d..342928c 100644 +--- a/Cargo.toml ++++ b/Cargo.toml +@@ -30,6 +30,8 @@ datafusion = { version = "54.0.0", default-features = false } + arrow = { version = "58.0.0", features = ["prettyprint", "ffi"] } + arrow-array = "58.0.0" + arrow-schema = "58.0.0" ++async-trait = "0.1" ++bytes = "1" + # Direct to name `chrono::TimeDelta` (the field type of lance's public + # `AutoCleanupParams`) and `chrono::DateTime`/`Utc` (index metadata + # timestamps); already in the graph transitively via lance. +@@ -37,8 +39,10 @@ chrono = { version = "0.4", default-features = false } + half = "2" + tokio = { version = "1", features = ["rt-multi-thread", "sync"] } + futures = "0.3" ++foyer = "=0.22.4" + log = "0.4" + libc = "0.2" ++object_store = "0.13.2" + pin-project = "1.0" + prost = "0.14" + snafu = "0.9" +diff --git a/README.md b/README.md +index 2056671..d9a5f2b 100644 +--- a/README.md ++++ b/README.md +@@ -68,6 +68,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] | 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 | ++| [x] | Data-file cache | Optional Foyer memory/disk cache for immutable `data/*.lance` reads | + + ## Building + +@@ -197,6 +198,27 @@ auto ds = lance::Dataset::open_with_session(session, "data.lance"); + auto stats = session.cache_stats(); + ``` + ++To add a process-local memory/disk cache for remote Lance data-file reads, ++create the session with Foyer configuration. The cache is deliberately narrow: ++whole-object, single-range, and batched range reads of direct `data/*.lance` ++children are cached. Conditional and versioned reads, plus manifests, deletion ++files, and index files, keep using Lance's normal paths. Use one shared session ++for datasets that share the cache directory. ++ ++```cpp ++lance::DataCacheOptions data_cache{ ++ "/var/cache/my-service/lance", ++ 512ULL * 1024 * 1024, // memory tier ++ 100ULL * 1024 * 1024 * 1024, // disk tier ++ 1ULL * 1024 * 1024, // range-cache block ++}; ++lance::Session session( ++ 6ULL * 1024 * 1024 * 1024, ++ 1ULL * 1024 * 1024 * 1024, ++ data_cache); ++auto ds = lance::Dataset::open_with_session(session, "s3://bucket/data.lance"); ++``` ++ + ### Open at a specific version + + `lance_dataset_open` takes a `version` argument — `0` means the latest, any +diff --git a/include/lance/lance.h b/include/lance/lance.h +index 3bf291f..e4c593e 100644 +--- a/include/lance/lance.h ++++ b/include/lance/lance.h +@@ -206,6 +206,36 @@ typedef struct LanceSessionCacheStats { + uint64_t metadata_cache_size_bytes; + } LanceSessionCacheStats; + ++/** ++ * Configuration for the optional Foyer cache of immutable Lance data files. ++ * ++ * Whole-object, single-range, and batched range reads of direct ++ * `data/.lance` children are cached. Conditional and versioned reads, ++ * plus metadata, deletion files, and index files, continue to use Lance's normal ++ * paths. ++ */ ++typedef struct LanceDataCacheOptions { ++ const char* directory; ++ /** Maximum raw data bytes retained by Foyer's in-memory tier. */ ++ uint64_t memory_capacity_bytes; ++ /** Maximum bytes allocated to Foyer's disk tier. */ ++ uint64_t disk_capacity_bytes; ++ /** Data-file range cache unit, in bytes. */ ++ uint64_t read_block_size_bytes; ++} LanceDataCacheOptions; ++ ++/** ++ * Cumulative Foyer data-cache statistics for one opened dataset handle. ++ * ++ * Successful reads are accumulated. Both fields measure bytes returned to the ++ * dataset reader. Their sum is the logical data-file range bytes observed by ++ * the Foyer wrapper; block-aligned origin read amplification is not included. ++ */ ++typedef struct LanceDataCacheStatistics { ++ uint64_t bytes_read_from_cache; ++ uint64_t bytes_read_from_remote; ++} LanceDataCacheStatistics; ++ + /** + * Create a session that can share metadata and index caches across datasets. + * +@@ -217,6 +247,25 @@ LanceSession* lance_session_new( + uint64_t metadata_cache_size_bytes + ); + ++/** ++ * Create a shared Lance session with a Foyer data-file cache. ++ * ++ * `data_cache_options` and its `directory` field must not be NULL. The cache ++ * directory and all capacities are process configuration and remain owned by ++ * the caller; their values are copied during this call. ++ * ++ * `read_block_size_bytes` must be a non-zero multiple of 4096. The memory ++ * capacity must hold at least one read block. The disk capacity must be a ++ * multiple of 4096 and hold at least two read blocks. ++ * ++ * @return Session handle, or NULL on error ++ */ ++LanceSession* lance_session_new_with_data_cache( ++ uint64_t index_cache_size_bytes, ++ uint64_t metadata_cache_size_bytes, ++ const LanceDataCacheOptions* data_cache_options ++); ++ + /** + * Close a session handle. Safe to call with NULL. Datasets previously opened + * with the session remain valid and retain the shared cache state. +@@ -273,6 +322,20 @@ LanceDataset* lance_dataset_open_with_session( + const LanceSession* session + ); + ++/** ++ * Copy this dataset handle's cumulative data-cache statistics. ++ * ++ * A dataset not opened with a data cache reports all-zero statistics. The ++ * snapshot belongs only to this dataset handle; the underlying cache may ++ * still be shared by other datasets through a session. ++ * ++ * @return 0 on success, -1 on error ++ */ ++int32_t lance_dataset_get_data_cache_statistics( ++ const LanceDataset* dataset, ++ LanceDataCacheStatistics* out_statistics ++); ++ + /** Close and free a dataset handle. Safe to call with NULL. */ + void lance_dataset_close(LanceDataset* dataset); + +diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp +index 6cf245f..c1102e4 100644 +--- a/include/lance/lance.hpp ++++ b/include/lance/lance.hpp +@@ -171,6 +171,13 @@ struct SqlColumn { + + // ─── Shared Session ────────────────────────────────────────────────────────── + ++struct DataCacheOptions { ++ std::string directory; ++ uint64_t memory_capacity_bytes; ++ uint64_t disk_capacity_bytes; ++ uint64_t read_block_size_bytes; ++}; ++ + class Session { + Handle handle_; + +@@ -180,6 +187,21 @@ class Session { + if (!handle_) check_error(); + } + ++ Session(uint64_t index_cache_size_bytes, ++ uint64_t metadata_cache_size_bytes, ++ const DataCacheOptions& data_cache_options) { ++ LanceDataCacheOptions options{ ++ data_cache_options.directory.c_str(), ++ data_cache_options.memory_capacity_bytes, ++ data_cache_options.disk_capacity_bytes, ++ data_cache_options.read_block_size_bytes, ++ }; ++ handle_ = Handle( ++ lance_session_new_with_data_cache( ++ index_cache_size_bytes, metadata_cache_size_bytes, &options)); ++ if (!handle_) check_error(); ++ } ++ + LanceSessionCacheStats cache_stats() const { + LanceSessionCacheStats stats{}; + if (lance_session_get_cache_stats(handle_.get(), &stats) != 0) +@@ -260,6 +282,13 @@ class Dataset { + return Dataset(ds); + } + ++ LanceDataCacheStatistics data_cache_statistics() const { ++ LanceDataCacheStatistics statistics{}; ++ if (lance_dataset_get_data_cache_statistics(handle_.get(), &statistics) != 0) ++ check_error(); ++ return statistics; ++ } ++ + /// Write an Arrow record batch stream to a Lance dataset and return the + /// open dataset at the committed version. + /// +diff --git a/src/data_cache.rs b/src/data_cache.rs +new file mode 100644 +index 0000000..430b4c8 +--- /dev/null ++++ b/src/data_cache.rs +@@ -0,0 +1,68 @@ ++// SPDX-License-Identifier: Apache-2.0 ++// SPDX-FileCopyrightText: Copyright The Lance Authors ++ ++//! Private bridge between a shared data-cache backend and dataset handles. ++ ++use std::fmt::Debug; ++use std::sync::Arc; ++ ++use lance::Dataset; ++use lance_core::Result; ++ ++use crate::dataset::LanceDataset; ++use crate::error::ffi_try; ++ ++/// Data-cache statistics owned by one opened dataset. ++#[repr(C)] ++#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] ++pub struct LanceDataCacheStatistics { ++ /// Requested bytes returned from usable data-cache entries. ++ pub bytes_read_from_cache: u64, ++ /// Requested bytes returned after a data-cache miss or fallback. ++ pub bytes_read_from_remote: u64, ++} ++ ++pub(crate) trait DatasetDataCache: Debug + Send + Sync { ++ fn snapshot(&self) -> LanceDataCacheStatistics; ++ ++ fn attach_fresh(&self, dataset: Dataset) -> (Dataset, Arc); ++} ++ ++pub(crate) trait DataCacheFactory: Debug + Send + Sync { ++ fn attach(&self, dataset: Dataset) -> (Dataset, Arc); ++} ++ ++/// Copy this dataset handle's cumulative data-cache statistics into ++/// `out_statistics`. ++/// ++/// A dataset not opened with a data cache reports all-zero statistics. ++#[unsafe(no_mangle)] ++pub unsafe extern "C" fn lance_dataset_get_data_cache_statistics( ++ dataset: *const LanceDataset, ++ out_statistics: *mut LanceDataCacheStatistics, ++) -> i32 { ++ ffi_try!( ++ unsafe { dataset_get_data_cache_statistics_inner(dataset, out_statistics) }, ++ neg ++ ) ++} ++ ++unsafe fn dataset_get_data_cache_statistics_inner( ++ dataset: *const LanceDataset, ++ out_statistics: *mut LanceDataCacheStatistics, ++) -> Result { ++ if dataset.is_null() || out_statistics.is_null() { ++ return Err(lance_core::Error::invalid_input_source( ++ "dataset and out_statistics must not be NULL".into(), ++ )); ++ } ++ let dataset = unsafe { &*dataset }; ++ let statistics = dataset ++ .data_cache ++ .as_ref() ++ .map_or_else(LanceDataCacheStatistics::default, |cache| cache.snapshot()); ++ unsafe { ++ std::ptr::write_unaligned(out_statistics, statistics); ++ } ++ Ok(0) ++} +diff --git a/src/dataset.rs b/src/dataset.rs +index cc1f87c..76fd39e 100644 +--- a/src/dataset.rs ++++ b/src/dataset.rs +@@ -14,6 +14,7 @@ use lance::Dataset; + use lance::dataset::builder::DatasetBuilder; + use lance_core::Result; + ++use crate::data_cache::DatasetDataCache; + use crate::error::{ffi_try, swallow_unwind}; + use crate::helpers; + use crate::runtime::block_on; +@@ -23,6 +24,7 @@ use crate::stream_guard::guarded_ffi_stream_from_reader; + /// Opaque handle representing an opened Lance dataset. + pub struct LanceDataset { + pub(crate) inner: RwLock>, ++ pub(crate) data_cache: Option>, + } + + impl LanceDataset { +@@ -182,8 +184,16 @@ unsafe fn open_dataset_inner( + } + + let dataset = block_on(builder.load())?; ++ let (dataset, data_cache) = ++ if let Some(factory) = session.and_then(|session| session.data_cache_factory.clone()) { ++ let (dataset, data_cache) = factory.attach(dataset); ++ (dataset, Some(data_cache)) ++ } else { ++ (dataset, None) ++ }; + let handle = LanceDataset { + inner: RwLock::new(Arc::new(dataset)), ++ data_cache, + }; + Ok(Box::into_raw(Box::new(handle))) + } +@@ -519,6 +529,7 @@ mod tests { + .unwrap(); + let handle = LanceDataset { + inner: RwLock::new(Arc::new(dataset)), ++ data_cache: None, + }; + (tmp, handle) + } +diff --git a/src/foyer_data_cache.rs b/src/foyer_data_cache.rs +new file mode 100644 +index 0000000..9a10e43 +--- /dev/null ++++ b/src/foyer_data_cache.rs +@@ -0,0 +1,1215 @@ ++// SPDX-License-Identifier: Apache-2.0 ++// SPDX-FileCopyrightText: Copyright The Lance Authors ++ ++//! Foyer-backed cache for immutable Lance data-file reads. ++ ++use std::collections::{BTreeMap, HashMap}; ++use std::ffi::c_char; ++use std::fmt::{Debug, Display, Formatter}; ++use std::ops::Range; ++use std::path::Path as FsPath; ++use std::sync::atomic::{AtomicU64, Ordering}; ++use std::sync::{Arc, Mutex, Weak}; ++ ++use async_trait::async_trait; ++use bytes::{Bytes, BytesMut}; ++use foyer::{ ++ BlockEngineConfig, DeviceBuilder, FsDeviceBuilder, HybridCache, HybridCacheBuilder, ++ HybridCachePolicy, PsyncIoEngineConfig, ++}; ++use futures::stream::BoxStream; ++use lance_io::object_store::WrappingObjectStore; ++use object_store::path::Path; ++use object_store::{ ++ CopyOptions, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload, ObjectMeta, ++ ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, ++ RenameOptions, Result, ++}; ++ ++use crate::data_cache::{DataCacheFactory, DatasetDataCache, LanceDataCacheStatistics}; ++use crate::error::ffi_try; ++use crate::helpers; ++use crate::runtime::block_on; ++use crate::session::{LanceSession, session_new_with_data_cache_factory}; ++ ++const CACHE_KEY_VERSION: &str = "lance-data-v1"; ++const FOYER_PAGE_SIZE: usize = 4096; ++ ++/// Configuration for the optional Foyer data-file cache. ++#[repr(C)] ++#[derive(Clone, Copy, Debug)] ++pub struct LanceDataCacheOptions { ++ /// UTF-8 directory used by Foyer for persistent cache storage. ++ pub directory: *const c_char, ++ /// Maximum data bytes retained by Foyer's in-memory tier. ++ pub memory_capacity_bytes: u64, ++ /// Maximum bytes retained by Foyer's disk tier. ++ pub disk_capacity_bytes: u64, ++ /// Read/cache unit. Must be a non-zero multiple of 4096. ++ pub read_block_size_bytes: u64, ++} ++ ++/// Create a shared Lance session with a Foyer cache for immutable data-file ++/// whole-object, single-range, and batched range reads. ++/// ++/// `data_cache_options` and its `directory` field must not be NULL. The ++/// pointed-to values are copied before this function returns. ++#[unsafe(no_mangle)] ++pub unsafe extern "C" fn lance_session_new_with_data_cache( ++ index_cache_size_bytes: u64, ++ metadata_cache_size_bytes: u64, ++ data_cache_options: *const LanceDataCacheOptions, ++) -> *mut LanceSession { ++ ffi_try!( ++ unsafe { ++ session_new_with_data_cache_inner( ++ index_cache_size_bytes, ++ metadata_cache_size_bytes, ++ data_cache_options, ++ ) ++ }, ++ null ++ ) ++} ++ ++unsafe fn session_new_with_data_cache_inner( ++ index_cache_size_bytes: u64, ++ metadata_cache_size_bytes: u64, ++ data_cache_options: *const LanceDataCacheOptions, ++) -> lance_core::Result<*mut LanceSession> { ++ if data_cache_options.is_null() { ++ return Err(lance_core::Error::invalid_input_source( ++ "data_cache_options must not be NULL".into(), ++ )); ++ } ++ let options = unsafe { &*data_cache_options }; ++ let directory = unsafe { helpers::parse_c_string(options.directory)? }.ok_or_else(|| { ++ lance_core::Error::invalid_input_source( ++ "data_cache_options.directory must not be NULL".into(), ++ ) ++ })?; ++ if directory.is_empty() { ++ return Err(lance_core::Error::invalid_input_source( ++ "data_cache_options.directory must not be empty".into(), ++ )); ++ } ++ ++ let memory_capacity = u64_to_usize(options.memory_capacity_bytes, "memory_capacity_bytes")?; ++ let disk_capacity = u64_to_usize(options.disk_capacity_bytes, "disk_capacity_bytes")?; ++ let read_block_size = u64_to_usize(options.read_block_size_bytes, "read_block_size_bytes")?; ++ validate_data_cache_sizes(memory_capacity, disk_capacity, read_block_size)?; ++ ++ let data_cache = block_on(FoyerDataCache::try_new( ++ FsPath::new(&directory), ++ memory_capacity, ++ disk_capacity, ++ read_block_size, ++ )) ++ .map_err(|error| { ++ lance_core::Error::io(format!( ++ "failed to initialize Foyer data cache at {directory:?}: {error}" ++ )) ++ })?; ++ session_new_with_data_cache_factory( ++ index_cache_size_bytes, ++ metadata_cache_size_bytes, ++ Some(Arc::new(data_cache)), ++ ) ++} ++ ++fn validate_data_cache_sizes( ++ memory_capacity: usize, ++ disk_capacity: usize, ++ read_block_size: usize, ++) -> lance_core::Result<()> { ++ if read_block_size == 0 || !read_block_size.is_multiple_of(FOYER_PAGE_SIZE) { ++ return Err(lance_core::Error::invalid_input_source( ++ format!( ++ "read_block_size_bytes={read_block_size} must be a non-zero multiple of {FOYER_PAGE_SIZE}" ++ ) ++ .into(), ++ )); ++ } ++ if memory_capacity < read_block_size { ++ return Err(lance_core::Error::invalid_input_source( ++ format!( ++ "memory_capacity_bytes={memory_capacity} must be at least read_block_size_bytes={read_block_size}" ++ ) ++ .into(), ++ )); ++ } ++ let minimum_disk_capacity = read_block_size.checked_mul(2).ok_or_else(|| { ++ lance_core::Error::invalid_input_source( ++ format!("read_block_size_bytes={read_block_size} is too large").into(), ++ ) ++ })?; ++ if !disk_capacity.is_multiple_of(FOYER_PAGE_SIZE) || disk_capacity < minimum_disk_capacity { ++ return Err(lance_core::Error::invalid_input_source( ++ format!( ++ "disk_capacity_bytes={disk_capacity} must be a multiple of {FOYER_PAGE_SIZE} and at least twice read_block_size_bytes={read_block_size}" ++ ) ++ .into(), ++ )); ++ } ++ Ok(()) ++} ++ ++fn u64_to_usize(value: u64, field: &'static str) -> lance_core::Result { ++ usize::try_from(value).map_err(|_| { ++ lance_core::Error::invalid_input_source( ++ format!("{field}={value} exceeds usize::MAX on this target").into(), ++ ) ++ }) ++} ++ ++/// Process-local owner of a Foyer hybrid cache. ++#[derive(Clone)] ++pub(crate) struct FoyerDataCache { ++ cache: HybridCache, ++ read_block_size: usize, ++ wrapped_stores: Arc>>, ++} ++ ++struct WrappedStore { ++ wrapper: Weak, ++ origin: Weak, ++} ++ ++impl Debug for FoyerDataCache { ++ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { ++ f.debug_struct("FoyerDataCache") ++ .field("read_block_size", &self.read_block_size) ++ .finish_non_exhaustive() ++ } ++} ++ ++impl FoyerDataCache { ++ pub(crate) async fn try_new( ++ directory: &FsPath, ++ memory_capacity: usize, ++ disk_capacity: usize, ++ read_block_size: usize, ++ ) -> std::result::Result { ++ let engine_block_size = read_block_size ++ .checked_mul(2) ++ .ok_or_else(|| foyer::Error::new(foyer::ErrorKind::Config, "read block size overflow"))? ++ .max(FOYER_PAGE_SIZE); ++ let memory_shards = (memory_capacity / read_block_size).clamp(1, 8); ++ let device = FsDeviceBuilder::new(directory) ++ .with_capacity(disk_capacity) ++ .build()?; ++ let engine = BlockEngineConfig::new(device).with_block_size(engine_block_size); ++ let cache = HybridCacheBuilder::new() ++ .with_name("lance_data") ++ .with_policy(HybridCachePolicy::WriteOnInsertion) ++ // Entries are already sent to storage on insertion. Avoid making ++ // the last Dataset drop wait for a final full-memory flush. ++ .with_flush_on_close(false) ++ .memory(memory_capacity) ++ .with_shards(memory_shards) ++ .with_weighter(|_key: &String, value: &Bytes| value.len().max(1)) ++ .storage() ++ .with_io_engine_config(PsyncIoEngineConfig::new()) ++ .with_engine_config(engine) ++ .build() ++ .await?; ++ Ok(Self { ++ cache, ++ read_block_size, ++ wrapped_stores: Arc::new(Mutex::new(HashMap::new())), ++ }) ++ } ++ ++ fn is_cacheable_data_file(location: &Path) -> bool { ++ let mut parts = location.as_ref().rsplit('/'); ++ matches!( ++ (parts.next(), parts.next()), ++ (Some(file), Some("data")) if file.ends_with(".lance") ++ ) ++ } ++ ++ fn key(&self, store_prefix: &str, location: &Path, block_index: u64) -> String { ++ format!( ++ "{CACHE_KEY_VERSION}\0{}\0{store_prefix}\0{}\0{block_index}", ++ self.read_block_size, ++ location.as_ref() ++ ) ++ } ++ ++ fn size_key(&self, store_prefix: &str, location: &Path) -> String { ++ format!( ++ "{CACHE_KEY_VERSION}\0{}\0{store_prefix}\0{}\0size", ++ self.read_block_size, ++ location.as_ref() ++ ) ++ } ++ ++ fn create_scope(&self) -> Arc { ++ Arc::new(DatasetFoyerDataCache { ++ cache: self.clone(), ++ statistics: Arc::new(FoyerDataCacheStatistics::default()), ++ }) ++ } ++ ++ fn unwrap_store(&self, store: Arc) -> Arc { ++ let identity = Arc::as_ptr(&store) as *const () as usize; ++ let origin = { ++ let mut wrapped_stores = self ++ .wrapped_stores ++ .lock() ++ .unwrap_or_else(|error| error.into_inner()); ++ let origin = wrapped_stores.get(&identity).and_then(|entry| { ++ let wrapper = entry.wrapper.upgrade()?; ++ if Arc::ptr_eq(&wrapper, &store) { ++ entry.origin.upgrade() ++ } else { ++ None ++ } ++ }); ++ if origin.is_none() { ++ wrapped_stores.remove(&identity); ++ } ++ origin ++ }; ++ match origin { ++ Some(origin) => origin, ++ None => store, ++ } ++ } ++ ++ fn remember_wrapper(&self, wrapper: &Arc, origin: &Arc) { ++ let identity = Arc::as_ptr(wrapper) as *const () as usize; ++ self.wrapped_stores ++ .lock() ++ .unwrap_or_else(|error| error.into_inner()) ++ .insert( ++ identity, ++ WrappedStore { ++ wrapper: Arc::downgrade(wrapper), ++ origin: Arc::downgrade(origin), ++ }, ++ ); ++ } ++ ++ fn forget_wrapper(&self, identity: usize) { ++ self.wrapped_stores ++ .lock() ++ .unwrap_or_else(|error| error.into_inner()) ++ .remove(&identity); ++ } ++} ++ ++#[derive(Debug, Default)] ++struct FoyerDataCacheStatistics { ++ bytes_read_from_cache: AtomicU64, ++ bytes_read_from_remote: AtomicU64, ++} ++ ++impl FoyerDataCacheStatistics { ++ fn record(&self, bytes_read_from_cache: u64, bytes_read_from_remote: u64) { ++ self.bytes_read_from_cache ++ .fetch_add(bytes_read_from_cache, Ordering::Relaxed); ++ self.bytes_read_from_remote ++ .fetch_add(bytes_read_from_remote, Ordering::Relaxed); ++ } ++ ++ fn snapshot(&self) -> LanceDataCacheStatistics { ++ LanceDataCacheStatistics { ++ bytes_read_from_cache: self.bytes_read_from_cache.load(Ordering::Relaxed), ++ bytes_read_from_remote: self.bytes_read_from_remote.load(Ordering::Relaxed), ++ } ++ } ++} ++ ++impl DataCacheFactory for FoyerDataCache { ++ fn attach(&self, dataset: lance::Dataset) -> (lance::Dataset, Arc) { ++ let scope = self.create_scope(); ++ let wrapper: Arc = scope.clone(); ++ let dataset = dataset.with_object_store_wrappers([wrapper]); ++ (dataset, scope) ++ } ++} ++ ++#[derive(Debug)] ++struct DatasetFoyerDataCache { ++ cache: FoyerDataCache, ++ statistics: Arc, ++} ++ ++impl DatasetDataCache for DatasetFoyerDataCache { ++ fn snapshot(&self) -> LanceDataCacheStatistics { ++ self.statistics.snapshot() ++ } ++ ++ fn attach_fresh(&self, dataset: lance::Dataset) -> (lance::Dataset, Arc) { ++ self.cache.attach(dataset) ++ } ++} ++ ++impl WrappingObjectStore for DatasetFoyerDataCache { ++ fn wrap(&self, store_prefix: &str, original: Arc) -> Arc { ++ // A derived Dataset can already contain this cache wrapper. Resolve ++ // that exact wrapper back to its origin before attaching fresh ++ // dataset-scoped counters. ++ let original = self.cache.unwrap_store(original); ++ let reader = DataCacheReader { ++ cache: self.cache.clone(), ++ store_prefix: store_prefix.to_owned(), ++ original: original.clone(), ++ statistics: self.statistics.clone(), ++ }; ++ let cached_store = ++ Arc::new_cyclic(|weak: &Weak| DataCacheObjectStore { ++ reader, ++ identity: weak.as_ptr() as usize, ++ }); ++ let wrapped: Arc = cached_store.clone(); ++ self.cache.remember_wrapper(&wrapped, &original); ++ wrapped ++ } ++} ++ ++#[derive(Debug)] ++struct DataCacheObjectStore { ++ reader: DataCacheReader, ++ identity: usize, ++} ++ ++#[derive(Clone, Debug)] ++struct DataCacheReader { ++ cache: FoyerDataCache, ++ store_prefix: String, ++ original: Arc, ++ statistics: Arc, ++} ++ ++impl Drop for DataCacheObjectStore { ++ fn drop(&mut self) { ++ self.reader.cache.forget_wrapper(self.identity); ++ } ++} ++ ++impl Display for DataCacheObjectStore { ++ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { ++ write!(f, "FoyerDataCache({})", self.reader.original) ++ } ++} ++ ++impl DataCacheObjectStore { ++ fn is_cache_safe_get(options: &GetOptions) -> bool { ++ !options.head ++ && options.if_match.is_none() ++ && options.if_none_match.is_none() ++ && options.if_modified_since.is_none() ++ && options.if_unmodified_since.is_none() ++ && options.version.is_none() ++ && options.extensions.is_empty() ++ } ++ ++ async fn cached_get(&self, location: &Path, options: GetOptions) -> Result { ++ // Fetch metadata separately so the returned GetResult retains the origin's identity while ++ // its payload uses the same block cache as get_ranges(). This also provides the object size ++ // needed to resolve bounded, offset, and suffix ranges. ++ let GetResult { ++ meta: metadata, ++ attributes, ++ .. ++ } = self ++ .reader ++ .original ++ .get_opts( ++ location, ++ GetOptions { ++ head: true, ++ ..Default::default() ++ }, ++ ) ++ .await?; ++ let object_size = metadata.size; ++ self.reader.cache.cache.insert( ++ self.reader ++ .cache ++ .size_key(&self.reader.store_prefix, location), ++ Bytes::copy_from_slice(&object_size.to_le_bytes()), ++ ); ++ ++ let range = match options.range.clone() { ++ Some(requested) => match requested.as_range(object_size) { ++ Ok(range) if !range.is_empty() => range, ++ // Preserve the origin's exact error for invalid or empty ranges. ++ _ => return self.reader.original.get_opts(location, options).await, ++ }, ++ None => 0..object_size, ++ }; ++ ++ let reader = self.reader.clone(); ++ let stream_location = location.clone(); ++ let stream_range = range.clone(); ++ let stream = futures::stream::try_unfold( ++ (reader, stream_location, stream_range), ++ |(reader, location, remaining)| async move { ++ if remaining.is_empty() { ++ return Ok(None); ++ } ++ ++ // Yield no more than the remainder of one cache block. The next block is not ++ // requested until the consumer polls again, so cancellation drops the pending ++ // range without downloading or retaining the rest of the object. ++ let block_size = reader.cache.read_block_size as u64; ++ let bytes_to_boundary = block_size - remaining.start % block_size; ++ let end = remaining ++ .start ++ .saturating_add(bytes_to_boundary) ++ .min(remaining.end); ++ let chunk_range = remaining.start..end; ++ let chunk = reader ++ .cached_ranges(&location, std::slice::from_ref(&chunk_range)) ++ .await? ++ .into_iter() ++ .next() ++ .ok_or_else(|| cache_error(format!("missing get result for {location}")))?; ++ Ok(Some((chunk, (reader, location, end..remaining.end)))) ++ }, ++ ); ++ let payload = GetResultPayload::Stream(Box::pin(stream)); ++ Ok(GetResult { ++ payload, ++ meta: metadata, ++ range, ++ attributes, ++ }) ++ } ++} ++ ++impl DataCacheReader { ++ async fn read_origin_ranges( ++ &self, ++ location: &Path, ++ ranges: &[Range], ++ ) -> Result> { ++ let bytes = self.original.get_ranges(location, ranges).await?; ++ self.statistics.record(0, total_bytes(&bytes)); ++ Ok(bytes) ++ } ++ ++ async fn object_size(&self, location: &Path) -> Result { ++ let key = self.cache.size_key(&self.store_prefix, location); ++ match self.cache.cache.get(&key).await { ++ Ok(Some(entry)) => match entry.value().as_ref().try_into() { ++ Ok(bytes) => return Ok(u64::from_le_bytes(bytes)), ++ Err(_) => log::warn!( ++ "Foyer data-cache size entry was malformed for {location}; refreshing it" ++ ), ++ }, ++ Ok(None) => {} ++ Err(error) => { ++ log::warn!("Foyer data-cache size lookup failed for {location}: {error}"); ++ } ++ } ++ ++ let size = self.original.head(location).await?.size; ++ self.cache ++ .cache ++ .insert(key, Bytes::copy_from_slice(&size.to_le_bytes())); ++ Ok(size) ++ } ++ ++ async fn cached_ranges(&self, location: &Path, ranges: &[Range]) -> Result> { ++ if ranges.is_empty() { ++ return Ok(Vec::new()); ++ } ++ if ranges.iter().any(|range| range.start >= range.end) { ++ return self.read_origin_ranges(location, ranges).await; ++ } ++ ++ let object_size = self.object_size(location).await?; ++ if ranges.iter().any(|range| range.start >= object_size) { ++ // Preserve the origin's exact error for ranges that start at or ++ // beyond EOF. ++ return self.read_origin_ranges(location, ranges).await; ++ } ++ let readable_ranges = ranges ++ .iter() ++ .map(|range| range.start..range.end.min(object_size)) ++ .collect::>(); ++ ++ let block_size = self.cache.read_block_size as u64; ++ let mut blocks = BTreeMap::>::new(); ++ for range in &readable_ranges { ++ let first = range.start / block_size; ++ let last = (range.end - 1) / block_size; ++ for block_index in first..=last { ++ blocks.entry(block_index).or_default(); ++ } ++ } ++ ++ for (block_index, block) in &mut blocks { ++ let key = self.cache.key(&self.store_prefix, location, *block_index); ++ match self.cache.cache.get(&key).await { ++ Ok(Some(entry)) => *block = Some(entry.value().clone()), ++ Ok(None) => {} ++ Err(error) => { ++ // Cache availability must not affect query correctness. ++ log::warn!("Foyer data-cache lookup failed for {location}: {error}"); ++ } ++ } ++ } ++ ++ let (bytes_read_from_cache, bytes_read_from_remote) = ++ requested_bytes_by_cache_status(&readable_ranges, block_size, &blocks); ++ let missing: Vec = blocks ++ .iter() ++ .filter_map(|(block_index, value)| value.is_none().then_some(*block_index)) ++ .collect(); ++ let mut runs = Vec::>::new(); ++ for block_index in missing { ++ let start = block_index ++ .checked_mul(block_size) ++ .ok_or_else(|| cache_error("data-cache block offset overflow"))?; ++ let end = start ++ .checked_add(block_size) ++ .ok_or_else(|| cache_error("data-cache block end overflow"))? ++ .min(object_size); ++ match runs.last_mut() { ++ Some(run) if run.end == start => run.end = end, ++ _ => runs.push(start..end), ++ } ++ } ++ ++ if !runs.is_empty() { ++ // Fetch all contiguous miss runs together so a large Lance read is ++ // not expanded into one remote request per cache block. ++ let fetched = self.original.get_ranges(location, &runs).await?; ++ for (run, bytes) in runs.into_iter().zip(fetched) { ++ let first_block = run.start / block_size; ++ for (offset, chunk) in bytes.chunks(self.cache.read_block_size).enumerate() { ++ let block_index = first_block + offset as u64; ++ let value = Bytes::copy_from_slice(chunk); ++ let key = self.cache.key(&self.store_prefix, location, block_index); ++ self.cache.cache.insert(key, value.clone()); ++ if let Some(block) = blocks.get_mut(&block_index) { ++ *block = Some(value); ++ } ++ } ++ } ++ } ++ ++ let assembled = readable_ranges ++ .iter() ++ .map(|range| assemble_range(location, range, block_size, &blocks)) ++ .collect::>>(); ++ match assembled { ++ Ok(bytes) => { ++ self.statistics ++ .record(bytes_read_from_cache, bytes_read_from_remote); ++ Ok(bytes) ++ } ++ Err(error) => { ++ // A malformed or incomplete cached entry must never turn a ++ // valid source read into a query failure. ++ log::warn!( ++ "Foyer data-cache entry was unusable for {location}; bypassing cache: {error}" ++ ); ++ let bytes = self.original.get_ranges(location, ranges).await?; ++ self.statistics.record(0, total_bytes(&bytes)); ++ Ok(bytes) ++ } ++ } ++ } ++} ++ ++fn total_bytes(ranges: &[Bytes]) -> u64 { ++ ranges.iter().fold(0_u64, |total, bytes| { ++ total.saturating_add(bytes.len() as u64) ++ }) ++} ++ ++fn requested_bytes_by_cache_status( ++ ranges: &[Range], ++ block_size: u64, ++ blocks: &BTreeMap>, ++) -> (u64, u64) { ++ let mut hit_bytes = 0_u64; ++ let mut miss_bytes = 0_u64; ++ for range in ranges { ++ let mut start = range.start; ++ while start < range.end { ++ let block_index = start / block_size; ++ let block_start = block_index * block_size; ++ let end = range.end.min(block_start.saturating_add(block_size)); ++ let bytes = end - start; ++ if blocks ++ .get(&block_index) ++ .is_some_and(|block| block.is_some()) ++ { ++ hit_bytes = hit_bytes.saturating_add(bytes); ++ } else { ++ miss_bytes = miss_bytes.saturating_add(bytes); ++ } ++ start = end; ++ } ++ } ++ (hit_bytes, miss_bytes) ++} ++ ++fn assemble_range( ++ location: &Path, ++ range: &Range, ++ block_size: u64, ++ blocks: &BTreeMap>, ++) -> Result { ++ if range.is_empty() { ++ return Ok(Bytes::new()); ++ } ++ let first = range.start / block_size; ++ let last = (range.end - 1) / block_size; ++ if first == last { ++ let block = blocks ++ .get(&first) ++ .and_then(Option::as_ref) ++ .ok_or_else(|| cache_error(format!("missing block {first} for {location}")))?; ++ let block_start = first * block_size; ++ let start = usize::try_from(range.start - block_start) ++ .map_err(|_| cache_error("data-cache slice start exceeds usize::MAX"))?; ++ let end = usize::try_from((range.end - block_start).min(block_size)) ++ .map_err(|_| cache_error("data-cache slice end exceeds usize::MAX"))? ++ .min(block.len()); ++ if start >= block.len() { ++ return Err(cache_error(format!( ++ "short data-cache block {first} for {location}: need {start}..{end}, got {} bytes", ++ block.len() ++ ))); ++ } ++ return Ok(block.slice(start..end)); ++ } ++ ++ let requested_len = usize::try_from(range.end - range.start) ++ .map_err(|_| cache_error(format!("range {range:?} for {location} exceeds usize::MAX")))?; ++ let mut output = BytesMut::with_capacity(requested_len); ++ for block_index in first..=last { ++ let block = blocks ++ .get(&block_index) ++ .and_then(Option::as_ref) ++ .ok_or_else(|| cache_error(format!("missing block {block_index} for {location}")))?; ++ let block_start = block_index * block_size; ++ let start = usize::try_from(range.start.saturating_sub(block_start)) ++ .map_err(|_| cache_error("data-cache slice start exceeds usize::MAX"))?; ++ let end_in_block = range.end.saturating_sub(block_start).min(block_size); ++ let end = usize::try_from(end_in_block) ++ .map_err(|_| cache_error("data-cache slice end exceeds usize::MAX"))?; ++ if start >= block.len() { ++ if !output.is_empty() { ++ break; ++ } ++ return Err(cache_error(format!( ++ "short data-cache block {block_index} for {location}: need {start}..{end}, got {} bytes", ++ block.len() ++ ))); ++ } ++ let actual_end = end.min(block.len()); ++ output.extend_from_slice(&block[start..actual_end]); ++ if actual_end < end { ++ break; ++ } ++ } ++ Ok(output.freeze()) ++} ++ ++fn cache_error(message: impl Into) -> object_store::Error { ++ object_store::Error::Generic { ++ store: "foyer_data_cache", ++ source: Box::new(std::io::Error::other(message.into())), ++ } ++} ++ ++#[async_trait] ++#[deny(clippy::missing_trait_methods)] ++impl ObjectStore for DataCacheObjectStore { ++ async fn put_opts( ++ &self, ++ location: &Path, ++ payload: PutPayload, ++ opts: PutOptions, ++ ) -> Result { ++ self.reader.original.put_opts(location, payload, opts).await ++ } ++ ++ async fn put_multipart_opts( ++ &self, ++ location: &Path, ++ opts: PutMultipartOptions, ++ ) -> Result> { ++ self.reader ++ .original ++ .put_multipart_opts(location, opts) ++ .await ++ } ++ ++ async fn get_opts(&self, location: &Path, options: GetOptions) -> Result { ++ if FoyerDataCache::is_cacheable_data_file(location) && Self::is_cache_safe_get(&options) { ++ self.cached_get(location, options).await ++ } else { ++ self.reader.original.get_opts(location, options).await ++ } ++ } ++ ++ async fn get_ranges(&self, location: &Path, ranges: &[Range]) -> Result> { ++ if FoyerDataCache::is_cacheable_data_file(location) { ++ self.reader.cached_ranges(location, ranges).await ++ } else { ++ self.reader.original.get_ranges(location, ranges).await ++ } ++ } ++ ++ fn delete_stream( ++ &self, ++ locations: BoxStream<'static, Result>, ++ ) -> BoxStream<'static, Result> { ++ self.reader.original.delete_stream(locations) ++ } ++ ++ fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result> { ++ self.reader.original.list(prefix) ++ } ++ ++ fn list_with_offset( ++ &self, ++ prefix: Option<&Path>, ++ offset: &Path, ++ ) -> BoxStream<'static, Result> { ++ self.reader.original.list_with_offset(prefix, offset) ++ } ++ ++ async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result { ++ self.reader.original.list_with_delimiter(prefix).await ++ } ++ ++ async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> { ++ self.reader.original.copy_opts(from, to, options).await ++ } ++ ++ async fn rename_opts(&self, from: &Path, to: &Path, options: RenameOptions) -> Result<()> { ++ self.reader.original.rename_opts(from, to, options).await ++ } ++} ++ ++#[cfg(test)] ++mod tests { ++ use std::sync::mpsc; ++ use std::time::Duration; ++ ++ use futures::StreamExt; ++ use lance_io::object_store::ChainedWrappingObjectStore; ++ use object_store::GetRange; ++ use object_store::memory::InMemory; ++ ++ use super::*; ++ ++ fn wrap_for_test( ++ cache: &FoyerDataCache, ++ original: Arc, ++ ) -> (Arc, Arc) { ++ let scope = cache.create_scope(); ++ (scope.wrap("memory://test", original), scope) ++ } ++ ++ #[tokio::test] ++ async fn caches_only_immutable_data_file_ranges() { ++ let directory = tempfile::tempdir().unwrap(); ++ let cache = FoyerDataCache::try_new(directory.path(), 256 * 1024, 1024 * 1024, 64 * 1024) ++ .await ++ .unwrap(); ++ let original = Arc::new(InMemory::new()); ++ let data_path = Path::from("table.lance/data/part-0.lance"); ++ let second_data_path = Path::from("table.lance/data/part-1.lance"); ++ let manifest_path = Path::from("table.lance/_versions/1.manifest"); ++ let data = Bytes::from((0..200_000).map(|value| value as u8).collect::>()); ++ let second_data = Bytes::from_static(b"second fragment"); ++ original.put(&data_path, data.clone().into()).await.unwrap(); ++ original ++ .put(&second_data_path, second_data.clone().into()) ++ .await ++ .unwrap(); ++ original ++ .put(&manifest_path, Bytes::from_static(b"manifest").into()) ++ .await ++ .unwrap(); ++ ++ let (wrapped, statistics) = wrap_for_test(&cache, original.clone()); ++ let ranges = vec![10..90_000, 65_000..140_000, 190_000..220_000]; ++ let second_data_range = 0..15; ++ let first = wrapped.get_ranges(&data_path, &ranges).await.unwrap(); ++ assert_eq!(first[0], data.slice(10..90_000)); ++ assert_eq!(first[1], data.slice(65_000..140_000)); ++ assert_eq!(first[2], data.slice(190_000..200_000)); ++ assert_eq!( ++ wrapped ++ .get_ranges(&second_data_path, std::slice::from_ref(&second_data_range)) ++ .await ++ .unwrap(), ++ vec![second_data.clone()] ++ ); ++ ++ assert_eq!( ++ statistics.snapshot(), ++ LanceDataCacheStatistics { ++ bytes_read_from_cache: 0, ++ bytes_read_from_remote: 175_005, ++ } ++ ); ++ ++ original.delete(&data_path).await.unwrap(); ++ original.delete(&second_data_path).await.unwrap(); ++ let second = wrapped.get_ranges(&data_path, &ranges).await.unwrap(); ++ assert_eq!(second, first); ++ assert_eq!( ++ wrapped ++ .get_ranges(&second_data_path, std::slice::from_ref(&second_data_range)) ++ .await ++ .unwrap(), ++ vec![second_data] ++ ); ++ ++ assert_eq!( ++ statistics.snapshot(), ++ LanceDataCacheStatistics { ++ bytes_read_from_cache: 175_005, ++ bytes_read_from_remote: 175_005, ++ } ++ ); ++ ++ assert_eq!( ++ wrapped.get_range(&manifest_path, 0..8).await.unwrap(), ++ Bytes::from_static(b"manifest") ++ ); ++ original.delete(&manifest_path).await.unwrap(); ++ assert!(wrapped.get_range(&manifest_path, 0..8).await.is_err()); ++ } ++ ++ #[tokio::test] ++ async fn caches_small_data_file_whole_object_reads() { ++ let directory = tempfile::tempdir().unwrap(); ++ let cache = FoyerDataCache::try_new(directory.path(), 128 * 1024, 1024 * 1024, 64 * 1024) ++ .await ++ .unwrap(); ++ let original = Arc::new(InMemory::new()); ++ let data_path = Path::from("table.lance/data/small.lance"); ++ let data = Bytes::from(vec![7; 42_000]); ++ original.put(&data_path, data.clone().into()).await.unwrap(); ++ ++ let (wrapped, statistics) = wrap_for_test(&cache, original); ++ let first = wrapped ++ .get(&data_path) ++ .await ++ .unwrap() ++ .bytes() ++ .await ++ .unwrap(); ++ assert_eq!(first, data); ++ assert_eq!( ++ statistics.snapshot(), ++ LanceDataCacheStatistics { ++ bytes_read_from_cache: 0, ++ bytes_read_from_remote: 42_000, ++ } ++ ); ++ ++ let second = wrapped ++ .get(&data_path) ++ .await ++ .unwrap() ++ .bytes() ++ .await ++ .unwrap(); ++ assert_eq!(second, data); ++ assert_eq!( ++ statistics.snapshot(), ++ LanceDataCacheStatistics { ++ bytes_read_from_cache: 42_000, ++ bytes_read_from_remote: 42_000, ++ } ++ ); ++ } ++ ++ #[tokio::test] ++ async fn streams_large_data_file_gets_with_bounded_read_ahead() { ++ let directory = tempfile::tempdir().unwrap(); ++ let block_size = 64 * 1024; ++ let cache = FoyerDataCache::try_new(directory.path(), 512 * 1024, 1024 * 1024, block_size) ++ .await ++ .unwrap(); ++ let original = Arc::new(InMemory::new()); ++ let data_path = Path::from("table.lance/data/large.lance"); ++ let data = Bytes::from(vec![7; 4 * block_size]); ++ original.put(&data_path, data.clone().into()).await.unwrap(); ++ ++ let (wrapped, statistics) = wrap_for_test(&cache, original); ++ let result = wrapped.get(&data_path).await.unwrap(); ++ assert_eq!(statistics.snapshot(), LanceDataCacheStatistics::default()); ++ ++ let mut stream = result.into_stream(); ++ let first = stream.next().await.unwrap().unwrap(); ++ assert_eq!(first, data.slice(..block_size)); ++ assert_eq!( ++ statistics.snapshot(), ++ LanceDataCacheStatistics { ++ bytes_read_from_cache: 0, ++ bytes_read_from_remote: block_size as u64, ++ } ++ ); ++ ++ drop(stream); ++ assert_eq!( ++ statistics.snapshot().bytes_read_from_remote, ++ block_size as u64 ++ ); ++ } ++ ++ #[tokio::test] ++ async fn caches_single_data_file_range_reads() { ++ let directory = tempfile::tempdir().unwrap(); ++ let cache = FoyerDataCache::try_new(directory.path(), 512 * 1024, 1024 * 1024, 64 * 1024) ++ .await ++ .unwrap(); ++ let original = Arc::new(InMemory::new()); ++ let data = Bytes::from((0..100_000).map(|value| value as u8).collect::>()); ++ let cases = [ ++ ( ++ Path::from("table.lance/data/bounded.lance"), ++ GetRange::Bounded(1_000..2_000), ++ 1_000..2_000, ++ ), ++ ( ++ Path::from("table.lance/data/offset.lance"), ++ GetRange::Offset(90_000), ++ 90_000..100_000, ++ ), ++ ( ++ Path::from("table.lance/data/suffix.lance"), ++ GetRange::Suffix(500), ++ 99_500..100_000, ++ ), ++ ]; ++ for (path, _, _) in &cases { ++ original.put(path, data.clone().into()).await.unwrap(); ++ } ++ ++ let (wrapped, statistics) = wrap_for_test(&cache, original); ++ for (path, requested, expected_range) in cases { ++ let expected = data.slice(expected_range.start as usize..expected_range.end as usize); ++ let before = statistics.snapshot(); ++ let first = wrapped ++ .get_opts(&path, GetOptions::new().with_range(Some(requested.clone()))) ++ .await ++ .unwrap() ++ .bytes() ++ .await ++ .unwrap(); ++ assert_eq!(first, expected); ++ assert_eq!( ++ statistics.snapshot().bytes_read_from_remote, ++ before.bytes_read_from_remote + expected.len() as u64 ++ ); ++ ++ let second = wrapped ++ .get_opts(&path, GetOptions::new().with_range(Some(requested))) ++ .await ++ .unwrap() ++ .bytes() ++ .await ++ .unwrap(); ++ assert_eq!(second, expected); ++ assert_eq!( ++ statistics.snapshot().bytes_read_from_cache, ++ before.bytes_read_from_cache + expected.len() as u64 ++ ); ++ } ++ ++ let before = statistics.snapshot(); ++ let conditional = GetOptions::new().with_if_match(Some("wrong-etag")); ++ assert!( ++ wrapped ++ .get_opts(&Path::from("table.lance/data/bounded.lance"), conditional) ++ .await ++ .is_err() ++ ); ++ assert_eq!(statistics.snapshot(), before); ++ ++ assert!( ++ wrapped ++ .get_range( ++ &Path::from("table.lance/data/bounded.lance"), ++ 100_000..100_001 ++ ) ++ .await ++ .is_err() ++ ); ++ assert_eq!(statistics.snapshot(), before); ++ } ++ ++ #[tokio::test] ++ async fn recovers_cached_data_from_disk() { ++ let directory = tempfile::tempdir().unwrap(); ++ let original = Arc::new(InMemory::new()); ++ let data_path = Path::from("table.lance/data/part-0.lance"); ++ let data = Bytes::from((0..100_000).map(|value| value as u8).collect::>()); ++ original.put(&data_path, data.clone().into()).await.unwrap(); ++ ++ let cache = FoyerDataCache::try_new(directory.path(), 128 * 1024, 1024 * 1024, 64 * 1024) ++ .await ++ .unwrap(); ++ let (wrapped, _) = wrap_for_test(&cache, original.clone()); ++ let requested_range = 10..90_000; ++ assert_eq!( ++ wrapped ++ .get_ranges(&data_path, std::slice::from_ref(&requested_range)) ++ .await ++ .unwrap(), ++ vec![data.slice(10..90_000)] ++ ); ++ drop(wrapped); ++ cache.cache.close().await.unwrap(); ++ drop(cache); ++ ++ original.delete(&data_path).await.unwrap(); ++ let recovered = ++ FoyerDataCache::try_new(directory.path(), 128 * 1024, 1024 * 1024, 64 * 1024) ++ .await ++ .unwrap(); ++ let (wrapped, statistics) = wrap_for_test(&recovered, original); ++ assert_eq!( ++ wrapped ++ .get_ranges(&data_path, std::slice::from_ref(&requested_range)) ++ .await ++ .unwrap(), ++ vec![data.slice(10..90_000)] ++ ); ++ assert_eq!(statistics.snapshot().bytes_read_from_cache, 89_990); ++ assert_eq!(statistics.snapshot().bytes_read_from_remote, 0); ++ } ++ ++ #[tokio::test] ++ async fn dataset_scopes_share_cache_without_sharing_statistics() { ++ let directory = tempfile::tempdir().unwrap(); ++ let cache = FoyerDataCache::try_new(directory.path(), 128 * 1024, 1024 * 1024, 64 * 1024) ++ .await ++ .unwrap(); ++ let original = Arc::new(InMemory::new()); ++ let data_path = Path::from("table.lance/data/part-0.lance"); ++ let data = Bytes::from((0..100_000).map(|value| value as u8).collect::>()); ++ original.put(&data_path, data.clone().into()).await.unwrap(); ++ ++ let source_scope = cache.create_scope(); ++ let source_store = source_scope.wrap("memory://test", original.clone()); ++ ++ // A restored Dataset is derived from an already-wrapped source ++ // Dataset. The fresh scope must unwrap to the registered origin rather ++ // than nesting over the source scope. ++ let restored_scope = cache.create_scope(); ++ let restored_store = restored_scope.wrap("memory://test", source_store.clone()); ++ let requested_range = 10..90_000; ++ assert_eq!( ++ restored_store ++ .get_ranges(&data_path, std::slice::from_ref(&requested_range)) ++ .await ++ .unwrap(), ++ vec![data.slice(10..90_000)] ++ ); ++ assert_eq!(source_scope.snapshot(), LanceDataCacheStatistics::default()); ++ assert_eq!(restored_scope.snapshot().bytes_read_from_remote, 89_990); ++ ++ original.delete(&data_path).await.unwrap(); ++ assert_eq!( ++ source_store ++ .get_ranges(&data_path, std::slice::from_ref(&requested_range)) ++ .await ++ .unwrap(), ++ vec![data.slice(10..90_000)] ++ ); ++ assert_eq!(source_scope.snapshot().bytes_read_from_cache, 89_990); ++ assert_eq!(restored_scope.snapshot().bytes_read_from_remote, 89_990); ++ ++ drop(restored_store); ++ drop(source_store); ++ assert!(cache.wrapped_stores.lock().unwrap().is_empty()); ++ } ++ ++ #[tokio::test] ++ async fn chained_scopes_drop_intermediate_store_without_deadlocking() { ++ let directory = tempfile::tempdir().unwrap(); ++ let cache = FoyerDataCache::try_new(directory.path(), 128 * 1024, 1024 * 1024, 64 * 1024) ++ .await ++ .unwrap(); ++ let first: Arc = cache.create_scope(); ++ let second: Arc = cache.create_scope(); ++ let chained = ChainedWrappingObjectStore::new(vec![first, second]); ++ let original: Arc = Arc::new(InMemory::new()); ++ let (sender, receiver) = mpsc::channel(); ++ ++ let thread = std::thread::spawn(move || { ++ sender ++ .send(chained.wrap("memory://test", original)) ++ .unwrap(); ++ }); ++ let wrapped = receiver ++ .recv_timeout(Duration::from_secs(2)) ++ .expect("chained cache wrappers deadlocked while dropping the intermediate store"); ++ thread.join().unwrap(); ++ ++ assert_eq!(cache.wrapped_stores.lock().unwrap().len(), 1); ++ drop(wrapped); ++ assert!(cache.wrapped_stores.lock().unwrap().is_empty()); ++ } ++ ++ #[tokio::test] ++ async fn truncates_ranges_at_eof_before_enumerating_cache_blocks() { ++ let directory = tempfile::tempdir().unwrap(); ++ let cache = FoyerDataCache::try_new(directory.path(), 128 * 1024, 1024 * 1024, 64 * 1024) ++ .await ++ .unwrap(); ++ let original = Arc::new(InMemory::new()); ++ let data_path = Path::from("table.lance/data/part-0.lance"); ++ let data = Bytes::from((0..100_000).map(|value| value as u8).collect::>()); ++ original.put(&data_path, data.into()).await.unwrap(); ++ ++ let requested_range = 99_990..300_000; ++ let expected = original ++ .get_ranges(&data_path, std::slice::from_ref(&requested_range)) ++ .await ++ .unwrap(); ++ let (wrapped, statistics) = wrap_for_test(&cache, original.clone()); ++ let actual = wrapped ++ .get_ranges(&data_path, std::slice::from_ref(&requested_range)) ++ .await ++ .unwrap(); ++ assert_eq!(actual, expected); ++ assert_eq!(statistics.snapshot().bytes_read_from_remote, 10); ++ ++ original.delete(&data_path).await.unwrap(); ++ assert_eq!( ++ wrapped ++ .get_ranges(&data_path, std::slice::from_ref(&requested_range)) ++ .await ++ .unwrap(), ++ expected ++ ); ++ assert_eq!(statistics.snapshot().bytes_read_from_cache, 10); ++ assert_eq!(statistics.snapshot().bytes_read_from_remote, 10); ++ } ++ ++ #[test] ++ fn recognizes_only_direct_data_children() { ++ assert!(FoyerDataCache::is_cacheable_data_file(&Path::from( ++ "dataset/data/part.lance" ++ ))); ++ assert!(!FoyerDataCache::is_cacheable_data_file(&Path::from( ++ "dataset/data/nested/part.lance" ++ ))); ++ assert!(!FoyerDataCache::is_cacheable_data_file(&Path::from( ++ "dataset/indices/index.lance" ++ ))); ++ assert!(!FoyerDataCache::is_cacheable_data_file(&Path::from( ++ "dataset/_versions/1.manifest" ++ ))); ++ } ++} +diff --git a/src/lib.rs b/src/lib.rs +index 8b212f5..923528a 100644 +--- a/src/lib.rs ++++ b/src/lib.rs +@@ -25,11 +25,13 @@ mod alter_columns; + mod async_dispatcher; + mod batch; + mod compact; ++mod data_cache; + mod data_statistics; + mod dataset; + mod delete; + mod drop_columns; + mod error; ++mod foyer_data_cache; + mod fragment_writer; + mod fts_query; + mod helpers; +@@ -51,6 +53,7 @@ pub use add_columns::*; + pub use alter_columns::*; + pub use batch::*; + pub use compact::*; ++pub use data_cache::{LanceDataCacheStatistics, lance_dataset_get_data_cache_statistics}; + pub use data_statistics::*; + pub use dataset::*; + pub use delete::*; +@@ -58,6 +61,7 @@ pub use drop_columns::*; + pub use error::{ + LanceErrorCode, lance_free_string, lance_last_error_code, lance_last_error_message, + }; ++pub use foyer_data_cache::{LanceDataCacheOptions, lance_session_new_with_data_cache}; + pub use fragment_writer::*; + pub use fts_query::*; + pub use index::*; +diff --git a/src/restore.rs b/src/restore.rs +index 7804b55..fa2d26d 100644 +--- a/src/restore.rs ++++ b/src/restore.rs +@@ -65,8 +65,16 @@ unsafe fn restore_inner(dataset: *const LanceDataset, version: u64) -> Result<*m + Ok::<_, lance_core::Error>(checked_out) + })?; + ++ let (restored, data_cache) = if let Some(data_cache) = &ds.data_cache { ++ let (restored, data_cache) = data_cache.attach_fresh(restored); ++ (restored, Some(data_cache)) ++ } else { ++ (restored, None) ++ }; ++ + let handle = LanceDataset { + inner: RwLock::new(Arc::new(restored)), ++ data_cache, + }; + Ok(Box::into_raw(Box::new(handle))) + } +diff --git a/src/session.rs b/src/session.rs +index 60a1623..9ed8cfe 100644 +--- a/src/session.rs ++++ b/src/session.rs +@@ -8,12 +8,14 @@ use std::sync::Arc; + use lance::session::Session; + use lance_core::Result; + ++use crate::data_cache::DataCacheFactory; + use crate::error::{ffi_try, swallow_unwind}; + use crate::runtime::block_on; + +-/// Opaque handle for sharing Lance metadata and index caches across datasets. ++/// Opaque handle for shared Lance caches across datasets. + pub struct LanceSession { + pub(crate) inner: Arc, ++ pub(crate) data_cache_factory: Option>, + } + + /// Snapshot of a session's metadata and index cache statistics. +@@ -47,6 +49,14 @@ pub extern "C" fn lance_session_new( + fn session_new_inner( + index_cache_size_bytes: u64, + metadata_cache_size_bytes: u64, ++) -> Result<*mut LanceSession> { ++ session_new_with_data_cache_factory(index_cache_size_bytes, metadata_cache_size_bytes, None) ++} ++ ++pub(crate) fn session_new_with_data_cache_factory( ++ index_cache_size_bytes: u64, ++ metadata_cache_size_bytes: u64, ++ data_cache_factory: Option>, + ) -> Result<*mut LanceSession> { + let index_cache_size_bytes = u64_to_usize(index_cache_size_bytes, "index_cache_size_bytes")?; + let metadata_cache_size_bytes = +@@ -58,6 +68,7 @@ fn session_new_inner( + ); + Ok(Box::into_raw(Box::new(LanceSession { + inner: Arc::new(session), ++ data_cache_factory, + }))) + } + +diff --git a/src/writer.rs b/src/writer.rs +index 1971510..ba51c87 100644 +--- a/src/writer.rs ++++ b/src/writer.rs +@@ -282,6 +282,7 @@ unsafe fn write_dataset_inner( + if !out_dataset.is_null() { + let handle = LanceDataset { + inner: RwLock::new(Arc::new(dataset)), ++ data_cache: None, + }; + // SAFETY: `out_dataset` is non-NULL (checked above) and the caller + // guarantees it points to caller-owned, writable storage of size +diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs +index 8805764..b4313f4 100644 +--- a/tests/c_api_test.rs ++++ b/tests/c_api_test.rs +@@ -96,10 +96,83 @@ fn create_large_dataset(num_rows: i32) -> (tempfile::TempDir, String) { + (tmp, uri) + } + ++/// Helper: create two fragments large enough for Lance's batched range-read ++/// path, which is the path wrapped by the Foyer data cache. ++fn create_large_multi_fragment_dataset(num_rows_per_fragment: i32) -> (tempfile::TempDir, String) { ++ let (tmp, uri) = create_large_dataset(num_rows_per_fragment); ++ let schema = Arc::new(Schema::new(vec![ ++ Field::new("id", DataType::Int32, false), ++ Field::new("value", DataType::Float32, true), ++ Field::new("label", DataType::Utf8, true), ++ ])); ++ let ids: Vec = (num_rows_per_fragment..2 * num_rows_per_fragment).collect(); ++ let values: Vec = ids.iter().map(|id| *id as f32 * 0.5).collect(); ++ let labels: Vec = ids.iter().map(|id| format!("row_{id}")).collect(); ++ let label_refs: Vec<&str> = labels.iter().map(String::as_str).collect(); ++ let batch = RecordBatch::try_new( ++ schema.clone(), ++ vec![ ++ Arc::new(Int32Array::from(ids)), ++ Arc::new(Float32Array::from(values)), ++ Arc::new(StringArray::from(label_refs)), ++ ], ++ ) ++ .unwrap(); ++ ++ lance_c::runtime::block_on(async { ++ let mut dataset = Dataset::open(&uri).await.unwrap(); ++ dataset ++ .append( ++ arrow::record_batch::RecordBatchIterator::new(vec![Ok(batch)], schema), ++ None, ++ ) ++ .await ++ .unwrap(); ++ }); ++ ++ (tmp, uri) ++} ++ + fn c_str(s: &str) -> CString { + CString::new(s).unwrap() + } + ++fn file_object_store_uri(path: &str) -> CString { ++ let path = path.replace('\\', "/"); ++ let leading_slash = if path.starts_with('/') { "" } else { "/" }; ++ c_str(&format!("file-object-store://{leading_slash}{path}")) ++} ++ ++fn create_data_cache_session() -> (tempfile::TempDir, *mut LanceSession) { ++ let directory = tempfile::tempdir().unwrap(); ++ let c_directory = c_str(directory.path().to_str().unwrap()); ++ let options = LanceDataCacheOptions { ++ directory: c_directory.as_ptr(), ++ memory_capacity_bytes: 8 * 1024 * 1024, ++ disk_capacity_bytes: 32 * 1024 * 1024, ++ read_block_size_bytes: 64 * 1024, ++ }; ++ let session = unsafe { lance_session_new_with_data_cache(0, 16 * 1024 * 1024, &options) }; ++ assert!(!session.is_null(), "data-cache session should be created"); ++ (directory, session) ++} ++ ++fn data_cache_statistics(dataset: *const LanceDataset) -> LanceDataCacheStatistics { ++ let mut statistics = LanceDataCacheStatistics::default(); ++ assert_eq!( ++ unsafe { lance_dataset_get_data_cache_statistics(dataset, &mut statistics) }, ++ 0 ++ ); ++ statistics ++} ++ ++fn scanned_row_count(dataset: *const LanceDataset) -> usize { ++ scan_all_rows(dataset) ++ .iter() ++ .map(RecordBatch::num_rows) ++ .sum() ++} ++ + #[derive(Default)] + struct CapturedScanStatistics { + calls: usize, +@@ -313,6 +386,120 @@ fn test_shared_session_rejects_null_inputs() { + } + } + ++#[test] ++fn test_session_with_data_cache_serves_repeated_scan() { ++ let (tmp, uri) = create_large_multi_fragment_dataset(10_000); ++ let c_uri = file_object_store_uri(&uri); ++ let (_cache_directory, session) = create_data_cache_session(); ++ ++ let dataset = ++ unsafe { lance_dataset_open_with_session(c_uri.as_ptr(), ptr::null(), 0, session) }; ++ assert!(!dataset.is_null(), "dataset open should succeed"); ++ ++ assert_eq!(data_cache_statistics(dataset), Default::default()); ++ assert_eq!(scanned_row_count(dataset), 20_000); ++ let first_statistics = data_cache_statistics(dataset); ++ assert!(first_statistics.bytes_read_from_remote > 0); ++ ++ let cached_dataset = ++ unsafe { lance_dataset_open_with_session(c_uri.as_ptr(), ptr::null(), 0, session) }; ++ assert!( ++ !cached_dataset.is_null(), ++ "second dataset open should succeed" ++ ); ++ unsafe { lance_session_close(session) }; ++ ++ for entry in std::fs::read_dir(tmp.path().join("large_ds/data")).unwrap() { ++ std::fs::remove_file(entry.unwrap().path()).unwrap(); ++ } ++ assert_eq!(scanned_row_count(cached_dataset), 20_000); ++ let cached_statistics = data_cache_statistics(cached_dataset); ++ assert!(cached_statistics.bytes_read_from_cache > 0); ++ assert_eq!(cached_statistics.bytes_read_from_remote, 0); ++ ++ assert_eq!(data_cache_statistics(dataset), first_statistics); ++ ++ unsafe { lance_dataset_close(cached_dataset) }; ++ unsafe { lance_dataset_close(dataset) }; ++} ++ ++#[test] ++fn test_dataset_data_cache_statistics_validates_inputs_and_defaults_to_zero() { ++ let (_tmp, uri) = create_test_dataset(); ++ let c_uri = c_str(&uri); ++ let dataset = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; ++ assert!(!dataset.is_null()); ++ ++ let mut statistics = LanceDataCacheStatistics::default(); ++ assert_eq!( ++ unsafe { lance_dataset_get_data_cache_statistics(dataset, &mut statistics) }, ++ 0 ++ ); ++ assert_eq!(statistics, LanceDataCacheStatistics::default()); ++ assert_eq!( ++ unsafe { lance_dataset_get_data_cache_statistics(ptr::null(), &mut statistics) }, ++ -1 ++ ); ++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); ++ assert_eq!( ++ unsafe { lance_dataset_get_data_cache_statistics(dataset, ptr::null_mut()) }, ++ -1 ++ ); ++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); ++ ++ unsafe { lance_dataset_close(dataset) }; ++} ++ ++#[test] ++fn test_session_with_data_cache_rejects_invalid_options() { ++ let session = unsafe { lance_session_new_with_data_cache(0, 0, ptr::null()) }; ++ assert!(session.is_null()); ++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); ++ ++ let null_directory = LanceDataCacheOptions { ++ directory: ptr::null(), ++ memory_capacity_bytes: 128 * 1024, ++ disk_capacity_bytes: 1024 * 1024, ++ read_block_size_bytes: 64 * 1024, ++ }; ++ let session = unsafe { lance_session_new_with_data_cache(0, 0, &null_directory) }; ++ assert!(session.is_null()); ++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); ++ ++ let empty_directory = c_str(""); ++ let empty_directory_options = LanceDataCacheOptions { ++ directory: empty_directory.as_ptr(), ++ memory_capacity_bytes: 128 * 1024, ++ disk_capacity_bytes: 1024 * 1024, ++ read_block_size_bytes: 64 * 1024, ++ }; ++ let session = unsafe { lance_session_new_with_data_cache(0, 0, &empty_directory_options) }; ++ assert!(session.is_null()); ++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); ++ ++ let cache_directory = tempfile::tempdir().unwrap(); ++ let c_cache_directory = c_str(cache_directory.path().to_str().unwrap()); ++ let unaligned_block = LanceDataCacheOptions { ++ directory: c_cache_directory.as_ptr(), ++ memory_capacity_bytes: 128 * 1024, ++ disk_capacity_bytes: 1024 * 1024, ++ read_block_size_bytes: 65_535, ++ }; ++ let session = unsafe { lance_session_new_with_data_cache(0, 0, &unaligned_block) }; ++ assert!(session.is_null()); ++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); ++ ++ let zero_capacity = LanceDataCacheOptions { ++ directory: c_cache_directory.as_ptr(), ++ memory_capacity_bytes: 0, ++ disk_capacity_bytes: 0, ++ read_block_size_bytes: 64 * 1024, ++ }; ++ let session = unsafe { lance_session_new_with_data_cache(0, 0, &zero_capacity) }; ++ assert!(session.is_null()); ++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); ++} ++ + #[test] + fn test_open_nonexistent() { + let c_uri = c_str("memory://nonexistent_dataset_xyz"); +@@ -2787,6 +2974,40 @@ fn test_dataset_restore_to_prior_version() { + unsafe { lance_dataset_close(ds) }; + } + ++#[test] ++fn test_restored_handle_has_independent_data_cache_statistics() { ++ let (_tmp, uri) = create_large_multi_fragment_dataset(10_000); ++ let c_uri = file_object_store_uri(&uri); ++ let (_cache_directory, session) = create_data_cache_session(); ++ let source = ++ unsafe { lance_dataset_open_with_session(c_uri.as_ptr(), ptr::null(), 0, session) }; ++ assert!(!source.is_null()); ++ ++ assert_eq!(scanned_row_count(source), 20_000); ++ let source_statistics = data_cache_statistics(source); ++ assert!(source_statistics.bytes_read_from_remote > 0); ++ ++ let restored = unsafe { lance_dataset_restore(source, 1) }; ++ assert!(!restored.is_null()); ++ assert_eq!(data_cache_statistics(restored), Default::default()); ++ let source_statistics_after_restore = data_cache_statistics(source); ++ ++ assert_eq!(scanned_row_count(restored), 10_000); ++ let restored_statistics = data_cache_statistics(restored); ++ assert!(restored_statistics.bytes_read_from_cache > 0); ++ ++ assert_eq!( ++ data_cache_statistics(source), ++ source_statistics_after_restore ++ ); ++ ++ unsafe { ++ lance_session_close(session); ++ lance_dataset_close(restored); ++ lance_dataset_close(source); ++ } ++} ++ + #[test] + fn test_dataset_restore_to_current_latest_writes_new_manifest() { + // Restoring to the current latest still writes a new manifest. The +diff --git a/tests/cpp/test_c_api.c b/tests/cpp/test_c_api.c +index c49ecfa..dd674eb 100644 +--- a/tests/cpp/test_c_api.c ++++ b/tests/cpp/test_c_api.c +@@ -126,6 +126,37 @@ static void test_shared_session(const char *uri) { + (unsigned long long)stats.metadata_cache_entries); + } + ++static void test_data_cache_session(const char *uri, const char *write_uri) { ++ printf(" test_data_cache_session... "); ++ ++ char cache_directory[4096]; ++ int path_len = snprintf(cache_directory, sizeof(cache_directory), ++ "%s_foyer_cache", write_uri); ++ ASSERT(path_len > 0 && (size_t)path_len < sizeof(cache_directory), ++ "cache directory path is too long"); ++ LanceDataCacheOptions options = { ++ .directory = cache_directory, ++ .memory_capacity_bytes = 128 * 1024, ++ .disk_capacity_bytes = 1024 * 1024, ++ .read_block_size_bytes = 64 * 1024, ++ }; ++ LanceSession *session = ++ lance_session_new_with_data_cache(0, 16 * 1024 * 1024, &options); ++ ASSERT(session != NULL, "data-cache session creation failed"); ++ ++ LanceDataset *ds = lance_dataset_open_with_session(uri, NULL, 0, session); ++ ASSERT(ds != NULL, "data-cache session dataset open failed"); ++ LanceDataCacheStatistics statistics; ++ memset(&statistics, 0, sizeof(statistics)); ++ ASSERT(lance_dataset_get_data_cache_statistics(ds, &statistics) == 0, ++ "data-cache dataset statistics failed"); ++ lance_session_close(session); ++ ASSERT(lance_dataset_count_rows(ds) > 0, ++ "dataset should remain valid after data-cache session close"); ++ lance_dataset_close(ds); ++ printf("OK\n"); ++} ++ + static void test_scan(const char *uri) { + printf(" test_scan... "); + +@@ -973,6 +1004,7 @@ int main(int argc, char **argv) { + + test_open_and_metadata(uri); + test_shared_session(uri); ++ test_data_cache_session(uri, write_uri); + test_scan(uri); + test_scan_with_limit(uri); + test_versions(uri); +diff --git a/tests/cpp/test_cpp_api.cpp b/tests/cpp/test_cpp_api.cpp +index 17b1ab6..e1aadbd 100644 +--- a/tests/cpp/test_cpp_api.cpp ++++ b/tests/cpp/test_cpp_api.cpp +@@ -99,6 +99,28 @@ static void test_shared_session(const std::string& uri) { + PASS(); + } + ++static void test_data_cache_session(const std::string& uri, ++ const std::string& write_uri) { ++ TEST(test_data_cache_session); ++ ++ lance::DataCacheOptions options{ ++ write_uri + "_foyer_cache", ++ 128 * 1024, ++ 1024 * 1024, ++ 64 * 1024, ++ }; ++ auto session = std::make_unique( ++ 0, 16 * 1024 * 1024, options); ++ auto ds = lance::Dataset::open_with_session(*session, uri); ++ auto statistics = ds.data_cache_statistics(); ++ assert(statistics.bytes_read_from_cache == 0); ++ assert(statistics.bytes_read_from_remote == 0); ++ session.reset(); ++ assert(ds.count_rows() > 0); ++ ++ PASS(); ++} ++ + static void test_dataset_schema(const std::string& uri) { + TEST(test_dataset_schema); + +@@ -922,6 +944,7 @@ int main(int argc, char** argv) { + + test_dataset_open(uri); + test_shared_session(uri); ++ test_data_cache_session(uri, write_uri); + test_dataset_schema(uri); + test_scanner_fluent(uri); + test_scanner_async_stream_ownership(uri); diff --git a/thirdparty/patches/lance-c-0.1.9-pr-74.patch b/thirdparty/patches/lance-c-0.1.9-pr-74.patch new file mode 100644 index 00000000000000..24c6d33457d84c --- /dev/null +++ b/thirdparty/patches/lance-c-0.1.9-pr-74.patch @@ -0,0 +1,1159 @@ +From b07f970bf2cf3f983cc6043bc72a0fe444c0fd4c Mon Sep 17 00:00:00 2001 +From: zhangstar333 +Date: Thu, 3 Sep 2026 17:16:34 +0800 +Subject: [PATCH 1/2] fts + +--- + include/lance/lance.h | 74 ++++++++--- + include/lance/lance.hpp | 44 +++++-- + src/fts_query.rs | 235 +++++++++++++++++++++++++++++------ + src/scanner.rs | 72 ++++++++--- + tests/c_api_test.rs | 264 +++++++++++++++++++++++++++++++++++++--- + 5 files changed, 595 insertions(+), 94 deletions(-) + +diff --git a/include/lance/lance.h b/include/lance/lance.h +index 3bf291f..0c76edc 100644 +--- a/include/lance/lance.h ++++ b/include/lance/lance.h +@@ -1695,39 +1695,85 @@ typedef enum { + LANCE_FTS_COVERAGE_INDEX_ONLY = 1, + } LanceFtsCoverageMode; + ++/** How the analyzed terms of one Match query are combined. */ ++typedef enum { ++ /** At least one analyzed term must match. */ ++ LANCE_FTS_MATCH_OPERATOR_OR = 0, ++ /** Every analyzed term must match. */ ++ LANCE_FTS_MATCH_OPERATOR_AND = 1, ++} LanceFtsMatchOperator; ++ ++/** ++ * Prepare an OR Match query context for one column. ++ * ++ * @deprecated Use lance_dataset_prepare_fts_match_query() to select the Match ++ * operator explicitly. This compatibility API is equivalent to ++ * LANCE_FTS_MATCH_OPERATOR_OR. ++ */ ++LanceFtsQueryContext* lance_dataset_prepare_fts_query( ++ const LanceDataset* dataset, ++ const char* column, ++ const char* query, ++ uint32_t max_fuzzy_distance, ++ int32_t coverage_mode ++); ++ + /** +- * Prepare an immutable, process-local FTS query context for one column. ++ * Prepare an immutable, process-local Match query context for one column. + * + * Preparation pins the dataset handle's current snapshot, enumerates all + * committed FTS segments for `column`, checks fragment coverage, opens those +- * segments, and computes one query-specific global BM25 scorer across their +- * indexed documents. The context can then be shared by any number of scanners +- * created from the exact same process-local dataset snapshot. It has no +- * serialization or cross-process transport format. Reopening the same URI and +- * manifest version creates a different identity and cannot reuse the context, +- * because storage options and object-store endpoints may differ. ++ * segments, and prepares one global BM25 scorer across their indexed ++ * documents. `match_operator` supports both AND and OR. ++ * ++ * The context can be shared by scanners created from the exact same ++ * process-local dataset snapshot. It has no serialization or cross-process ++ * transport format. Reopening the same URI and manifest version creates a ++ * different identity and cannot reuse the context because storage options and ++ * object-store endpoints may differ. + * + * In LANCE_FTS_COVERAGE_INDEX_ONLY mode, unindexed fragments are allowed and + * excluded from both the scorer corpus and query results. In STRICT mode any + * unindexed fragment makes this call fail. + * +- * Prepared contexts currently support exact Match queries only. +- * `max_fuzzy_distance` must be zero because fuzzy execution requires its +- * canonical expanded vocabulary to be prepared together with the scorer. +- * This restriction does not apply to lance_scanner_full_text_search(). +- * +- * @param max_fuzzy_distance Must be zero for prepared query contexts. ++ * @param match_operator Fixed-width LanceFtsMatchOperator discriminant. ++ * @param max_fuzzy_distance Reserved for prepared fuzzy matching and currently ++ * must be 0. The parameter is retained so enabling ++ * canonical cross-segment fuzzy vocabulary injection ++ * later does not require another C ABI change. + * @param coverage_mode Fixed-width LanceFtsCoverageMode discriminant. + * @return Context handle on success, or NULL on error. + */ +-LanceFtsQueryContext* lance_dataset_prepare_fts_query( ++LanceFtsQueryContext* lance_dataset_prepare_fts_match_query( + const LanceDataset* dataset, + const char* column, + const char* query, ++ int32_t match_operator, + uint32_t max_fuzzy_distance, + int32_t coverage_mode + ); + ++/** ++ * Prepare an immutable, process-local Phrase query context for one column. ++ * ++ * The selected FTS index must store token positions. `slop == 0` requires an ++ * exact phrase; a positive value permits that many intervening positions. ++ * Dataset identity, coverage, sharing, and segment-scoped execution follow the ++ * same contract as lance_dataset_prepare_fts_match_query(). ++ * ++ * @param slop Maximum number of intervening token positions permitted between ++ * adjacent phrase terms. ++ * @param coverage_mode Fixed-width LanceFtsCoverageMode discriminant. ++ * @return Context handle on success, or NULL on error. ++ */ ++LanceFtsQueryContext* lance_dataset_prepare_fts_phrase_query( ++ const LanceDataset* dataset, ++ const char* column, ++ const char* query, ++ uint32_t slop, ++ int32_t coverage_mode ++); ++ + /** + * Close a context handle. NULL-safe. Scanners that already attached this + * context retain shared ownership and remain valid. +diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp +index 6cf245f..973216a 100644 +--- a/include/lance/lance.hpp ++++ b/include/lance/lance.hpp +@@ -128,6 +128,11 @@ enum class FtsCoverageMode : int32_t { + IndexOnly = LANCE_FTS_COVERAGE_INDEX_ONLY, + }; + ++enum class FtsMatchOperator : int32_t { ++ Or = LANCE_FTS_MATCH_OPERATOR_OR, ++ And = LANCE_FTS_MATCH_OPERATOR_AND, ++}; ++ + /// Tunable parameters for Dataset::write. Numeric fields default-out via 0; + /// `data_storage_version` defaults out via `std::nullopt`. + /// +@@ -764,18 +769,43 @@ class Dataset { + /// Create a Scanner builder for this dataset. + Scanner scan() const; + +- /// Prepare a query-specific global BM25 scorer over the committed FTS +- /// segments of this pinned snapshot. IndexOnly permits unindexed fragments; +- /// Strict rejects them. Prepared contexts currently require +- /// `max_fuzzy_distance == 0`. The context can only be attached to scanners +- /// created from this exact process-local dataset snapshot. ++ /// Compatibility wrapper for an OR Match query. ++ [[deprecated("Use prepare_fts_match_query() to select the Match operator")]] + FtsQueryContext prepare_fts_query( + const std::string& column, + const std::string& query, + uint32_t max_fuzzy_distance = 0, + FtsCoverageMode coverage_mode = FtsCoverageMode::Strict) const { +- auto* context = lance_dataset_prepare_fts_query( +- handle_.get(), column.c_str(), query.c_str(), max_fuzzy_distance, ++ return prepare_fts_match_query(column, query, FtsMatchOperator::Or, ++ max_fuzzy_distance, coverage_mode); ++ } ++ ++ /// Prepare a Match query with a global BM25 scorer. AND and OR are ++ /// supported. `max_fuzzy_distance` is reserved and currently must be zero; ++ /// keeping it here avoids another API change when canonical cross-segment ++ /// fuzzy vocabulary injection becomes available. ++ FtsQueryContext prepare_fts_match_query( ++ const std::string& column, ++ const std::string& query, ++ FtsMatchOperator match_operator = FtsMatchOperator::Or, ++ uint32_t max_fuzzy_distance = 0, ++ FtsCoverageMode coverage_mode = FtsCoverageMode::Strict) const { ++ auto* context = lance_dataset_prepare_fts_match_query( ++ handle_.get(), column.c_str(), query.c_str(), ++ static_cast(match_operator), max_fuzzy_distance, ++ static_cast(coverage_mode)); ++ if (!context) check_error(); ++ return FtsQueryContext(context); ++ } ++ ++ /// Prepare a Phrase query. Its FTS index must store token positions. ++ FtsQueryContext prepare_fts_phrase_query( ++ const std::string& column, ++ const std::string& query, ++ uint32_t slop = 0, ++ FtsCoverageMode coverage_mode = FtsCoverageMode::Strict) const { ++ auto* context = lance_dataset_prepare_fts_phrase_query( ++ handle_.get(), column.c_str(), query.c_str(), slop, + static_cast(coverage_mode)); + if (!context) check_error(); + return FtsQueryContext(context); +diff --git a/src/fts_query.rs b/src/fts_query.rs +index cd194c7..3bf7f3e 100644 +--- a/src/fts_query.rs ++++ b/src/fts_query.rs +@@ -14,7 +14,9 @@ use lance_core::{Error, Result}; + use lance_index::IndexCriteria; + use lance_index::metrics::NoOpMetricsCollector; + use lance_index::scalar::FullTextSearchQuery; +-use lance_index::scalar::inverted::query::{FtsQuery, collect_query_tokens}; ++use lance_index::scalar::inverted::query::{ ++ FtsQuery, MatchQuery, Operator, PhraseQuery, collect_query_tokens, ++}; + use lance_index::scalar::inverted::{InvertedIndex, MemBM25Scorer, build_global_bm25_scorer}; + use lance_table::format::IndexMetadata; + use uuid::Uuid; +@@ -48,12 +50,53 @@ impl TryFrom for LanceFtsCoverageMode { + } + } + ++/// Operator used to combine the analyzed terms of a Match query. ++#[repr(i32)] ++#[derive(Clone, Copy, Debug, PartialEq, Eq)] ++pub enum LanceFtsMatchOperator { ++ /// At least one analyzed term must match. ++ Or = 0, ++ /// Every analyzed term must match. ++ And = 1, ++} ++ ++impl TryFrom for LanceFtsMatchOperator { ++ type Error = Error; ++ ++ fn try_from(value: i32) -> Result { ++ match value { ++ 0 => Ok(Self::Or), ++ 1 => Ok(Self::And), ++ _ => Err(Error::invalid_input(format!( ++ "invalid match_operator {value}; expected 0 (OR) or 1 (AND)" ++ ))), ++ } ++ } ++} ++ ++impl From for Operator { ++ fn from(value: LanceFtsMatchOperator) -> Self { ++ match value { ++ LanceFtsMatchOperator::Or => Self::Or, ++ LanceFtsMatchOperator::And => Self::And, ++ } ++ } ++} ++ ++/// Query-specific state that must be shared by every segment-scoped scan. ++pub(crate) enum PreparedFtsQuery { ++ /// Exact Match queries share one corpus-wide scorer. ++ Match(Arc), ++ /// Phrase does not expand terms, so a shared global scorer is sufficient. ++ Phrase(Arc), ++} ++ + /// Rust-owned immutable state behind [`LanceFtsQueryContext`]. + pub(crate) struct FtsQueryContextInner { + pub(crate) dataset: Arc, + pub(crate) query: FullTextSearchQuery, + pub(crate) segments: Vec, +- pub(crate) scorer: Arc, ++ pub(crate) prepared: PreparedFtsQuery, + } + + impl FtsQueryContextInner { +@@ -87,7 +130,7 @@ fn invalid_input(message: impl Into) -> Error { + async fn prepare_fts_query_context( + dataset: Arc, + column: String, +- query_text: String, ++ query: FullTextSearchQuery, + coverage_mode: LanceFtsCoverageMode, + ) -> Result { + let logical_index = dataset +@@ -193,34 +236,78 @@ async fn prepare_fts_query_context( + ))); + } + +- let query = FullTextSearchQuery::new(query_text).with_column(column.clone())?; +- let match_query = match &query.query { +- FtsQuery::Match(query) => query, ++ let prepared = match &query.query { ++ FtsQuery::Match(match_query) => { ++ let mut tokenizer = indices[0].tokenizer(); ++ let query_tokens = collect_query_tokens(&match_query.terms, &mut tokenizer); ++ let params = query ++ .params() ++ .with_fuzziness(match_query.fuzziness) ++ .with_max_expansions(match_query.max_expansions) ++ .with_prefix_length(match_query.prefix_length); ++ PreparedFtsQuery::Match(Arc::new( ++ build_global_bm25_scorer(&indices, &query_tokens, ¶ms).await?, ++ )) ++ } ++ FtsQuery::Phrase(phrase_query) => { ++ if !expected_params.has_positions() { ++ return Err(invalid_input(format!( ++ "FTS index '{}' for column '{column}' does not store token positions required by Phrase queries; recreate the index with positions enabled", ++ logical_index.name ++ ))); ++ } ++ let mut tokenizer = indices[0].tokenizer(); ++ let query_tokens = collect_query_tokens(&phrase_query.terms, &mut tokenizer); ++ let params = query.params().with_phrase_slop(Some(phrase_query.slop)); ++ PreparedFtsQuery::Phrase(Arc::new( ++ build_global_bm25_scorer(&indices, &query_tokens, ¶ms).await?, ++ )) ++ } + _ => { + return Err(Error::internal( +- "prepared FTS query unexpectedly produced a non-Match query".to_string(), ++ "prepared FTS query must be a single-column Match or Phrase query".to_string(), + )); + } + }; +- let mut tokenizer = indices[0].tokenizer(); +- let query_tokens = collect_query_tokens(&match_query.terms, &mut tokenizer); +- let params = query +- .params() +- .with_fuzziness(match_query.fuzziness) +- .with_max_expansions(match_query.max_expansions) +- .with_prefix_length(match_query.prefix_length); +- let scorer = Arc::new(build_global_bm25_scorer(&indices, &query_tokens, ¶ms).await?); + + Ok(FtsQueryContextInner { + dataset, + query, + segments, +- scorer, ++ prepared, + }) + } + +-/// Prepare a process-local global BM25 scorer and the committed segment list +-/// for one single-column Match query against the dataset's pinned snapshot. ++unsafe fn parse_query_inputs( ++ dataset: *const LanceDataset, ++ column: *const c_char, ++ query: *const c_char, ++ coverage_mode: i32, ++) -> Result<(Arc, String, String, LanceFtsCoverageMode)> { ++ if dataset.is_null() || column.is_null() || query.is_null() { ++ return Err(invalid_input("dataset, column, and query must not be NULL")); ++ } ++ let column = unsafe { helpers::parse_c_string(column)? } ++ .filter(|value| !value.is_empty()) ++ .ok_or_else(|| invalid_input("column must not be empty"))? ++ .to_string(); ++ let query = unsafe { helpers::parse_c_string(query)? } ++ .filter(|value| !value.is_empty()) ++ .ok_or_else(|| invalid_input("query must not be empty"))? ++ .to_string(); ++ let coverage_mode = LanceFtsCoverageMode::try_from(coverage_mode)?; ++ let snapshot = unsafe { &*dataset }.snapshot(); ++ Ok((snapshot, column, query, coverage_mode)) ++} ++ ++fn into_context(inner: FtsQueryContextInner) -> *mut LanceFtsQueryContext { ++ Box::into_raw(Box::new(LanceFtsQueryContext { ++ inner: Arc::new(inner), ++ })) ++} ++ ++/// Compatibility API for an OR Match query. ++#[deprecated(note = "use lance_dataset_prepare_fts_match_query to select the Match operator")] + #[unsafe(no_mangle)] + pub unsafe extern "C" fn lance_dataset_prepare_fts_query( + dataset: *const LanceDataset, +@@ -231,46 +318,120 @@ pub unsafe extern "C" fn lance_dataset_prepare_fts_query( + ) -> *mut LanceFtsQueryContext { + ffi_try!( + unsafe { +- prepare_fts_query_inner(dataset, column, query, max_fuzzy_distance, coverage_mode) ++ prepare_fts_match_query_inner( ++ dataset, ++ column, ++ query, ++ LanceFtsMatchOperator::Or as i32, ++ max_fuzzy_distance, ++ coverage_mode, ++ ) ++ }, ++ null ++ ) ++} ++ ++/// Prepare a process-local Match query context. AND and OR are supported. ++/// `max_fuzzy_distance` is retained for the future prepared-fuzzy path but ++/// must be zero with the currently pinned Lance revision. ++#[unsafe(no_mangle)] ++pub unsafe extern "C" fn lance_dataset_prepare_fts_match_query( ++ dataset: *const LanceDataset, ++ column: *const c_char, ++ query: *const c_char, ++ match_operator: i32, ++ max_fuzzy_distance: u32, ++ coverage_mode: i32, ++) -> *mut LanceFtsQueryContext { ++ ffi_try!( ++ unsafe { ++ prepare_fts_match_query_inner( ++ dataset, ++ column, ++ query, ++ match_operator, ++ max_fuzzy_distance, ++ coverage_mode, ++ ) + }, + null + ) + } + +-unsafe fn prepare_fts_query_inner( ++unsafe fn prepare_fts_match_query_inner( + dataset: *const LanceDataset, + column: *const c_char, + query: *const c_char, ++ match_operator: i32, + max_fuzzy_distance: u32, + coverage_mode: i32, + ) -> Result<*mut LanceFtsQueryContext> { +- if dataset.is_null() || column.is_null() || query.is_null() { +- return Err(invalid_input("dataset, column, and query must not be NULL")); +- } +- let column = unsafe { helpers::parse_c_string(column)? } +- .filter(|value| !value.is_empty()) +- .ok_or_else(|| invalid_input("column must not be empty"))? +- .to_string(); +- let query = unsafe { helpers::parse_c_string(query)? } +- .filter(|value| !value.is_empty()) +- .ok_or_else(|| invalid_input("query must not be empty"))? +- .to_string(); +- let coverage_mode = LanceFtsCoverageMode::try_from(coverage_mode)?; ++ let (snapshot, column, query_text, coverage_mode) = ++ unsafe { parse_query_inputs(dataset, column, query, coverage_mode)? }; ++ let operator: Operator = LanceFtsMatchOperator::try_from(match_operator)?.into(); ++ // The parameter remains in the public API so callers do not need another ++ // ABI change when Lance-C moves to a Lance revision that can inject the ++ // same canonical fuzzy vocabulary into every segment-scoped scan. The ++ // pinned Lance revision can share only the scorer, so accepting fuzzy here ++ // would allow different segments to choose different capped expansions. + if max_fuzzy_distance != 0 { + return Err(invalid_input(format!( +- "max_fuzzy_distance must be 0 for prepared FTS query contexts, got {max_fuzzy_distance}; fuzzy queries require a canonical prepared BM25 vocabulary" ++ "max_fuzzy_distance must be 0 for prepared FTS with the pinned Lance revision, got {max_fuzzy_distance}; the parameter is reserved until canonical fuzzy vocabulary injection is available" + ))); + } +- let snapshot = unsafe { &*dataset }.snapshot(); ++ let query = FullTextSearchQuery::new_query( ++ MatchQuery::new(query_text) ++ .with_column(Some(column.clone())) ++ .with_operator(operator) ++ .with_fuzziness(Some(0)) ++ .into(), ++ ); + let inner = block_on(prepare_fts_query_context( + snapshot, + column, + query, + coverage_mode, + ))?; +- Ok(Box::into_raw(Box::new(LanceFtsQueryContext { +- inner: Arc::new(inner), +- }))) ++ Ok(into_context(inner)) ++} ++ ++/// Prepare a process-local Phrase query context. ++#[unsafe(no_mangle)] ++pub unsafe extern "C" fn lance_dataset_prepare_fts_phrase_query( ++ dataset: *const LanceDataset, ++ column: *const c_char, ++ query: *const c_char, ++ slop: u32, ++ coverage_mode: i32, ++) -> *mut LanceFtsQueryContext { ++ ffi_try!( ++ unsafe { prepare_fts_phrase_query_inner(dataset, column, query, slop, coverage_mode) }, ++ null ++ ) ++} ++ ++unsafe fn prepare_fts_phrase_query_inner( ++ dataset: *const LanceDataset, ++ column: *const c_char, ++ query: *const c_char, ++ slop: u32, ++ coverage_mode: i32, ++) -> Result<*mut LanceFtsQueryContext> { ++ let (snapshot, column, query_text, coverage_mode) = ++ unsafe { parse_query_inputs(dataset, column, query, coverage_mode)? }; ++ let query = FullTextSearchQuery::new_query( ++ PhraseQuery::new(query_text) ++ .with_column(Some(column.clone())) ++ .with_slop(slop) ++ .into(), ++ ); ++ let inner = block_on(prepare_fts_query_context( ++ snapshot, ++ column, ++ query, ++ coverage_mode, ++ ))?; ++ Ok(into_context(inner)) + } + + /// Close a context handle. NULL-safe. Scanners that already attached the +diff --git a/src/scanner.rs b/src/scanner.rs +index 0c29b17..cbf1b13 100644 +--- a/src/scanner.rs ++++ b/src/scanner.rs +@@ -18,7 +18,7 @@ use lance::Dataset; + use lance::dataset::scanner::{ + DatasetRecordBatchStream, ExecutionStatsCallback, ExecutionSummaryCounts, + }; +-use lance::io::exec::fts::{FlatMatchQueryExec, MatchQueryExec}; ++use lance::io::exec::fts::{FlatMatchQueryExec, MatchQueryExec, PhraseQueryExec}; + use lance_core::Result; + use lance_index::scalar::FullTextSearchQuery; + use lance_io::stream::RecordBatchStream; +@@ -33,7 +33,8 @@ use crate::error::{ + panic_payload_message, set_lance_error, set_last_error, swallow_unwind, + }; + use crate::fts_query::{ +- FtsQueryContextInner, LanceFtsQueryContext, clone_context, parse_segment_uuids, ++ FtsQueryContextInner, LanceFtsQueryContext, PreparedFtsQuery, clone_context, ++ parse_segment_uuids, + }; + use crate::helpers; + use crate::runtime::{RT, block_on}; +@@ -328,17 +329,17 @@ impl PreparedScanner { + let (plan, rewritten) = rewrite_prepared_fts_plan( + plan, + &distributed_fts.segments, +- &distributed_fts.context.scorer, ++ &distributed_fts.context.prepared, + selected_segments_have_current_fragments, + )?; +- if rewritten.match_query_execs > 1 ++ if rewritten.indexed_query_execs > 1 + || rewritten.flat_match_query_execs > 1 +- || rewritten.match_query_execs + rewritten.flat_match_query_execs == 0 +- || (selected_segments_have_current_fragments && rewritten.match_query_execs != 1) ++ || rewritten.indexed_query_execs + rewritten.flat_match_query_execs == 0 ++ || (selected_segments_have_current_fragments && rewritten.indexed_query_execs != 1) + { + return Err(lance_core::Error::internal(format!( +- "unexpected prepared FTS plan for selected segments with current fragment coverage {selected_segments_have_current_fragments}: rewrote {} MatchQueryExec node(s) and removed {} FlatMatchQueryExec node(s)", +- rewritten.match_query_execs, rewritten.flat_match_query_execs ++ "unexpected prepared FTS plan for selected segments with current fragment coverage {selected_segments_have_current_fragments}: rewrote {} indexed FTS query node(s) and removed {} FlatMatchQueryExec node(s)", ++ rewritten.indexed_query_execs, rewritten.flat_match_query_execs + ))); + } + let stream = lance_datafusion::exec::execute_plan( +@@ -420,14 +421,14 @@ fn segments_have_current_fragments( + + #[derive(Default)] + struct PreparedFtsPlanRewriteCounts { +- match_query_execs: usize, ++ indexed_query_execs: usize, + flat_match_query_execs: usize, + } + + fn rewrite_prepared_fts_plan( + plan: Arc, + segments: &[IndexMetadata], +- scorer: &Arc, ++ prepared: &PreparedFtsQuery, + selected_segments_have_current_fragments: bool, + ) -> Result<(Arc, PreparedFtsPlanRewriteCounts)> { + // Lance's ordinary FTS planner adds a flat-search branch for fragments not +@@ -438,7 +439,7 @@ fn rewrite_prepared_fts_plan( + return Ok(( + Arc::new(EmptyExec::new(plan.schema())), + PreparedFtsPlanRewriteCounts { +- match_query_execs: 0, ++ indexed_query_execs: 0, + flat_match_query_execs: 1, + }, + )); +@@ -454,11 +455,11 @@ fn rewrite_prepared_fts_plan( + let (new_child, child_rewritten) = rewrite_prepared_fts_plan( + Arc::clone(child), + segments, +- scorer, ++ prepared, + selected_segments_have_current_fragments, + )?; + new_children.push(new_child); +- rewritten.match_query_execs += child_rewritten.match_query_execs; ++ rewritten.indexed_query_execs += child_rewritten.indexed_query_execs; + rewritten.flat_match_query_execs += child_rewritten.flat_match_query_execs; + } + plan.with_new_children(new_children).map_err(|error| { +@@ -469,10 +470,15 @@ fn rewrite_prepared_fts_plan( + }; + + if let Some(exec) = rebuilt.downcast_ref::() { +- rewritten.match_query_execs += 1; ++ rewritten.indexed_query_execs += 1; + if !selected_segments_have_current_fragments { + return Ok((Arc::new(EmptyExec::new(rebuilt.schema())), rewritten)); + } ++ let PreparedFtsQuery::Match(scorer) = prepared else { ++ return Err(lance_core::Error::internal( ++ "prepared Phrase state cannot be attached to MatchQueryExec".to_string(), ++ )); ++ }; + let replacement = MatchQueryExec::new_with_segments( + Arc::clone(exec.dataset()), + exec.query().clone(), +@@ -483,6 +489,26 @@ fn rewrite_prepared_fts_plan( + .with_base_scorer(Arc::clone(scorer)); + return Ok((Arc::new(replacement), rewritten)); + } ++ if let Some(exec) = rebuilt.downcast_ref::() { ++ rewritten.indexed_query_execs += 1; ++ if !selected_segments_have_current_fragments { ++ return Ok((Arc::new(EmptyExec::new(rebuilt.schema())), rewritten)); ++ } ++ let PreparedFtsQuery::Phrase(scorer) = prepared else { ++ return Err(lance_core::Error::internal( ++ "prepared Match state cannot be attached to PhraseQueryExec".to_string(), ++ )); ++ }; ++ let replacement = PhraseQueryExec::new_with_segments( ++ Arc::clone(exec.dataset()), ++ exec.query().clone(), ++ exec.params().clone(), ++ exec.prefilter_source().clone(), ++ segments.to_vec(), ++ ) ++ .with_base_scorer(Arc::clone(scorer)); ++ return Ok((Arc::new(replacement), rewritten)); ++ } + Ok((rebuilt, rewritten)) + } + +@@ -2152,7 +2178,8 @@ mod tests { + use crate::dataset::{lance_dataset_close, lance_dataset_open}; + use crate::error::{lance_last_error_code, lance_last_error_message}; + use crate::fts_query::{ +- LanceFtsCoverageMode, lance_dataset_prepare_fts_query, lance_fts_query_context_close, ++ LanceFtsCoverageMode, LanceFtsMatchOperator, lance_dataset_prepare_fts_match_query, ++ lance_fts_query_context_close, + }; + use std::ffi::{CStr, CString}; + use std::sync::atomic::{AtomicI32, AtomicUsize}; +@@ -2276,10 +2303,11 @@ mod tests { + let column = CString::new("name").unwrap(); + let query = CString::new("a").unwrap(); + let context = unsafe { +- lance_dataset_prepare_fts_query( ++ lance_dataset_prepare_fts_match_query( + dataset, + column.as_ptr(), + query.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, + 0, + LanceFtsCoverageMode::IndexOnly as i32, + ) +@@ -2293,7 +2321,6 @@ mod tests { + let prepared = unsafe { &*scanner }.build_scanner().unwrap(); + let distributed = prepared.distributed_fts.as_ref().unwrap(); + let segments = distributed.segments.clone(); +- let scorer = Arc::clone(&distributed.context.scorer); + let plan = block_on(prepared.scanner.create_plan()).unwrap(); + assert_eq!( + prepared_fts_plan_shape(&plan), +@@ -2303,9 +2330,14 @@ mod tests { + + let has_current_fragments = + segments_have_current_fragments(&distributed.context.dataset, &segments).unwrap(); +- let (rewritten, counts) = +- rewrite_prepared_fts_plan(plan, &segments, &scorer, has_current_fragments).unwrap(); +- assert_eq!(counts.match_query_execs, 1); ++ let (rewritten, counts) = rewrite_prepared_fts_plan( ++ plan, ++ &segments, ++ &distributed.context.prepared, ++ has_current_fragments, ++ ) ++ .unwrap(); ++ assert_eq!(counts.indexed_query_execs, 1); + assert_eq!(counts.flat_match_query_execs, 1); + assert_eq!(prepared_fts_plan_shape(&rewritten), (1, 0, 0)); + +diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs +index 8805764..a3e4ef7 100644 +--- a/tests/c_api_test.rs ++++ b/tests/c_api_test.rs +@@ -5762,6 +5762,203 @@ fn load_fts_segment_uuids(uri: &str, column: &str) -> Vec<[u8; 16]> { + }) + } + ++#[test] ++#[allow(deprecated)] ++fn test_prepared_fts_match_phrase_and_legacy_compatibility() { ++ let tmp = tempfile::tempdir().unwrap(); ++ let uri = tmp ++ .path() ++ .join("prepared_fts_queries") ++ .to_str() ++ .unwrap() ++ .to_string(); ++ let schema = Arc::new(Schema::new(vec![ ++ Field::new("id", DataType::Int32, false), ++ Field::new("text", DataType::Utf8, false), ++ ])); ++ let batch = RecordBatch::try_new( ++ schema.clone(), ++ vec![ ++ Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5])), ++ Arc::new(StringArray::from(vec![ ++ "quick brown fox", ++ "quick blue fox", ++ "slow brown fox", ++ "quik brown fox", ++ "quick red brown fox", ++ ])), ++ ], ++ ) ++ .unwrap(); ++ lance_c::runtime::block_on(async { ++ Dataset::write( ++ arrow::record_batch::RecordBatchIterator::new(vec![Ok(batch)], schema), ++ &uri, ++ None, ++ ) ++ .await ++ .unwrap(); ++ }); ++ ++ let uri_c = c_str(&uri); ++ let column = c_str("text"); ++ let index_params = ++ c_str(r#"{"base_tokenizer":"simple","language":"English","with_position":true}"#); ++ let dataset = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; ++ assert_eq!( ++ unsafe { ++ lance_dataset_create_scalar_index( ++ dataset, ++ column.as_ptr(), ++ ptr::null(), ++ LanceScalarIndexType::Inverted as i32, ++ index_params.as_ptr(), ++ false, ++ ) ++ }, ++ 0 ++ ); ++ ++ let query = c_str("quick brown"); ++ let exact_or = unsafe { ++ lance_dataset_prepare_fts_match_query( ++ dataset, ++ column.as_ptr(), ++ query.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, ++ 0, ++ LanceFtsCoverageMode::Strict as i32, ++ ) ++ }; ++ assert!(!exact_or.is_null()); ++ assert_eq!(collect_context_fts_scores(dataset, exact_or, None).len(), 5); ++ unsafe { lance_fts_query_context_close(exact_or) }; ++ ++ let legacy_or = unsafe { ++ lance_dataset_prepare_fts_query( ++ dataset, ++ column.as_ptr(), ++ query.as_ptr(), ++ 0, ++ LanceFtsCoverageMode::Strict as i32, ++ ) ++ }; ++ assert!(!legacy_or.is_null()); ++ assert_eq!( ++ collect_context_fts_scores(dataset, legacy_or, None).len(), ++ 5 ++ ); ++ unsafe { lance_fts_query_context_close(legacy_or) }; ++ ++ let exact_and = unsafe { ++ lance_dataset_prepare_fts_match_query( ++ dataset, ++ column.as_ptr(), ++ query.as_ptr(), ++ LanceFtsMatchOperator::And as i32, ++ 0, ++ LanceFtsCoverageMode::Strict as i32, ++ ) ++ }; ++ assert!(!exact_and.is_null()); ++ let exact_and_scores = collect_context_fts_scores(dataset, exact_and, None); ++ let mut exact_and_ids = exact_and_scores.keys().copied().collect::>(); ++ exact_and_ids.sort_unstable(); ++ assert_eq!(exact_and_ids, vec![1, 5]); ++ unsafe { lance_fts_query_context_close(exact_and) }; ++ ++ let fuzzy_and = unsafe { ++ lance_dataset_prepare_fts_match_query( ++ dataset, ++ column.as_ptr(), ++ query.as_ptr(), ++ LanceFtsMatchOperator::And as i32, ++ 1, ++ LanceFtsCoverageMode::Strict as i32, ++ ) ++ }; ++ assert!(fuzzy_and.is_null()); ++ let message = take_last_error_message(); ++ assert!( ++ message.contains("max_fuzzy_distance must be 0"), ++ "{message}" ++ ); ++ ++ let phrase = unsafe { ++ lance_dataset_prepare_fts_phrase_query( ++ dataset, ++ column.as_ptr(), ++ query.as_ptr(), ++ 0, ++ LanceFtsCoverageMode::Strict as i32, ++ ) ++ }; ++ assert!(!phrase.is_null(), "{}", take_last_error_message()); ++ let phrase_scores = collect_context_fts_scores(dataset, phrase, None); ++ assert_eq!(phrase_scores.keys().copied().collect::>(), vec![1]); ++ unsafe { lance_fts_query_context_close(phrase) }; ++ ++ let phrase_with_slop = unsafe { ++ lance_dataset_prepare_fts_phrase_query( ++ dataset, ++ column.as_ptr(), ++ query.as_ptr(), ++ 1, ++ LanceFtsCoverageMode::Strict as i32, ++ ) ++ }; ++ assert!(!phrase_with_slop.is_null(), "{}", take_last_error_message()); ++ let phrase_with_slop_scores = collect_context_fts_scores(dataset, phrase_with_slop, None); ++ let mut phrase_with_slop_ids = phrase_with_slop_scores.keys().copied().collect::>(); ++ phrase_with_slop_ids.sort_unstable(); ++ assert_eq!(phrase_with_slop_ids, vec![1, 5]); ++ unsafe { lance_fts_query_context_close(phrase_with_slop) }; ++ ++ unsafe { lance_dataset_close(dataset) }; ++} ++ ++#[test] ++fn test_prepared_fts_phrase_requires_positions() { ++ let (_tmp, uri) = create_test_dataset(); ++ let uri_c = c_str(&uri); ++ let column = c_str("name"); ++ let query = c_str("alice smith"); ++ let index_params = c_str(r#"{"base_tokenizer":"simple","language":"English"}"#); ++ let dataset = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; ++ assert_eq!( ++ unsafe { ++ lance_dataset_create_scalar_index( ++ dataset, ++ column.as_ptr(), ++ ptr::null(), ++ LanceScalarIndexType::Inverted as i32, ++ index_params.as_ptr(), ++ false, ++ ) ++ }, ++ 0 ++ ); ++ ++ let context = unsafe { ++ lance_dataset_prepare_fts_phrase_query( ++ dataset, ++ column.as_ptr(), ++ query.as_ptr(), ++ 0, ++ LanceFtsCoverageMode::Strict as i32, ++ ) ++ }; ++ assert!(context.is_null()); ++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); ++ let message = take_last_error_message(); ++ assert!( ++ message.contains("does not store token positions"), ++ "{message}" ++ ); ++ ++ unsafe { lance_dataset_close(dataset) }; ++} ++ + #[test] + fn test_prepared_fts_row_id_output_is_explicit() { + let (_tmp, uri) = create_test_dataset(); +@@ -5785,10 +5982,11 @@ fn test_prepared_fts_row_id_output_is_explicit() { + 0 + ); + let context = unsafe { +- lance_dataset_prepare_fts_query( ++ lance_dataset_prepare_fts_match_query( + dataset, + column.as_ptr(), + query.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, + 0, + LanceFtsCoverageMode::Strict as i32, + ) +@@ -5880,10 +6078,11 @@ fn test_prepare_fts_query_index_only_allows_unindexed_fragment() { + + let dataset = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + let strict = unsafe { +- lance_dataset_prepare_fts_query( ++ lance_dataset_prepare_fts_match_query( + dataset, + column.as_ptr(), + query.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, + 0, + LanceFtsCoverageMode::Strict as i32, + ) +@@ -5898,10 +6097,11 @@ fn test_prepare_fts_query_index_only_allows_unindexed_fragment() { + assert!(message.contains("unindexed fragments"), "{message}"); + + let context = unsafe { +- lance_dataset_prepare_fts_query( ++ lance_dataset_prepare_fts_match_query( + dataset, + column.as_ptr(), + query.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, + 0, + LanceFtsCoverageMode::IndexOnly as i32, + ) +@@ -5976,10 +6176,11 @@ fn test_prepared_fts_index_only_empty_segment_returns_empty_shard() { + let query = c_str("alice"); + let dataset = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + let context = unsafe { +- lance_dataset_prepare_fts_query( ++ lance_dataset_prepare_fts_match_query( + dataset, + column.as_ptr(), + query.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, + 0, + LanceFtsCoverageMode::IndexOnly as i32, + ) +@@ -6075,10 +6276,11 @@ fn test_prepared_fts_global_scorer_is_shared_across_segment_splits() { + + let dataset = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + let context = unsafe { +- lance_dataset_prepare_fts_query( ++ lance_dataset_prepare_fts_match_query( + dataset, + column.as_ptr(), + query.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, + 0, + LanceFtsCoverageMode::Strict as i32, + ) +@@ -6186,7 +6388,7 @@ fn test_prepared_fts_global_scorer_is_shared_across_segment_splits() { + } + + #[test] +-fn test_prepare_fts_query_rejects_null_empty_invalid_mode_and_fuzzy() { ++fn test_prepare_fts_queries_reject_invalid_inputs() { + let (_tmp, uri) = create_test_dataset(); + let uri_c = c_str(&uri); + let dataset = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; +@@ -6196,10 +6398,11 @@ fn test_prepare_fts_query_rejects_null_empty_invalid_mode_and_fuzzy() { + + assert!( + unsafe { +- lance_dataset_prepare_fts_query( ++ lance_dataset_prepare_fts_match_query( + ptr::null(), + column.as_ptr(), + query.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, + 0, + LanceFtsCoverageMode::Strict as i32, + ) +@@ -6208,10 +6411,11 @@ fn test_prepare_fts_query_rejects_null_empty_invalid_mode_and_fuzzy() { + ); + assert!( + unsafe { +- lance_dataset_prepare_fts_query( ++ lance_dataset_prepare_fts_match_query( + dataset, + empty.as_ptr(), + query.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, + 0, + LanceFtsCoverageMode::Strict as i32, + ) +@@ -6219,20 +6423,39 @@ fn test_prepare_fts_query_rejects_null_empty_invalid_mode_and_fuzzy() { + .is_null() + ); + assert!( +- unsafe { lance_dataset_prepare_fts_query(dataset, column.as_ptr(), empty.as_ptr(), 0, 0) } +- .is_null() ++ unsafe { ++ lance_dataset_prepare_fts_match_query( ++ dataset, ++ column.as_ptr(), ++ empty.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, ++ 0, ++ LanceFtsCoverageMode::Strict as i32, ++ ) ++ } ++ .is_null() + ); + assert!( +- unsafe { lance_dataset_prepare_fts_query(dataset, column.as_ptr(), query.as_ptr(), 0, 99) } +- .is_null() ++ unsafe { ++ lance_dataset_prepare_fts_match_query( ++ dataset, ++ column.as_ptr(), ++ query.as_ptr(), ++ LanceFtsMatchOperator::Or as i32, ++ 0, ++ 99, ++ ) ++ } ++ .is_null() + ); + assert!( + unsafe { +- lance_dataset_prepare_fts_query( ++ lance_dataset_prepare_fts_match_query( + dataset, + column.as_ptr(), + query.as_ptr(), +- 1, ++ 99, ++ 0, + LanceFtsCoverageMode::Strict as i32, + ) + } +@@ -6244,9 +6467,18 @@ fn test_prepare_fts_query_rejects_null_empty_invalid_mode_and_fuzzy() { + .to_string_lossy() + .into_owned() + }; ++ assert!(message.contains("invalid match_operator"), "{message}"); + assert!( +- message.contains("max_fuzzy_distance must be 0"), +- "{message}" ++ unsafe { ++ lance_dataset_prepare_fts_phrase_query( ++ dataset, ++ column.as_ptr(), ++ ptr::null(), ++ 0, ++ LanceFtsCoverageMode::Strict as i32, ++ ) ++ } ++ .is_null() + ); + let scanner = unsafe { lance_scanner_new(dataset, ptr::null(), ptr::null()) }; + assert_eq!( + +From 9bb38749c9de185664511def3cbeb699ee60a38b Mon Sep 17 00:00:00 2001 +From: zhangstar333 +Date: Thu, 3 Sep 2026 17:49:02 +0800 +Subject: [PATCH 2/2] update slop i32 + +--- + include/lance/lance.h | 6 +++--- + include/lance/lance.hpp | 5 +++-- + src/fts_query.rs | 6 ++++-- + tests/c_api_test.rs | 14 ++++++++++++++ + 4 files changed, 24 insertions(+), 7 deletions(-) + +diff --git a/include/lance/lance.h b/include/lance/lance.h +index 0c76edc..25c0430 100644 +--- a/include/lance/lance.h ++++ b/include/lance/lance.h +@@ -1761,8 +1761,8 @@ LanceFtsQueryContext* lance_dataset_prepare_fts_match_query( + * Dataset identity, coverage, sharing, and segment-scoped execution follow the + * same contract as lance_dataset_prepare_fts_match_query(). + * +- * @param slop Maximum number of intervening token positions permitted between +- * adjacent phrase terms. ++ * @param slop Maximum non-negative number of intervening token positions ++ * permitted between adjacent phrase terms. + * @param coverage_mode Fixed-width LanceFtsCoverageMode discriminant. + * @return Context handle on success, or NULL on error. + */ +@@ -1770,7 +1770,7 @@ LanceFtsQueryContext* lance_dataset_prepare_fts_phrase_query( + const LanceDataset* dataset, + const char* column, + const char* query, +- uint32_t slop, ++ int32_t slop, + int32_t coverage_mode + ); + +diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp +index 973216a..1e69a06 100644 +--- a/include/lance/lance.hpp ++++ b/include/lance/lance.hpp +@@ -798,11 +798,12 @@ class Dataset { + return FtsQueryContext(context); + } + +- /// Prepare a Phrase query. Its FTS index must store token positions. ++ /// Prepare a Phrase query. Its FTS index must store token positions and ++ /// slop must be non-negative. + FtsQueryContext prepare_fts_phrase_query( + const std::string& column, + const std::string& query, +- uint32_t slop = 0, ++ int32_t slop = 0, + FtsCoverageMode coverage_mode = FtsCoverageMode::Strict) const { + auto* context = lance_dataset_prepare_fts_phrase_query( + handle_.get(), column.c_str(), query.c_str(), slop, +diff --git a/src/fts_query.rs b/src/fts_query.rs +index 3bf7f3e..c9d3a43 100644 +--- a/src/fts_query.rs ++++ b/src/fts_query.rs +@@ -401,7 +401,7 @@ pub unsafe extern "C" fn lance_dataset_prepare_fts_phrase_query( + dataset: *const LanceDataset, + column: *const c_char, + query: *const c_char, +- slop: u32, ++ slop: i32, + coverage_mode: i32, + ) -> *mut LanceFtsQueryContext { + ffi_try!( +@@ -414,9 +414,11 @@ unsafe fn prepare_fts_phrase_query_inner( + dataset: *const LanceDataset, + column: *const c_char, + query: *const c_char, +- slop: u32, ++ slop: i32, + coverage_mode: i32, + ) -> Result<*mut LanceFtsQueryContext> { ++ let slop = u32::try_from(slop) ++ .map_err(|_| invalid_input(format!("slop must be non-negative, got {slop}")))?; + let (snapshot, column, query_text, coverage_mode) = + unsafe { parse_query_inputs(dataset, column, query, coverage_mode)? }; + let query = FullTextSearchQuery::new_query( +diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs +index a3e4ef7..9411e15 100644 +--- a/tests/c_api_test.rs ++++ b/tests/c_api_test.rs +@@ -5914,6 +5914,20 @@ fn test_prepared_fts_match_phrase_and_legacy_compatibility() { + assert_eq!(phrase_with_slop_ids, vec![1, 5]); + unsafe { lance_fts_query_context_close(phrase_with_slop) }; + ++ let negative_phrase_slop = unsafe { ++ lance_dataset_prepare_fts_phrase_query( ++ dataset, ++ column.as_ptr(), ++ query.as_ptr(), ++ -1, ++ LanceFtsCoverageMode::Strict as i32, ++ ) ++ }; ++ assert!(negative_phrase_slop.is_null()); ++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); ++ let message = take_last_error_message(); ++ assert!(message.contains("slop must be non-negative"), "{message}"); ++ + unsafe { lance_dataset_close(dataset) }; + } + diff --git a/thirdparty/vars.sh b/thirdparty/vars.sh index fa7643d62c33bb..886a0eeff9f858 100644 --- a/thirdparty/vars.sh +++ b/thirdparty/vars.sh @@ -603,10 +603,10 @@ PUGIXML_SOURCE=pugixml-1.15 PUGIXML_MD5SUM="3b894c29455eb33a40b165c6e2de5895" # lance-c -LANCE_C_DOWNLOAD="https://github.com/lance-format/lance-c/archive/refs/tags/v0.1.8.tar.gz" -LANCE_C_NAME="lance-c-v0.1.8.tar.gz" -LANCE_C_SOURCE="lance-c-0.1.8" -LANCE_C_MD5SUM="2a4af9398cdec19d5d379a27353b1266" +LANCE_C_DOWNLOAD="https://github.com/lance-format/lance-c/archive/refs/tags/v0.1.9.tar.gz" +LANCE_C_NAME="lance-c-v0.1.9.tar.gz" +LANCE_C_SOURCE="lance-c-0.1.9" +LANCE_C_MD5SUM="7138ed44e92d4bc91d5b522a6b92ed64" # all thirdparties which need to be downloaded is set in array TP_ARCHIVES export TP_ARCHIVES=(