This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 f89181c2dfe [improvement](build) Add merged lance-c multi-vector API
patch (#68056)
f89181c2dfe is described below
commit f89181c2dfe60d00d68fe03b47b35217f1bd68c8
Author: Gabriel <[email protected]>
AuthorDate: Wed Sep 16 21:44:55 2026 +0800
[improvement](build) Add merged lance-c multi-vector API patch (#68056)
### What problem does this PR solve?
Related PR: #67741; consumer integration: #68028
Add merged
[lance-format/lance-c#83](https://github.com/lance-format/lance-c/pull/83)
to the Lance-C third-party patch chain on master. This provides the
C/C++ multi-vector query API and its upstream scoring, validation and
regression tests as a prerequisite for Doris multi-vector search
integration.
The patch records upstream commit
`0a30ee6c5a9d1455ceb36f4745acb79e53a00461`. Overlapping context is
adapted to the existing patch chain while retaining PR #79's
scalar-segment execution path and tests. The multi-vector implementation
and integration tests are byte-identical to the merged upstream commit.
The release archive and dependency versions remain unchanged from
#67741.
Apply PR #83 with a separate completion marker so source trees already
carrying the existing patches also receive the new API, and repeat runs
do not reapply it. Rebuild Lance-C before consuming the new API. This PR
changes only third-party patch integration and does not expose new Doris
SQL syntax.
### Validation
- Applied the actual Lance-C patch-driver block to clean v0.1.9 sources
and to sources already patched through #67741; both produce identical
source trees. Repeat execution leaves the sources unchanged. All patches
apply with `--fuzz=0`.
- Compared the resulting source tree with the previously validated
integrated chain and checked the multi-vector implementation/tests
against the merged upstream commit.
- `cargo test --locked`: 410 tests passed, including 14 multi-vector
integration tests; three opt-in C/C++ compile/run and static-transport
tests were ignored in this run.
- `cargo fmt --check` and `bash -n thirdparty/download-thirdparty.sh`
passed.
- Completed local self-review of patch provenance, conflict adaptation,
existing API behavior, ownership/resource bounds and cached-source
application before committing.
A complete Doris build and SQL regression run are not part of this local
dependency-only validation.
### Release note
None
### Check List (For Author)
- Test
- [x] Unit Test
- [x] Manual test: patch application, cached-source upgrade, repeat
execution and source equivalence
- Behavior changed:
- [x] Yes: the third-party library gains the upstream multi-vector API;
existing API signatures remain unchanged.
- Does this need documentation?
- [x] No: upstream API documentation is included in the patch.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
thirdparty/download-thirdparty.sh | 6 +
thirdparty/patches/lance-c-0.1.9-pr-83.patch | 1914 ++++++++++++++++++++++++++
2 files changed, 1920 insertions(+)
diff --git a/thirdparty/download-thirdparty.sh
b/thirdparty/download-thirdparty.sh
index 324320fa751..f55df80e340 100755
--- a/thirdparty/download-thirdparty.sh
+++ b/thirdparty/download-thirdparty.sh
@@ -764,6 +764,12 @@ if [[ " ${TP_ARCHIVES[*]} " =~ " LANCE_C " ]]; then
done
touch "${PATCHED_MARK}"
fi
+ # The base marker may already exist in cached sources; apply PR #83
independently.
+ if [[ ! -f "${PATCHED_MARK}_pr83" ]]; then
+ patch --batch --forward --reject-file=- --fuzz=0
--no-backup-if-mismatch -s \
+ -p1 <"${TP_PATCH_DIR}/${LANCE_C_SOURCE}-pr-83.patch"
+ touch "${PATCHED_MARK}_pr83"
+ fi
cd -
echo "Finished patching ${LANCE_C_SOURCE}"
fi
diff --git a/thirdparty/patches/lance-c-0.1.9-pr-83.patch
b/thirdparty/patches/lance-c-0.1.9-pr-83.patch
new file mode 100644
index 00000000000..86c0c23e227
--- /dev/null
+++ b/thirdparty/patches/lance-c-0.1.9-pr-83.patch
@@ -0,0 +1,1914 @@
+From 0a30ee6c5a9d1455ceb36f4745acb79e53a00461 Mon Sep 17 00:00:00 2001
+Subject: [PATCH] feat: support multi-vector queries through C and C++ APIs
(#83)
+
+Upstream: https://github.com/lance-format/lance-c/pull/83
+Commit: 0a30ee6c5a9d1455ceb36f4745acb79e53a00461
+
+Adapted to the v0.1.9 community patch chain used by Doris. Resolve context
+conflicts with PR #73 and retain PR #79 scalar_segment fields, execution
+branch, and tests alongside the upstream multi-vector additions. The
+multi-vector implementation and its integration tests are unchanged.
+
+diff --git a/README.md b/README.md
+index d9a5f2b..29c89c4 100644
+--- a/README.md
++++ b/README.md
+@@ -70,6 +70,40 @@ Based on the [liblance
RFC](https://github.com/lance-format/lance/discussions/60
+ | [x] | Filter pushdown | `lance_scanner_set_substrait_filter()` accepts a
serialized Substrait `ExtendedExpression`;
`lance_scanner_additional_sql_filter()` adds SQL predicates with AND before
scanning starts |
+ | [x] | Data-file cache | Optional Foyer memory/disk cache for immutable
`data/*.lance` reads |
+
++## Multi-vector search
++
++Use `lance_scanner_nearest_multivector` or the C++
`Scanner::nearest_multivector`
++method for a `List<FixedSizeList<float16|float32|float64, D>>` column:
++
++```cpp
++const float query[] = {1.0f, 0.0f, 0.0f, 1.0f};
++auto scanner = dataset.scan();
++scanner.nearest_multivector("embeddings", query, 2, 2, LANCE_DTYPE_FLOAT32,
10)
++ .metric(LANCE_METRIC_COSINE)
++ .prefilter(true);
++```
++
++The copied, row-major matrix is **one query** containing two subvectors.
Results
++rank logical rows by the sum of each query subvector's minimum distance to a
++stored subvector. Empty or null outer rows do not rank. Inner vectors must be
++non-nullable; actual stored null or non-finite elements encountered during
++scoring fail the stream. Float types and dimensions must match the column.
++Cosine pairs with zero norm have undefined distance and are ignored. A row is
++excluded if any query subvector has no defined match; a zero-norm query
subvector
++therefore produces no results. Column names use Lance field-path syntax,
++including nested paths such as `payload.embeddings` and backtick-quoted names.
++
++L2 is the default on every fragment. Cosine multi-vector indexes are supported
++by the pinned Lance version; incompatible metrics use exact search. Indexed
++candidates are refined against stored values (`refine_factor` defaults to 1).
++ANN candidate selection remains approximate. Limit and offset apply after
++restoring distance order, including fragment-scoped searches. Strict row
batching
++is applied after that final result window, preserving full batches except the
last.
++
++Queries accept at most 128 subvectors. Both `num_vectors * k` and
++`refine_factor * k` must be at most 100,000 to bound plan expansion and
candidate
++allocation. The existing single-vector API and its defaults are unchanged.
++
+ ## Building
+
+ There are four supported entry points; pick whichever matches your toolchain.
+diff --git a/include/lance/lance.h b/include/lance/lance.h
+index a201bfb..e404612 100644
+--- a/include/lance/lance.h
++++ b/include/lance/lance.h
+@@ -1847,6 +1847,21 @@ int32_t lance_scanner_nearest(
+ uint32_t k
+ );
+
++/**
++ * Set one multi-vector query on a
List<FixedSizeList<float16|float32|float64>> column.
++ * Inner vectors must be non-nullable and contain no null elements; the outer
list may be nullable.
++ * query_data contains dimension * num_vectors aligned elements in row-major
order.
++ * Both sizes and k must be positive. At most 128 query subvectors are
accepted;
++ * num_vectors * k and refine_factor * k must each be at most 100000.
++ * Values are copied before returning. The default metric is L2 on every
fragment.
++ * Scores sum each query vector's minimum distance; refinement defaults to 1.
++ * Returns 0 on success, -1 on error. Stored invalid elements fail during
execution.
++ */
++int32_t lance_scanner_nearest_multivector(
++ LanceScanner* scanner, const char* column, const void* query_data,
++ size_t dimension, size_t num_vectors, LanceDataType element_type,
uint32_t k
++);
++
+ /**
+ * Set both the minimum and maximum vector-index partition-search bounds.
+ *
+diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp
+index a231a00..146c6b1 100644
+--- a/include/lance/lance.hpp
++++ b/include/lance/lance.hpp
+@@ -1465,6 +1465,16 @@ public:
+ return *this;
+ }
+
++ /// One multi-vector query, copied from dimension * num_vectors row-major
elements.
++ Scanner& nearest_multivector(const std::string& column, const void*
query_data,
++ size_t dimension, size_t num_vectors,
++ LanceDataType element_type, uint32_t k) {
++ if (lance_scanner_nearest_multivector(handle_.get(), column.c_str(),
query_data,
++ dimension, num_vectors,
element_type, k) != 0)
++ check_error();
++ return *this;
++ }
++
+ /// Replace both minimum and maximum partition-search bounds.
+ Scanner& nprobes(uint32_t nprobes) {
+ if (lance_scanner_set_nprobes(handle_.get(), nprobes) != 0)
check_error();
+diff --git a/src/lib.rs b/src/lib.rs
+index 197da5f..55f6ecf 100644
+--- a/src/lib.rs
++++ b/src/lib.rs
+@@ -39,6 +39,7 @@ mod index;
+ mod index_model;
+ mod index_segment;
+ mod merge_insert;
++mod multivector;
+ mod restore;
+ pub mod runtime;
+ mod scalar_segment;
+diff --git a/src/multivector.rs b/src/multivector.rs
+new file mode 100644
+index 0000000..c08df28
+--- /dev/null
++++ b/src/multivector.rs
+@@ -0,0 +1,514 @@
++// SPDX-License-Identifier: Apache-2.0
++// SPDX-FileCopyrightText: Copyright The Lance Authors
++
++//! Correct multi-vector scoring before the pinned Lance plan's candidate
limits.
++
++use std::collections::HashMap;
++use std::sync::Arc;
++
++use arrow_array::types::{Float16Type, Float32Type, Float64Type};
++use arrow_array::{
++ Array, ArrayRef, ArrowPrimitiveType, BooleanArray, FixedSizeListArray,
Float32Array, ListArray,
++ RecordBatch, UInt64Array,
++};
++use arrow_schema::{DataType, SchemaRef};
++use datafusion::error::{DataFusionError, Result};
++use datafusion::execution::context::TaskContext;
++use datafusion::physical_plan::{
++ DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
SendableRecordBatchStream,
++ stream::RecordBatchStreamAdapter,
++};
++use futures::{StreamExt, TryStreamExt, stream};
++use lance::io::exec::KNNVectorDistanceExec;
++use lance_linalg::distance::{Cosine, DistanceType, Dot, L2};
++
++// Lance creates one ANN branch per query vector,
++// each overfetching 10 * k candidates before scoring; wire bytes alone
cannot bound this work.
++pub(crate) const MAX_QUERY_VECTORS: usize = 128;
++pub(crate) const MAX_QUERY_VECTOR_CANDIDATES: usize = 100_000;
++
++fn invalid(message: impl Into<String>) -> DataFusionError {
++ DataFusionError::Execution(message.into())
++}
++
++/// Rewrite inside TopK/refinement, before any score can discard a candidate.
++pub(crate) fn rewrite(plan: Arc<dyn ExecutionPlan>) -> Result<Arc<dyn
ExecutionPlan>> {
++ let children = plan
++ .children()
++ .into_iter()
++ .map(|child| rewrite(child.clone()))
++ .collect::<Result<Vec<_>>>()?;
++ let plan = if children.is_empty() {
++ plan
++ } else {
++ plan.with_new_children(children)?
++ };
++ let mode = if let Some(exact) =
plan.downcast_ref::<KNNVectorDistanceExec>() {
++ if exact.is_batch {
++ return Err(invalid(
++ "expected one logical multi-vector query, not batch queries",
++ ));
++ }
++ Some(Scoring::Exact {
++ query: exact.query.clone(),
++ column: exact.column.clone(),
++ metric: exact.distance_type,
++ })
++ // This pinned Lance node is not publicly re-exported, so match its
stable plan name.
++ } else if plan.name() == "MultivectorScoringExec" {
++ Some(Scoring::Indexed)
++ } else {
++ None
++ };
++ Ok(match mode {
++ Some(mode) => Arc::new(MultiVectorScoreExec {
++ original: plan,
++ mode,
++ }),
++ None => plan,
++ })
++}
++
++/// Apply the final distance-ordered window without invalidating output
batching.
++pub(crate) fn apply_result_window(
++ plan: Arc<dyn ExecutionPlan>,
++ offset: usize,
++ limit: Option<usize>,
++) -> Result<Arc<dyn ExecutionPlan>> {
++ use datafusion::physical_expr::{PhysicalSortExpr, expressions};
++ use datafusion::physical_plan::{
++ coalesce_partitions::CoalescePartitionsExec, limit::GlobalLimitExec,
sorts::sort::SortExec,
++ };
++ if plan
++ .downcast_ref::<lance_datafusion::exec::StrictBatchSizeExec>()
++ .is_some()
++ {
++ // Offset can split a previously strict batch. Keep Lance's final
rechunker
++ // outside the window, preserving its resolved batch size, including
defaults.
++ let input = apply_result_window(plan.children()[0].clone(), offset,
limit)?;
++ return plan.with_new_children(vec![input]);
++ }
++ let sort = PhysicalSortExpr {
++ expr: expressions::col("_distance", plan.schema().as_ref())?,
++ options: arrow::compute::SortOptions {
++ descending: false,
++ nulls_first: false,
++ },
++ };
++ // Fragment-scoped payload takes can reorder batches. Restore distance
order
++ // before the window; the nearest plan already bounds candidate rows by k.
++ let sorted = Arc::new(SortExec::new(
++ [sort].into(),
++ Arc::new(CoalescePartitionsExec::new(plan)),
++ ));
++ Ok(Arc::new(GlobalLimitExec::new(sorted, offset, limit)))
++}
++
++#[derive(Clone, Debug)]
++enum Scoring {
++ Exact {
++ query: ArrayRef,
++ column: String,
++ metric: DistanceType,
++ },
++ Indexed,
++}
++
++#[derive(Debug)]
++struct MultiVectorScoreExec {
++ original: Arc<dyn ExecutionPlan>,
++ mode: Scoring,
++}
++
++impl DisplayAs for MultiVectorScoreExec {
++ fn fmt_as(&self, _: DisplayFormatType, f: &mut std::fmt::Formatter) ->
std::fmt::Result {
++ write!(f, "MultiVectorScore: {}", self.original.name())
++ }
++}
++
++impl ExecutionPlan for MultiVectorScoreExec {
++ fn name(&self) -> &str {
++ "MultiVectorScoreExec"
++ }
++ fn properties(&self) -> &Arc<PlanProperties> {
++ self.original.properties()
++ }
++ fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
++ self.original.children()
++ }
++ fn required_input_distribution(&self) ->
Vec<datafusion::physical_expr::Distribution> {
++ self.original.required_input_distribution()
++ }
++ fn with_new_children(
++ self: Arc<Self>,
++ children: Vec<Arc<dyn ExecutionPlan>>,
++ ) -> Result<Arc<dyn ExecutionPlan>> {
++ Ok(Arc::new(Self {
++ original: self.original.clone().with_new_children(children)?,
++ mode: self.mode.clone(),
++ }))
++ }
++ fn execute(
++ &self,
++ partition: usize,
++ context: Arc<TaskContext>,
++ ) -> Result<SendableRecordBatchStream> {
++ let schema = self.schema();
++ match &self.mode {
++ Scoring::Exact {
++ query,
++ column,
++ metric,
++ } => {
++ let input = self.children()[0].execute(partition, context)?;
++ let query = query.clone();
++ let column = column.clone();
++ let metric = *metric;
++ let output_schema = schema.clone();
++ let output = input
++ .map(move |batch| {
++ let query = query.clone();
++ let column = column.clone();
++ let schema = output_schema.clone();
++ async move {
++ let batch = batch?;
++ tokio::task::spawn_blocking(move || {
++ exact_batch(batch, query, &column, metric,
schema)
++ })
++ .await
++ .map_err(|e|
DataFusionError::External(Box::new(e)))?
++ }
++ })
++
.buffered(lance_core::utils::tokio::get_num_compute_intensive_cpus());
++ Ok(Box::pin(RecordBatchStreamAdapter::new(schema, output)))
++ }
++ Scoring::Indexed => {
++ let inputs = self
++ .children()
++ .into_iter()
++ .map(|child| child.execute(partition, context.clone()))
++ .collect::<Result<Vec<_>>>()?;
++ let output_schema = schema.clone();
++ let output =
++ stream::once(async move { indexed_batch(inputs,
output_schema).await });
++ Ok(Box::pin(RecordBatchStreamAdapter::new(schema, output)))
++ }
++ }
++ }
++}
++
++fn row_distance<T: ArrowPrimitiveType>(
++ query: &dyn Array,
++ vectors: &FixedSizeListArray,
++ metric: DistanceType,
++) -> Result<Option<f32>>
++where
++ T::Native: L2 + Cosine + Dot + Into<f64>,
++{
++ let q = query
++ .as_any()
++ .downcast_ref::<arrow_array::PrimitiveArray<T>>()
++ .ok_or_else(|| invalid("multi-vector query element type mismatch"))?;
++ let values = vectors
++ .values()
++ .as_any()
++ .downcast_ref::<arrow_array::PrimitiveArray<T>>()
++ .ok_or_else(|| invalid("multi-vector stored element type mismatch"))?;
++ if vectors.null_count() != 0
++ || values.null_count() != 0
++ || values
++ .values()
++ .iter()
++ .any(|v| !Into::<f64>::into(*v).is_finite())
++ {
++ return Err(invalid(
++ "multi-vector stored subvectors must contain only finite,
non-null elements",
++ ));
++ }
++ let dimension = vectors.value_length() as usize;
++ let distance = metric.func();
++ // Subtracting each small distance from 1 rounds it away before TopK. Sum
minima
++ // directly, using f64 only for the accumulator; the base kernels and
output remain f32.
++ let mut score = 0.0f64;
++ for query_vector in q.values().chunks_exact(dimension) {
++ let best = values
++ .values()
++ .chunks_exact(dimension)
++ .map(|vector| distance(query_vector, vector))
++ // Finite zero-norm vectors have undefined cosine distance.
Ignore those
++ // pairs; a query with no defined match masks this row, not the
whole scan.
++ .filter(|distance| !distance.is_nan())
++ .min_by(f32::total_cmp);
++ let Some(best) = best else {
++ return Ok(None);
++ };
++ score += best as f64;
++ }
++ let score = score as f32;
++ if !score.is_finite() {
++ return Err(invalid("multi-vector distance is not finite"));
++ }
++ Ok(Some(score))
++}
++
++fn vector_column(batch: &RecordBatch, column: &str) -> Result<ArrayRef> {
++ if let Some(array) = batch.column_by_name(column) {
++ return Ok(array.clone());
++ }
++ // The planner resolves field paths, including quoted dotted names. Its
private
++ // KNN resolver is not exported, so use the same parser and struct
traversal here.
++ let parts = lance_core::datatypes::parse_field_path(column)
++ .map_err(|e| invalid(format!("invalid vector column path '{column}':
{e}")))?;
++ let root = parts
++ .first()
++ .ok_or_else(|| invalid("empty vector column path"))?;
++ let mut array = batch
++ .column_by_name(root)
++ .cloned()
++ .ok_or_else(|| invalid(format!("missing vector column '{column}'")))?;
++ for part in &parts[1..] {
++ array = array
++ .as_any()
++ .downcast_ref::<arrow_array::StructArray>()
++ .and_then(|parent| parent.column_by_name(part))
++ .cloned()
++ .ok_or_else(|| {
++ invalid(format!(
++ "missing struct field '{part}' in vector column
'{column}'"
++ ))
++ })?;
++ }
++ Ok(array)
++}
++
++fn exact_batch(
++ batch: RecordBatch,
++ query: ArrayRef,
++ column: &str,
++ metric: DistanceType,
++ schema: SchemaRef,
++) -> Result<RecordBatch> {
++ if batch.num_rows() == 0 {
++ return Ok(RecordBatch::new_empty(schema));
++ }
++ let array = vector_column(&batch, column)?;
++ let vectors = array
++ .as_any()
++ .downcast_ref::<ListArray>()
++ .ok_or_else(|| invalid("multi-vector scoring requires a List
column"))?;
++ let row_ids = batch.column_by_name("_rowid");
++ let mut scores = Vec::with_capacity(batch.num_rows());
++ for (i, vector) in vectors.iter().enumerate() {
++ if row_ids.is_some_and(|ids| ids.is_null(i)) {
++ scores.push(None);
++ continue;
++ }
++ let Some(vector) = vector else {
++ scores.push(None);
++ continue;
++ };
++ let vector = vector
++ .as_any()
++ .downcast_ref::<FixedSizeListArray>()
++ .ok_or_else(|| invalid("multi-vector row must be a
FixedSizeList"))?;
++ if vector.is_empty() {
++ scores.push(None);
++ continue;
++ }
++ let score = match query.data_type() {
++ DataType::Float16 => row_distance::<Float16Type>(query.as_ref(),
vector, metric),
++ DataType::Float32 => row_distance::<Float32Type>(query.as_ref(),
vector, metric),
++ DataType::Float64 => row_distance::<Float64Type>(query.as_ref(),
vector, metric),
++ _ => Err(invalid("unsupported multi-vector element type")),
++ }?;
++ scores.push(score);
++ }
++ let mask = BooleanArray::from_iter(scores.iter().map(|score|
Some(score.is_some())));
++ let distances: ArrayRef = Arc::new(Float32Array::from(scores));
++ let columns = schema
++ .fields()
++ .iter()
++ .map(|field| {
++ if field.name() == "_distance" {
++ Ok(distances.clone())
++ } else {
++ batch
++ .column_by_name(field.name())
++ .cloned()
++ .ok_or_else(|| invalid(format!("missing score output
column {}", field.name())))
++ }
++ })
++ .collect::<Result<Vec<_>>>()?;
++ Ok(arrow::compute::filter_record_batch(
++ &RecordBatch::try_new(schema, columns)?,
++ &mask,
++ )?)
++}
++
++async fn indexed_batch(
++ inputs: Vec<SendableRecordBatchStream>,
++ schema: SchemaRef,
++) -> Result<RecordBatch> {
++ // A query child may emit several batches. Reduce its entire stream
exactly once;
++ // treating each batch as a query adds spurious missing-query
contributions.
++ let queries = futures::future::try_join_all(inputs.into_iter().map(|mut
input| async move {
++ let mut rows = HashMap::<u64, f32>::new();
++ let mut maximum: Option<f32> = None;
++ while let Some(batch) = input.try_next().await? {
++ let ids = batch
++ .column_by_name("_rowid")
++ .and_then(|a| a.as_any().downcast_ref::<UInt64Array>())
++ .ok_or_else(|| invalid("indexed multi-vector scorer requires
row IDs"))?;
++ let distances = batch
++ .column_by_name("_distance")
++ .and_then(|a| a.as_any().downcast_ref::<Float32Array>())
++ .ok_or_else(|| invalid("indexed multi-vector scorer requires
distances"))?;
++ for i in 0..batch.num_rows() {
++ let distance = distances.value(i);
++ if ids.is_null(i) || distances.is_null(i) ||
!distance.is_finite() {
++ return Err(invalid(
++ "indexed multi-vector candidate has invalid row ID or
distance",
++ ));
++ }
++ maximum = Some(maximum.map_or(distance, |old|
old.max(distance)));
++ rows.entry(ids.value(i))
++ .and_modify(|old| *old = old.min(distance))
++ .or_insert(distance);
++ }
++ }
++ Ok::<_, DataFusionError>((rows, maximum.unwrap_or(1.0)))
++ }))
++ .await?;
++ let mut results = HashMap::<u64, f64>::new();
++ let mut missed = 0.0f64;
++ for (rows, maximum) in queries {
++ for (id, score) in &mut results {
++ *score += *rows.get(id).unwrap_or(&maximum) as f64;
++ }
++ for (id, score) in rows {
++ results.entry(id).or_insert(score as f64 + missed);
++ }
++ missed += maximum as f64;
++ }
++ let (ids, scores): (Vec<_>, Vec<_>) = results.into_iter().unzip();
++ Ok(RecordBatch::try_new(
++ schema,
++ vec![
++ Arc::new(Float32Array::from_iter_values(
++ scores.into_iter().map(|score| score as f32),
++ )),
++ Arc::new(UInt64Array::from(ids)),
++ ],
++ )?)
++}
++
++pub(crate) fn validate_query(values: &dyn Array) -> Result<()> {
++ fn finite<T: ArrowPrimitiveType>(values: &dyn Array) -> bool
++ where
++ T::Native: Into<f64>,
++ {
++ values
++ .as_any()
++ .downcast_ref::<arrow_array::PrimitiveArray<T>>()
++ .is_some_and(|array| {
++ array.null_count() == 0
++ && array
++ .values()
++ .iter()
++ .all(|v| Into::<f64>::into(*v).is_finite())
++ })
++ }
++ let valid = match values.data_type() {
++ DataType::Float16 => finite::<Float16Type>(values),
++ DataType::Float32 => finite::<Float32Type>(values),
++ DataType::Float64 => finite::<Float64Type>(values),
++ _ => false,
++ };
++ if valid {
++ Ok(())
++ } else {
++ Err(invalid(
++ "multi-vector query must contain only finite, non-null elements",
++ ))
++ }
++}
++
++#[cfg(test)]
++mod tests {
++ use super::*;
++ use arrow_schema::{Field, Schema};
++ use lance_datafusion::exec::{
++ LanceExecutionOptions, OneShotExec, StrictBatchSizeExec, execute_plan,
++ };
++
++ #[test]
++ fn result_window_preserves_strict_batching_and_distance_order() {
++ crate::runtime::block_on(async {
++ for size in [2, 3] {
++ for (offset, limit) in [
++ (1, Some(4)),
++ (1, Some(3)),
++ (5, Some(4)),
++ (6, Some(4)),
++ (1, None),
++ ] {
++ let schema = Arc::new(Schema::new(vec![Field::new(
++ "_distance",
++ DataType::Float32,
++ false,
++ )]));
++ let batches = [[5., 0.], [4., 1.], [3., 2.]]
++ .into_iter()
++ .map(|values| {
++ RecordBatch::try_new(
++ schema.clone(),
++
vec![Arc::new(Float32Array::from(values.to_vec()))],
++ )
++ .map_err(DataFusionError::from)
++ })
++ .collect::<Vec<_>>();
++ let input = Arc::new(OneShotExec::new(Box::pin(
++ RecordBatchStreamAdapter::new(schema,
stream::iter(batches)),
++ )));
++ let plan = Arc::new(StrictBatchSizeExec::new(input,
size));
++ let plan = apply_result_window(plan, offset,
limit).unwrap();
++ let batches: Vec<_> = execute_plan(
++ plan,
++ LanceExecutionOptions {
++ batch_size: Some(2),
++ ..Default::default()
++ },
++ )
++ .unwrap()
++ .try_collect()
++ .await
++ .unwrap();
++ let actual: Vec<_> = batches
++ .iter()
++ .flat_map(|b| {
++ b.column(0)
++ .as_any()
++ .downcast_ref::<Float32Array>()
++ .unwrap()
++ .values()
++ .to_vec()
++ })
++ .collect();
++ let expected: Vec<_> = (0..6)
++ .skip(offset)
++ .take(limit.unwrap_or(6))
++ .map(|i| i as f32)
++ .collect();
++ assert_eq!(actual, expected);
++ assert_eq!(
++ batches
++ .iter()
++ .map(RecordBatch::num_rows)
++ .collect::<Vec<_>>(),
++
expected.chunks(size).map(<[f32]>::len).collect::<Vec<_>>()
++ );
++ }
++ }
++ });
++ }
++}
+diff --git a/src/scanner.rs b/src/scanner.rs
+index bda554d..f4b3f36 100644
+--- a/src/scanner.rs
++++ b/src/scanner.rs
+@@ -363,8 +363,18 @@ impl LanceScanner {
+ if let Some(cols) = &self.columns {
+ scanner.project(cols)?;
+ }
++ let multi_vector = self.nearest.as_ref().is_some_and(|query| {
++ matches!(
++ query.query.data_type(),
++ arrow_schema::DataType::FixedSizeList(_, _)
++ )
++ });
+ if self.limit.is_some() || self.offset.is_some() {
+ scanner.limit(self.limit, self.offset)?;
++ if multi_vector {
++ // Retain Lance's window validation, but defer truncation
until the final sort.
++ scanner.limit(None, None)?;
++ }
+ }
+ if let Some(bs) = self.batch_size {
+ scanner.batch_size(bs);
+@@ -440,7 +450,27 @@ impl LanceScanner {
+ if let Some(query_parallelism) = self.query_parallelism {
+ scanner.query_parallelism(query_parallelism);
+ }
+- if let Some(rf) = self.refine_factor {
++ if multi_vector {
++ if matches!(
++ self.metric_override,
++ Some(crate::index::LanceMetricType::Hamming)
++ ) {
++ return Err(lance_core::Error::invalid_input_source(
++ "multi-vector queries support only l2, cosine, and
dot metrics".into(),
++ ));
++ }
++ let refine = self.refine_factor.unwrap_or(1);
++ if refine == 0
++ || n.k as usize
++ > crate::multivector::MAX_QUERY_VECTOR_CANDIDATES /
refine as usize
++ {
++ return Err(lance_core::Error::invalid_input_source(
++ "multi-vector refined candidate count must be in
1..=100000".into(),
++ ));
++ }
++ // Validate actual stored values and refine candidate scores
before TopK.
++ scanner.refine(refine);
++ } else if let Some(rf) = self.refine_factor {
+ scanner.refine(rf);
+ }
+ if let Some(ef) = self.ef {
+@@ -448,6 +478,9 @@ impl LanceScanner {
+ }
+ if let Some(m) = self.metric_override {
+ scanner.distance_metric(m.to_distance());
++ } else if multi_vector {
++ // Resolve the same default on indexed and uncovered
fragments.
++
scanner.distance_metric(lance_linalg::distance::DistanceType::L2);
+ }
+ if let Some(ui) = self.use_index {
+ scanner.use_index(ui);
+@@ -506,6 +539,12 @@ impl LanceScanner {
+ scanner,
+ distributed_fts,
+ scalar_segment,
++ multi_vector_window: multi_vector.then_some((
++ self.offset.unwrap_or(0) as usize,
++ self.limit.map(|n| n as usize),
++ )),
++ batch_size: self.batch_size,
++ scan_statistics_callback: self.scan_statistics_callback.clone(),
+ })
+ }
+ }
+@@ -521,6 +560,9 @@ struct PreparedScanner {
+ scanner: lance::dataset::scanner::Scanner,
+ distributed_fts: Option<PreparedFtsExecution>,
+ scalar_segment: Option<PreparedScalarSegment>,
++ multi_vector_window: Option<(usize, Option<usize>)>,
++ batch_size: Option<usize>,
++ scan_statistics_callback: Option<ExecutionStatsCallback>,
+ }
+
+ impl PreparedScanner {
+@@ -532,6 +574,19 @@ impl PreparedScanner {
+ .try_into_stream()
+ .await;
+ }
++ if let Some((offset, limit)) = self.multi_vector_window {
++ let plan =
crate::multivector::rewrite(self.scanner.create_plan().await?)?;
++ let plan = crate::multivector::apply_result_window(plan, offset,
limit)?;
++ let stream = lance_datafusion::exec::execute_plan(
++ plan,
++ lance_datafusion::exec::LanceExecutionOptions {
++ batch_size: self.batch_size,
++ execution_stats_callback: self.scan_statistics_callback,
++ ..Default::default()
++ },
++ )?;
++ return Ok(DatasetRecordBatchStream::new(stream));
++ }
+ let Some(distributed_fts) = self.distributed_fts else {
+ return self.scanner.try_into_stream().await;
+ };
+@@ -2744,6 +2799,21 @@ unsafe fn scanner_nearest_inner(
+ }
+ let column_str = unsafe { helpers::parse_c_string(column)? }.unwrap();
+
++ let query = unsafe { decode_query_values(query_data, query_len,
element_type)? };
++
++ s.nearest = Some(NearestQuery {
++ column: column_str.to_string(),
++ query,
++ k,
++ });
++ Ok(0)
++}
++
++unsafe fn decode_query_values(
++ query_data: *const c_void,
++ query_len: usize,
++ element_type: i32,
++) -> Result<arrow_array::ArrayRef> {
+ let dtype = match element_type {
+ 0 => LanceDataType::Float32,
+ 1 => LanceDataType::Float16,
+@@ -2782,9 +2852,112 @@ unsafe fn scanner_nearest_inner(
+ }
+ };
+
++ Ok(query)
++}
++
++/// Set one multi-vector query, supplied as a row-major matrix of
floating-point values.
++/// The caller must supply dimension * num_vectors aligned elements matching
the column type.
++#[unsafe(no_mangle)]
++pub unsafe extern "C" fn lance_scanner_nearest_multivector(
++ scanner: *mut LanceScanner,
++ column: *const c_char,
++ query_data: *const c_void,
++ dimension: usize,
++ num_vectors: usize,
++ element_type: i32,
++ k: u32,
++) -> i32 {
++ scanner_poison_check!(scanner, -1);
++ scanner_ffi_try!(scanner, unsafe {
++ nearest_multivector_inner(
++ scanner,
++ column,
++ query_data,
++ dimension,
++ num_vectors,
++ element_type,
++ k,
++ )
++ },)
++}
++
++unsafe fn nearest_multivector_inner(
++ scanner: *mut LanceScanner,
++ column: *const c_char,
++ query_data: *const c_void,
++ dimension: usize,
++ num_vectors: usize,
++ element_type: i32,
++ k: u32,
++) -> Result<i32> {
++ use arrow_schema::{DataType, Field};
++ let invalid = |message: &str|
lance_core::Error::invalid_input_source(message.into());
++ if scanner.is_null() || column.is_null() || query_data.is_null() {
++ return Err(invalid("scanner, column, and query_data must not be
NULL"));
++ }
++ if dimension == 0 || dimension > i32::MAX as usize || num_vectors == 0 ||
k == 0 {
++ return Err(invalid(
++ "dimension, num_vectors, and k must be positive; dimension must
fit int32",
++ ));
++ }
++ if num_vectors > crate::multivector::MAX_QUERY_VECTORS
++ || num_vectors > crate::multivector::MAX_QUERY_VECTOR_CANDIDATES / k
as usize
++ {
++ return Err(invalid(
++ "multi-vector query exceeds 128 subvectors or 100000
subvector-candidates",
++ ));
++ }
++ let (data_type, width) = match element_type {
++ 0 => (DataType::Float32, 4),
++ 1 => (DataType::Float16, 2),
++ 2 => (DataType::Float64, 8),
++ _ => {
++ return Err(invalid(
++ "multi-vector queries require float16, float32, or float64",
++ ));
++ }
++ };
++ let count = dimension
++ .checked_mul(num_vectors)
++ .filter(|count| *count <= isize::MAX as usize / width)
++ .ok_or_else(|| invalid("query matrix byte size overflows"))?;
++ let s = unsafe { &mut *scanner };
++ if s.fts_query.is_some() || s.fts_context.is_some() {
++ return Err(invalid(
++ "nearest and full-text search are mutually exclusive",
++ ));
++ }
++ let column = unsafe { helpers::parse_c_string(column)? }.unwrap();
++ let field = s
++ .dataset
++ .schema()
++ .field(column)
++ .ok_or_else(|| invalid("multi-vector column does not exist"))?;
++ match field.data_type() {
++ DataType::List(child) if !child.is_nullable() => match
child.data_type() {
++ DataType::FixedSizeList(element, dim)
++ if *dim == dimension as i32 && *element.data_type() ==
data_type => {}
++ _ => return Err(invalid("multi-vector dimension/type mismatch")),
++ },
++ _ => {
++ return Err(invalid(
++ "multi-vector column must be List of non-nullable
FixedSizeList",
++ ));
++ }
++ }
++ // A primitive array is interpreted as one vector by Lance. Preserve
matrix shape even
++ // for a single subvector. Lance does not preserve element nullability in
its schema.
++ let values = unsafe { decode_query_values(query_data, count,
element_type)? };
++ crate::multivector::validate_query(values.as_ref())?;
++ let query = arrow_array::FixedSizeListArray::try_new(
++ Arc::new(Field::new("item", data_type, false)),
++ dimension as i32,
++ values,
++ None,
++ )?;
+ s.nearest = Some(NearestQuery {
+- column: column_str.to_string(),
+- query,
++ column: column.to_string(),
++ query: Arc::new(query),
+ k,
+ });
+ Ok(0)
+diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs
+index 04699da..24bef9f 100644
+--- a/tests/c_api_test.rs
++++ b/tests/c_api_test.rs
+@@ -13388,3 +13388,28 @@ fn
test_scalar_segment_requires_explicit_domain_and_checks_uuid() {
+ lance_dataset_close(ds);
+ }
+ }
++
++#[test]
++fn test_multivector_nearest_rejects_null_handle() {
++ let column = c_str("vectors");
++ let query = [1.0f32, 0.0];
++ let status = unsafe {
++ lance_scanner_nearest_multivector(
++ ptr::null_mut(),
++ column.as_ptr(),
++ query.as_ptr().cast(),
++ 2,
++ 1,
++ 0,
++ 1,
++ )
++ };
++ assert_eq!(status, -1);
++ let error = lance_last_error_message();
++ assert!(!error.is_null());
++ let message = unsafe { std::ffi::CStr::from_ptr(error) }
++ .to_string_lossy()
++ .into_owned();
++ unsafe { lance_free_string(error) };
++ assert!(message.contains("NULL"));
++}
+diff --git a/tests/cpp/test_cpp_api.cpp b/tests/cpp/test_cpp_api.cpp
+index fc6fc82..11badc6 100644
+--- a/tests/cpp/test_cpp_api.cpp
++++ b/tests/cpp/test_cpp_api.cpp
+@@ -427,6 +427,20 @@ static void test_nearest_smoke(const std::string& uri) {
+ PASS();
+ }
+
++static void test_multivector_rejects_flat_column(const std::string& uri) {
++ TEST(test_multivector_rejects_flat_column);
++ auto scanner = lance::Dataset::open(uri).scan();
++ const float query[8] = {1.0f, 0.0f, 0.0f, 0.0f, 0.0f, 0.0f, 0.0f, 0.0f};
++ bool caught = false;
++ try {
++ scanner.nearest_multivector("embedding", query, 8, 1,
LANCE_DTYPE_FLOAT32, 1);
++ } catch (const lance::Error&) {
++ caught = true;
++ }
++ assert(caught);
++ PASS();
++}
++
+ static void test_index_segments_smoke(const std::string& /*uri*/) {
+ TEST(test_index_segments_smoke);
+
+@@ -971,6 +985,7 @@ int main(int argc, char** argv) {
+ test_error_exception(uri);
+ test_index_lifecycle(uri);
+ test_nearest_smoke(uri);
++ test_multivector_rejects_flat_column(uri);
+ test_index_segments_smoke(uri);
+ test_index_segment_builder(uri);
+ test_vector_models_and_reusable_segments(uri);
+diff --git a/tests/multivector_test.rs b/tests/multivector_test.rs
+new file mode 100644
+index 0000000..f02e850
+--- /dev/null
++++ b/tests/multivector_test.rs
+@@ -0,0 +1,965 @@
++// SPDX-License-Identifier: Apache-2.0
++// SPDX-FileCopyrightText: Copyright The Lance Authors
++
++use std::ffi::{CString, c_void};
++use std::ptr;
++use std::sync::Arc;
++
++use arrow::buffer::{NullBuffer, OffsetBuffer};
++use arrow::ffi_stream::{ArrowArrayStreamReader, FFI_ArrowArrayStream};
++use arrow::record_batch::RecordBatchIterator;
++use arrow_array::{Array, FixedSizeListArray, Float32Array, Int32Array,
ListArray, RecordBatch};
++use arrow_schema::{DataType, Field, Schema};
++use lance::Dataset;
++use lance_c::*;
++
++fn fixture() -> (tempfile::TempDir, CString) {
++ let dir = tempfile::tempdir().unwrap();
++ let path = dir.path().join("vectors.lance");
++ let element = Arc::new(Field::new("item", DataType::Float32, false));
++ let vectors = FixedSizeListArray::try_new(
++ element,
++ 2,
++ Arc::new(Float32Array::from(vec![
++ 1., 0., 0., 1., 2., 0., 3., 0., 0., 3.,
++ ])),
++ None,
++ )
++ .unwrap();
++ let child = Arc::new(Field::new("item", vectors.data_type().clone(),
false));
++ let rows = ListArray::try_new(
++ child,
++ OffsetBuffer::new(vec![0i32, 2, 3, 5, 5, 5].into()),
++ Arc::new(vectors),
++ Some(NullBuffer::from(vec![true, true, true, true, false])),
++ )
++ .unwrap();
++ let schema = Arc::new(Schema::new(vec![
++ Field::new("id", DataType::Int32, false),
++ Field::new("vectors", rows.data_type().clone(), true),
++ ]));
++ let batch = RecordBatch::try_new(
++ schema.clone(),
++ vec![
++ Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5])),
++ Arc::new(rows),
++ ],
++ )
++ .unwrap();
++ lance_c::runtime::block_on(async {
++ Dataset::write(
++ RecordBatchIterator::new(vec![Ok(batch)], schema),
++ path.to_str().unwrap(),
++ None,
++ )
++ .await
++ .unwrap();
++ });
++ (dir, CString::new(path.to_str().unwrap()).unwrap())
++}
++
++unsafe fn collect(scanner: *mut LanceScanner) -> Vec<(i32, f32)> {
++ let mut stream = FFI_ArrowArrayStream::empty();
++ let status = unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream)
};
++ if status != 0 {
++ panic!(
++ "{}",
++ unsafe { std::ffi::CStr::from_ptr(lance_last_error_message())
}.to_string_lossy()
++ );
++ }
++ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
++ reader
++ .flat_map(|batch| {
++ let batch = batch.unwrap();
++ let ids = batch
++ .column_by_name("id")
++ .unwrap()
++ .as_any()
++ .downcast_ref::<Int32Array>()
++ .unwrap();
++ let distances = batch
++ .column_by_name("_distance")
++ .unwrap()
++ .as_any()
++ .downcast_ref::<Float32Array>()
++ .unwrap();
++ (0..batch.num_rows())
++ .map(|i| (ids.value(i), distances.value(i)))
++ .collect::<Vec<_>>()
++ })
++ .collect()
++}
++
++#[test]
++fn multivector_search_scores_logical_rows_and_excludes_empty_and_null_rows() {
++ let (_dir, uri) = fixture();
++ let column = CString::new("vectors").unwrap();
++ let q = [1.0f32, 0., 0., 1.];
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ assert!(!ds.is_null());
++ for count in [1, 2] {
++ let scan = lance_scanner_new(ds, ptr::null(), ptr::null());
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ q.as_ptr() as *const c_void,
++ 2,
++ count,
++ 0,
++ 10
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_use_index(scan, false), 0);
++ assert_eq!(lance_scanner_set_prefilter(scan, true), 0);
++ let rows = collect(scan);
++ let expected = if count == 2 {
++ vec![(1, 0.), (2, 6.), (3, 8.)]
++ } else {
++ vec![(1, 0.), (2, 1.), (3, 4.)]
++ };
++ assert_eq!(rows, expected);
++ lance_scanner_close(scan);
++ }
++ lance_dataset_close(ds);
++ }
++}
++
++#[test]
++fn multivector_rejects_invalid_shape_and_preserves_previous_query() {
++ let (_dir, uri) = fixture();
++ let column = CString::new("vectors").unwrap();
++ let q = [1.0f32, 0., 0., 1.];
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ let scan = lance_scanner_new(ds, ptr::null(), ptr::null());
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ q.as_ptr() as *const c_void,
++ 2,
++ 2,
++ 0,
++ 10
++ ),
++ 0
++ );
++ for (dim, count, dtype, k) in [
++ (0, 2, 0, 10),
++ (2, 0, 0, 10),
++ (3, 1, 0, 10),
++ (2, 2, 3, 10),
++ (2, 2, 4, 10),
++ (2, 2, 1, 10),
++ (2, 2, 0, 0),
++ (usize::MAX, 2, 0, 10),
++ (2, usize::MAX, 0, 10),
++ ] {
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ q.as_ptr() as *const c_void,
++ dim,
++ count,
++ dtype,
++ k
++ ),
++ -1
++ );
++ }
++ assert_eq!(
++ lance_scanner_nearest_multivector(scan, column.as_ptr(),
ptr::null(), 2, 2, 0, 10),
++ -1
++ );
++ assert_eq!(lance_scanner_set_use_index(scan, false), 0);
++ for (limit, offset) in [(-1, 0), (10, -1)] {
++ assert_eq!(lance_scanner_set_limit(scan, limit), 0);
++ assert_eq!(lance_scanner_set_offset(scan, offset), 0);
++ let mut stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(lance_scanner_to_arrow_stream(scan, &mut stream), -1);
++ }
++ assert_eq!(lance_scanner_set_limit(scan, 10), 0);
++ assert_eq!(lance_scanner_set_offset(scan, 0), 0);
++ assert_eq!(collect(scan), vec![(1, 0.), (2, 6.), (3, 8.)]);
++ lance_scanner_close(scan);
++ lance_dataset_close(ds);
++ }
++}
++
++#[test]
++fn fragment_scoped_indexed_search_orders_candidates_before_offset() {
++ use lance::index::DatasetIndexExt;
++ use lance::index::vector::VectorIndexParams;
++ use lance_index::IndexType;
++ use lance_linalg::distance::MetricType;
++
++ let (_dir, uri) = fixture();
++ lance_c::runtime::block_on(async {
++ let mut ds = Dataset::open(uri.to_str().unwrap()).await.unwrap();
++ let batch = ds.scan().try_into_batch().await.unwrap();
++ ds.create_index(
++ &["vectors"],
++ IndexType::Vector,
++ None,
++ &VectorIndexParams::ivf_flat(1, MetricType::Cosine),
++ false,
++ )
++ .await
++ .unwrap();
++ ds.append(
++ RecordBatchIterator::new(vec![Ok(batch.clone())], batch.schema()),
++ None,
++ )
++ .await
++ .unwrap();
++ });
++ let column = CString::new("vectors").unwrap();
++ let filter = CString::new("id >= 2").unwrap();
++ let q = [1.0f32, 0., 0., 1.];
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ for _ in 0..10 {
++ let scan = lance_scanner_new(ds, ptr::null(), filter.as_ptr());
++ assert_eq!(
++ lance_scanner_set_fragment_ids(scan, [0u64, 1].as_ptr(), 2),
++ 0
++ );
++ assert_eq!(lance_scanner_set_prefilter(scan, true), 0);
++ assert_eq!(lance_scanner_set_batch_size(scan, 1), 0);
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ q.as_ptr() as *const c_void,
++ 2,
++ 2,
++ 0,
++ 4
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_metric(scan, 1), 0);
++ assert_eq!(lance_scanner_set_refine_factor(scan, 1), 0);
++ assert_eq!(lance_scanner_set_offset(scan, 1), 0);
++ assert_eq!(lance_scanner_set_limit(scan, 2), 0);
++ assert_eq!(collect(scan), vec![(3, 0.), (2, 1.)]);
++ lance_scanner_close(scan);
++ }
++ lance_dataset_close(ds);
++ }
++}
++
++fn custom_fixture(
++ rows: Vec<Vec<Option<f32>>>,
++ dim: i32,
++ indexed: bool,
++) -> (tempfile::TempDir, CString) {
++ custom_fixture_at(rows, dim, indexed, "vectors")
++}
++
++fn custom_fixture_at(
++ rows: Vec<Vec<Option<f32>>>,
++ dim: i32,
++ indexed: bool,
++ column: &str,
++) -> (tempfile::TempDir, CString) {
++ use lance::index::DatasetIndexExt;
++ use lance::index::vector::VectorIndexParams;
++ use lance_index::IndexType;
++ use lance_linalg::distance::MetricType;
++ let dir = tempfile::tempdir().unwrap();
++ let path = dir.path().join("vectors.lance");
++ let mut offsets = vec![0i32];
++ let mut values = Vec::new();
++ for row in &rows {
++ values.extend_from_slice(row);
++ offsets.push(values.len() as i32 / dim);
++ }
++ let vectors = FixedSizeListArray::try_new(
++ Arc::new(Field::new("item", DataType::Float32, true)),
++ dim,
++ Arc::new(Float32Array::from(values)),
++ None,
++ )
++ .unwrap();
++ let rows_array = ListArray::try_new(
++ Arc::new(Field::new("item", vectors.data_type().clone(), false)),
++ OffsetBuffer::new(offsets.into()),
++ Arc::new(vectors),
++ None,
++ )
++ .unwrap();
++ let parts = lance_core::datatypes::parse_field_path(column).unwrap();
++ let mut array: arrow_array::ArrayRef = Arc::new(rows_array);
++ for name in parts[1..].iter().rev() {
++ let field = Arc::new(Field::new(name, array.data_type().clone(),
true));
++ let labels: arrow_array::ArrayRef = Arc::new(Int32Array::from(vec![7;
rows.len()]));
++ array = Arc::new(arrow_array::StructArray::from(vec![
++ (field, array),
++ (
++ Arc::new(Field::new("label", DataType::Int32, false)),
++ labels,
++ ),
++ ]));
++ }
++ let schema = Arc::new(Schema::new(vec![
++ Field::new("id", DataType::Int32, false),
++ Field::new(&parts[0], array.data_type().clone(), true),
++ ]));
++ let batch = RecordBatch::try_new(
++ schema.clone(),
++ vec![
++ Arc::new(Int32Array::from_iter_values(1..=rows.len() as i32)),
++ array,
++ ],
++ )
++ .unwrap();
++ lance_c::runtime::block_on(async {
++ let mut ds = Dataset::write(
++ RecordBatchIterator::new(vec![Ok(batch)], schema),
++ path.to_str().unwrap(),
++ None,
++ )
++ .await
++ .unwrap();
++ if indexed {
++ ds.create_index(
++ &[column],
++ IndexType::Vector,
++ None,
++ &VectorIndexParams::ivf_flat(1, MetricType::Cosine),
++ false,
++ )
++ .await
++ .unwrap();
++ }
++ });
++ (dir, CString::new(path.to_str().unwrap()).unwrap())
++}
++
++#[test]
++fn exact_top_one_preserves_small_distances_before_truncation() {
++ let (_dir, uri) = custom_fixture(vec![vec![Some(0.00015)],
vec![Some(0.0001)]], 1, false);
++ let column = CString::new("vectors").unwrap();
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ for count in [1, 2] {
++ for batch_size in [1, 1024] {
++ let scan = lance_scanner_new(ds, ptr::null(), ptr::null());
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ [0.0f32, 0.0].as_ptr().cast(),
++ 1,
++ count,
++ 0,
++ 1
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_use_index(scan, false), 0);
++ assert_eq!(lance_scanner_set_batch_size(scan, batch_size), 0);
++ let rows = collect(scan);
++ assert_eq!(rows.len(), 1);
++ assert_eq!(rows[0].0, 2);
++ assert!((rows[0].1 - count as f32 * 1e-8).abs() < 1e-14,
"{rows:?}");
++ lance_scanner_close(scan);
++ }
++ }
++ lance_dataset_close(ds);
++ }
++}
++
++#[test]
++fn indexed_top_one_is_independent_of_child_batch_boundaries() {
++ let rows = vec![
++ vec![Some(1.), Some(0.)],
++ vec![Some(0.), Some(1.)],
++ vec![Some(1.), Some(1.)],
++ vec![Some(2.), Some(1.)],
++ vec![Some(1.), Some(2.)],
++ ];
++ let (_dir, uri) = custom_fixture(rows, 2, true);
++ let column = CString::new("vectors").unwrap();
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ for batch_size in [1, 2, 1024] {
++ let scan = lance_scanner_new(ds, ptr::null(), ptr::null());
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ [1.0f32, 0., 0., 1.].as_ptr().cast(),
++ 2,
++ 2,
++ 0,
++ 1
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_metric(scan, 1), 0);
++ assert_eq!(lance_scanner_set_refine_factor(scan, 1), 0);
++ assert_eq!(lance_scanner_set_batch_size(scan, batch_size), 0);
++ let result = collect(scan);
++ assert_eq!(result[0].0, 3, "batch_size={batch_size}");
++ assert!((result[0].1 - (2.0 - 2.0f32.sqrt())).abs() < 1e-6);
++ lance_scanner_close(scan);
++ }
++ lance_dataset_close(ds);
++ }
++}
++
++#[test]
++fn actual_null_and_nonfinite_stored_elements_fail_search() {
++ let column = CString::new("vectors").unwrap();
++ for value in [
++ None,
++ Some(f32::NAN),
++ Some(f32::INFINITY),
++ Some(f32::NEG_INFINITY),
++ ] {
++ let (_dir, uri) = custom_fixture(vec![vec![value, Some(0.)]], 2,
false);
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ let scan = lance_scanner_new(ds, ptr::null(), ptr::null());
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ [0.0f32, 0.].as_ptr().cast(),
++ 2,
++ 1,
++ 0,
++ 1
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_use_index(scan, false), 0);
++ let mut stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(lance_scanner_to_arrow_stream(scan, &mut stream), 0);
++ let mut reader = ArrowArrayStreamReader::from_raw(&mut
stream).unwrap();
++ let result = reader.next();
++ assert!(
++ matches!(result, Some(Err(_))),
++ "value={value:?}: {result:?}"
++ );
++ drop(reader);
++ lance_scanner_close(scan);
++ lance_dataset_close(ds);
++ }
++ }
++}
++
++#[test]
++fn rejects_excessive_multivector_plan_width_and_candidate_work() {
++ let (_dir, uri) = fixture();
++ let column = CString::new("vectors").unwrap();
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ let scan = lance_scanner_new(ds, ptr::null(), ptr::null());
++ let q = [0.0f32; 258];
++ for (count, k) in [(129, 1), (128, 782), (1, 100001)] {
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ q.as_ptr().cast(),
++ 2,
++ count,
++ 0,
++ k
++ ),
++ -1
++ );
++ }
++ lance_scanner_close(scan);
++ lance_dataset_close(ds);
++ }
++}
++
++#[test]
++fn omitted_metric_is_l2_on_indexed_and_unindexed_fragments() {
++ let (_dir, uri) = custom_fixture(vec![vec![Some(2.), Some(0.)]], 2, true);
++ lance_c::runtime::block_on(async {
++ let mut ds = Dataset::open(uri.to_str().unwrap()).await.unwrap();
++ let old = ds.scan().try_into_batch().await.unwrap();
++ let vectors = FixedSizeListArray::try_new(
++ Arc::new(Field::new("item", DataType::Float32, true)),
++ 2,
++ Arc::new(Float32Array::from(vec![1.0, 0.1])),
++ None,
++ )
++ .unwrap();
++ let lists = ListArray::try_new(
++ Arc::new(Field::new("item", vectors.data_type().clone(), false)),
++ OffsetBuffer::new(vec![0i32, 1].into()),
++ Arc::new(vectors),
++ None,
++ )
++ .unwrap();
++ let batch = RecordBatch::try_new(
++ old.schema(),
++ vec![Arc::new(Int32Array::from(vec![2])), Arc::new(lists)],
++ )
++ .unwrap();
++ ds.append(
++ RecordBatchIterator::new(vec![Ok(batch)], old.schema()),
++ None,
++ )
++ .await
++ .unwrap();
++ });
++ let column = CString::new("vectors").unwrap();
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ for (fragment, id, score) in [(0u64, 1, 1.0f32), (1u64, 2, 0.01f32)] {
++ let scan = lance_scanner_new(ds, ptr::null(), ptr::null());
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ [1.0f32, 0.].as_ptr().cast(),
++ 2,
++ 1,
++ 0,
++ 1
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_prefilter(scan, true), 0);
++ assert_eq!(lance_scanner_set_fragment_ids(scan, &fragment, 1), 0);
++ let rows = collect(scan);
++ assert_eq!(rows[0].0, id);
++ assert!(
++ (rows[0].1 - score).abs() < 1e-6,
++ "fragment={fragment}: {rows:?}"
++ );
++ lance_scanner_close(scan);
++ }
++ lance_dataset_close(ds);
++ }
++}
++
++#[test]
++fn query_values_and_refinement_are_validated_before_execution() {
++ let (_dir, uri) = fixture();
++ let column = CString::new("vectors").unwrap();
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ let scan = lance_scanner_new(ds, ptr::null(), ptr::null());
++ for value in [f32::NAN, f32::INFINITY, f32::NEG_INFINITY] {
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ [value, 0.0f32].as_ptr().cast(),
++ 2,
++ 1,
++ 0,
++ 1
++ ),
++ -1
++ );
++ }
++ let query = [1.0f32; 256];
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ query.as_ptr().cast(),
++ 2,
++ 128,
++ 0,
++ 781
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_metric(scan, 3), 0);
++ let mut invalid_metric_stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(
++ lance_scanner_to_arrow_stream(scan, &mut invalid_metric_stream),
++ -1
++ );
++ assert_eq!(lance_scanner_set_metric(scan, 0), 0);
++ assert_eq!(lance_scanner_set_refine_factor(scan, u32::MAX), 0);
++ let mut stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(lance_scanner_to_arrow_stream(scan, &mut stream), -1);
++ lance_scanner_close(scan);
++ lance_dataset_close(ds);
++ }
++}
++
++#[test]
++fn float16_and_float64_queries_support_all_distance_metrics() {
++ let (dir, source) = fixture();
++ for (code, dtype) in [(1, DataType::Float16), (2, DataType::Float64)] {
++ let path = dir.path().join(format!("typed_{code}.lance"));
++ lance_c::runtime::block_on(async {
++ let source =
Dataset::open(source.to_str().unwrap()).await.unwrap();
++ let batch = source.scan().try_into_batch().await.unwrap();
++ let vector_type = DataType::List(Arc::new(Field::new(
++ "item",
++ DataType::FixedSizeList(Arc::new(Field::new("item", dtype,
true)), 2),
++ false,
++ )));
++ let vectors =
++
arrow::compute::cast(batch.column_by_name("vectors").unwrap(), &vector_type)
++ .unwrap();
++ let schema = Arc::new(Schema::new(vec![
++ Field::new("id", DataType::Int32, false),
++ Field::new("vectors", vector_type, true),
++ ]));
++ let batch =
++ RecordBatch::try_new(schema.clone(),
vec![batch.column(0).clone(), vectors])
++ .unwrap();
++ Dataset::write(
++ RecordBatchIterator::new(vec![Ok(batch)], schema),
++ path.to_str().unwrap(),
++ None,
++ )
++ .await
++ .unwrap();
++ });
++ let uri = CString::new(path.to_str().unwrap()).unwrap();
++ let column = CString::new("vectors").unwrap();
++ let f16_query = [1.0f32, 0., 0., 1.].map(half::f16::from_f32);
++ let f64_query = [1.0f64, 0., 0., 1.];
++ let query = if code == 1 {
++ f16_query.as_ptr().cast()
++ } else {
++ f64_query.as_ptr().cast()
++ };
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ for (metric, expected) in [(0, [0., 6., 8.]), (1, [0., 1., 0.]),
(2, [0., 0., -4.])] {
++ let scanner = lance_scanner_new(ds, ptr::null(), ptr::null());
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scanner,
++ column.as_ptr(),
++ query,
++ 2,
++ 2,
++ code,
++ 10
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_metric(scanner, metric), 0);
++ assert_eq!(lance_scanner_set_use_index(scanner, false), 0);
++ let mut rows = collect(scanner);
++ rows.sort_by_key(|row| row.0);
++ assert_eq!(
++ rows,
++ vec![(1, expected[0]), (2, expected[1]), (3, expected[2])]
++ );
++ lance_scanner_close(scanner);
++ }
++ lance_dataset_close(ds);
++ }
++ }
++}
++
++#[test]
++fn cosine_zero_norm_rows_do_not_abort_exact_or_refined_search() {
++ let rows = vec![
++ vec![Some(0.), Some(0.)],
++ vec![Some(1.), Some(0.)],
++ vec![Some(0.), Some(0.), Some(0.), Some(1.)],
++ ];
++ for indexed in [false, true] {
++ let (_dir, uri) = custom_fixture(rows.clone(), 2, indexed);
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ for query in [[1.0f32, 0., 0., 1.], [0., 0., 0., 1.]] {
++ for count in [1, 2] {
++ let scan = lance_scanner_new(ds, ptr::null(),
ptr::null());
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ c"vectors".as_ptr(),
++ query.as_ptr().cast(),
++ 2,
++ count,
++ 0,
++ 3
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_use_index(scan, indexed), 0);
++ assert_eq!(lance_scanner_set_metric(scan, 1), 0);
++ assert_eq!(lance_scanner_set_refine_factor(scan, 2), 0);
++ let mut actual = collect(scan);
++ actual.sort_by_key(|row| row.0);
++ let expected = if query[0] == 0. {
++ vec![]
++ } else if count == 1 {
++ vec![(2, 0.), (3, 1.)]
++ } else {
++ vec![(2, 1.), (3, 1.)]
++ };
++ assert_eq!(
++ actual, expected,
++ "indexed={indexed}, query={query:?}, count={count}"
++ );
++ lance_scanner_close(scan);
++ }
++ }
++ lance_dataset_close(ds);
++ }
++ }
++}
++
++#[test]
++fn nested_and_quoted_columns_support_exact_and_refined_search() {
++ for column in [
++ "payload.vectors",
++ "payload.`vectors.with.dot`",
++ "payload.inner.`vectors.with.dot`",
++ ] {
++ let rows = vec![
++ vec![Some(1.), Some(0.), Some(0.), Some(1.)],
++ vec![Some(2.), Some(0.)],
++ vec![Some(1.), Some(1.)],
++ ];
++ for indexed in [false, true] {
++ let (_dir, uri) = custom_fixture_at(rows.clone(), 2, indexed,
column);
++ let column = CString::new(column).unwrap();
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ let scan = lance_scanner_new(ds, ptr::null(), ptr::null());
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ [1.0f32, 0., 0., 1.].as_ptr().cast(),
++ 2,
++ 2,
++ 0,
++ 3
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_use_index(scan, indexed), 0);
++ assert_eq!(lance_scanner_set_metric(scan, 1), 0);
++ assert_eq!(lance_scanner_set_refine_factor(scan, 2), 0);
++ let actual = collect(scan);
++ assert_eq!(actual.len(), 3);
++ assert_eq!(
++ actual.iter().map(|row| row.0).collect::<Vec<_>>(),
++ vec![1, 3, 2]
++ );
++ assert_eq!(actual[0].1, 0.);
++ assert!((actual[1].1 - (2. - 2.0f32.sqrt())).abs() < 1e-6);
++ assert_eq!(actual[2].1, 1.);
++ lance_scanner_close(scan);
++ lance_dataset_close(ds);
++ }
++ }
++ }
++}
++
++#[test]
++fn
nested_projections_preserve_schema_and_values_in_exact_indexed_and_hybrid_search()
{
++ for column in ["payload.vectors", "payload.`vectors.with.dot`"] {
++ for indexed in [false, true] {
++ for appended in [false, true] {
++ let (_dir, uri) = custom_fixture_at(
++ vec![vec![Some(1.), Some(0.)], vec![Some(0.), Some(1.)]],
++ 2,
++ indexed,
++ column,
++ );
++ if appended {
++ lance_c::runtime::block_on(async {
++ let mut ds =
Dataset::open(uri.to_str().unwrap()).await.unwrap();
++ let batch = ds.scan().try_into_batch().await.unwrap();
++ ds.append(
++ RecordBatchIterator::new(vec![Ok(batch.clone())],
batch.schema()),
++ None,
++ )
++ .await
++ .unwrap();
++ });
++ }
++ for projection in [
++ vec![],
++ vec!["id"],
++ vec!["payload.label"],
++ vec!["id", column],
++ ] {
++ let expected = lance_c::runtime::block_on(async {
++ let ds =
Dataset::open(uri.to_str().unwrap()).await.unwrap();
++ let mut scanner = ds.scan();
++ scanner.scan_in_order(true);
++ if !projection.is_empty() {
++ scanner.project(&projection).unwrap();
++ }
++ scanner.try_into_batch().await.unwrap()
++ });
++ // The appended fragment repeats the same two values;
distance order groups
++ // both exact matches before the orthogonal rows,
regardless of tie order.
++ let indices = arrow_array::UInt32Array::from(if appended {
++ vec![0, 2, 1, 3]
++ } else {
++ vec![0, 1]
++ });
++ let expected = RecordBatch::try_new(
++ expected.schema(),
++ expected
++ .columns()
++ .iter()
++ .map(|array| {
++ arrow::compute::take(array.as_ref(),
&indices, None).unwrap()
++ })
++ .collect(),
++ )
++ .unwrap();
++ unsafe {
++ let names: Vec<_> = projection
++ .iter()
++ .map(|name| CString::new(*name).unwrap())
++ .collect();
++ let mut columns: Vec<_> = names.iter().map(|name|
name.as_ptr()).collect();
++ columns.push(ptr::null());
++ let ds = lance_dataset_open(uri.as_ptr(),
ptr::null(), 0);
++ let scan = lance_scanner_new(
++ ds,
++ if projection.is_empty() {
++ ptr::null()
++ } else {
++ columns.as_ptr()
++ },
++ ptr::null(),
++ );
++ let column = CString::new(column).unwrap();
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ column.as_ptr(),
++ [1.0f32, 0.].as_ptr().cast(),
++ 2,
++ 1,
++ 0,
++ 10
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_use_index(scan,
indexed), 0);
++ assert_eq!(lance_scanner_set_metric(scan, 1), 0);
++ assert_eq!(lance_scanner_set_batch_size(scan, 1), 0);
++ let mut stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(lance_scanner_to_arrow_stream(scan, &mut
stream), 0);
++ let batches = ArrowArrayStreamReader::from_raw(&mut
stream)
++ .unwrap()
++ .collect::<Result<Vec<_>, _>>()
++ .unwrap();
++ lance_scanner_close(scan);
++ lance_dataset_close(ds);
++ let actual =
++
arrow::compute::concat_batches(&batches[0].schema(), &batches).unwrap();
++ let distances = actual
++ .column_by_name("_distance")
++ .unwrap()
++ .as_any()
++ .downcast_ref::<Float32Array>()
++ .unwrap();
++ assert_eq!(
++ distances.values().as_ref(),
++ if appended {
++ &[0., 0., 1., 1.][..]
++ } else {
++ &[0., 1.][..]
++ }
++ );
++ let positions: Vec<_> = actual
++ .schema()
++ .fields()
++ .iter()
++ .enumerate()
++ .filter_map(|(i, field)| (field.name() !=
"_distance").then_some(i))
++ .collect();
++ assert_eq!(
++ actual.project(&positions).unwrap(),
++ expected,
++ "column={column:?} indexed={indexed}
appended={appended} projection={projection:?}"
++ );
++ }
++ }
++ }
++ }
++ }
++}
++
++#[test]
++fn strict_batches_apply_after_multivector_offset_and_limit() {
++ let (_dir, uri) = custom_fixture(
++ (1..=6).map(|i| vec![Some(i as f32), Some(0.)]).collect(),
++ 2,
++ false,
++ );
++ unsafe {
++ let ds = lance_dataset_open(uri.as_ptr(), ptr::null(), 0);
++ for batch_size in [Some(2), None] {
++ for (offset, limit) in [(1, 4), (1, 3), (5, 4), (6, 4), (0, 6)] {
++ let scan = lance_scanner_new(ds, ptr::null(), ptr::null());
++ assert_eq!(
++ lance_scanner_nearest_multivector(
++ scan,
++ c"vectors".as_ptr(),
++ [1.0f32, 0.].as_ptr().cast(),
++ 2,
++ 1,
++ 0,
++ 6
++ ),
++ 0
++ );
++ assert_eq!(lance_scanner_set_use_index(scan, false), 0);
++ if let Some(size) = batch_size {
++ assert_eq!(lance_scanner_set_batch_size(scan, size), 0);
++ }
++ assert_eq!(lance_scanner_set_strict_batch_size(scan, true),
0);
++ assert_eq!(lance_scanner_set_offset(scan, offset), 0);
++ assert_eq!(lance_scanner_set_limit(scan, limit), 0);
++ let mut stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(lance_scanner_to_arrow_stream(scan, &mut stream),
0);
++ let batches = ArrowArrayStreamReader::from_raw(&mut stream)
++ .unwrap()
++ .collect::<Result<Vec<_>, _>>()
++ .unwrap();
++ lance_scanner_close(scan);
++ let sizes: Vec<_> =
batches.iter().map(RecordBatch::num_rows).collect();
++ let ids: Vec<_> = batches
++ .iter()
++ .flat_map(|b| {
++ b.column_by_name("id")
++ .unwrap()
++ .as_any()
++ .downcast_ref::<Int32Array>()
++ .unwrap()
++ .values()
++ .to_vec()
++ })
++ .collect();
++ let expected_ids: Vec<_> =
++ (1..=6).skip(offset as usize).take(limit as
usize).collect();
++ let expected_sizes: Vec<_> = expected_ids
++ .chunks(batch_size.unwrap_or(8192) as usize)
++ .map(<[i32]>::len)
++ .collect();
++ assert_eq!(ids, expected_ids);
++ assert_eq!(
++ sizes, expected_sizes,
++ "batch_size={batch_size:?} offset={offset} limit={limit}"
++ );
++ }
++ }
++ lance_dataset_close(ds);
++ }
++}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
