github-actions[bot] commented on code in PR #68028:
URL: https://github.com/apache/doris/pull/68028#discussion_r4017606877
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -1235,6 +1265,26 @@ Status LanceTableReader::_fill_block_from_record_batch(
->read_column_from_arrow(*columns[output_idx],
normalized_column.get(), 0, row_count,
_ctz));
+ if (_search_kind == SearchKind::VECTOR && field->name() ==
LANCE_DISTANCE_COLUMN) {
+ const auto& query =
_scan_params->lance_scan_params.external_search_request
+
.search_query.vector_search.query_vector;
+ if (query.__isset.num_vectors) {
+ // Lance's exact score is 1 - sum(1 - distance). Expose
sum(min distance)
+ // so one logical row has the same scoring contract for
every query size.
+ IColumn* output = columns[output_idx].get();
+ if (auto* nullable =
check_and_get_column<ColumnNullable>(*output)) {
+ output = &nullable->get_nested_column();
+ }
+ auto* distances =
check_and_get_column<ColumnFloat32>(*output);
+ if (distances == nullptr) {
+ return Status::InternalError("Lance multi-vector
distance must be FLOAT");
+ }
+ auto& data = distances->get_data();
+ for (size_t i = data.size() - row_count; i < data.size();
++i) {
+ data[i] += static_cast<float>(query.num_vectors - 1);
Review Comment:
[P1] Compute this normalization before Lance selects TopK. The pinned scorer
first forms each term as f32 `1 - distance`; distinct small L2 distances (for
example about `1e-8` and `2.25e-8`) both round to the same value before Lance
sorts and fetches `k`. Adding the constant while reading the already-truncated
result cannot recover either the score bits or a nearer row discarded as a tie,
including with `use_index=false`. Please patch/repin the scorer to accumulate
`sum(min(distance))` directly and cover a `top_k=1` small-distance case.
##########
thirdparty/patches/lance-c-0.1.9-multivector.patch:
##########
@@ -0,0 +1,508 @@
+diff --git a/src/scanner.rs b/src/scanner.rs
+--- a/src/scanner.rs
++++ b/src/scanner.rs
+@@ -226,8 +226,18 @@
+ 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);
+@@ -300,6 +310,12 @@
+ Ok(PreparedScanner {
+ scanner,
+ distributed_fts,
++ 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(),
+ })
+ }
+ }
+@@ -314,10 +330,45 @@
+ struct PreparedScanner {
+ scanner: lance::dataset::scanner::Scanner,
+ distributed_fts: Option<PreparedFtsExecution>,
++ multi_vector_window: Option<(usize, Option<usize>)>,
++ batch_size: Option<usize>,
++ scan_statistics_callback: Option<ExecutionStatsCallback>,
+ }
+
+ impl PreparedScanner {
+ async fn try_into_stream(self) -> Result<DatasetRecordBatchStream> {
++ if let Some((offset, limit)) = self.multi_vector_window {
++ use datafusion::physical_expr::{PhysicalSortExpr, expressions};
++ use datafusion::physical_plan::{
++ coalesce_partitions::CoalescePartitionsExec,
limit::GlobalLimitExec,
++ sorts::sort::SortExec,
++ };
++ let plan = self.scanner.create_plan().await?;
++ let sort = PhysicalSortExpr {
++ expr: expressions::col("_distance", plan.schema().as_ref())?,
++ options: arrow::compute::SortOptions {
++ descending: false,
++ nulls_first: false,
++ },
++ };
++ // Fragment-scoped Lance plans can reorder candidate batches
during payload take.
++ // Apply the result window only after restoring distance order
across all partitions.
++ // The nearest plan already bounds the candidate rows by k.
++ let sorted = Arc::new(SortExec::new(
++ [sort].into(),
++ Arc::new(CoalescePartitionsExec::new(plan)),
++ ));
++ let plan = Arc::new(GlobalLimitExec::new(sorted, offset, limit));
++ let stream = lance_datafusion::exec::execute_plan(
++ plan,
++ lance_datafusion::exec::LanceExecutionOptions {
++ batch_size: self.batch_size,
Review Comment:
[P1] Avoid exposing these output batch boundaries to
`MultivectorScoringExec`. DataFusion 54's TopK emits its fetched rows in
`session_config.batch_size()` chunks, while the pinned scorer treats every
received batch as another complete query stream and advances `missed_sim_sum`
for it. Once a subquery returns more rows than the batch size (for example
`batch_size=1`, `top_k=1`), scores and candidates depend on fragmentation and
refinement cannot recover a row dropped by the scorer's TopK. Aggregate each
child stream across all of its batches before multi-vector scoring, and add a
truncating small-batch test.
##########
docs/lance-multivector-search.md:
##########
@@ -0,0 +1,105 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Lance multi-vector search
+
+A Lance `List<FixedSizeList<T, D>>` column stores a variable number of
D-dimensional
+subvectors in each table row. Doris exposes it as `ARRAY<ARRAY<FLOAT>>` for
Float16
+and Float32, or `ARRAY<ARRAY<DOUBLE>>` for Float64.
+
+`vector_search` accepts a non-empty JSON matrix for such a column. Every inner
+array must contain exactly D finite numeric values representable in the
column's
+element type. A matrix with one inner array is still one multi-vector query.
+This is not a batch of independent searches or a search across several columns.
+Ordinary `FixedSizeList` columns continue to accept a one-dimensional JSON
array.
+
+```sql
+SELECT id, _distance
+FROM vector_search(
+ "table" = "lance_catalog.default.documents",
+ "column" = "embeddings",
+ "query_vector" = "[[1,0],[0,1]]",
+ "top_k" = "10",
+ "metric" = "cosine",
+ "filter" = "id > 100"
+)
+ORDER BY _distance, id;
+```
+
+## Row-level score and filtering
+
+For query subvectors Q and a row's subvectors V, Doris reports:
+
+`_distance = sum(q in Q, min(v in V, distance(q, v)))`
+
+Smaller is better. The base distance is squared Euclidean distance for `l2`,
+`1 - cosine_similarity` for `cosine`, or `1 - dot_product` for `dot`. Each
query
+subvector contributes once; the same stored subvector may match several query
+subvectors. The score is a sum, not an average, so duplicating a query
subvector
+changes the score. `_distance` uses Doris FLOAT, including for Float64 inputs.
+
+`top_k` and `offset` count table rows, not subvectors. The TVF `filter` is
applied
+before candidate search; an outer SQL WHERE filters the search's results.
+Distributed fragments return row candidates for the global TopK. Ties at the
+TopK boundary have no guaranteed row order.
+
+`use_index=false` computes exact scores over the selected rows. With indexes,
+candidate selection is approximate. Doris always refines multi-vector
candidates
+against their original values so indexed and unindexed rows use the same score.
+`refine_factor` can increase the candidate pool beyond this default refinement.
+It does not turn ANN candidate selection into an exhaustive search.
+
+The pinned Lance version supports multi-vector indexes with the cosine metric.
+Use `use_index=false` for L2 or dot searches. `nprobes`, `ef`, and
`refine_factor`
Review Comment:
[P1] Resolve one metric for every split when `metric` is omitted. FE treats
the unset value as L2 while deciding not to create a cosine index-segment
split, but it leaves `use_index=true` and BE leaves Lance's metric unset. On
the partially indexed layout documented here, a covered fragment can therefore
adopt the cosine index while an appended uncovered fragment defaults to L2, and
the global TopN compares incomparable exact scores. Materialize the chosen
metric in every scanner request (or disable index use consistently) and cover
the omitted-metric mixed-coverage case.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -537,13 +538,27 @@ Status
LanceTableReader::_validate_external_search_request() const {
return Status::NotSupported("unsupported Lance query vector
element type: {}",
static_cast<int>(query_vector.element_type));
}
+ const bool multi_vector = query_vector.__isset.num_vectors;
+ if (multi_vector != (request.schema_version == 2) ||
+ (multi_vector && query_vector.num_vectors <= 0)) {
Review Comment:
[P1] Bound `num_vectors` (or a checked `num_vectors * candidate_k` work
budget) here, and enforce the same limit in FE and the C API. The pinned
planner creates one ANN-plus-sort child per query subvector and lets each child
produce up to `10 * k` candidates. A valid dimension-one Float32 matrix with
100,000 subvectors is only about 400 KiB on this boundary, but it constructs
100,000 plan branches and can feed roughly `1,000,000 * k` candidates before
the final limit, allowing one query to exhaust BE CPU/memory.
##########
thirdparty/patches/lance-c-0.1.9-multivector.patch:
##########
@@ -0,0 +1,508 @@
+diff --git a/src/scanner.rs b/src/scanner.rs
+--- a/src/scanner.rs
++++ b/src/scanner.rs
+@@ -226,8 +226,18 @@
+ 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);
+@@ -300,6 +310,12 @@
+ Ok(PreparedScanner {
+ scanner,
+ distributed_fts,
++ 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(),
+ })
+ }
+ }
+@@ -314,10 +330,45 @@
+ struct PreparedScanner {
+ scanner: lance::dataset::scanner::Scanner,
+ distributed_fts: Option<PreparedFtsExecution>,
++ multi_vector_window: Option<(usize, Option<usize>)>,
++ batch_size: Option<usize>,
++ scan_statistics_callback: Option<ExecutionStatsCallback>,
+ }
+
+ impl PreparedScanner {
+ async fn try_into_stream(self) -> Result<DatasetRecordBatchStream> {
++ if let Some((offset, limit)) = self.multi_vector_window {
++ use datafusion::physical_expr::{PhysicalSortExpr, expressions};
++ use datafusion::physical_plan::{
++ coalesce_partitions::CoalescePartitionsExec,
limit::GlobalLimitExec,
++ sorts::sort::SortExec,
++ };
++ let plan = self.scanner.create_plan().await?;
++ let sort = PhysicalSortExpr {
++ expr: expressions::col("_distance", plan.schema().as_ref())?,
++ options: arrow::compute::SortOptions {
++ descending: false,
++ nulls_first: false,
++ },
++ };
++ // Fragment-scoped Lance plans can reorder candidate batches
during payload take.
++ // Apply the result window only after restoring distance order
across all partitions.
++ // The nearest plan already bounds the candidate rows by k.
++ let sorted = Arc::new(SortExec::new(
++ [sort].into(),
++ Arc::new(CoalescePartitionsExec::new(plan)),
++ ));
++ let plan = Arc::new(GlobalLimitExec::new(sorted, 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;
+ };
+@@ -1978,6 +2029,21 @@
+ }
+ 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,
+@@ -2016,9 +2082,104 @@
+ }
+ };
+
++ 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",
++ ));
++ }
++ 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)
Review Comment:
[P2] Validate actual primitive values in the scoring path before accepting
this schema. The pinned exact scorer reads
`multivector.values().as_primitive().values()` without consulting the validity
bitmap or checking finiteness, so an externally written row such as `[[NULL,
0]]` can consume the null slot's backing zero and rank at a plausible distance
instead of failing; NaN/Infinity also reach scoring. The documentation
precondition does not prevent silent wrong rows. Keep tolerating Lance's
reconstructed nullable flag, but reject actual null/non-finite elements
explicitly and test those external-writer cases.
--
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]
