This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 017d1a48 perf(table): create pk-vector exact-fallback readers on
demand (#535)
017d1a48 is described below
commit 017d1a4819a9935ebf2a3927c0458bc788e772ec
Author: Junrui Lee <[email protected]>
AuthorDate: Sun Jul 19 09:00:35 2026 +0800
perf(table): create pk-vector exact-fallback readers on demand (#535)
---
crates/paimon/src/table/pk_vector_orchestrator.rs | 215 ++++++++++------
crates/paimon/src/table/vector_search_builder.rs | 117 +++------
crates/paimon/src/vindex/pkvector/bucket.rs | 294 +++++++++++++---------
3 files changed, 341 insertions(+), 285 deletions(-)
diff --git a/crates/paimon/src/table/pk_vector_orchestrator.rs
b/crates/paimon/src/table/pk_vector_orchestrator.rs
index 6004a98c..e8da709e 100644
--- a/crates/paimon/src/table/pk_vector_orchestrator.rs
+++ b/crates/paimon/src/table/pk_vector_orchestrator.rs
@@ -34,9 +34,10 @@ use crate::table::data_file_reader::DataFileReader;
use crate::table::pk_vector_indexed_split_read::PkVectorIndexedSplit;
use crate::table::source::{DataSplit, DataSplitBuilder, RowRange};
use crate::vindex::pkvector::ann::PkVectorAnnSearcher;
-use crate::vindex::pkvector::bucket::{bucket_search, BucketActiveFile,
BucketAnnSegment};
+use crate::vindex::pkvector::bucket::{
+ bucket_search, BucketActiveFile, BucketAnnSegment, ExactReaderFuture,
+};
use crate::vindex::pkvector::metric::{java_float_compare, VectorSearchMetric};
-use crate::vindex::pkvector::reader::PkVectorReader;
use crate::vindex::pkvector::result::PkVectorSearchResult;
fn data_invalid(message: impl Into<String>) -> crate::Error {
@@ -46,6 +47,33 @@ fn data_invalid(message: impl Into<String>) -> crate::Error {
}
}
+/// Coerce a closure into the higher-ranked per-file exact-reader factory
shape so
+/// its returned future borrows for exactly the file argument's lifetime.
Closure
+/// return types cannot express this borrow through inference alone, so the
bound
+/// is supplied here.
+fn as_bucket_factory<F>(f: F) -> F
+where
+ F: for<'f> FnMut(&'f BucketActiveFile) -> ExactReaderFuture<'f> + Send,
+{
+ f
+}
+
+/// Coerce a closure into the split-scoped exact-reader factory shape expected
by
+/// [`PkVectorOrchestrator::search_candidates`], binding the returned future's
+/// borrow to the file argument's lifetime. Callers building a factory closure
use
+/// this so the higher-ranked bound is supplied where inference cannot.
+pub(crate) fn as_split_exact_reader_factory<F>(f: F) -> F
+where
+ F: for<'s, 'f> FnMut(
+ usize,
+ &'s PkVectorSearchSplit,
+ &'f BucketActiveFile,
+ ) -> ExactReaderFuture<'f>
+ + Send,
+{
+ f
+}
+
/// Validate a hit's physical row position against its data file, mirroring the
/// bounds Java `PrimaryKeyVectorResult.splits()` enforces per candidate: the
/// position must be non-negative, within the file's row count, and fit in an
@@ -293,11 +321,11 @@ impl PkVectorOrchestrator {
metric: VectorSearchMetric,
limit: usize,
ann_searcher: Option<&dyn PkVectorAnnSearcher>,
- exact_reader_factory: &mut (dyn FnMut(
+ exact_reader_factory: &mut (dyn for<'s, 'f> FnMut(
usize,
- &PkVectorSearchSplit,
- &BucketActiveFile,
- ) -> crate::Result<Box<dyn PkVectorReader>>
+ &'s PkVectorSearchSplit,
+ &'f BucketActiveFile,
+ ) -> ExactReaderFuture<'f>
+ Send),
search_options: &HashMap<String, String>,
skip_exact_fallback: bool,
@@ -323,8 +351,11 @@ impl PkVectorOrchestrator {
for (split_index, split) in splits.iter().enumerate() {
let dvs = build_bucket_dv_map(&self.reader, split).await?;
// Wrap the split-scoped factory into bucket_search's per-file
signature.
- let mut bucket_factory =
- |file: &BucketActiveFile| exact_reader_factory(split_index,
split, file);
+ // The coercion helper ties the produced future's borrow to the
file
+ // argument, which closure inference cannot express on its own.
+ let mut bucket_factory = as_bucket_factory(|file:
&BucketActiveFile| {
+ exact_reader_factory(split_index, split, file)
+ });
let residual_ranges = residual_by_split.map(|per_split|
&per_split[split_index]);
let results = bucket_search(
ann_searcher,
@@ -338,7 +369,8 @@ impl PkVectorOrchestrator {
search_options,
skip_exact_fallback,
residual_ranges,
- )?;
+ )
+ .await?;
for PkVectorSearchResult {
data_file_name,
row_position,
@@ -634,6 +666,7 @@ mod e2e_tests {
use crate::table::schema_manager::SchemaManager;
use crate::table::source::DeletionFile;
use crate::vindex::pkvector::reader::test_support::ArrayReader;
+ use crate::vindex::pkvector::reader::PkVectorReader;
use arrow_array::{Array, Float32Array, Int32Array, Int64Array,
RecordBatch};
use bytes::Bytes;
use futures::TryStreamExt;
@@ -800,6 +833,30 @@ mod e2e_tests {
}
}
+ /// Coerce a closure into the higher-ranked split-scoped exact-reader
factory
+ /// shape so its returned future borrows for exactly the file argument's
+ /// lifetime. Closure return types cannot express this borrow through
inference
+ /// alone, so the bound is supplied here.
+ fn as_split_factory<F>(f: F) -> F
+ where
+ F: for<'s, 'f> FnMut(
+ usize,
+ &'s PkVectorSearchSplit,
+ &'f BucketActiveFile,
+ ) -> ExactReaderFuture<'f>
+ + Send,
+ {
+ as_split_exact_reader_factory(f)
+ }
+
+ /// Same coercion for the per-file factory used by
`materialize_via_splits`.
+ fn as_file_factory<F>(f: F) -> F
+ where
+ F: for<'f> FnMut(&'f BucketActiveFile) -> ExactReaderFuture<'f> + Send,
+ {
+ f
+ }
+
fn column_by_name<'a>(batch: &'a RecordBatch, name: &str) -> Option<&'a
Arc<dyn Array>> {
batch
.schema()
@@ -886,14 +943,14 @@ mod e2e_tests {
metric: VectorSearchMetric,
limit: usize,
ann: Option<&dyn PkVectorAnnSearcher>,
- factory: &mut (dyn FnMut(&BucketActiveFile) -> crate::Result<Box<dyn
PkVectorReader>>
- + Send),
+ factory: &mut (dyn for<'f> FnMut(&'f BucketActiveFile) ->
ExactReaderFuture<'f> + Send),
opts: &HashMap<String, String>,
) -> crate::Result<Vec<RecordBatch>> {
let orch = PkVectorOrchestrator::new(reader.clone());
// Wrap the per-file factory into the split-scoped shape
search_candidates
// expects; the split index/split are unused here.
- let mut wrapped = |_: usize, _: &PkVectorSearchSplit, f:
&BucketActiveFile| factory(f);
+ let mut wrapped =
+ as_split_factory(|_: usize, _: &PkVectorSearchSplit, f:
&BucketActiveFile| factory(f));
let survivors = orch
.search_candidates(
splits,
@@ -924,12 +981,11 @@ mod e2e_tests {
let file_io = FileIOBuilder::new("memory").build().unwrap();
let reader = make_reader(file_io, "memory:/pkvo_zero");
let splits: Vec<PkVectorSearchSplit> = Vec::new();
- let mut factory = |_: usize,
- _: &PkVectorSearchSplit,
- _: &BucketActiveFile|
- -> crate::Result<Box<dyn PkVectorReader>> {
- unreachable!("no bucket search on eager-rejected input")
- };
+ let mut factory = as_split_factory(
+ |_: usize, _: &PkVectorSearchSplit, _: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!("no bucket search on
eager-rejected input") })
+ },
+ );
let opts = HashMap::new();
let err = PkVectorOrchestrator::new(reader)
.search_candidates(
@@ -954,12 +1010,11 @@ mod e2e_tests {
let file_io = FileIOBuilder::new("memory").build().unwrap();
let reader = make_reader(file_io, "memory:/pkvo_empty_query");
let splits: Vec<PkVectorSearchSplit> = Vec::new();
- let mut factory = |_: usize,
- _: &PkVectorSearchSplit,
- _: &BucketActiveFile|
- -> crate::Result<Box<dyn PkVectorReader>> {
- unreachable!("no bucket search on eager-rejected input")
- };
+ let mut factory = as_split_factory(
+ |_: usize, _: &PkVectorSearchSplit, _: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!("no bucket search on
eager-rejected input") })
+ },
+ );
let opts = HashMap::new();
let err = PkVectorOrchestrator::new(reader)
.search_candidates(
@@ -1013,13 +1068,15 @@ mod e2e_tests {
}],
};
// Exact fallback scans only "exact.mosaic": pos0 {1,0} d=1.0, pos1
{2,0} d=4.0.
- let mut factory = |f: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
+ let mut factory = as_file_factory(|f: &BucketActiveFile| ->
ExactReaderFuture<'_> {
let vectors = match f.file_name.as_str() {
"exact.mosaic" => vec![Some(vec![1.0, 0.0]), Some(vec![2.0,
0.0])],
other => panic!("unexpected exact scan of covered file
{other}"),
};
- Ok(Box::new(ArrayReader::new(2, vectors)))
- };
+ Box::pin(async move {
+ Ok(Box::new(ArrayReader::new(2, vectors)) as Box<dyn
PkVectorReader>)
+ })
+ });
let opts = HashMap::new();
let batches = materialize_via_splits(
make_reader(file_io, table_path),
@@ -1094,7 +1151,7 @@ mod e2e_tests {
// b0: x = 1,4,6 -> d = 1,16,36. b1: x = 2,3,5 -> d = 4,9,25.
// Global best 3: d1 (b0 pos0 id10), d4 (b1 pos0 id20), d9 (b1 pos1
id21).
- let mut factory = |f: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
+ let mut factory = as_file_factory(|f: &BucketActiveFile| ->
ExactReaderFuture<'_> {
let vectors = match f.file_name.as_str() {
"b0.mosaic" => vec![
Some(vec![1.0, 0.0]),
@@ -1108,8 +1165,10 @@ mod e2e_tests {
],
other => panic!("unexpected file {other}"),
};
- Ok(Box::new(ArrayReader::new(2, vectors)))
- };
+ Box::pin(async move {
+ Ok(Box::new(ArrayReader::new(2, vectors)) as Box<dyn
PkVectorReader>)
+ })
+ });
let opts = HashMap::new();
let batches = materialize_via_splits(
make_reader(file_io, table_path),
@@ -1163,7 +1222,7 @@ mod e2e_tests {
};
// pos0 {1,0} d=1, pos1 {2,0} d=4 (DELETED), pos2 {3,0} d=9, pos3
{0,0} d=0.
- let mut factory = |f: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
+ let mut factory = as_file_factory(|f: &BucketActiveFile| ->
ExactReaderFuture<'_> {
let vectors = match f.file_name.as_str() {
"d.mosaic" => vec![
Some(vec![1.0, 0.0]),
@@ -1173,8 +1232,10 @@ mod e2e_tests {
],
other => panic!("unexpected file {other}"),
};
- Ok(Box::new(ArrayReader::new(2, vectors)))
- };
+ Box::pin(async move {
+ Ok(Box::new(ArrayReader::new(2, vectors)) as Box<dyn
PkVectorReader>)
+ })
+ });
let opts = HashMap::new();
let batches = materialize_via_splits(
make_reader(file_io, table_path),
@@ -1224,7 +1285,7 @@ mod e2e_tests {
};
// pos0 {3,0} d=9, pos1 {1,0} d=1, pos2 {2,0} d=4. Best-first =
[1,2,0].
- let mut factory = |f: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
+ let mut factory = as_file_factory(|f: &BucketActiveFile| ->
ExactReaderFuture<'_> {
let vectors = match f.file_name.as_str() {
"o.mosaic" => vec![
Some(vec![3.0, 0.0]),
@@ -1233,8 +1294,10 @@ mod e2e_tests {
],
other => panic!("unexpected file {other}"),
};
- Ok(Box::new(ArrayReader::new(2, vectors)))
- };
+ Box::pin(async move {
+ Ok(Box::new(ArrayReader::new(2, vectors)) as Box<dyn
PkVectorReader>)
+ })
+ });
let opts = HashMap::new();
let batches = materialize_via_splits(
make_reader(file_io, table_path),
@@ -1283,20 +1346,21 @@ mod e2e_tests {
active_files: vec![active("c.mosaic", 3)],
};
// pos0 {3,0} d=9, pos1 {1,0} d=1, pos2 {2,0} d=4.
- let mut factory = |_: usize,
- _: &PkVectorSearchSplit,
- f: &BucketActiveFile|
- -> crate::Result<Box<dyn PkVectorReader>> {
- assert_eq!(f.file_name, "c.mosaic");
- Ok(Box::new(ArrayReader::new(
- 2,
- vec![
- Some(vec![3.0, 0.0]),
- Some(vec![1.0, 0.0]),
- Some(vec![2.0, 0.0]),
- ],
- )))
- };
+ let mut factory = as_split_factory(
+ |_: usize, _: &PkVectorSearchSplit, f: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ assert_eq!(f.file_name, "c.mosaic");
+ Box::pin(async {
+ Ok(Box::new(ArrayReader::new(
+ 2,
+ vec![
+ Some(vec![3.0, 0.0]),
+ Some(vec![1.0, 0.0]),
+ Some(vec![2.0, 0.0]),
+ ],
+ )) as Box<dyn PkVectorReader>)
+ })
+ },
+ );
let opts = HashMap::new();
let cands = PkVectorOrchestrator::new(make_reader(file_io, table_path))
.search_candidates(
@@ -1348,20 +1412,21 @@ mod e2e_tests {
active_files: vec![active("r.mosaic", 3)],
};
// pos0 {3,0} d=9, pos1 {1,0} d=1, pos2 {2,0} d=4.
- let mut factory = |_: usize,
- _: &PkVectorSearchSplit,
- f: &BucketActiveFile|
- -> crate::Result<Box<dyn PkVectorReader>> {
- assert_eq!(f.file_name, "r.mosaic");
- Ok(Box::new(ArrayReader::new(
- 2,
- vec![
- Some(vec![3.0, 0.0]),
- Some(vec![1.0, 0.0]),
- Some(vec![2.0, 0.0]),
- ],
- )))
- };
+ let mut factory = as_split_factory(
+ |_: usize, _: &PkVectorSearchSplit, f: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ assert_eq!(f.file_name, "r.mosaic");
+ Box::pin(async {
+ Ok(Box::new(ArrayReader::new(
+ 2,
+ vec![
+ Some(vec![3.0, 0.0]),
+ Some(vec![1.0, 0.0]),
+ Some(vec![2.0, 0.0]),
+ ],
+ )) as Box<dyn PkVectorReader>)
+ })
+ },
+ );
// Allow only positions 0 and 2 for "r.mosaic"; pos1 (the best hit) is
// excluded by the residual.
let mut allowed = RoaringTreemap::new();
@@ -1415,12 +1480,11 @@ mod e2e_tests {
ann_segments: Vec::new(),
active_files: vec![active("m.mosaic", 2)],
};
- let mut factory = |_: usize,
- _: &PkVectorSearchSplit,
- _: &BucketActiveFile|
- -> crate::Result<Box<dyn PkVectorReader>> {
- unreachable!("length guard must fire before any bucket search")
- };
+ let mut factory = as_split_factory(
+ |_: usize, _: &PkVectorSearchSplit, _: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!("length guard must fire before
any bucket search") })
+ },
+ );
// Two residual maps for a single split.
let residual_by_split: Vec<HashMap<String, RoaringTreemap>> =
vec![HashMap::new(), HashMap::new()];
@@ -1465,12 +1529,11 @@ mod e2e_tests {
ann_segments: Vec::new(),
active_files: vec![active("f.mosaic", 2)],
};
- let mut factory = |_: usize,
- _: &PkVectorSearchSplit,
- _: &BucketActiveFile|
- -> crate::Result<Box<dyn PkVectorReader>> {
- unreachable!("fast mode must not read exact")
- };
+ let mut factory = as_split_factory(
+ |_: usize, _: &PkVectorSearchSplit, _: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!("fast mode must not read exact")
})
+ },
+ );
let opts = HashMap::new();
let cands = PkVectorOrchestrator::new(make_reader(file_io, table_path))
.search_candidates(
diff --git a/crates/paimon/src/table/vector_search_builder.rs
b/crates/paimon/src/table/vector_search_builder.rs
index facc1f99..2ce2f8df 100644
--- a/crates/paimon/src/table/vector_search_builder.rs
+++ b/crates/paimon/src/table/vector_search_builder.rs
@@ -32,8 +32,8 @@ use crate::table::global_index_scanner::{
use crate::table::pk_vector_data_file_reader::DataFilePkVectorReaderFactory;
use crate::table::pk_vector_indexed_split_read::PkVectorIndexedSplitRead;
use crate::table::pk_vector_orchestrator::{
- build_indexed_splits, validate_row_position, PkVectorCandidate,
PkVectorOrchestrator,
- PkVectorSearchSplit,
+ as_split_exact_reader_factory, build_indexed_splits,
validate_row_position, PkVectorCandidate,
+ PkVectorOrchestrator, PkVectorSearchSplit,
};
use crate::table::pk_vector_position_read::{
PKEY_VECTOR_POSITION_COLUMN, PKEY_VECTOR_SCORE_COLUMN,
@@ -47,9 +47,8 @@ use crate::table::{
use crate::vector_search::{GlobalIndexIOMeta, SearchResult, VectorSearch};
use crate::vindex::is_vindex_index_type;
use crate::vindex::pkvector::ann::VindexAnnSearcher;
-use crate::vindex::pkvector::bucket::{covered_source_files, BucketActiveFile,
BucketAnnSegment};
+use crate::vindex::pkvector::bucket::{BucketActiveFile, BucketAnnSegment,
ExactReaderFuture};
use crate::vindex::pkvector::metric::VectorSearchMetric;
-use crate::vindex::pkvector::reader::PkVectorReader;
use crate::vindex::reader::VindexVectorGlobalIndexReader;
use arrow_array::{Array, FixedSizeListArray, Float32Array, Int64Array,
ListArray, RecordBatch};
use arrow_select::interleave::interleave_record_batch;
@@ -425,8 +424,8 @@ impl<'a> VectorSearchBuilder<'a> {
// when a filter is set; otherwise `None` leaves the search
unfiltered. The
// residual reader projects the predicate columns plus `_ROW_ID` (used
to
// recover file-local physical positions) and carries no pushdown,
matching
- // `residual_positions_by_file`. Computed before the exact-reader
preload so
- // the preload can skip files the residual allow-list leaves empty.
+ // `residual_positions_by_file`. A file the allow-list leaves empty is
+ // skipped by the bucket search without opening an exact reader.
let residual_by_split: Option<Vec<HashMap<String, RoaringTreemap>>> =
match &self.filter {
Some(filter) => {
let file_predicates = FilePredicates {
@@ -465,50 +464,32 @@ impl<'a> VectorSearchBuilder<'a> {
None => None,
};
- // Exact-fallback readers, keyed by (split_index, file_name). In FAST
mode
- // the kernel never invokes the factory, so skip the in-memory column
read
- // entirely. Otherwise preload only the *uncovered* active files:
files an
- // ANN segment already covers never reach the exact fallback, so
reading
- // their vector column here would be wasted IO/memory. Mirrors Java,
which
- // creates a `PkVectorReader` lazily only for uncovered files. When a
- // residual filter leaves a file's allow-list empty (or absent) the
bucket
- // search skips it, so its reader is not preloaded either.
- let mut exact_readers: HashMap<(usize, String), Box<dyn
PkVectorReader>> = HashMap::new();
- if !skip_exact_fallback {
- for (split_index, split) in plan.splits.iter().enumerate() {
- let covered = covered_source_files(&split.ann_segments,
&split.active_files);
- let factory = DataFilePkVectorReaderFactory::new(
- reader.clone(),
- split.data_split.clone(),
- vector_field.clone(),
- )?;
- for active in &split.active_files {
- if covered.contains(&active.file_name) {
- continue;
- }
- if !should_preload_exact_reader(
- residual_by_split.as_deref(),
- split_index,
- &active.file_name,
- ) {
- continue;
- }
- let r = factory.create(active).await?;
- exact_readers.insert((split_index,
active.file_name.clone()), r);
- }
- }
- }
- let mut factory = |split_index: usize,
- _split: &PkVectorSearchSplit,
- file: &BucketActiveFile|
- -> crate::Result<Box<dyn PkVectorReader>> {
- exact_readers
- .remove(&(split_index, file.file_name.clone()))
- .ok_or_else(|| crate::Error::DataInvalid {
- message: format!("no preloaded exact reader for {}",
file.file_name),
- source: None,
+ // Build the exact-fallback vector reader on demand: the kernel calls
this
+ // only for a file it actually searches (uncovered by ANN,
residual-allowed,
+ // and only when the search mode is not FAST). Everything the future
needs is
+ // cloned/owned up front so it borrows neither the split nor the file
across
+ // the await.
+ let reader_for_factory = reader.clone();
+ let vector_field_for_factory = vector_field.clone();
+ let mut factory = as_split_exact_reader_factory(
+ move |_split_index: usize,
+ split: &PkVectorSearchSplit,
+ file: &BucketActiveFile|
+ -> ExactReaderFuture<'_> {
+ let reader = reader_for_factory.clone();
+ let vector_field = vector_field_for_factory.clone();
+ let data_split = split.data_split.clone();
+ let active = BucketActiveFile {
+ file_name: file.file_name.clone(),
+ row_count: file.row_count,
+ };
+ Box::pin(async move {
+ let factory =
+ DataFilePkVectorReaderFactory::new(reader, data_split,
vector_field)?;
+ factory.create(&active).await
})
- };
+ },
+ );
let candidates = PkVectorOrchestrator::new(reader)
.search_candidates(
@@ -1022,24 +1003,6 @@ fn is_vector_global_index_file(index_file:
&IndexFileMeta) -> bool {
VectorIndexBackend::from_index_type(&index_file.index_type).is_some()
}
-/// Whether the exact-fallback reader for `file_name` in split `split_index`
-/// should be preloaded. With a residual filter, a file absent from the split's
-/// allow-list or with an empty allow-list has no candidate rows, so the bucket
-/// search skips it and preloading its vector column would be wasted IO.
-fn should_preload_exact_reader(
- residual_by_split: Option<&[HashMap<String, RoaringTreemap>]>,
- split_index: usize,
- file_name: &str,
-) -> bool {
- match residual_by_split {
- None => true,
- Some(per_split) => per_split
- .get(split_index)
- .and_then(|m| m.get(file_name))
- .is_some_and(|allowed| !allowed.is_empty()),
- }
-}
-
/// Compute, per data file in `split`, the set of physical row positions whose
/// rows satisfy the residual predicate. Mirrors the row-collecting half of
Java
/// `PrimaryKeyVectorRead`'s `executeFilter`: because
@@ -2232,26 +2195,6 @@ mod tests {
assert_eq!(find_field_id_by_name(&fields, "nonexistent"), None);
}
- #[test]
- fn should_preload_skips_empty_or_absent_residual_files() {
- use roaring::RoaringTreemap;
- use std::collections::HashMap;
- // No filter -> always preload.
- assert!(should_preload_exact_reader(None, 0, "f0"));
- // Filter present: file with a non-empty allow-list -> preload.
- let mut m0: HashMap<String, RoaringTreemap> = HashMap::new();
- m0.insert("f0".to_string(), RoaringTreemap::from_iter([0u64]));
- let per_split = vec![m0];
- assert!(should_preload_exact_reader(Some(&per_split), 0, "f0"));
- // Filter present: file absent -> skip.
- assert!(!should_preload_exact_reader(Some(&per_split), 0, "missing"));
- // Filter present: file with empty allow-list -> skip.
- let mut m1: HashMap<String, RoaringTreemap> = HashMap::new();
- m1.insert("f1".to_string(), RoaringTreemap::new());
- let per_split2 = vec![m1];
- assert!(!should_preload_exact_reader(Some(&per_split2), 0, "f1"));
- }
-
#[test]
fn test_raw_vector_score_matches_java_metric_semantics() {
let l2 = compute_raw_vector_score(&[1.0, 2.0], &[1.0, 4.0],
RawVectorMetric::L2);
diff --git a/crates/paimon/src/vindex/pkvector/bucket.rs
b/crates/paimon/src/vindex/pkvector/bucket.rs
index 604ac7dc..137d7fc0 100644
--- a/crates/paimon/src/vindex/pkvector/bucket.rs
+++ b/crates/paimon/src/vindex/pkvector/bucket.rs
@@ -19,6 +19,8 @@ use std::cmp::Ordering;
use std::collections::{BinaryHeap, HashMap, HashSet};
use std::sync::Arc;
+use futures::future::BoxFuture;
+
use super::ann::PkVectorAnnSearcher;
use super::data_invalid;
use super::exact::exact_search;
@@ -28,6 +30,10 @@ use super::result::PkVectorSearchResult;
use crate::deletion_vector::DeletionVector;
use crate::spec::PkVectorSourceMeta;
+/// Future returned by the exact-fallback reader factory: builds the sequential
+/// vector reader for one data file on demand.
+pub(crate) type ExactReaderFuture<'a> = BoxFuture<'a, crate::Result<Box<dyn
PkVectorReader>>>;
+
/// One ANN segment to be searched by the bucket kernel. `source_meta` resolves
/// segment ordinals back to physical `(data file, position)` and drives
live-row
/// masking; the remaining fields address the segment's index file for the ANN
@@ -110,11 +116,10 @@ fn add_candidate(heap: &mut BinaryHeap<WorstFirst>,
candidate: PkVectorSearchRes
/// Active data files whose rows are already covered by an ANN segment's source
/// metadata, matched by both file name AND row count. The bucket exact
fallback
-/// skips these files, so a caller that preloads exact readers should preload
-/// only the *uncovered* active files (`active_files` minus this set) rather
than
-/// reading every active file's vector column up front. A source naming an
-/// inactive file, or one whose row count disagrees with the active file, is
not
-/// covered here; `bucket_search` rejects the row-count mismatch separately.
+/// skips these files (an ANN segment already covers their rows), so they never
+/// need an exact reader. A source naming an inactive file, or one whose row
+/// count disagrees with the active file, is not covered here; `bucket_search`
+/// rejects the row-count mismatch separately.
pub(crate) fn covered_source_files(
ann_segments: &[BucketAnnSegment],
active_files: &[BucketActiveFile],
@@ -150,14 +155,13 @@ pub(crate) fn covered_source_files(
/// with an empty set) has no allowed rows and produces no candidates. Mirrors
Java
/// `rowRangesByFile`.
#[allow(clippy::too_many_arguments)]
-pub(crate) fn bucket_search(
+pub(crate) async fn bucket_search(
ann_searcher: Option<&dyn PkVectorAnnSearcher>,
ann_segments: &[BucketAnnSegment],
active_files: &[BucketActiveFile],
deletion_vectors: &HashMap<String, Arc<DeletionVector>>,
- exact_reader_factory: &mut dyn FnMut(
- &BucketActiveFile,
- ) -> crate::Result<Box<dyn PkVectorReader>>,
+ exact_reader_factory: &mut (dyn for<'a> FnMut(&'a BucketActiveFile) ->
ExactReaderFuture<'a>
+ + Send),
query: &[f32],
metric: VectorSearchMetric,
limit: usize,
@@ -219,8 +223,7 @@ pub(crate) fn bucket_search(
let active_source_files: HashSet<String> =
files_by_name.keys().map(|name| name.to_string()).collect();
// Active files whose rows an ANN segment already covers; the exact
fallback
- // skips them. Same rule the caller's exact-reader preload uses, so both
agree
- // on which files still need an exact reader.
+ // skips them, so the lazy exact-reader factory is never invoked for those
files.
let covered = covered_source_files(ann_segments, active_files);
for segment in ann_segments {
@@ -290,7 +293,7 @@ pub(crate) fn bucket_search(
},
}
};
- let mut reader = exact_reader_factory(file)?;
+ let mut reader = exact_reader_factory(file).await?;
for result in exact_search(
&file.file_name,
reader.as_mut(),
@@ -316,7 +319,6 @@ mod tests {
use crate::vindex::pkvector::ann::PkVectorAnnSearcher;
use crate::vindex::pkvector::reader::test_support::ArrayReader;
use roaring::RoaringBitmap;
- use std::cell::RefCell;
fn meta(files: &[(&str, i64)]) -> PkVectorSourceMeta {
PkVectorSourceMeta::new(
@@ -336,6 +338,17 @@ mod tests {
}
}
+ /// Coerce a closure into the higher-ranked exact-reader factory shape so
its
+ /// returned future borrows for exactly the argument's lifetime. Closure
return
+ /// types cannot express this borrow through inference alone, so the bound
is
+ /// supplied here.
+ fn as_factory<F>(f: F) -> F
+ where
+ F: for<'a> FnMut(&'a BucketActiveFile) -> ExactReaderFuture<'a> + Send,
+ {
+ f
+ }
+
/// Fake ANN searcher returning preset results and recording calls.
struct FakeAnnSearcher {
result: Vec<PkVectorSearchResult>,
@@ -356,10 +369,11 @@ mod tests {
}
}
- #[test]
- fn test_rejects_non_positive_limit() {
- let mut factory =
- |_: &BucketActiveFile| -> crate::Result<Box<dyn PkVectorReader>> {
unreachable!() };
+ #[tokio::test]
+ async fn test_rejects_non_positive_limit() {
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!() })
+ });
let err = bucket_search(
None,
&[],
@@ -373,12 +387,13 @@ mod tests {
false,
None,
)
+ .await
.unwrap_err();
assert!(err.to_string().contains("positive"));
}
- #[test]
- fn test_bounded_heap_evicts_by_best_first_tiebreak_over_limit() {
+ #[tokio::test]
+ async fn test_bounded_heap_evicts_by_best_first_tiebreak_over_limit() {
// All candidates share distance 1.0, so eviction is decided purely by
the
// BEST_FIRST tie-break (data_file_name ASC, then row_position ASC).
Feed
// more than `limit` ANN hits and assert the kept set is the smallest
@@ -399,8 +414,9 @@ mod tests {
hit("data-1", 1),
],
};
- let mut factory =
- |_: &BucketActiveFile| -> crate::Result<Box<dyn PkVectorReader>> {
unreachable!() };
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!() })
+ });
let results = bucket_search(
Some(&ann),
&[segment],
@@ -414,6 +430,7 @@ mod tests {
false,
None,
)
+ .await
.unwrap();
// Top-3 BEST_FIRST: (data-1,0), (data-1,1), (data-1,2) — the larger
// data_file_name "data-2" entries are evicted despite equal distance.
@@ -426,8 +443,8 @@ mod tests {
);
}
- #[test]
- fn nan_ann_hit_never_evicts_finite_candidate_from_top1() {
+ #[tokio::test]
+ async fn nan_ann_hit_never_evicts_finite_candidate_from_top1() {
// The core failure mode: an ANN hit with a negative-NaN distance must
not
// win the single bucket Top-1 slot over a finite hit. Under
f32::total_cmp
// the -NaN would rank best and evict the finite candidate here in the
@@ -449,8 +466,9 @@ mod tests {
},
],
};
- let mut factory =
- |_: &BucketActiveFile| -> crate::Result<Box<dyn PkVectorReader>> {
unreachable!() };
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!() })
+ });
let results = bucket_search(
Some(&ann),
&[segment],
@@ -464,14 +482,15 @@ mod tests {
false,
None,
)
+ .await
.unwrap();
assert_eq!(results.len(), 1);
assert_eq!(results[0].row_position, 1);
assert_eq!(results[0].distance, -1.0);
}
- #[test]
- fn test_merges_ann_and_exact_without_rescanning_covered_files() {
+ #[tokio::test]
+ async fn test_merges_ann_and_exact_without_rescanning_covered_files() {
// data-1 is ANN-covered; data-2 is exact fallback. Factory must never
be
// called for data-1.
let segment = BucketAnnSegment::for_test(meta(&[("data-1", 2)]));
@@ -482,15 +501,17 @@ mod tests {
distance: 0.5,
}],
};
- let calls = RefCell::new(Vec::<String>::new());
- let mut factory = |f: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
- calls.borrow_mut().push(f.file_name.clone());
+ let calls = std::sync::Mutex::new(Vec::<String>::new());
+ let mut factory = as_factory(|f: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ calls.lock().unwrap().push(f.file_name.clone());
// data-2 vectors: pos0 {1,0} dist 1.0, pos1 {3,0} dist 9.0
- Ok(Box::new(ArrayReader::new(
- 2,
- vec![Some(vec![1.0, 0.0]), Some(vec![3.0, 0.0])],
- )))
- };
+ Box::pin(async {
+ Ok(Box::new(ArrayReader::new(
+ 2,
+ vec![Some(vec![1.0, 0.0]), Some(vec![3.0, 0.0])],
+ )) as Box<dyn PkVectorReader>)
+ })
+ });
let results = bucket_search(
Some(&ann),
&[segment],
@@ -504,6 +525,7 @@ mod tests {
false,
None,
)
+ .await
.unwrap();
assert_eq!(
results,
@@ -520,22 +542,24 @@ mod tests {
},
]
);
- assert_eq!(calls.borrow().as_slice(), &["data-2".to_string()]);
+ assert_eq!(calls.lock().unwrap().as_slice(), &["data-2".to_string()]);
}
- #[test]
- fn test_exact_fallback_merges_files_and_applies_deletion_vectors() {
+ #[tokio::test]
+ async fn test_exact_fallback_merges_files_and_applies_deletion_vectors() {
// No ANN. data-1 pos0 {0,0} deleted; remaining candidates merge
across files.
- let calls = RefCell::new(0);
- let mut factory = |f: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
- *calls.borrow_mut() += 1;
+ let calls = std::sync::Mutex::new(0usize);
+ let mut factory = as_factory(|f: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ *calls.lock().unwrap() += 1;
let vectors = match f.file_name.as_str() {
"data-1" => vec![Some(vec![0.0, 0.0]), Some(vec![2.0, 0.0])],
"data-2" => vec![Some(vec![1.0, 0.0]), None],
_ => unreachable!(),
};
- Ok(Box::new(ArrayReader::new(2, vectors)))
- };
+ Box::pin(async move {
+ Ok(Box::new(ArrayReader::new(2, vectors)) as Box<dyn
PkVectorReader>)
+ })
+ });
let mut dvs: HashMap<String, Arc<DeletionVector>> = HashMap::new();
let mut bm = RoaringBitmap::new();
bm.insert(0); // data-1 position 0 deleted
@@ -554,6 +578,7 @@ mod tests {
false,
None,
)
+ .await
.unwrap();
// Candidates: data-2 pos0 {1,0} dist 1.0; data-1 pos1 {2,0} dist 4.0.
// (data-1 pos0 deleted, data-2 pos1 null.)
@@ -574,10 +599,11 @@ mod tests {
);
}
- #[test]
- fn test_rejects_duplicate_active_file_name() {
- let mut factory =
- |_: &BucketActiveFile| -> crate::Result<Box<dyn PkVectorReader>> {
unreachable!() };
+ #[tokio::test]
+ async fn test_rejects_duplicate_active_file_name() {
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!() })
+ });
let err = bucket_search(
None,
&[],
@@ -591,18 +617,20 @@ mod tests {
false,
None,
)
+ .await
.unwrap_err();
assert!(err.to_string().contains("duplicate") ||
err.to_string().contains("Duplicate"));
}
- #[test]
- fn test_rejects_ann_source_row_count_mismatch_for_active_file() {
+ #[tokio::test]
+ async fn test_rejects_ann_source_row_count_mismatch_for_active_file() {
let ann = FakeAnnSearcher { result: vec![] };
// Segment references data-1 with 2 rows, but the active file has 3
rows.
// An active source with a mismatched row count is still a hard error.
let segment = BucketAnnSegment::for_test(meta(&[("data-1", 2)]));
- let mut factory =
- |_: &BucketActiveFile| -> crate::Result<Box<dyn PkVectorReader>> {
unreachable!() };
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!() })
+ });
let err = bucket_search(
Some(&ann),
&[segment],
@@ -616,14 +644,15 @@ mod tests {
false,
None,
)
+ .await
.unwrap_err();
assert!(
err.to_string().contains("does not match") ||
err.to_string().contains("ANN source")
);
}
- #[test]
- fn test_skips_inactive_ann_source_and_searches_active_ones() {
+ #[tokio::test]
+ async fn test_skips_inactive_ann_source_and_searches_active_ones() {
// Segment covers [data-1, data-2] but only data-1 is still active
// (data-2 was compacted away). Java master skips the inactive source
// instead of failing the whole query; data-2 is neither covered (so it
@@ -637,11 +666,11 @@ mod tests {
distance: 0.5,
}],
};
- let calls = RefCell::new(Vec::<String>::new());
- let mut factory = |f: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
- calls.borrow_mut().push(f.file_name.clone());
- unreachable!("only data-1 is active and it is ANN-covered")
- };
+ let calls = std::sync::Mutex::new(Vec::<String>::new());
+ let mut factory = as_factory(|f: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ calls.lock().unwrap().push(f.file_name.clone());
+ Box::pin(async { unreachable!("only data-1 is active and it is
ANN-covered") })
+ });
let results = bucket_search(
Some(&ann),
&[segment],
@@ -655,6 +684,7 @@ mod tests {
false,
None,
)
+ .await
.unwrap();
assert_eq!(
results,
@@ -665,14 +695,15 @@ mod tests {
}]
);
// No exact fallback ran: data-1 is ANN-covered, data-2 is not active.
- assert!(calls.borrow().is_empty());
+ assert!(calls.lock().unwrap().is_empty());
}
- #[test]
- fn test_rejects_segments_without_ann_searcher() {
+ #[tokio::test]
+ async fn test_rejects_segments_without_ann_searcher() {
let segment = BucketAnnSegment::for_test(meta(&[("data-1", 2)]));
- let mut factory =
- |_: &BucketActiveFile| -> crate::Result<Box<dyn PkVectorReader>> {
unreachable!() };
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!() })
+ });
let err = bucket_search(
None,
&[segment],
@@ -686,6 +717,7 @@ mod tests {
false,
None,
)
+ .await
.unwrap_err();
assert!(
err.to_string().contains("ANN search is not configured")
@@ -693,12 +725,13 @@ mod tests {
);
}
- #[test]
- fn test_skip_exact_fallback_does_not_call_factory() {
+ #[tokio::test]
+ async fn test_skip_exact_fallback_does_not_call_factory() {
// No ANN segments, two active files. With skip_exact_fallback = true
the
// factory must never be called and the result is empty.
- let mut factory =
- |_: &BucketActiveFile| -> crate::Result<Box<dyn PkVectorReader>> {
unreachable!() };
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!() })
+ });
let results = bucket_search(
None,
&[],
@@ -712,12 +745,13 @@ mod tests {
true, // skip_exact_fallback
None,
)
+ .await
.unwrap();
assert!(results.is_empty());
}
- #[test]
- fn test_rejects_duplicate_ann_segment_path() {
+ #[tokio::test]
+ async fn test_rejects_duplicate_ann_segment_path() {
let seg1 = BucketAnnSegment {
source_meta: meta(&[("data-1", 2)]),
path: "duplicate-path".to_string(),
@@ -731,8 +765,9 @@ mod tests {
index_meta: vec![4, 5, 6],
};
let ann = FakeAnnSearcher { result: vec![] };
- let mut factory =
- |_: &BucketActiveFile| -> crate::Result<Box<dyn PkVectorReader>> {
unreachable!() };
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!() })
+ });
let err = bucket_search(
Some(&ann),
&[seg1, seg2],
@@ -746,6 +781,7 @@ mod tests {
false,
None,
)
+ .await
.unwrap_err();
assert!(
err.to_string().contains("duplicate-path")
@@ -753,8 +789,8 @@ mod tests {
);
}
- #[test]
- fn test_rejects_source_file_covered_by_multiple_segments() {
+ #[tokio::test]
+ async fn test_rejects_source_file_covered_by_multiple_segments() {
let seg1 = BucketAnnSegment {
source_meta: meta(&[("data-1", 2)]),
path: "segment-1".to_string(),
@@ -768,8 +804,9 @@ mod tests {
index_meta: vec![4, 5, 6],
};
let ann = FakeAnnSearcher { result: vec![] };
- let mut factory =
- |_: &BucketActiveFile| -> crate::Result<Box<dyn PkVectorReader>> {
unreachable!() };
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!() })
+ });
let err = bucket_search(
Some(&ann),
&[seg1, seg2],
@@ -783,6 +820,7 @@ mod tests {
false,
None,
)
+ .await
.unwrap_err();
assert!(
err.to_string().contains("data-1")
@@ -792,10 +830,11 @@ mod tests {
);
}
- #[test]
- fn test_negative_active_row_count_rejected() {
- let mut factory =
- |_: &BucketActiveFile| -> crate::Result<Box<dyn PkVectorReader>> {
unreachable!() };
+ #[tokio::test]
+ async fn test_negative_active_row_count_rejected() {
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async { unreachable!() })
+ });
let err = bucket_search(
None,
&[],
@@ -809,6 +848,7 @@ mod tests {
false,
None,
)
+ .await
.unwrap_err();
assert!(err.to_string().contains("row count") ||
err.to_string().contains("-1"));
}
@@ -850,21 +890,23 @@ mod tests {
t
}
- #[test]
- fn test_exact_residual_allow_list_restricts_positions() {
+ #[tokio::test]
+ async fn test_exact_residual_allow_list_restricts_positions() {
// No ANN. data-1 has 3 rows: pos0 {1,0} dist 1.0, pos1 {2,0} dist 4.0,
// pos2 {3,0} dist 9.0. residual allows only {0, 2} -> pos1 excluded
even
// though it is not deletion-vector deleted.
- let mut factory = |_: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
- Ok(Box::new(ArrayReader::new(
- 2,
- vec![
- Some(vec![1.0, 0.0]),
- Some(vec![2.0, 0.0]),
- Some(vec![3.0, 0.0]),
- ],
- )))
- };
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async {
+ Ok(Box::new(ArrayReader::new(
+ 2,
+ vec![
+ Some(vec![1.0, 0.0]),
+ Some(vec![2.0, 0.0]),
+ Some(vec![3.0, 0.0]),
+ ],
+ )) as Box<dyn PkVectorReader>)
+ })
+ });
let mut residual: HashMap<String, roaring::RoaringTreemap> =
HashMap::new();
residual.insert("data-1".into(), treemap(&[0, 2]));
let results = bucket_search(
@@ -880,6 +922,7 @@ mod tests {
false,
Some(&residual),
)
+ .await
.unwrap();
assert_eq!(
results,
@@ -898,18 +941,20 @@ mod tests {
);
}
- #[test]
- fn test_exact_residual_file_absent_from_map_is_skipped_without_reading() {
+ #[tokio::test]
+ async fn
test_exact_residual_file_absent_from_map_is_skipped_without_reading() {
// residual covers only data-1; data-2 has no entry -> no allowed
rows, so
// data-2 is skipped entirely (its factory reader is never built).
- let calls = RefCell::new(Vec::<String>::new());
- let mut factory = |f: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
- calls.borrow_mut().push(f.file_name.clone());
- Ok(Box::new(ArrayReader::new(
- 2,
- vec![Some(vec![1.0, 0.0]), Some(vec![2.0, 0.0])],
- )))
- };
+ let calls = std::sync::Mutex::new(Vec::<String>::new());
+ let mut factory = as_factory(|f: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ calls.lock().unwrap().push(f.file_name.clone());
+ Box::pin(async {
+ Ok(Box::new(ArrayReader::new(
+ 2,
+ vec![Some(vec![1.0, 0.0]), Some(vec![2.0, 0.0])],
+ )) as Box<dyn PkVectorReader>)
+ })
+ });
let mut residual: HashMap<String, roaring::RoaringTreemap> =
HashMap::new();
residual.insert("data-1".into(), treemap(&[0, 1]));
let results = bucket_search(
@@ -925,21 +970,22 @@ mod tests {
false,
Some(&residual),
)
+ .await
.unwrap();
// Only data-1 rows appear; data-2 was never read.
assert!(results.iter().all(|r| r.data_file_name == "data-1"));
- assert_eq!(calls.borrow().as_slice(), &["data-1".to_string()]);
+ assert_eq!(calls.lock().unwrap().as_slice(), &["data-1".to_string()]);
}
- #[test]
- fn test_exact_residual_empty_set_file_is_skipped_without_reading() {
+ #[tokio::test]
+ async fn test_exact_residual_empty_set_file_is_skipped_without_reading() {
// data-1 has an entry but it is empty -> no allowed rows, skipped
without
// reading. Mirrors a file with no residual matches.
- let calls = RefCell::new(0);
- let mut factory = |_: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
- *calls.borrow_mut() += 1;
- unreachable!("data-1 has an empty allow set and must not be read")
- };
+ let calls = std::sync::Mutex::new(0usize);
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ *calls.lock().unwrap() += 1;
+ Box::pin(async { unreachable!("data-1 has an empty allow set and
must not be read") })
+ });
let mut residual: HashMap<String, roaring::RoaringTreemap> =
HashMap::new();
residual.insert("data-1".into(), treemap(&[]));
let results = bucket_search(
@@ -955,25 +1001,28 @@ mod tests {
false,
Some(&residual),
)
+ .await
.unwrap();
assert!(results.is_empty());
- assert_eq!(*calls.borrow(), 0);
+ assert_eq!(*calls.lock().unwrap(), 0);
}
- #[test]
- fn test_exact_residual_intersects_with_deletion_vector() {
+ #[tokio::test]
+ async fn test_exact_residual_intersects_with_deletion_vector() {
// residual allows {0, 1, 2} but the deletion vector deletes pos0; the
// surviving candidates are the residual-allowed AND not-deleted rows.
- let mut factory = |_: &BucketActiveFile| -> crate::Result<Box<dyn
PkVectorReader>> {
- Ok(Box::new(ArrayReader::new(
- 2,
- vec![
- Some(vec![1.0, 0.0]),
- Some(vec![2.0, 0.0]),
- Some(vec![3.0, 0.0]),
- ],
- )))
- };
+ let mut factory = as_factory(|_: &BucketActiveFile| ->
ExactReaderFuture<'_> {
+ Box::pin(async {
+ Ok(Box::new(ArrayReader::new(
+ 2,
+ vec![
+ Some(vec![1.0, 0.0]),
+ Some(vec![2.0, 0.0]),
+ Some(vec![3.0, 0.0]),
+ ],
+ )) as Box<dyn PkVectorReader>)
+ })
+ });
let mut dvs: HashMap<String, Arc<DeletionVector>> = HashMap::new();
let mut bm = RoaringBitmap::new();
bm.insert(0); // pos0 deleted
@@ -993,6 +1042,7 @@ mod tests {
false,
Some(&residual),
)
+ .await
.unwrap();
assert_eq!(
results.iter().map(|r| r.row_position).collect::<Vec<_>>(),
