This is an automated email from the ASF dual-hosted git repository.
zhangstar333 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 952bfcbb40f [chore](lance) update some patch about lance (#67262)
952bfcbb40f is described below
commit 952bfcbb40fa756cb66a7d8e70d559496eb344c8
Author: zhangstar333 <[email protected]>
AuthorDate: Tue Sep 1 19:56:29 2026 +0800
[chore](lance) update some patch about lance (#67262)
update some patch about lance
---
thirdparty/download-thirdparty.sh | 6 +-
thirdparty/patches/lance-c-0.1.7-pr-64.patch | 1522 --------------------------
thirdparty/patches/lance-c-0.1.8-pr-69.patch | 653 +++++++++++
thirdparty/vars.sh | 8 +-
4 files changed, 660 insertions(+), 1529 deletions(-)
diff --git a/thirdparty/download-thirdparty.sh
b/thirdparty/download-thirdparty.sh
index b377cbf08a7..95c67e2cdad 100755
--- a/thirdparty/download-thirdparty.sh
+++ b/thirdparty/download-thirdparty.sh
@@ -803,12 +803,12 @@ if [[ " ${TP_ARCHIVES[*]} " =~ " PAIMON_CPP " ]]; then
echo "Finished patching ${PAIMON_CPP_SOURCE}"
fi
-# Patch lance-c with the scan execution statistics API from upstream PR #64.
+# Apply Doris lance-c patches.
if [[ " ${TP_ARCHIVES[*]} " =~ " LANCE_C " ]]; then
- if [[ "${LANCE_C_SOURCE}" == "lance-c-0.1.7" ]]; then
+ if [[ "${LANCE_C_SOURCE}" == "lance-c-0.1.8" ]]; then
cd "${TP_SOURCE_DIR}/${LANCE_C_SOURCE}"
if [[ ! -f "${PATCHED_MARK}" ]]; then
- patch -p1 <"${TP_PATCH_DIR}/lance-c-0.1.7-pr-64.patch"
+ patch -p1 <"${TP_PATCH_DIR}/lance-c-0.1.8-pr-69.patch"
touch "${PATCHED_MARK}"
fi
cd -
diff --git a/thirdparty/patches/lance-c-0.1.7-pr-64.patch
b/thirdparty/patches/lance-c-0.1.7-pr-64.patch
deleted file mode 100644
index 17f91bc7c31..00000000000
--- a/thirdparty/patches/lance-c-0.1.7-pr-64.patch
+++ /dev/null
@@ -1,1522 +0,0 @@
-From e3320c1e7d5c72234b3b44e9e7e9a93a72fe488c Mon Sep 17 00:00:00 2001
-From: zhangstar333 <[email protected]>
-Date: Mon, 24 Aug 2026 16:37:33 +0800
-Subject: [PATCH 1/3] update
-
----
- include/lance/lance.h | 73 ++++++++++++++++
- include/lance/lance.hpp | 10 +++
- src/scanner.rs | 179 +++++++++++++++++++++++++++++++++++++-
- tests/c_api_test.rs | 188 +++++++++++++++++++++++++++++++++++++++-
- 4 files changed, 448 insertions(+), 2 deletions(-)
-
-diff --git a/include/lance/lance.h b/include/lance/lance.h
-index 986905b..c6c3985 100644
---- a/include/lance/lance.h
-+++ b/include/lance/lance.h
-@@ -863,6 +863,79 @@ int32_t lance_scanner_set_substrait_filter(
- size_t len
- );
-
-+/** Type of a dynamically named scan metric. */
-+typedef enum {
-+ LANCE_SCAN_METRIC_COUNT = 0,
-+ LANCE_SCAN_METRIC_TIME_NANOSECONDS = 1,
-+} LanceScanMetricKind;
-+
-+/**
-+ * Borrowed view of one dynamically named scan metric.
-+ *
-+ * `name` is not NUL-terminated. `name` and this structure are valid only for
-+ * the duration of the LanceScanStatisticsCallback invocation.
-+ */
-+typedef struct {
-+ const char* name;
-+ size_t name_len;
-+ LanceScanMetricKind kind;
-+ uint64_t value;
-+} LanceScanMetric;
-+
-+/**
-+ * Borrowed view of the execution statistics for one finalized scan.
-+ *
-+ * The fixed fields are stable summary metrics. `metrics` contains additional
-+ * implementation-specific counters and timings. Those names are not a stable
-+ * API and are intended for diagnostics and profiles. Dynamic metrics are
-+ * best-effort and may be omitted if they cannot be materialized. `metrics` is
-+ * NULL when `metrics_len` is zero.
-+ */
-+typedef struct {
-+ uint64_t iops;
-+ uint64_t requests;
-+ uint64_t bytes_read;
-+ uint64_t indices_loaded;
-+ uint64_t index_partitions_loaded;
-+ uint64_t index_comparisons;
-+ const LanceScanMetric* metrics;
-+ size_t metrics_len;
-+} LanceScanStatistics;
-+
-+/**
-+ * Receives scan statistics when a stream reaches EOF, fails, or is released.
-+ *
-+ * The statistics and all nested pointers are borrowed and valid only for the
-+ * duration of this call. The callback may run on the thread that consumes or
-+ * releases the scan stream and must therefore be thread-safe. It must return
-+ * normally without throwing an exception or unwinding, and must not call any
-+ * `lance_scanner_*` function with the originating scanner.
-+ *
-+ * Scan statistics are diagnostic and best-effort. The callback must handle
its
-+ * own errors and must not use them to abort or throw across this FFI
boundary.
-+ */
-+typedef void (*LanceScanStatisticsCallback)(
-+ void* callback_ctx,
-+ const LanceScanStatistics* statistics
-+);
-+
-+/**
-+ * Register the execution-statistics callback for this scanner.
-+ *
-+ * Must be called before starting the scan; registering after the scan starts
-+ * returns an error. `callback` must not be NULL. `callback_ctx` may be NULL.
A
-+ * non-NULL `callback_ctx` must remain valid, and `callback` must remain
valid,
-+ * until the stream reaches EOF, fails, or is released. Replaces a previously
-+ * registered callback.
-+ *
-+ * @return 0 on success, -1 on error
-+ */
-+int32_t lance_scanner_set_statistics_callback(
-+ LanceScanner* scanner,
-+ LanceScanStatisticsCallback callback,
-+ void* callback_ctx
-+);
-+
- /** Close and free a scanner handle. */
- void lance_scanner_close(LanceScanner* scanner);
-
-diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp
-index afce358..9440a23 100644
---- a/include/lance/lance.hpp
-+++ b/include/lance/lance.hpp
-@@ -1127,6 +1127,16 @@ class Scanner {
- return substrait_filter(bytes.data(), bytes.size());
- }
-
-+ /// Register a callback for scan execution statistics before starting the
scan.
-+ /// The callback may run on the thread that consumes or releases the
stream. It
-+ /// must be thread-safe, must not throw, and must not re-enter the
originating
-+ /// scanner. A non-null callback context must outlive the exported stream.
-+ Scanner& statistics_callback(LanceScanStatisticsCallback callback, void*
callback_ctx) {
-+ if (lance_scanner_set_statistics_callback(handle_.get(), callback,
callback_ctx) != 0)
-+ check_error();
-+ return *this;
-+ }
-+
- /// Restrict the next k-NN query to a subset of vector index segments.
- /// Pass `len` 16-byte UUIDs concatenated as a single byte buffer
- /// (total bytes = `len * 16`). Pass len=0 (and any pointer) to clear.
-diff --git a/src/scanner.rs b/src/scanner.rs
-index f44b82f..d95089e 100644
---- a/src/scanner.rs
-+++ b/src/scanner.rs
-@@ -14,7 +14,9 @@ use arrow::ffi_stream::FFI_ArrowArrayStream;
- use arrow_schema::SchemaRef;
- use futures::{FutureExt, Stream, StreamExt};
- use lance::Dataset;
--use lance::dataset::scanner::DatasetRecordBatchStream;
-+use lance::dataset::scanner::{
-+ DatasetRecordBatchStream, ExecutionStatsCallback, ExecutionSummaryCounts,
-+};
- use lance_core::Result;
- use lance_index::scalar::FullTextSearchQuery;
- use lance_io::stream::RecordBatchStream;
-@@ -69,6 +71,8 @@ pub struct LanceScanner {
- // the spawned async task can poison the handle from outside this call
- // frame via `poison_flag()`.
- poisoned: Arc<AtomicBool>,
-+ scan_statistics_callback: Option<ExecutionStatsCallback>,
-+ scan_started: AtomicBool,
- // Materialized on first iteration call
- stream: Option<Pin<Box<DatasetRecordBatchStream>>>,
- #[allow(dead_code)]
-@@ -122,6 +126,8 @@ impl LanceScanner {
- prefilter: false,
- fts_query: None,
- poisoned: Arc::new(AtomicBool::new(false)),
-+ scan_statistics_callback: None,
-+ scan_started: AtomicBool::new(false),
- stream: None,
- schema: None,
- }
-@@ -157,6 +163,7 @@ impl LanceScanner {
-
- /// Build the underlying Scanner and open a stream.
- fn materialize_stream(&mut self) -> Result<()> {
-+ self.scan_started.store(true, Ordering::Release);
- let mut scanner = self.dataset.scan();
- if let Some(cols) = &self.columns {
- scanner.project(cols)?;
-@@ -212,6 +219,9 @@ impl LanceScanner {
- if let Some(fts) = &self.fts_query {
- scanner.full_text_search(fts.clone())?;
- }
-+ if let Some(callback) = &self.scan_statistics_callback {
-+ scanner.scan_stats_callback(callback.clone());
-+ }
- let stream = block_on(scanner.try_into_stream())?;
- self.schema = Some(stream.schema());
- self.stream = Some(Box::pin(stream));
-@@ -220,6 +230,7 @@ impl LanceScanner {
-
- /// Build a Scanner (without materializing) and return it.
- fn build_scanner(&self) -> Result<lance::dataset::scanner::Scanner> {
-+ self.scan_started.store(true, Ordering::Release);
- let mut scanner = self.dataset.scan();
- if let Some(cols) = &self.columns {
- scanner.project(cols)?;
-@@ -274,10 +285,122 @@ impl LanceScanner {
- if let Some(fts) = &self.fts_query {
- scanner.full_text_search(fts.clone())?;
- }
-+ if let Some(callback) = &self.scan_statistics_callback {
-+ scanner.scan_stats_callback(callback.clone());
-+ }
- Ok(scanner)
- }
- }
-
-+/// Type of a dynamically named scan metric.
-+#[repr(i32)]
-+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-+pub enum LanceScanMetricKind {
-+ /// Monotonically accumulated counter.
-+ Count = 0,
-+ /// Accumulated duration in nanoseconds.
-+ TimeNanoseconds = 1,
-+}
-+
-+/// Borrowed view of one dynamically named scan metric.
-+///
-+/// `name` is not NUL-terminated. Both `name` and this structure are valid
only
-+/// for the duration of the scan statistics callback.
-+#[repr(C)]
-+#[derive(Clone, Copy, Debug)]
-+pub struct LanceScanMetric {
-+ pub name: *const c_char,
-+ pub name_len: usize,
-+ pub kind: LanceScanMetricKind,
-+ pub value: u64,
-+}
-+
-+/// Borrowed view of the execution statistics for one completed scan.
-+///
-+/// The fixed fields are stable summary metrics. `metrics` contains additional
-+/// implementation-specific counters and timings and is valid only for the
-+/// duration of the callback.
-+#[repr(C)]
-+#[derive(Clone, Copy, Debug)]
-+pub struct LanceScanStatistics {
-+ pub iops: u64,
-+ pub requests: u64,
-+ pub bytes_read: u64,
-+ pub indices_loaded: u64,
-+ pub index_partitions_loaded: u64,
-+ pub index_comparisons: u64,
-+ pub metrics: *const LanceScanMetric,
-+ pub metrics_len: usize,
-+}
-+
-+/// Callback invoked when a scan stream reaches EOF, fails, or is released.
-+///
-+/// The callback is an FFI boundary and must return normally without unwinding
-+/// or throwing an exception. It must not call back into `lance_scanner_*`
with
-+/// the originating scanner.
-+pub type LanceScanStatisticsCallback =
-+ Option<unsafe extern "C" fn(ctx: *mut c_void, statistics: *const
LanceScanStatistics)>;
-+
-+struct SendScanStatisticsCallback {
-+ callback: unsafe extern "C" fn(*mut c_void, *const LanceScanStatistics),
-+ ctx: *mut c_void,
-+}
-+
-+// SAFETY: The C API requires the callback and its context to remain valid and
-+// safe to invoke from the thread that consumes or releases the scan stream.
-+unsafe impl Send for SendScanStatisticsCallback {}
-+unsafe impl Sync for SendScanStatisticsCallback {}
-+
-+impl SendScanStatisticsCallback {
-+ fn invoke(&self, counts: &ExecutionSummaryCounts) {
-+ // Dynamic profile metrics are best-effort. Use fallible reservation
so
-+ // allocation failure omits them instead of aborting the embedding
process.
-+ let mut metrics = Vec::new();
-+ if let Some(metrics_len) =
counts.all_counts.len().checked_add(counts.all_times.len())
-+ && metrics.try_reserve_exact(metrics_len).is_ok()
-+ {
-+ metrics.extend(
-+ counts
-+ .all_counts
-+ .iter()
-+ .map(|(name, value)| LanceScanMetric {
-+ name: name.as_ptr().cast(),
-+ name_len: name.len(),
-+ kind: LanceScanMetricKind::Count,
-+ value: *value as u64,
-+ }),
-+ );
-+ metrics.extend(
-+ counts
-+ .all_times
-+ .iter()
-+ .map(|(name, value)| LanceScanMetric {
-+ name: name.as_ptr().cast(),
-+ name_len: name.len(),
-+ kind: LanceScanMetricKind::TimeNanoseconds,
-+ value: *value as u64,
-+ }),
-+ );
-+ }
-+
-+ let statistics = LanceScanStatistics {
-+ iops: counts.iops as u64,
-+ requests: counts.requests as u64,
-+ bytes_read: counts.bytes_read as u64,
-+ indices_loaded: counts.indices_loaded as u64,
-+ index_partitions_loaded: counts.parts_loaded as u64,
-+ index_comparisons: counts.index_comparisons as u64,
-+ metrics: if metrics.is_empty() {
-+ ptr::null()
-+ } else {
-+ metrics.as_ptr()
-+ },
-+ metrics_len: metrics.len(),
-+ };
-+ unsafe { (self.callback)(self.ctx, &statistics) };
-+ }
-+}
-+
- // ---------------------------------------------------------------------------
- // Poison check shared by all `lance_scanner_*` entry points
- // ---------------------------------------------------------------------------
-@@ -529,6 +652,60 @@ unsafe fn scanner_set_substrait_filter_inner(
- Ok(0)
- }
-
-+/// Register a callback that receives execution statistics when the scan
stream
-+/// reaches EOF, fails, or is released.
-+///
-+/// The callback and `callback_ctx` must remain valid until the scan stream is
-+/// finalized. Metric names and arrays passed to the callback are borrowed and
-+/// must be copied if the caller needs to retain them. The callback must be
-+/// thread-safe, must return normally without unwinding or throwing an
-+/// exception, and must not call `lance_scanner_*` with the originating
scanner.
-+#[unsafe(no_mangle)]
-+pub unsafe extern "C" fn lance_scanner_set_statistics_callback(
-+ scanner: *mut LanceScanner,
-+ callback: LanceScanStatisticsCallback,
-+ callback_ctx: *mut c_void,
-+) -> i32 {
-+ scanner_poison_check!(scanner, -1);
-+ ffi_try!(
-+ unsafe { scanner_set_statistics_callback_inner(scanner, callback,
callback_ctx) },
-+ neg
-+ )
-+}
-+
-+unsafe fn scanner_set_statistics_callback_inner(
-+ scanner: *mut LanceScanner,
-+ callback: LanceScanStatisticsCallback,
-+ callback_ctx: *mut c_void,
-+) -> Result<i32> {
-+ if scanner.is_null() {
-+ return Err(lance_core::Error::invalid_input_source(
-+ "scanner is NULL".into(),
-+ ));
-+ }
-+ let Some(callback) = callback else {
-+ return Err(lance_core::Error::invalid_input_source(
-+ "statistics callback is NULL".into(),
-+ ));
-+ };
-+
-+ let s = unsafe { &mut *scanner };
-+ if s.scan_started.load(Ordering::Acquire) {
-+ return Err(lance_core::Error::invalid_input_source(
-+ "statistics callback must be registered before the scan
starts".into(),
-+ ));
-+ }
-+
-+ let callback = SendScanStatisticsCallback {
-+ callback,
-+ ctx: callback_ctx,
-+ };
-+ s.scan_statistics_callback = Some(Arc::new(move |counts:
&ExecutionSummaryCounts| {
-+ callback.invoke(counts);
-+ }));
-+ Ok(0)
-+}
-+
- /// Close and free a scanner handle.
- ///
- /// Best-effort (issue #61): this drops a possibly-live
-diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs
-index d6cd928..db35fea 100644
---- a/tests/c_api_test.rs
-+++ b/tests/c_api_test.rs
-@@ -6,7 +6,7 @@
- //! These tests call the `extern "C"` functions directly from Rust,
- //! validating the C API contract without needing a C compiler.
-
--use std::ffi::{CString, c_char};
-+use std::ffi::{CString, c_char, c_void};
- use std::process::Command;
- use std::ptr;
- use std::sync::Arc;
-@@ -99,6 +99,58 @@ fn c_str(s: &str) -> CString {
- CString::new(s).unwrap()
- }
-
-+#[derive(Default)]
-+struct CapturedScanStatistics {
-+ calls: usize,
-+ iops: u64,
-+ requests: u64,
-+ bytes_read: u64,
-+ indices_loaded: u64,
-+ index_partitions_loaded: u64,
-+ index_comparisons: u64,
-+ metrics: Vec<(String, LanceScanMetricKind, u64)>,
-+}
-+
-+unsafe extern "C" fn capture_scan_statistics(
-+ callback_ctx: *mut c_void,
-+ statistics: *const LanceScanStatistics,
-+) {
-+ assert!(!callback_ctx.is_null());
-+ assert!(!statistics.is_null());
-+ let captured = unsafe { &mut
*callback_ctx.cast::<CapturedScanStatistics>() };
-+ let statistics = unsafe { &*statistics };
-+ let metrics = if statistics.metrics_len == 0 {
-+ &[]
-+ } else {
-+ assert!(!statistics.metrics.is_null());
-+ unsafe { std::slice::from_raw_parts(statistics.metrics,
statistics.metrics_len) }
-+ };
-+
-+ captured.calls += 1;
-+ captured.iops = statistics.iops;
-+ captured.requests = statistics.requests;
-+ captured.bytes_read = statistics.bytes_read;
-+ captured.indices_loaded = statistics.indices_loaded;
-+ captured.index_partitions_loaded = statistics.index_partitions_loaded;
-+ captured.index_comparisons = statistics.index_comparisons;
-+ captured.metrics = metrics
-+ .iter()
-+ .map(|metric| {
-+ let name = if metric.name_len == 0 {
-+ &[]
-+ } else {
-+ assert!(!metric.name.is_null());
-+ unsafe { std::slice::from_raw_parts(metric.name.cast::<u8>(),
metric.name_len) }
-+ };
-+ (
-+ std::str::from_utf8(name).unwrap().to_owned(),
-+ metric.kind,
-+ metric.value,
-+ )
-+ })
-+ .collect();
-+}
-+
- /// Helper: build a tiny dataset whose `value` column is nullable AND contains
- /// at least one NULL. Used by tests that need to exercise upstream's
- /// nullability-tightening pre-scan failure path.
-@@ -284,6 +336,140 @@ fn test_scanner_to_arrow_stream() {
- unsafe { lance_dataset_close(ds) };
- }
-
-+#[test]
-+fn test_scanner_statistics_callback_with_next_multi_fragment() {
-+ let (_tmp, uri) = create_multi_fragment_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+ assert_eq!(unsafe { lance_dataset_fragment_count(ds) }, 2);
-+
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
-+ assert!(!scanner.is_null());
-+ let mut captured = CapturedScanStatistics::default();
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics),
-+ (&mut captured as *mut CapturedScanStatistics).cast(),
-+ )
-+ },
-+ 0
-+ );
-+
-+ loop {
-+ let mut batch = ptr::null_mut();
-+ match unsafe { lance_scanner_next(scanner, &mut batch) } {
-+ 0 => unsafe { lance_batch_free(batch) },
-+ 1 => break,
-+ status => panic!("scanner_next returned error: {status}"),
-+ }
-+ }
-+
-+ assert_eq!(captured.calls, 1);
-+ assert!(captured.bytes_read > 0);
-+ assert!(captured.requests > 0);
-+ assert!(captured.metrics.iter().all(|(name, _, _)| !name.is_empty()));
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_scanner_statistics_callback_with_arrow_stream() {
-+ let (_tmp, uri) = create_test_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
-+ assert!(!scanner.is_null());
-+ let mut captured = CapturedScanStatistics::default();
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics),
-+ (&mut captured as *mut CapturedScanStatistics).cast(),
-+ )
-+ },
-+ 0
-+ );
-+
-+ let mut stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream)
}, 0);
-+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
-+ assert_eq!(reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
5);
-+ assert_eq!(captured.calls, 1);
-+ assert!(captured.bytes_read > 0);
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_scanner_statistics_callback_rejects_null_inputs() {
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ ptr::null_mut(),
-+ Some(capture_scan_statistics),
-+ ptr::null_mut(),
-+ )
-+ },
-+ -1
-+ );
-+ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument);
-+
-+ let (_tmp, uri) = create_test_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
-+ assert_eq!(
-+ unsafe { lance_scanner_set_statistics_callback(scanner, None,
ptr::null_mut()) },
-+ -1
-+ );
-+ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument);
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_scanner_statistics_callback_rejects_registration_after_scan_started()
{
-+ let (_tmp, uri) = create_test_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
-+ assert!(!scanner.is_null());
-+
-+ let mut batch = ptr::null_mut();
-+ assert_eq!(unsafe { lance_scanner_next(scanner, &mut batch) }, 0);
-+ assert!(!batch.is_null());
-+ unsafe { lance_batch_free(batch) };
-+
-+ let mut captured = CapturedScanStatistics::default();
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics),
-+ (&mut captured as *mut CapturedScanStatistics).cast(),
-+ )
-+ },
-+ -1
-+ );
-+ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument);
-+ let error = take_last_error_message();
-+ assert!(error.contains("before the scan starts"), "{error}");
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ assert_eq!(captured.calls, 0);
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
- #[test]
- fn test_scanner_with_filter() {
- let (_tmp, uri) = create_test_dataset();
-
-From 8e6e92140031c06184da079b8f6c194799c0de88 Mon Sep 17 00:00:00 2001
-From: zhangstar333 <[email protected]>
-Date: Mon, 24 Aug 2026 17:04:51 +0800
-Subject: [PATCH 2/3] formatter
-
----
- include/lance/lance.h | 23 ++++---
- include/lance/lance.hpp | 9 +--
- src/scanner.rs | 23 ++++---
- tests/c_api_test.rs | 134 +++++++++++++++++++++++++++++++++++++++-
- 4 files changed, 165 insertions(+), 24 deletions(-)
-
-diff --git a/include/lance/lance.h b/include/lance/lance.h
-index c6c3985..1a6822b 100644
---- a/include/lance/lance.h
-+++ b/include/lance/lance.h
-@@ -883,7 +883,7 @@ typedef struct {
- } LanceScanMetric;
-
- /**
-- * Borrowed view of the execution statistics for one finalized scan.
-+ * Borrowed view of the execution statistics for one fully consumed scan.
- *
- * The fixed fields are stable summary metrics. `metrics` contains additional
- * implementation-specific counters and timings. Those names are not a stable
-@@ -903,13 +903,13 @@ typedef struct {
- } LanceScanStatistics;
-
- /**
-- * Receives scan statistics when a stream reaches EOF, fails, or is released.
-+ * Receives scan statistics after a stream is fully consumed to EOF.
- *
- * The statistics and all nested pointers are borrowed and valid only for the
-- * duration of this call. The callback may run on the thread that consumes or
-- * releases the scan stream and must therefore be thread-safe. It must return
-- * normally without throwing an exception or unwinding, and must not call any
-- * `lance_scanner_*` function with the originating scanner.
-+ * duration of this call. The callback may run on the thread that observes EOF
-+ * and must therefore be thread-safe. It must return normally without throwing
-+ * an exception or unwinding, and must not call any `lance_scanner_*` function
-+ * with the originating scanner.
- *
- * Scan statistics are diagnostic and best-effort. The callback must handle
its
- * own errors and must not use them to abort or throw across this FFI
boundary.
-@@ -925,8 +925,15 @@ typedef void (*LanceScanStatisticsCallback)(
- * Must be called before starting the scan; registering after the scan starts
- * returns an error. `callback` must not be NULL. `callback_ctx` may be NULL.
A
- * non-NULL `callback_ctx` must remain valid, and `callback` must remain
valid,
-- * until the stream reaches EOF, fails, or is released. Replaces a previously
-- * registered callback.
-+ * until the callback returns or, if the callback has not run, until the
owning
-+ * scan stream is released. For `lance_scanner_next` and
-+ * `lance_scanner_poll_next`, the scanner owns the stream. For an exported
-+ * ArrowArrayStream, the Arrow stream owns it independently of the scanner.
-+ *
-+ * The callback is invoked exactly once when the stream is fully consumed to
-+ * EOF. It is not guaranteed to run if execution fails, the scan is cancelled,
-+ * or the scanner / ArrowArrayStream is released before EOF. Replaces a
-+ * previously registered callback.
- *
- * @return 0 on success, -1 on error
- */
-diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp
-index 9440a23..4268801 100644
---- a/include/lance/lance.hpp
-+++ b/include/lance/lance.hpp
-@@ -1127,10 +1127,11 @@ class Scanner {
- return substrait_filter(bytes.data(), bytes.size());
- }
-
-- /// Register a callback for scan execution statistics before starting the
scan.
-- /// The callback may run on the thread that consumes or releases the
stream. It
-- /// must be thread-safe, must not throw, and must not re-enter the
originating
-- /// scanner. A non-null callback context must outlive the exported stream.
-+ /// Register a callback for scan statistics after successful full
exhaustion.
-+ /// The callback is not guaranteed on error, cancellation, or early
release. It
-+ /// may run on the thread that observes EOF, must be thread-safe, must
not throw,
-+ /// and must not re-enter the originating scanner. The callback and a
non-null
-+ /// context must remain valid until the callback returns or the stream is
released.
- Scanner& statistics_callback(LanceScanStatisticsCallback callback, void*
callback_ctx) {
- if (lance_scanner_set_statistics_callback(handle_.get(), callback,
callback_ctx) != 0)
- check_error();
-diff --git a/src/scanner.rs b/src/scanner.rs
-index d95089e..414c269 100644
---- a/src/scanner.rs
-+++ b/src/scanner.rs
-@@ -315,7 +315,7 @@ pub struct LanceScanMetric {
- pub value: u64,
- }
-
--/// Borrowed view of the execution statistics for one completed scan.
-+/// Borrowed view of the execution statistics for one fully consumed scan.
- ///
- /// The fixed fields are stable summary metrics. `metrics` contains additional
- /// implementation-specific counters and timings and is valid only for the
-@@ -333,7 +333,7 @@ pub struct LanceScanStatistics {
- pub metrics_len: usize,
- }
-
--/// Callback invoked when a scan stream reaches EOF, fails, or is released.
-+/// Callback invoked after a scan stream is fully consumed to EOF.
- ///
- /// The callback is an FFI boundary and must return normally without unwinding
- /// or throwing an exception. It must not call back into `lance_scanner_*`
with
-@@ -347,7 +347,7 @@ struct SendScanStatisticsCallback {
- }
-
- // SAFETY: The C API requires the callback and its context to remain valid and
--// safe to invoke from the thread that consumes or releases the scan stream.
-+// safe to invoke from the thread that observes the scan stream's EOF.
- unsafe impl Send for SendScanStatisticsCallback {}
- unsafe impl Sync for SendScanStatisticsCallback {}
-
-@@ -652,14 +652,17 @@ unsafe fn scanner_set_substrait_filter_inner(
- Ok(0)
- }
-
--/// Register a callback that receives execution statistics when the scan
stream
--/// reaches EOF, fails, or is released.
-+/// Register a callback that receives execution statistics after the scan
stream
-+/// is fully consumed to EOF.
- ///
--/// The callback and `callback_ctx` must remain valid until the scan stream is
--/// finalized. Metric names and arrays passed to the callback are borrowed and
--/// must be copied if the caller needs to retain them. The callback must be
--/// thread-safe, must return normally without unwinding or throwing an
--/// exception, and must not call `lance_scanner_*` with the originating
scanner.
-+/// The callback is not guaranteed to run if execution fails, the scan is
-+/// cancelled, or the scanner / exported Arrow stream is released before EOF.
-+/// The callback and `callback_ctx` must remain valid until the callback
returns
-+/// or, if it has not run, until the owning scan stream is released. Metric
names
-+/// and arrays passed to the callback are borrowed and must be copied if the
-+/// caller needs to retain them. The callback must be thread-safe, must return
-+/// normally without unwinding or throwing an exception, and must not call
-+/// `lance_scanner_*` with the originating scanner.
- #[unsafe(no_mangle)]
- pub unsafe extern "C" fn lance_scanner_set_statistics_callback(
- scanner: *mut LanceScanner,
-diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs
-index db35fea..c1bb71d 100644
---- a/tests/c_api_test.rs
-+++ b/tests/c_api_test.rs
-@@ -372,7 +372,43 @@ fn
test_scanner_statistics_callback_with_next_multi_fragment() {
- assert!(captured.requests > 0);
- assert!(captured.metrics.iter().all(|(name, _, _)| !name.is_empty()));
-
-+ let mut batch = ptr::null_mut();
-+ assert_eq!(unsafe { lance_scanner_next(scanner, &mut batch) }, 1);
-+ assert!(batch.is_null());
-+ assert_eq!(captured.calls, 1, "callback must run exactly once");
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_scanner_statistics_callback_not_called_on_early_scanner_close() {
-+ let (_tmp, uri) = create_test_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
-+ assert!(!scanner.is_null());
-+ let mut captured = CapturedScanStatistics::default();
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics),
-+ (&mut captured as *mut CapturedScanStatistics).cast(),
-+ )
-+ },
-+ 0
-+ );
-+
-+ let mut batch = ptr::null_mut();
-+ assert_eq!(unsafe { lance_scanner_next(scanner, &mut batch) }, 0);
-+ assert!(!batch.is_null());
-+ unsafe { lance_batch_free(batch) };
-+
- unsafe { lance_scanner_close(scanner) };
-+ assert_eq!(captured.calls, 0);
- unsafe { lance_dataset_close(ds) };
- }
-
-@@ -398,9 +434,15 @@ fn test_scanner_statistics_callback_with_arrow_stream() {
- );
-
- let mut stream = FFI_ArrowArrayStream::empty();
-- assert_eq!(unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream)
}, 0);
-+ assert_eq!(
-+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) },
-+ 0
-+ );
- let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
-- assert_eq!(reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
5);
-+ assert_eq!(
-+ reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
-+ 5
-+ );
- assert_eq!(captured.calls, 1);
- assert!(captured.bytes_read > 0);
-
-@@ -408,6 +450,74 @@ fn test_scanner_statistics_callback_with_arrow_stream() {
- unsafe { lance_dataset_close(ds) };
- }
-
-+#[test]
-+fn
test_scanner_statistics_callback_not_called_on_early_arrow_stream_release() {
-+ let (_tmp, uri) = create_test_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
-+ assert!(!scanner.is_null());
-+ let mut captured = CapturedScanStatistics::default();
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics),
-+ (&mut captured as *mut CapturedScanStatistics).cast(),
-+ )
-+ },
-+ 0
-+ );
-+
-+ let mut stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) },
-+ 0
-+ );
-+ let mut reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
-+ assert!(reader.next().unwrap().is_ok());
-+ drop(reader);
-+
-+ assert_eq!(captured.calls, 0);
-+ unsafe { lance_scanner_close(scanner) };
-+ assert_eq!(captured.calls, 0);
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_scanner_statistics_callback_not_called_on_materialization_error() {
-+ let (_tmp, uri) = create_test_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ let bad_filter = c_str("NOT A VALID >>> FILTER ???");
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(),
bad_filter.as_ptr()) };
-+ assert!(!scanner.is_null());
-+ let mut captured = CapturedScanStatistics::default();
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics),
-+ (&mut captured as *mut CapturedScanStatistics).cast(),
-+ )
-+ },
-+ 0
-+ );
-+
-+ let mut batch = ptr::null_mut();
-+ assert_eq!(unsafe { lance_scanner_next(scanner, &mut batch) }, -1);
-+ assert!(batch.is_null());
-+ assert_eq!(captured.calls, 0);
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ assert_eq!(captured.calls, 0);
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
- #[test]
- fn test_scanner_statistics_callback_rejects_null_inputs() {
- assert_eq!(
-@@ -1367,6 +1477,17 @@ fn test_poll_next_basic() {
- let c_uri = c_str(&uri_clone);
- let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0)
};
- let scanner = unsafe { lance_scanner_new(ds, ptr::null(),
ptr::null()) };
-+ let mut captured = CapturedScanStatistics::default();
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics),
-+ (&mut captured as *mut CapturedScanStatistics).cast(),
-+ )
-+ },
-+ 0
-+ );
-
- use std::sync::atomic::{AtomicBool, Ordering};
- static WOKE: AtomicBool = AtomicBool::new(false);
-@@ -1401,6 +1522,15 @@ fn test_poll_next_basic() {
- assert!(iterations < 1000, "poll loop should not spin forever");
- }
- assert_eq!(total_rows, 5);
-+ assert_eq!(captured.calls, 1);
-+
-+ let mut batch: *mut LanceBatch = ptr::null_mut();
-+ assert_eq!(
-+ unsafe { lance_scanner_poll_next(scanner, test_waker,
ptr::null_mut(), &mut batch) },
-+ LancePollStatus::Finished
-+ );
-+ assert!(batch.is_null());
-+ assert_eq!(captured.calls, 1, "callback must run exactly once");
-
- unsafe { lance_scanner_close(scanner) };
- unsafe { lance_dataset_close(ds) };
-
-From fa168ef99951d0396e50c060e540dee93e14e2be Mon Sep 17 00:00:00 2001
-From: zhangstar333 <[email protected]>
-Date: Mon, 24 Aug 2026 20:47:41 +0800
-Subject: [PATCH 3/3] update
-
----
- include/lance/lance.h | 55 ++++++++---
- include/lance/lance.hpp | 22 ++++-
- src/scanner.rs | 51 +++++++---
- tests/c_api_test.rs | 188 +++++++++++++++++++++++++++++++------
- tests/cpp/test_c_api.c | 39 +++++++-
- tests/cpp/test_cpp_api.cpp | 28 +++++-
- 6 files changed, 319 insertions(+), 64 deletions(-)
-
-diff --git a/include/lance/lance.h b/include/lance/lance.h
-index 1a6822b..5b12f3d 100644
---- a/include/lance/lance.h
-+++ b/include/lance/lance.h
-@@ -873,7 +873,8 @@ typedef enum {
- * Borrowed view of one dynamically named scan metric.
- *
- * `name` is not NUL-terminated. `name` and this structure are valid only for
-- * the duration of the LanceScanStatisticsCallback invocation.
-+ * the duration of the LanceScanStatisticsCallback invocation. Metric order is
-+ * unspecified.
- */
- typedef struct {
- const char* name;
-@@ -905,11 +906,22 @@ typedef struct {
- /**
- * Receives scan statistics after a stream is fully consumed to EOF.
- *
-- * The statistics and all nested pointers are borrowed and valid only for the
-- * duration of this call. The callback may run on the thread that observes EOF
-- * and must therefore be thread-safe. It must return normally without throwing
-- * an exception or unwinding, and must not call any `lance_scanner_*` function
-- * with the originating scanner.
-+ * `statistics` is non-NULL. It and all nested pointers are borrowed and valid
-+ * only for the duration of this call. The callback may run on the thread that
-+ * observes EOF and must therefore be thread-safe. It must return normally
-+ * without throwing an exception or unwinding, and must not call any
-+ * `lance_scanner_*` function with the originating scanner.
-+ *
-+ * From callback entry until the enclosing operation that observes EOF has
-+ * returned to its caller, the callback must not directly or indirectly cause
-+ * `get_schema`, `get_next`, `get_last_error`, or `release` to be called on
any
-+ * ArrowArrayStream derived from the originating scanner, nor cause such a
-+ * stream to be moved, destroyed, or otherwise accessed. This includes
signaling
-+ * or scheduling another thread to act based only on callback completion: the
-+ * callback returns before the enclosing stream operation does. Such
interaction
-+ * is reentrant and has undefined behavior. Normal access may resume only
after
-+ * the enclosing ArrowArrayStream `get_next`, `lance_scanner_next`, or
-+ * `lance_scanner_poll_next` call returns to its caller.
- *
- * Scan statistics are diagnostic and best-effort. The callback must handle
its
- * own errors and must not use them to abort or throw across this FFI
boundary.
-@@ -925,15 +937,27 @@ typedef void (*LanceScanStatisticsCallback)(
- * Must be called before starting the scan; registering after the scan starts
- * returns an error. `callback` must not be NULL. `callback_ctx` may be NULL.
A
- * non-NULL `callback_ctx` must remain valid, and `callback` must remain
valid,
-- * until the callback returns or, if the callback has not run, until the
owning
-- * scan stream is released. For `lance_scanner_next` and
-- * `lance_scanner_poll_next`, the scanner owns the stream. For an exported
-- * ArrowArrayStream, the Arrow stream owns it independently of the scanner.
-- *
-- * The callback is invoked exactly once when the stream is fully consumed to
-- * EOF. It is not guaranteed to run if execution fails, the scan is cancelled,
-- * or the scanner / ArrowArrayStream is released before EOF. Replaces a
-- * previously registered callback.
-+ * until all of the following are true: the scanner is closed, every in-flight
-+ * `lance_scanner_scan_async` call has delivered its completion callback, and
-+ * every ArrowArrayStream derived from the scanner has been released. The
-+ * registration remains installed after a callback returns and applies to
-+ * streams created later from the same scanner. For `lance_scanner_next` and
-+ * `lance_scanner_poll_next`, the scanner owns the stream. Exported and
-+ * asynchronous ArrowArrayStreams own their registrations independently of the
-+ * scanner and may invoke the callback after the scanner is closed. Concurrent
-+ * streams may invoke the callback concurrently.
-+ *
-+ * The callback is invoked exactly once for each derived stream that is fully
-+ * consumed to EOF. It is not guaranteed to run for a stream if execution
fails,
-+ * the scan is cancelled, or the scanner / ArrowArrayStream is released before
-+ * EOF. Before scanning starts, a new registration replaces the previous one;
-+ * after a successful replacement, the previous callback and context are no
-+ * longer retained and may be retired.
-+ *
-+ * From callback entry until the enclosing EOF-observing operation returns,
the
-+ * callback must not directly or indirectly cause interaction with any
-+ * ArrowArrayStream derived from this scanner; see LanceScanStatisticsCallback
-+ * for the complete reentrancy restriction.
- *
- * @return 0 on success, -1 on error
- */
-@@ -950,6 +974,7 @@ void lance_scanner_close(LanceScanner* scanner);
-
- /**
- * Materialize the scan as an ArrowArrayStream (blocking).
-+ * The scanner remains valid, and each call creates an independent stream.
- *
- * Reading the exported stream may surface a mid-iteration panic as one
- * error through the Arrow C stream contract (nonzero get_next plus
-diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp
-index 4268801..8aa97e2 100644
---- a/include/lance/lance.hpp
-+++ b/include/lance/lance.hpp
-@@ -1128,10 +1128,22 @@ class Scanner {
- }
-
- /// Register a callback for scan statistics after successful full
exhaustion.
-- /// The callback is not guaranteed on error, cancellation, or early
release. It
-- /// may run on the thread that observes EOF, must be thread-safe, must
not throw,
-- /// and must not re-enter the originating scanner. The callback and a
non-null
-- /// context must remain valid until the callback returns or the stream is
released.
-+ /// The registration applies to every stream derived from this scanner,
including
-+ /// concurrent streams and streams created after an earlier callback
returns. The
-+ /// callback is not guaranteed on error, cancellation, or early release.
It may
-+ /// run on the thread that observes EOF, must be thread-safe, must not
throw, and
-+ /// must not re-enter the originating scanner. The callback and a
non-null context
-+ /// must remain valid until the scanner is closed, all async scan
requests have
-+ /// delivered their completion callbacks, and all derived streams are
released.
-+ /// From callback entry until the enclosing operation that observes EOF
has
-+ /// returned to its caller, the callback must not directly or indirectly
cause
-+ /// any ArrowArrayStream derived from this Scanner to be accessed, called,
-+ /// released, moved, or destroyed. This includes signaling or scheduling
another
-+ /// thread to act based only on callback completion: the callback returns
before
-+ /// the enclosing stream operation does. Such interaction is reentrant
and has
-+ /// undefined behavior. Normal access may resume only after the enclosing
-+ /// ArrowArrayStream `get_next`, `lance_scanner_next`, or
-+ /// `lance_scanner_poll_next` call returns.
- Scanner& statistics_callback(LanceScanStatisticsCallback callback, void*
callback_ctx) {
- if (lance_scanner_set_statistics_callback(handle_.get(), callback,
callback_ctx) != 0)
- check_error();
-@@ -1153,7 +1165,7 @@ class Scanner {
- return index_segments(reinterpret_cast<const uint8_t*>(uuids.data()),
uuids.size());
- }
-
-- /// Materialize the scan as an ArrowArrayStream (blocking).
-+ /// Materialize an independent ArrowArrayStream (blocking). The scanner
remains valid.
- void to_arrow_stream(ArrowArrayStream* out) {
- if (lance_scanner_to_arrow_stream(handle_.get(), out) != 0)
- check_error();
-diff --git a/src/scanner.rs b/src/scanner.rs
-index 414c269..ef9d290 100644
---- a/src/scanner.rs
-+++ b/src/scanner.rs
-@@ -305,7 +305,7 @@ pub enum LanceScanMetricKind {
- /// Borrowed view of one dynamically named scan metric.
- ///
- /// `name` is not NUL-terminated. Both `name` and this structure are valid
only
--/// for the duration of the scan statistics callback.
-+/// for the duration of the scan statistics callback. Metric order is
unspecified.
- #[repr(C)]
- #[derive(Clone, Copy, Debug)]
- pub struct LanceScanMetric {
-@@ -318,8 +318,10 @@ pub struct LanceScanMetric {
- /// Borrowed view of the execution statistics for one fully consumed scan.
- ///
- /// The fixed fields are stable summary metrics. `metrics` contains additional
--/// implementation-specific counters and timings and is valid only for the
--/// duration of the callback.
-+/// implementation-specific counters and timings whose names are not a stable
API
-+/// and are intended only for diagnostics and profiles. Dynamic metrics are
-+/// best-effort and may be omitted if they cannot be materialized. `metrics`
is
-+/// null when `metrics_len` is zero and is valid only for the callback
duration.
- #[repr(C)]
- #[derive(Clone, Copy, Debug)]
- pub struct LanceScanStatistics {
-@@ -333,13 +335,20 @@ pub struct LanceScanStatistics {
- pub metrics_len: usize,
- }
-
--/// Callback invoked after a scan stream is fully consumed to EOF.
-+/// Callback invoked once for each derived scan stream that is fully consumed
to EOF.
- ///
- /// The callback is an FFI boundary and must return normally without unwinding
- /// or throwing an exception. It must not call back into `lance_scanner_*`
with
--/// the originating scanner.
-+/// the originating scanner. From callback entry until the enclosing operation
-+/// that observes EOF has returned to its caller, the callback must not
directly
-+/// or indirectly cause any Arrow C stream derived from the originating
scanner to
-+/// be called, released, moved, destroyed, or otherwise accessed. This
includes
-+/// signaling or scheduling another thread to act based only on callback
completion:
-+/// the callback returns before the enclosing stream operation does. Such
interaction
-+/// is reentrant and has undefined behavior. Normal access may resume only
after the
-+/// enclosing `get_next`, `lance_scanner_next`, or `lance_scanner_poll_next`
returns.
- pub type LanceScanStatisticsCallback =
-- Option<unsafe extern "C" fn(ctx: *mut c_void, statistics: *const
LanceScanStatistics)>;
-+ Option<unsafe extern "C" fn(callback_ctx: *mut c_void, statistics: *const
LanceScanStatistics)>;
-
- struct SendScanStatisticsCallback {
- callback: unsafe extern "C" fn(*mut c_void, *const LanceScanStatistics),
-@@ -347,7 +356,9 @@ struct SendScanStatisticsCallback {
- }
-
- // SAFETY: The C API requires the callback and its context to remain valid and
--// safe to invoke from the thread that observes the scan stream's EOF.
-+// safe to invoke until the scanner is closed, every in-flight asynchronous
scan
-+// has delivered its completion callback, and every derived stream has been
-+// released. Concurrent derived streams may invoke the callback concurrently.
- unsafe impl Send for SendScanStatisticsCallback {}
- unsafe impl Sync for SendScanStatisticsCallback {}
-
-@@ -657,12 +668,24 @@ unsafe fn scanner_set_substrait_filter_inner(
- ///
- /// The callback is not guaranteed to run if execution fails, the scan is
- /// cancelled, or the scanner / exported Arrow stream is released before EOF.
--/// The callback and `callback_ctx` must remain valid until the callback
returns
--/// or, if it has not run, until the owning scan stream is released. Metric
names
--/// and arrays passed to the callback are borrowed and must be copied if the
--/// caller needs to retain them. The callback must be thread-safe, must return
--/// normally without unwinding or throwing an exception, and must not call
--/// `lance_scanner_*` with the originating scanner.
-+/// The registration applies to every stream derived from this scanner,
including
-+/// streams created after an earlier callback has returned. The callback and
-+/// `callback_ctx` must remain valid until the scanner is closed, every
in-flight
-+/// asynchronous scan has delivered its completion callback, and every derived
-+/// stream has been released.
-+/// Metric names and arrays passed to the callback are borrowed and must be
copied
-+/// if the caller needs to retain them. The callback must be thread-safe, must
-+/// return normally without unwinding or throwing an exception, and must not
call
-+/// `lance_scanner_*` with the originating scanner. From callback entry until
the
-+/// enclosing operation that observes EOF has returned to its caller, the
callback
-+/// must not directly or indirectly cause any Arrow C stream derived from that
-+/// scanner to be called, released, moved, destroyed, or otherwise accessed.
This
-+/// includes signaling or scheduling another thread to act based only on
callback
-+/// completion: the callback returns before the enclosing stream operation
does.
-+/// Such interaction is reentrant and has undefined behavior. Normal access
may
-+/// resume only after the enclosing `get_next`, `lance_scanner_next`, or
-+/// `lance_scanner_poll_next` returns. Replacing the registration before
scanning
-+/// starts immediately releases the previous registration.
- #[unsafe(no_mangle)]
- pub unsafe extern "C" fn lance_scanner_set_statistics_callback(
- scanner: *mut LanceScanner,
-@@ -732,7 +755,7 @@ pub unsafe extern "C" fn lance_scanner_close(scanner: *mut
LanceScanner) {
- /// Materialize the scan as an Arrow C Data Interface `ArrowArrayStream`.
- ///
- /// This is the preferred API for simple integrations — blocks the calling
thread.
--/// The scanner is consumed by this call and should not be used afterward
(close it).
-+/// The scanner remains valid and may be used to create additional streams.
- ///
- /// The exported stream is panic-guarded (issue #61): a panic during export
- /// poisons the scanner — this call returns -1 with `LANCE_ERR_PANIC`, and
-diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs
-index c1bb71d..daa7425 100644
---- a/tests/c_api_test.rs
-+++ b/tests/c_api_test.rs
-@@ -10,6 +10,7 @@ use std::ffi::{CString, c_char, c_void};
- use std::process::Command;
- use std::ptr;
- use std::sync::Arc;
-+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering as AtomicOrdering};
-
- use arrow::ffi::from_ffi;
- use arrow::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
-@@ -151,6 +152,29 @@ unsafe extern "C" fn capture_scan_statistics(
- .collect();
- }
-
-+#[derive(Default)]
-+struct AtomicScanStatisticsCapture {
-+ calls: AtomicUsize,
-+ invalid_statistics: AtomicBool,
-+}
-+
-+unsafe extern "C" fn capture_scan_statistics_atomically(
-+ callback_ctx: *mut c_void,
-+ statistics: *const LanceScanStatistics,
-+) {
-+ if callback_ctx.is_null() {
-+ return;
-+ }
-+ let captured = unsafe {
&*callback_ctx.cast::<AtomicScanStatisticsCapture>() };
-+ if statistics.is_null() {
-+ captured
-+ .invalid_statistics
-+ .store(true, AtomicOrdering::SeqCst);
-+ return;
-+ }
-+ captured.calls.fetch_add(1, AtomicOrdering::SeqCst);
-+}
-+
- /// Helper: build a tiny dataset whose `value` column is nullable AND contains
- /// at least one NULL. Used by tests that need to exercise upstream's
- /// nullability-tightening pre-scan failure path.
-@@ -413,7 +437,7 @@ fn
test_scanner_statistics_callback_not_called_on_early_scanner_close() {
- }
-
- #[test]
--fn test_scanner_statistics_callback_with_arrow_stream() {
-+fn test_scanner_statistics_callback_applies_to_reused_scanner() {
- let (_tmp, uri) = create_test_dataset();
- let c_uri = c_str(&uri);
- let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-@@ -433,20 +457,87 @@ fn test_scanner_statistics_callback_with_arrow_stream() {
- 0
- );
-
-- let mut stream = FFI_ArrowArrayStream::empty();
-+ let mut first_stream = FFI_ArrowArrayStream::empty();
- assert_eq!(
-- unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) },
-+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut first_stream) },
- 0
- );
-- let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
-+ let first_reader = unsafe { ArrowArrayStreamReader::from_raw(&mut
first_stream) }.unwrap();
- assert_eq!(
-- reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
-+ first_reader
-+ .map(|batch| batch.unwrap().num_rows())
-+ .sum::<usize>(),
- 5
- );
- assert_eq!(captured.calls, 1);
-- assert!(captured.bytes_read > 0);
-
-+ let mut second_stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut second_stream) },
-+ 0
-+ );
- unsafe { lance_scanner_close(scanner) };
-+
-+ let second_reader = unsafe { ArrowArrayStreamReader::from_raw(&mut
second_stream) }.unwrap();
-+ assert_eq!(
-+ second_reader
-+ .map(|batch| batch.unwrap().num_rows())
-+ .sum::<usize>(),
-+ 5
-+ );
-+ assert_eq!(captured.calls, 2);
-+
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_scanner_statistics_callback_supports_concurrent_exported_streams() {
-+ struct SendableArrowStream(FFI_ArrowArrayStream);
-+ unsafe impl Send for SendableArrowStream {}
-+
-+ fn consume_stream(mut stream: SendableArrowStream) -> usize {
-+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream.0)
}.unwrap();
-+ reader.map(|batch| batch.unwrap().num_rows()).sum()
-+ }
-+
-+ let (_tmp, uri) = create_test_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
-+ assert!(!scanner.is_null());
-+ let captured = Arc::new(AtomicScanStatisticsCapture::default());
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics_atomically),
-+ Arc::as_ptr(&captured).cast_mut().cast(),
-+ )
-+ },
-+ 0
-+ );
-+
-+ let mut first_stream = FFI_ArrowArrayStream::empty();
-+ let mut second_stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut first_stream) },
-+ 0
-+ );
-+ assert_eq!(
-+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut second_stream) },
-+ 0
-+ );
-+ unsafe { lance_scanner_close(scanner) };
-+
-+ let first = std::thread::spawn(move ||
consume_stream(SendableArrowStream(first_stream)));
-+ let second = std::thread::spawn(move ||
consume_stream(SendableArrowStream(second_stream)));
-+ assert_eq!(first.join().unwrap(), 5);
-+ assert_eq!(second.join().unwrap(), 5);
-+ assert_eq!(captured.calls.load(AtomicOrdering::SeqCst), 2);
-+ assert!(!captured.invalid_statistics.load(AtomicOrdering::SeqCst));
-+
- unsafe { lance_dataset_close(ds) };
- }
-
-@@ -546,6 +637,55 @@ fn test_scanner_statistics_callback_rejects_null_inputs()
{
- unsafe { lance_dataset_close(ds) };
- }
-
-+#[test]
-+fn test_scanner_statistics_callback_replaces_registration_before_scan() {
-+ let (_tmp, uri) = create_test_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
-+ assert!(!scanner.is_null());
-+
-+ let mut replaced = CapturedScanStatistics::default();
-+ let mut active = CapturedScanStatistics::default();
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics),
-+ (&mut replaced as *mut CapturedScanStatistics).cast(),
-+ )
-+ },
-+ 0
-+ );
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics),
-+ (&mut active as *mut CapturedScanStatistics).cast(),
-+ )
-+ },
-+ 0
-+ );
-+
-+ let mut stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) },
-+ 0
-+ );
-+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
-+ assert_eq!(
-+ reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
-+ 5
-+ );
-+ assert_eq!(replaced.calls, 0);
-+ assert_eq!(active.calls, 1);
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
- #[test]
- fn test_scanner_statistics_callback_rejects_registration_after_scan_started()
{
- let (_tmp, uri) = create_test_dataset();
-@@ -814,6 +954,17 @@ fn test_scanner_scan_async() {
-
- let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
- assert!(!scanner.is_null());
-+ let captured = Arc::new(AtomicScanStatisticsCapture::default());
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_set_statistics_callback(
-+ scanner,
-+ Some(capture_scan_statistics_atomically),
-+ Arc::as_ptr(&captured).cast_mut().cast(),
-+ )
-+ },
-+ 0
-+ );
-
- // Synchronization primitive for the async callback.
- struct CallbackResult {
-@@ -845,6 +996,7 @@ fn test_scanner_scan_async() {
- on_complete,
- Arc::as_ptr(&pair_clone) as *mut std::ffi::c_void,
- );
-+ lance_scanner_close(scanner);
- }
-
- // Wait for callback.
-@@ -861,8 +1013,9 @@ fn test_scanner_scan_async() {
- let reader = unsafe { ArrowArrayStreamReader::from_raw(ffi_stream)
}.unwrap();
- let total_rows: usize = reader.map(|r| r.unwrap().num_rows()).sum();
- assert_eq!(total_rows, 5);
-+ assert_eq!(captured.calls.load(AtomicOrdering::SeqCst), 1);
-+ assert!(!captured.invalid_statistics.load(AtomicOrdering::SeqCst));
-
-- unsafe { lance_scanner_close(scanner) };
- unsafe { lance_dataset_close(ds) };
- }
-
-@@ -1477,18 +1630,6 @@ fn test_poll_next_basic() {
- let c_uri = c_str(&uri_clone);
- let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0)
};
- let scanner = unsafe { lance_scanner_new(ds, ptr::null(),
ptr::null()) };
-- let mut captured = CapturedScanStatistics::default();
-- assert_eq!(
-- unsafe {
-- lance_scanner_set_statistics_callback(
-- scanner,
-- Some(capture_scan_statistics),
-- (&mut captured as *mut CapturedScanStatistics).cast(),
-- )
-- },
-- 0
-- );
--
- use std::sync::atomic::{AtomicBool, Ordering};
- static WOKE: AtomicBool = AtomicBool::new(false);
- unsafe extern "C" fn test_waker(_ctx: *mut std::ffi::c_void) {
-@@ -1522,15 +1663,6 @@ fn test_poll_next_basic() {
- assert!(iterations < 1000, "poll loop should not spin forever");
- }
- assert_eq!(total_rows, 5);
-- assert_eq!(captured.calls, 1);
--
-- let mut batch: *mut LanceBatch = ptr::null_mut();
-- assert_eq!(
-- unsafe { lance_scanner_poll_next(scanner, test_waker,
ptr::null_mut(), &mut batch) },
-- LancePollStatus::Finished
-- );
-- assert!(batch.is_null());
-- assert_eq!(captured.calls, 1, "callback must run exactly once");
-
- unsafe { lance_scanner_close(scanner) };
- unsafe { lance_dataset_close(ds) };
-diff --git a/tests/cpp/test_c_api.c b/tests/cpp/test_c_api.c
-index b3b78f0..4499cb9 100644
---- a/tests/cpp/test_c_api.c
-+++ b/tests/cpp/test_c_api.c
-@@ -39,6 +39,36 @@
- }
\
- } while (0)
-
-+typedef struct {
-+ uint64_t calls;
-+ uint64_t bytes_read;
-+ int invalid;
-+} ScanStatisticsCapture;
-+
-+static void capture_scan_statistics(
-+ void *callback_ctx,
-+ const LanceScanStatistics *statistics
-+) {
-+ if (callback_ctx == NULL) return;
-+ ScanStatisticsCapture *captured = (ScanStatisticsCapture *)callback_ctx;
-+ if (statistics == NULL ||
-+ (statistics->metrics_len > 0 && statistics->metrics == NULL)) {
-+ captured->invalid = 1;
-+ return;
-+ }
-+ for (size_t i = 0; i < statistics->metrics_len; ++i) {
-+ const LanceScanMetric *metric = &statistics->metrics[i];
-+ if ((metric->name_len > 0 && metric->name == NULL) ||
-+ (metric->kind != LANCE_SCAN_METRIC_COUNT &&
-+ metric->kind != LANCE_SCAN_METRIC_TIME_NANOSECONDS)) {
-+ captured->invalid = 1;
-+ return;
-+ }
-+ }
-+ captured->calls += 1;
-+ captured->bytes_read = statistics->bytes_read;
-+}
-+
- static void test_open_and_metadata(const char *uri) {
- printf(" test_open_and_metadata... ");
-
-@@ -84,10 +114,14 @@ static void test_scan(const char *uri) {
- /* Full scan via ArrowArrayStream */
- LanceScanner *scanner = lance_scanner_new(ds, NULL, NULL);
- ASSERT(scanner != NULL, "scanner creation failed");
-+ ScanStatisticsCapture captured = {0};
-+ int32_t rc = lance_scanner_set_statistics_callback(
-+ scanner, capture_scan_statistics, &captured);
-+ ASSERT(rc == 0, "statistics callback registration failed");
-
- struct ArrowArrayStream stream;
- memset(&stream, 0, sizeof(stream));
-- int32_t rc = lance_scanner_to_arrow_stream(scanner, &stream);
-+ rc = lance_scanner_to_arrow_stream(scanner, &stream);
- ASSERT(rc == 0, "to_arrow_stream failed");
-
- /* Read schema from stream */
-@@ -113,6 +147,9 @@ static void test_scan(const char *uri) {
- }
-
- ASSERT(total_rows == expected_rows, "row count mismatch");
-+ ASSERT(captured.calls == 1, "statistics callback count mismatch");
-+ ASSERT(captured.bytes_read > 0, "statistics should report bytes read");
-+ ASSERT(captured.invalid == 0, "statistics callback received invalid
data");
- printf("rows=%llu... ", (unsigned long long)total_rows);
-
- if (stream.release) stream.release(&stream);
-diff --git a/tests/cpp/test_cpp_api.cpp b/tests/cpp/test_cpp_api.cpp
-index f8ae701..3293bfb 100644
---- a/tests/cpp/test_cpp_api.cpp
-+++ b/tests/cpp/test_cpp_api.cpp
-@@ -25,6 +25,25 @@
- #define TEST(name) printf(" %s... ", #name)
- #define PASS() printf("OK\n")
-
-+struct ScanStatisticsCapture {
-+ uint64_t calls = 0;
-+ uint64_t bytes_read = 0;
-+ bool invalid = false;
-+};
-+
-+static void capture_scan_statistics(
-+ void* callback_ctx,
-+ const LanceScanStatistics* statistics) noexcept {
-+ if (!callback_ctx) return;
-+ auto* captured = static_cast<ScanStatisticsCapture*>(callback_ctx);
-+ if (!statistics || (statistics->metrics_len > 0 && !statistics->metrics))
{
-+ captured->invalid = true;
-+ return;
-+ }
-+ captured->calls += 1;
-+ captured->bytes_read = statistics->bytes_read;
-+}
-+
- static void test_dataset_open(const std::string& uri) {
- TEST(test_dataset_open);
-
-@@ -70,7 +89,11 @@ static void test_scanner_fluent(const std::string& uri) {
-
- // Fluent builder pattern.
- auto scanner = ds.scan();
-- scanner.limit(5).offset(0).batch_size(2);
-+ ScanStatisticsCapture captured;
-+ scanner.limit(5)
-+ .offset(0)
-+ .batch_size(2)
-+ .statistics_callback(capture_scan_statistics, &captured);
-
- ArrowArrayStream stream;
- memset(&stream, 0, sizeof(stream));
-@@ -89,6 +112,9 @@ static void test_scanner_fluent(const std::string& uri) {
- }
-
- assert(total == 5);
-+ assert(captured.calls == 1);
-+ assert(captured.bytes_read > 0);
-+ assert(!captured.invalid);
- printf("rows=%llu... ", (unsigned long long)total);
-
- if (stream.release) stream.release(&stream);
diff --git a/thirdparty/patches/lance-c-0.1.8-pr-69.patch
b/thirdparty/patches/lance-c-0.1.8-pr-69.patch
new file mode 100644
index 00000000000..ee38eb33bcb
--- /dev/null
+++ b/thirdparty/patches/lance-c-0.1.8-pr-69.patch
@@ -0,0 +1,653 @@
+From a4d6e489c627fe4b0e49d9a2991c9436313debc0 Mon Sep 17 00:00:00 2001
+From: zhangstar333 <[email protected]>
+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<IndexMetadata>,
+ pub(crate) scorer: Arc<MemBM25Scorer>,
++ 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 <[email protected]>
+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<IndexMetadata>,
+ pub(crate) scorer: Arc<MemBM25Scorer>,
+- 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<bool> {
++ 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::<Result<std::collections::HashSet<_>>>()?;
++ 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<dyn ExecutionPlan>,
+ segments: &[IndexMetadata],
+ scorer: &Arc<lance_index::scalar::inverted::MemBM25Scorer>,
+-) -> Result<(Arc<dyn ExecutionPlan>, usize)> {
++ selected_segments_have_current_fragments: bool,
++) -> Result<(Arc<dyn ExecutionPlan>, 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::<FlatMatchQueryExec>().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::<MatchQueryExec>() {
++ 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<dyn ExecutionPlan>) -> (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::<MatchQueryExec>() {
++ match_query_execs += 1;
++ if matches!(exec.prefilter_source(),
PreFilterSource::FilteredRowIds(_)) {
++ filtered_row_id_prefilters += 1;
++ }
++ }
++ if plan.downcast_ref::<FlatMatchQueryExec>().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::<u8>(),
++ 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/vars.sh b/thirdparty/vars.sh
index c9fec64e032..84f4a73edaa 100644
--- a/thirdparty/vars.sh
+++ b/thirdparty/vars.sh
@@ -589,10 +589,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.7.tar.gz";
-LANCE_C_NAME="lance-c-v0.1.7.tar.gz"
-LANCE_C_SOURCE="lance-c-0.1.7"
-LANCE_C_MD5SUM="15ef7cd20a2e1606384251cb2d41d42f"
+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"
# all thirdparties which need to be downloaded is set in array TP_ARCHIVES
export TP_ARCHIVES=(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
