github-actions[bot] commented on code in PR #68028:
URL: https://github.com/apache/doris/pull/68028#discussion_r4037625351
##########
gensrc/thrift/PlanNodes.thrift:
##########
@@ -467,6 +467,8 @@ struct TSearchVector {
1: optional TVectorElementType element_type
2: optional i32 dimension
3: optional binary values
+ // Present only for one multi-vector query; values contains a row-major
matrix.
+ 4: optional i32 num_vectors
Review Comment:
[P2] Fence this request from old smooth-upgrade BEs. Keeping schema version
1 preserves decoding, but it does not make the matrix payload executable: an
old BE ignores this optional field, validates values as one dimension-sized
vector, and rejects every query with two or more subvectors. LanceScanNode
currently checks old BEs only for additional projected types, so success
depends on split placement during an upgrade. Please add a multi-vector
capability/old-BE guard there and a mixed-version test.
##########
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))
Review Comment:
[P2] Preserve nullable struct ancestors while resolving this leaf. Arrow
keeps a StructArray's validity separately from its children, and column_by_name
returns the raw child, so a row with payload = NULL can still expose populated
payload.vectors storage here. exact_batch then checks only the leaf ListArray
and can rank that logically NULL row; indexed refinement reuses the same
scorer. Please AND every traversed ancestor mask into the leaf validity and
cover exact/refined searches with a null parent whose child buffer contains a
near vector.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
