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 1731787677f [chore](lance) update lance version to tag 0.1.7 (#67115)
1731787677f is described below
commit 1731787677f0199ccdb4fe6318f9116310627c52
Author: zhangstar333 <[email protected]>
AuthorDate: Tue Aug 25 22:36:06 2026 +0800
[chore](lance) update lance version to tag 0.1.7 (#67115)
### What problem does this PR solve?
Problem Summary:
1. update lance version to tag 0.1.7
2. group lance scan parameters in Thrift
---
gensrc/thrift/PlanNodes.thrift | 13 +-
thirdparty/download-thirdparty.sh | 6 +-
thirdparty/patches/lance-c-0.1.6-doris.patch | 632 -----------
thirdparty/patches/lance-c-0.1.7-pr-64.patch | 1522 ++++++++++++++++++++++++++
thirdparty/vars.sh | 8 +-
5 files changed, 1536 insertions(+), 645 deletions(-)
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index b2702dd0ed4..1eaa53251bc 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -497,6 +497,12 @@ struct TExternalSearchRequest {
1: optional i32 schema_version = 1
}
+struct TLanceScanParams {
+ 1: optional binary lance_substrait_filter
+ 2: optional TExternalSearchRequest external_search_request
+ 3: optional map<string, string> lance_storage_options
+}
+
struct TFileScanRangeParams {
// deprecated, move to TFileScanRange
1: optional Types.TFileType file_type;
@@ -580,12 +586,7 @@ struct TFileScanRangeParams {
// HMS catalog property hive.parquet.time-zone. When absent, format_v2
keeps INT96 wall-clock
// values unchanged. When present, only INT96 TIMESTAMP values are
converted with this zone.
36: optional string hive_parquet_time_zone
- // Serialized Substrait ExtendedExpression executed by the native Lance
scanner. Set at
- // ScanNode level so it is not serialized once per fragment split.
- 37: optional binary lance_substrait_filter
- // Provider-independent search request. Set at ScanNode level so all
ranges use the same logical
- // query. The first implementation uses one whole-dataset range for Lance
vector search.
- 38: optional TExternalSearchRequest external_search_request
+ 37: optional TLanceScanParams lance_scan_params
}
struct TFileRangeDesc {
diff --git a/thirdparty/download-thirdparty.sh
b/thirdparty/download-thirdparty.sh
index 830da72eacf..8d65b8ce275 100755
--- a/thirdparty/download-thirdparty.sh
+++ b/thirdparty/download-thirdparty.sh
@@ -819,12 +819,12 @@ if [[ " ${TP_ARCHIVES[*]} " =~ " PAIMON_CPP " ]]; then
echo "Finished patching ${PAIMON_CPP_SOURCE}"
fi
-# Patch lance-c for fragment-scoped nearest-neighbor search and row-ID-based
fetching.
+# Patch lance-c with the scan execution statistics API from upstream PR #64.
if [[ " ${TP_ARCHIVES[*]} " =~ " LANCE_C " ]]; then
- if [[ "${LANCE_C_SOURCE}" == "lance-c-0.1.6" ]]; then
+ if [[ "${LANCE_C_SOURCE}" == "lance-c-0.1.7" ]]; then
cd "${TP_SOURCE_DIR}/${LANCE_C_SOURCE}"
if [[ ! -f "${PATCHED_MARK}" ]]; then
- patch -p1 <"${TP_PATCH_DIR}/lance-c-0.1.6-doris.patch"
+ patch -p1 <"${TP_PATCH_DIR}/lance-c-0.1.7-pr-64.patch"
touch "${PATCHED_MARK}"
fi
cd -
diff --git a/thirdparty/patches/lance-c-0.1.6-doris.patch
b/thirdparty/patches/lance-c-0.1.6-doris.patch
deleted file mode 100644
index cc49293d9de..00000000000
--- a/thirdparty/patches/lance-c-0.1.6-doris.patch
+++ /dev/null
@@ -1,632 +0,0 @@
-diff --git a/include/lance/lance.h b/include/lance/lance.h
-index 1de72fa..a0a8b5e 100644
---- a/include/lance/lance.h
-+++ b/include/lance/lance.h
-@@ -752,0 +753,25 @@ int32_t lance_dataset_take(
-+/**
-+ * Take rows by dataset row IDs.
-+ *
-+ * Row IDs are values from the `_rowid` scanner column, not zero-based row
-+ * offsets. They must belong to the same dataset snapshot used for this read.
-+ * Missing or deleted row IDs may be omitted from the result. For found rows,
-+ * input order and duplicates are preserved.
-+ *
-+ * @param dataset Open dataset snapshot.
-+ * @param row_ids Array of dataset row IDs. May be NULL only when
-+ * `num_row_ids` is zero.
-+ * @param num_row_ids Length of `row_ids`.
-+ * @param columns NULL-terminated column names, or NULL for all. The
-+ * system column `_rowid` may be requested explicitly.
-+ * @param out Pointer to caller-allocated ArrowArrayStream.
-+ * @return 0 on success, -1 on error.
-+ */
-+int32_t lance_dataset_take_rows(
-+ const LanceDataset* dataset,
-+ const uint64_t* row_ids,
-+ size_t num_row_ids,
-+ const char* const* columns,
-+ struct ArrowArrayStream* out
-+);
-+
-diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp
-index 40aa9e3..d54dc5a 100644
---- a/include/lance/lance.hpp
-+++ b/include/lance/lance.hpp
-@@ -654,0 +655,24 @@ public:
-+ /// Take rows by dataset row IDs. Results exported as ArrowArrayStream.
-+ void take_rows(const uint64_t* row_ids, size_t num_row_ids,
-+ const std::vector<std::string>& columns,
-+ ArrowArrayStream* out) const {
-+ std::vector<const char*> col_ptrs;
-+ for (auto& c : columns) col_ptrs.push_back(c.c_str());
-+ col_ptrs.push_back(nullptr);
-+ const char* const* cols_ptr = columns.empty() ? nullptr :
col_ptrs.data();
-+
-+ if (lance_dataset_take_rows(
-+ handle_.get(), row_ids, num_row_ids, cols_ptr, out) != 0) {
-+ check_error();
-+ }
-+ }
-+
-+ /// Take all columns by dataset row IDs.
-+ void take_rows(const uint64_t* row_ids, size_t num_row_ids,
-+ ArrowArrayStream* out) const {
-+ if (lance_dataset_take_rows(
-+ handle_.get(), row_ids, num_row_ids, nullptr, out) != 0) {
-+ check_error();
-+ }
-+ }
-+
-diff --git a/src/dataset.rs b/src/dataset.rs
-index 91a735f..364be09 100644
---- a/src/dataset.rs
-+++ b/src/dataset.rs
-@@ -44,0 +45,20 @@ impl LanceDataset {
-+fn projection_from_columns(
-+ dataset: &Dataset,
-+ columns: Option<&[String]>,
-+) -> Result<lance::dataset::ProjectionRequest> {
-+ match columns {
-+ Some(columns) => {
-+ let schema = dataset
-+ .schema()
-+ .project_preserve_system_columns(columns)
-+ .map_err(|err| {
-+ lance_core::Error::invalid_input(format!("invalid columns
{columns:?}: {err}"))
-+ })?;
-+ Ok(lance::dataset::ProjectionRequest::from_schema(schema))
-+ }
-+ None => Ok(lance::dataset::ProjectionRequest::from_schema(
-+ dataset.schema().clone(),
-+ )),
-+ }
-+}
-+
-@@ -247,4 +267 @@ unsafe fn dataset_take_inner(
-- let projection = match &col_names {
-- Some(cols) =>
lance::dataset::ProjectionRequest::from_columns(cols.iter(), snap.schema()),
-- None =>
lance::dataset::ProjectionRequest::from_schema(snap.schema().clone()),
-- };
-+ let projection = projection_from_columns(&snap, col_names.as_deref())?;
-@@ -263,0 +281,69 @@ unsafe fn dataset_take_inner(
-+/// Take rows by dataset row IDs, returning results as an ArrowArrayStream.
-+///
-+/// - `row_ids`: array of dataset row IDs, such as values returned in the
-+/// `_rowid` scanner column
-+/// - `num_row_ids`: length of the row ID array
-+/// - `columns`: NULL-terminated column name array, or NULL for all columns
-+/// - `out`: pointer to a stack-allocated `ArrowArrayStream`
-+///
-+/// `row_ids` may be NULL only when `num_row_ids` is zero. Row IDs must belong
-+/// to the same dataset snapshot used for this read. Missing or deleted row
IDs
-+/// may be omitted from the result by the upstream Lance implementation.
-+///
-+/// Returns 0 on success, -1 on error.
-+#[unsafe(no_mangle)]
-+pub unsafe extern "C" fn lance_dataset_take_rows(
-+ dataset: *const LanceDataset,
-+ row_ids: *const u64,
-+ num_row_ids: usize,
-+ columns: *const *const c_char,
-+ out: *mut FFI_ArrowArrayStream,
-+) -> i32 {
-+ ffi_try!(
-+ unsafe { dataset_take_rows_inner(dataset, row_ids, num_row_ids,
columns, out) },
-+ neg
-+ )
-+}
-+
-+unsafe fn dataset_take_rows_inner(
-+ dataset: *const LanceDataset,
-+ row_ids: *const u64,
-+ num_row_ids: usize,
-+ columns: *const *const c_char,
-+ out: *mut FFI_ArrowArrayStream,
-+) -> Result<i32> {
-+ if dataset.is_null() {
-+ return Err(lance_core::Error::invalid_input("dataset must not be
NULL"));
-+ }
-+ if out.is_null() {
-+ return Err(lance_core::Error::invalid_input("out must not be NULL"));
-+ }
-+ if num_row_ids > 0 && row_ids.is_null() {
-+ return Err(lance_core::Error::invalid_input(format!(
-+ "row_ids must not be NULL when num_row_ids = {num_row_ids}"
-+ )));
-+ }
-+
-+ let ds = unsafe { &*dataset };
-+ let row_id_slice = if num_row_ids == 0 {
-+ &[]
-+ } else {
-+ unsafe { std::slice::from_raw_parts(row_ids, num_row_ids) }
-+ };
-+ let col_names = unsafe { helpers::parse_c_string_array(columns)? };
-+
-+ let snap = ds.snapshot();
-+ let projection = projection_from_columns(&snap, col_names.as_deref())?;
-+
-+ let batch = block_on(snap.take_rows(row_id_slice, projection))?;
-+
-+ // Match lance_dataset_take: export the single RecordBatch as an Arrow
stream.
-+ let schema = batch.schema();
-+ let reader =
arrow::record_batch::RecordBatchIterator::new(vec![Ok(batch)], schema);
-+ let ffi_stream = FFI_ArrowArrayStream::new(Box::new(reader));
-+ unsafe {
-+ std::ptr::write_unaligned(out, ffi_stream);
-+ }
-+ Ok(0)
-+}
-+
-diff --git a/src/scanner.rs b/src/scanner.rs
-index c1c2f67..de646be 100644
---- a/src/scanner.rs
-+++ b/src/scanner.rs
-@@ -160,0 +161,6 @@ impl LanceScanner {
-+ // Lance validates fragment-scoped nearest searches when nearest() is
-+ // configured. Such searches are supported when the fragment scan is
-+ // the input to a prefilter, so this flag must be set first.
-+ if self.prefilter {
-+ scanner.prefilter(true);
-+ }
-@@ -178,3 +183,0 @@ impl LanceScanner {
-- if self.prefilter {
-- scanner.prefilter(true);
-- }
-@@ -221,0 +225,5 @@ impl LanceScanner {
-+ // nearest() checks the current prefilter setting before accepting a
-+ // fragment-scoped search. Enable it before installing the query.
-+ if self.prefilter {
-+ scanner.prefilter(true);
-+ }
-@@ -239,3 +246,0 @@ impl LanceScanner {
-- if self.prefilter {
-- scanner.prefilter(true);
-- }
-diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs
-index 2babda6..9ee5af9 100644
---- a/tests/c_api_test.rs
-+++ b/tests/c_api_test.rs
-@@ -9,0 +10 @@ use std::ffi::{CString, c_char};
-+use std::process::Command;
-@@ -18 +19 @@ use arrow::record_batch::RecordBatchReader;
--use arrow_array::{Array, Float32Array, Int32Array, RecordBatch, StringArray};
-+use arrow_array::{Array, Float32Array, Int32Array, RecordBatch, StringArray,
UInt64Array};
-@@ -393,0 +395,94 @@ fn test_dataset_take() {
-+#[test]
-+fn test_dataset_take_rows_empty_and_null_validation() {
-+ 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 mut empty_stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_dataset_take_rows(ds, ptr::null(), 0, ptr::null(),
&mut empty_stream) },
-+ 0
-+ );
-+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut empty_stream)
}.unwrap();
-+ assert_eq!(
-+ reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
-+ 0
-+ );
-+
-+ let mut invalid_stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_dataset_take_rows(ds, ptr::null(), 1, ptr::null(),
&mut invalid_stream) },
-+ -1
-+ );
-+ let message = unsafe {
std::ffi::CStr::from_ptr(lance_last_error_message()) }.to_string_lossy();
-+ assert!(
-+ message.contains("row_ids must not be NULL when num_row_ids = 1"),
-+ "unexpected error: {message}"
-+ );
-+
-+ let row_id = 0_u64;
-+ assert_eq!(
-+ unsafe {
-+ lance_dataset_take_rows(ptr::null(), &row_id, 1, ptr::null(),
&mut invalid_stream)
-+ },
-+ -1
-+ );
-+ assert_eq!(
-+ unsafe { lance_dataset_take_rows(ds, &row_id, 1, ptr::null(),
ptr::null_mut()) },
-+ -1
-+ );
-+
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_dataset_take_rows_invalid_column() {
-+ const CHILD_ENV: &str = "LANCE_C_TEST_TAKE_ROWS_INVALID_COLUMN_CHILD";
-+
-+ if std::env::var_os(CHILD_ENV).is_some() {
-+ 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 row_id = 0_u64;
-+ let invalid_column = c_str("unknown_column");
-+ let columns = [invalid_column.as_ptr(), ptr::null()];
-+ let mut stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_dataset_take_rows(ds, &row_id, 1,
columns.as_ptr(), &mut stream) },
-+ -1
-+ );
-+ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument);
-+
-+ let message_ptr = lance_last_error_message();
-+ assert!(!message_ptr.is_null());
-+ let message = unsafe { std::ffi::CStr::from_ptr(message_ptr) }
-+ .to_string_lossy()
-+ .into_owned();
-+ unsafe { lance_free_string(message_ptr) };
-+ assert!(
-+ message.contains("unknown_column"),
-+ "unexpected error: {message}"
-+ );
-+
-+ unsafe { lance_dataset_close(ds) };
-+ return;
-+ }
-+
-+ let output = Command::new(std::env::current_exe().unwrap())
-+ .arg("--exact")
-+ .arg("test_dataset_take_rows_invalid_column")
-+ .arg("--nocapture")
-+ .env(CHILD_ENV, "1")
-+ .output()
-+ .unwrap();
-+ assert!(
-+ output.status.success(),
-+ "invalid-column subprocess failed\nstdout:\n{}\nstderr:\n{}",
-+ String::from_utf8_lossy(&output.stdout),
-+ String::from_utf8_lossy(&output.stderr)
-+ );
-+}
-+
-@@ -2497,0 +2593,73 @@ fn create_vector_dataset(num_rows: i32, dim: i32) ->
(tempfile::TempDir, String)
-+/// Create two vector fragments with deterministic vectors. Every component of
-+/// row `id` is `id as f32`, making nearest-neighbor expectations unambiguous.
-+fn create_multi_fragment_vector_dataset(
-+ rows_per_fragment: i32,
-+ dim: i32,
-+ enable_stable_row_ids: bool,
-+) -> (tempfile::TempDir, String) {
-+ use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
-+
-+ let tmp = tempfile::tempdir().unwrap();
-+ let uri = tmp
-+ .path()
-+ .join("multi_fragment_vec_ds")
-+ .to_str()
-+ .unwrap()
-+ .to_string();
-+ let schema = Arc::new(Schema::new(vec![
-+ Field::new("id", DataType::Int32, false),
-+ Field::new(
-+ "embedding",
-+ DataType::FixedSizeList(Arc::new(Field::new("item",
DataType::Float32, true)), dim),
-+ false,
-+ ),
-+ ]));
-+
-+ let make_batch = |first_id: i32| {
-+ let ids: Vec<i32> = (first_id..first_id +
rows_per_fragment).collect();
-+ let mut embeddings = FixedSizeListBuilder::new(Float32Builder::new(),
dim);
-+ for id in &ids {
-+ for _ in 0..dim {
-+ embeddings.values().append_value(*id as f32);
-+ }
-+ embeddings.append(true);
-+ }
-+ RecordBatch::try_new(
-+ schema.clone(),
-+ vec![
-+ Arc::new(Int32Array::from(ids)),
-+ Arc::new(embeddings.finish()),
-+ ],
-+ )
-+ .unwrap()
-+ };
-+
-+ let first = make_batch(0);
-+ let second = make_batch(rows_per_fragment);
-+ lance_c::runtime::block_on(async {
-+ Dataset::write(
-+ arrow::record_batch::RecordBatchIterator::new(vec![Ok(first)],
schema.clone()),
-+ &uri,
-+ Some(lance::dataset::WriteParams {
-+ enable_stable_row_ids,
-+ ..Default::default()
-+ }),
-+ )
-+ .await
-+ .unwrap();
-+ Dataset::write(
-+ arrow::record_batch::RecordBatchIterator::new(vec![Ok(second)],
schema),
-+ &uri,
-+ Some(lance::dataset::WriteParams {
-+ mode: lance::dataset::WriteMode::Append,
-+ enable_stable_row_ids,
-+ ..Default::default()
-+ }),
-+ )
-+ .await
-+ .unwrap();
-+ });
-+
-+ (tmp, uri)
-+}
-+
-@@ -2723,0 +2892,99 @@ fn test_scanner_nearest_brute_force() {
-+fn
assert_dataset_take_rows_from_multi_fragment_ann_result(enable_stable_row_ids:
bool) {
-+ let (_tmp, uri) = create_multi_fragment_vector_dataset(32, 8,
enable_stable_row_ids);
-+ let uri_c = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ // A non-NULL array whose first element is NULL is an explicit empty
-+ // projection. The ANN result should therefore contain only _distance and
-+ // the explicitly requested _rowid system column.
-+ let no_columns: [*const c_char; 1] = [ptr::null()];
-+ let scanner = unsafe { lance_scanner_new(ds, no_columns.as_ptr(),
ptr::null()) };
-+ assert!(!scanner.is_null());
-+ assert_eq!(unsafe { lance_scanner_with_row_id(scanner, true) }, 0);
-+
-+ let column = c_str("embedding");
-+ let query = [40.0_f32; 8];
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_nearest(
-+ scanner,
-+ column.as_ptr(),
-+ query.as_ptr().cast(),
-+ query.len(),
-+ LanceDataType::Float32 as i32,
-+ 1,
-+ )
-+ },
-+ 0
-+ );
-+ assert_eq!(unsafe { lance_scanner_set_use_index(scanner, false) }, 0);
-+
-+ let mut ann_stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut ann_stream) },
-+ 0
-+ );
-+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut ann_stream)
}.unwrap();
-+ let ann_batches = reader.map(|batch| batch.unwrap()).collect::<Vec<_>>();
-+ assert_eq!(
-+ ann_batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
-+ 1
-+ );
-+ assert_eq!(ann_batches[0].num_columns(), 2);
-+
-+ let distance = ann_batches[0]
-+ .column_by_name("_distance")
-+ .unwrap()
-+ .as_any()
-+ .downcast_ref::<Float32Array>()
-+ .unwrap()
-+ .value(0);
-+ assert_eq!(distance, 0.0);
-+ let row_id = ann_batches[0]
-+ .column_by_name("_rowid")
-+ .unwrap()
-+ .as_any()
-+ .downcast_ref::<UInt64Array>()
-+ .unwrap()
-+ .value(0);
-+ if !enable_stable_row_ids {
-+ assert_ne!(
-+ row_id >> 32,
-+ 0,
-+ "expected an address-style row ID from the second fragment"
-+ );
-+ }
-+
-+ let id_column = c_str("id");
-+ let columns = [id_column.as_ptr(), ptr::null()];
-+ let mut take_stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_dataset_take_rows(ds, &row_id, 1, columns.as_ptr(),
&mut take_stream) },
-+ 0
-+ );
-+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut take_stream)
}.unwrap();
-+ let batches = reader.map(|batch| batch.unwrap()).collect::<Vec<_>>();
-+ assert_eq!(batches.len(), 1);
-+ let ids = batches[0]
-+ .column_by_name("id")
-+ .unwrap()
-+ .as_any()
-+ .downcast_ref::<Int32Array>()
-+ .unwrap();
-+ assert_eq!(ids.values(), &[40]);
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_dataset_take_rows_from_multi_fragment_ann_result() {
-+ assert_dataset_take_rows_from_multi_fragment_ann_result(false);
-+}
-+
-+#[test]
-+fn
test_dataset_take_rows_from_multi_fragment_ann_result_with_stable_row_ids() {
-+ assert_dataset_take_rows_from_multi_fragment_ann_result(true);
-+}
-+
-@@ -2855,0 +3123,132 @@ fn test_scanner_nearest_filter_postfilter() {
-+#[test]
-+fn test_scanner_nearest_prefilter_with_fragment_ids_next() {
-+ let (_tmp, uri) = create_multi_fragment_vector_dataset(32, 8, false);
-+ let uri_c = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ let mut fragment_ids = vec![0; unsafe { lance_dataset_fragment_count(ds)
} as usize];
-+ assert_eq!(fragment_ids.len(), 2);
-+ assert_eq!(
-+ unsafe { lance_dataset_fragment_ids(ds, fragment_ids.as_mut_ptr()) },
-+ 0
-+ );
-+
-+ let filter = c_str("id >= 40");
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(),
filter.as_ptr()) };
-+ assert_eq!(
-+ unsafe { lance_scanner_set_fragment_ids(scanner,
fragment_ids[1..].as_ptr(), 1) },
-+ 0
-+ );
-+
-+ // Match the Doris call order: nearest is configured before prefilter.
-+ let column = c_str("embedding");
-+ let query = [40.0_f32; 8];
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_nearest(
-+ scanner,
-+ column.as_ptr(),
-+ query.as_ptr().cast(),
-+ query.len(),
-+ LanceDataType::Float32 as i32,
-+ 5,
-+ )
-+ },
-+ 0
-+ );
-+ assert_eq!(unsafe { lance_scanner_set_prefilter(scanner, true) }, 0);
-+ assert_eq!(unsafe { lance_scanner_set_use_index(scanner, false) }, 0);
-+
-+ let batches = scan_all_rows_from_scanner(scanner);
-+ let mut ids = batches
-+ .iter()
-+ .flat_map(|batch| {
-+ batch
-+ .column_by_name("id")
-+ .unwrap()
-+ .as_any()
-+ .downcast_ref::<Int32Array>()
-+ .unwrap()
-+ .values()
-+ .iter()
-+ .copied()
-+ .collect::<Vec<_>>()
-+ })
-+ .collect::<Vec<_>>();
-+ ids.sort_unstable();
-+ assert_eq!(ids, vec![40, 41, 42, 43, 44]);
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_scanner_nearest_prefilter_with_fragment_ids_arrow_stream() {
-+ let (_tmp, uri) = create_multi_fragment_vector_dataset(32, 8, false);
-+ let uri_c = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ let mut fragment_ids = vec![0; unsafe { lance_dataset_fragment_count(ds)
} as usize];
-+ assert_eq!(fragment_ids.len(), 2);
-+ assert_eq!(
-+ unsafe { lance_dataset_fragment_ids(ds, fragment_ids.as_mut_ptr()) },
-+ 0
-+ );
-+
-+ let filter = c_str("id >= 60");
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(),
filter.as_ptr()) };
-+ assert_eq!(
-+ unsafe { lance_scanner_set_fragment_ids(scanner,
fragment_ids[1..].as_ptr(), 1) },
-+ 0
-+ );
-+
-+ let column = c_str("embedding");
-+ let query = [60.0_f32; 8];
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_nearest(
-+ scanner,
-+ column.as_ptr(),
-+ query.as_ptr().cast(),
-+ query.len(),
-+ LanceDataType::Float32 as i32,
-+ 10,
-+ )
-+ },
-+ 0
-+ );
-+ assert_eq!(unsafe { lance_scanner_set_prefilter(scanner, true) }, 0);
-+ assert_eq!(unsafe { lance_scanner_set_use_index(scanner, false) }, 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 mut ids = reader
-+ .flat_map(|batch| {
-+ let batch = batch.unwrap();
-+ batch
-+ .column_by_name("id")
-+ .unwrap()
-+ .as_any()
-+ .downcast_ref::<Int32Array>()
-+ .unwrap()
-+ .values()
-+ .iter()
-+ .copied()
-+ .collect::<Vec<_>>()
-+ })
-+ .collect::<Vec<_>>();
-+ ids.sort_unstable();
-+ assert_eq!(ids, vec![60, 61, 62, 63]);
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-diff --git a/tests/cpp/test_cpp_api.cpp b/tests/cpp/test_cpp_api.cpp
-index 43491f2..444fbdc 100644
---- a/tests/cpp/test_cpp_api.cpp
-+++ b/tests/cpp/test_cpp_api.cpp
-@@ -124,0 +125,29 @@ static void test_dataset_take(const std::string& uri) {
-+static void test_dataset_take_rows(const std::string& uri) {
-+ TEST(test_dataset_take_rows);
-+
-+ auto ds = lance::Dataset::open(uri);
-+
-+ // The smoke fixture has one fragment, so its first row IDs are 0, 1, 2.
-+ uint64_t row_ids[] = {0, 1, 2};
-+ ArrowArrayStream stream;
-+ memset(&stream, 0, sizeof(stream));
-+ ds.take_rows(row_ids, 3, &stream);
-+
-+ uint64_t total = 0;
-+ while (true) {
-+ ArrowArray arr;
-+ memset(&arr, 0, sizeof(arr));
-+ int rc = stream.get_next(&stream, &arr);
-+ assert(rc == 0);
-+ if (!arr.release) break;
-+ total += (uint64_t)arr.length;
-+ arr.release(&arr);
-+ }
-+
-+ assert(total == 3);
-+ printf("rows=%llu... ", (unsigned long long)total);
-+
-+ if (stream.release) stream.release(&stream);
-+ PASS();
-+}
-+
-@@ -695,0 +725 @@ int main(int argc, char** argv) {
-+ test_dataset_take_rows(uri);
diff --git a/thirdparty/patches/lance-c-0.1.7-pr-64.patch
b/thirdparty/patches/lance-c-0.1.7-pr-64.patch
new file mode 100644
index 00000000000..17f91bc7c31
--- /dev/null
+++ b/thirdparty/patches/lance-c-0.1.7-pr-64.patch
@@ -0,0 +1,1522 @@
+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/vars.sh b/thirdparty/vars.sh
index 50834061fb5..27fca98f3f9 100644
--- a/thirdparty/vars.sh
+++ b/thirdparty/vars.sh
@@ -583,10 +583,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.6.tar.gz";
-LANCE_C_NAME="lance-c-v0.1.6.tar.gz"
-LANCE_C_SOURCE="lance-c-0.1.6"
-LANCE_C_MD5SUM="1599faa2532d9ce963db1188f7435a56"
+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"
# 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]
