upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-09-21 13:37:46 -05:00
Code Issues Packages Projects Releases Wiki Activity

Propagate OperationResult in PayloadFieldIndex::filter (#8647)

Browse Source
This commit is contained in:
xzfc
2026-05-08 13:46:39 +02:00
committed by timvisee
parent eda654085d
commit a306702563
29 changed files with 327 additions and 279 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/segment/src/index/field_index/bool_index/mod.rs
+8 -8
View File
@@ -150,7 +150,7 @@ impl PayloadFieldIndex for BoolIndex {
&'a self,
condition: &'a crate::types::FieldCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
match self {
BoolIndex::Mmap(index) => index.filter(condition, hw_counter),
}
@@ -311,8 +311,8 @@ mod tests {
let count = index
.filter(&match_bool(match_on), &hw_counter)
.unwrap()
.unwrap()
.count();
.process_results(|it| it.count())
.unwrap();
assert_eq!(count, expected_count);
}
@@ -372,14 +372,14 @@ mod tests {
let point_offsets = new_index
.filter(&match_bool(false), &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert_eq!(point_offsets, vec![1, 2, 3, 5, 6, 10]);
let point_offsets = new_index
.filter(&match_bool(true), &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert_eq!(point_offsets, vec![0, 2, 3, 4, 6, 11]);
@@ -409,7 +409,7 @@ mod tests {
let point_offsets = index
.filter(&match_bool(false), &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert_eq!(point_offsets, vec![idx]);
@@ -418,13 +418,13 @@ mod tests {
let point_offsets = index
.filter(&match_bool(true), &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert_eq!(point_offsets, vec![idx]);
let point_offsets = index
.filter(&match_bool(false), &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert!(point_offsets.is_empty());
}
lib/segment/src/index/field_index/bool_index/mutable_bool_index.rs
+5 -4
View File
@@ -376,8 +376,8 @@ impl PayloadFieldIndex for MutableBoolIndex {
&'a self,
condition: &'a FieldCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
Ok(match &condition.r#match {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
match &condition.r#match {
Some(Match::Value(MatchValue {
value: ValueVariants::Bool(value),
})) => {
@@ -389,11 +389,12 @@ impl PayloadFieldIndex for MutableBoolIndex {
hw_counter.new_accumulator(),
u8::BITS as usize,
|i| i.payload_index_io_read_counter(),
);
)
.map(Ok);
Some(Box::new(iter))
}
_ => None,
})
}
}
fn estimate_cardinality(
lib/segment/src/index/field_index/field_index_base.rs
+2 -2
View File
@@ -52,7 +52,7 @@ pub trait PayloadFieldIndex {
&'a self,
condition: &'a FieldCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>>;
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>>;
/// Return estimation of amount of points which satisfy given condition.
/// Returns `Ok(None)` if the condition does not match the index type
@@ -245,7 +245,7 @@ impl FieldIndex {
&'a self,
condition: &'a FieldCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
self.get_payload_field_index().filter(condition, hw_counter)
}
lib/segment/src/index/field_index/full_text_index/inverted_index/immutable_inverted_index.rs
+4 -4
View File
@@ -262,11 +262,11 @@ impl InvertedIndex for ImmutableInvertedIndex {
&'a self,
query: ParsedQuery,
_hw_counter: &'a HardwareCounterCell,
) -> Box<dyn Iterator<Item = PointOffsetType> + 'a> {
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a> {
match query {
ParsedQuery::AllTokens(tokens) => Box::new(self.filter_has_all(tokens)),
ParsedQuery::Phrase(tokens) => Box::new(self.filter_has_phrase(tokens)),
ParsedQuery::AnyTokens(tokens) => Box::new(self.filter_has_any(tokens)),
ParsedQuery::AllTokens(tokens) => Box::new(self.filter_has_all(tokens).map(Ok)),
ParsedQuery::Phrase(tokens) => Box::new(self.filter_has_phrase(tokens).map(Ok)),
ParsedQuery::AnyTokens(tokens) => Box::new(self.filter_has_any(tokens).map(Ok)),
}
}
lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/mod.rs
+6 -9
View File
@@ -190,7 +190,7 @@ impl MmapInvertedIndex {
pub fn filter_has_all<'a>(
&'a self,
tokens: TokenSet,
) -> Box<dyn Iterator<Item = PointOffsetType> + 'a> {
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a> {
// in case of mmap immutable index, deleted points are still in the postings
let filter = move |idx| self.is_active(idx);
@@ -198,7 +198,7 @@ impl MmapInvertedIndex {
postings: &'a MmapPostings<V>,
tokens: TokenSet,
filter: impl Fn(u32) -> bool + 'a,
) -> Box<dyn Iterator<Item = PointOffsetType> + 'a> {
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a> {
let postings_opt: Option<Vec<_>> = tokens
.tokens()
.iter()
@@ -215,10 +215,7 @@ impl MmapInvertedIndex {
return Box::new(std::iter::empty());
}
Box::new(intersect_compressed_postings_iterator(
posting_readers,
filter,
))
Box::new(intersect_compressed_postings_iterator(posting_readers, filter).map(Ok))
}
match &self.storage.postings {
@@ -456,11 +453,11 @@ impl InvertedIndex for MmapInvertedIndex {
&'a self,
query: ParsedQuery,
_hw_counter: &HardwareCounterCell,
) -> Box<dyn Iterator<Item = PointOffsetType> + 'a> {
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a> {
match query {
ParsedQuery::AllTokens(tokens) => self.filter_has_all(tokens),
ParsedQuery::Phrase(phrase) => Box::new(self.filter_has_phrase(phrase)),
ParsedQuery::AnyTokens(tokens) => Box::new(self.filter_has_any(tokens)),
ParsedQuery::Phrase(phrase) => Box::new(self.filter_has_phrase(phrase).map(Ok)),
ParsedQuery::AnyTokens(tokens) => Box::new(self.filter_has_any(tokens).map(Ok)),
}
}
lib/segment/src/index/field_index/full_text_index/inverted_index/mod.rs
+10 -3
View File
@@ -214,7 +214,7 @@ pub trait InvertedIndex {
&'a self,
query: ParsedQuery,
hw_counter: &'a HardwareCounterCell,
) -> Box<dyn Iterator<Item = PointOffsetType> + 'a>;
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>;
fn get_posting_len(&self, token_id: TokenId, hw_counter: &HardwareCounterCell)
-> Option<usize>;
@@ -695,10 +695,17 @@ mod tests {
// In this case both queries would filter to an empty set of documents.
continue;
};
let mut_filtered = mut_index.filter(mut_query, hw_counter).collect::<Vec<_>>();
let imm_filtered = mmap_index.filter(imm_query, hw_counter).collect::<Vec<_>>();
let mut_filtered = mut_index
.filter(mut_query, hw_counter)
.map(|r| r.unwrap())
.collect::<Vec<_>>();
let imm_filtered = mmap_index
.filter(imm_query, hw_counter)
.map(|r| r.unwrap())
.collect::<Vec<_>>();
let imm_mmap_filtered = imm_mmap_index
.filter(imm_mmap_query, hw_counter)
.map(|r| r.unwrap())
.collect::<Vec<_>>();
assert_eq!(mut_filtered, imm_filtered);
lib/segment/src/index/field_index/full_text_index/inverted_index/mutable_inverted_index.rs
+4 -4
View File
@@ -214,11 +214,11 @@ impl InvertedIndex for MutableInvertedIndex {
&self,
query: ParsedQuery,
_hw_counter: &HardwareCounterCell,
) -> Box<dyn Iterator<Item = PointOffsetType> + '_> {
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + '_> {
match query {
ParsedQuery::AllTokens(tokens) => Box::new(self.filter_has_all(tokens)),
ParsedQuery::Phrase(phrase) => self.filter_has_phrase(phrase),
ParsedQuery::AnyTokens(tokens) => Box::new(self.filter_has_any(tokens)),
ParsedQuery::AllTokens(tokens) => Box::new(self.filter_has_all(tokens).map(Ok)),
ParsedQuery::Phrase(phrase) => Box::new(self.filter_has_phrase(phrase).map(Ok)),
ParsedQuery::AnyTokens(tokens) => Box::new(self.filter_has_any(tokens).map(Ok)),
}
}
lib/segment/src/index/field_index/full_text_index/mutable_text_index.rs
+8 -8
View File
@@ -344,7 +344,7 @@ mod tests {
let search_res: Vec<_> = index
.filter(&filter_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect();
assert_eq!(search_res, vec![0, 4]);
@@ -352,7 +352,7 @@ mod tests {
let search_res: Vec<_> = index
.filter(&filter_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect();
assert_eq!(search_res, vec![2]);
@@ -360,7 +360,7 @@ mod tests {
let search_res: Vec<_> = index
.filter(&filter_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect();
assert_eq!(search_res, vec![4]);
@@ -372,7 +372,7 @@ mod tests {
index
.filter(&filter_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.next()
.is_none()
);
@@ -410,7 +410,7 @@ mod tests {
let search_res: Vec<_> = index
.filter(&filter_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect();
assert_eq!(search_res, vec![0]);
@@ -418,7 +418,7 @@ mod tests {
let search_res: Vec<_> = index
.filter(&filter_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect();
assert_eq!(search_res, vec![0, 1, 3, 4]);
@@ -428,7 +428,7 @@ mod tests {
let search_res: Vec<_> = index
.filter(&filter_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect();
assert!(search_res.is_empty());
assert_eq!(index.count_indexed_points(), 3);
@@ -438,7 +438,7 @@ mod tests {
let search_res: Vec<_> = index
.filter(&filter_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect();
assert_eq!(search_res, vec![1, 4]);
assert_eq!(index.count_indexed_points(), 2);
lib/segment/src/index/field_index/full_text_index/tests/mod.rs
+28 -6
View File
@@ -183,7 +183,10 @@ fn test_prefix_search() {
.unwrap();
}
let res: Vec<_> = index.query("ROBO", &hw_counter).collect();
let res: Vec<_> = index
.query("ROBO", &hw_counter)
.map(|r| r.unwrap())
.collect();
let query = index.parse_text_query("ROBO", &hw_counter).unwrap();
@@ -193,7 +196,10 @@ fn test_prefix_search() {
assert_eq!(res.len(), 3);
let res: Vec<_> = index.query("q231", &hw_counter).collect();
let res: Vec<_> = index
.query("q231", &hw_counter)
.map(|r| r.unwrap())
.collect();
assert!(res.is_empty());
assert!(index.parse_text_query("q231", &hw_counter).is_none());
@@ -257,7 +263,10 @@ fn test_phrase_matching() {
assert!(index.check_match(&text_query, 1));
assert!(index.check_match(&text_query, 2));
let text_results: Vec<_> = index.filter_query(text_query, &hw_counter).collect();
let text_results: Vec<_> = index
.filter_query(text_query, &hw_counter)
.map(|r| r.unwrap())
.collect();
// Should match documents 0, 1, and 2 (all contain "quick", "brown", "fox")
assert_eq!(text_results.len(), 3);
@@ -272,7 +281,10 @@ fn test_phrase_matching() {
assert!(index.check_match(&phrase_query, 0));
assert!(index.check_match(&phrase_query, 2));
let phrase_results: Vec<_> = index.filter_query(phrase_query, &hw_counter).collect();
let phrase_results: Vec<_> = index
.filter_query(phrase_query, &hw_counter)
.map(|r| r.unwrap())
.collect();
// Should only match documents 0 and 2 (contain "quick brown fox" in that exact order)
assert_eq!(phrase_results.len(), 2);
@@ -284,7 +296,10 @@ fn test_phrase_matching() {
let missing_query = index
.parse_phrase_query("fox brown quick", &hw_counter)
.unwrap();
let missing_results: Vec<_> = index.filter_query(missing_query, &hw_counter).collect();
let missing_results: Vec<_> = index
.filter_query(missing_query, &hw_counter)
.map(|r| r.unwrap())
.collect();
// Should match no documents (no document contains this exact phrase)
assert_eq!(missing_results.len(), 0);
@@ -301,7 +316,10 @@ fn test_phrase_matching() {
assert!(index.check_match(&phrase_query, 4));
// Should only match document 4
let filter_results: Vec<_> = index.filter_query(phrase_query, &hw_counter).collect();
let filter_results: Vec<_> = index
.filter_query(phrase_query, &hw_counter)
.map(|r| r.unwrap())
.collect();
assert_eq!(filter_results.len(), 1);
assert!(filter_results.contains(&4));
};
@@ -370,6 +388,7 @@ fn test_ascii_folding_in_full_text_index_word() {
let results_enabled: Vec<_> = index_enabled
.filter_query(query_enabled, &hw_counter)
.map(|r| r.unwrap())
.collect();
assert!(results_enabled.contains(&0));
@@ -378,6 +397,7 @@ fn test_ascii_folding_in_full_text_index_word() {
if let Some(query_disabled) = query_disabled_opt {
let results_disabled: Vec<_> = index_disabled
.filter_query(query_disabled, &hw_counter)
.map(|r| r.unwrap())
.collect();
assert!(!results_disabled.contains(&0));
}
@@ -387,6 +407,7 @@ fn test_ascii_folding_in_full_text_index_word() {
assert!(index_enabled.check_match(&query_acento, 0));
let results_acento: Vec<_> = index_enabled
.filter_query(query_acento, &hw_counter)
.map(|r| r.unwrap())
.collect();
assert!(results_acento.contains(&0));
@@ -395,6 +416,7 @@ fn test_ascii_folding_in_full_text_index_word() {
.unwrap();
let results_acento2: Vec<_> = index_disabled
.filter_query(query_acento2, &hw_counter)
.map(|r| r.unwrap())
.collect();
assert!(results_acento2.contains(&0));
}
lib/segment/src/index/field_index/full_text_index/tests/test_congruence.rs
+5
View File
@@ -352,9 +352,11 @@ fn test_congruence(
assert_eq!(
index_a
.filter_query(parsed_query_a, &hw_counter)
.map(|r| r.unwrap())
.collect::<HashSet<_>>(),
index_b
.filter_query(parsed_query_b, &hw_counter)
.map(|r| r.unwrap())
.collect::<HashSet<_>>(),
);
}
@@ -394,9 +396,11 @@ fn test_congruence(
assert_eq!(
index_a
.filter_query(parsed_query_a, &hw_counter)
.map(|r| r.unwrap())
.collect::<HashSet<_>>(),
index_b
.filter_query(parsed_query_b, &hw_counter)
.map(|r| r.unwrap())
.collect::<HashSet<_>>(),
);
}
@@ -453,6 +457,7 @@ fn check_phrase<const KEYWORD_COUNT: usize>(
let result = index
.filter_query(parsed_query, &hw_counter)
.map(|r| r.unwrap())
.collect::<HashSet<_>>();
assert!(!result.is_empty());
lib/segment/src/index/field_index/full_text_index/text_index.rs
+6 -6
View File
@@ -114,7 +114,7 @@ impl FullTextIndex {
&'a self,
query: ParsedQuery,
hw_counter: &'a HardwareCounterCell,
) -> Box<dyn Iterator<Item = PointOffsetType> + 'a> {
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a> {
match self {
Self::Mutable(index) => index.inverted_index.filter(query, hw_counter),
Self::Immutable(index) => index.inverted_index.filter(query, hw_counter),
@@ -303,7 +303,7 @@ impl FullTextIndex {
&'a self,
query: &'a str,
hw_counter: &'a HardwareCounterCell,
) -> Box<dyn Iterator<Item = PointOffsetType> + 'a> {
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a> {
let Some(parsed_query) = self.parse_text_query(query, hw_counter) else {
return Box::new(std::iter::empty());
};
@@ -469,20 +469,20 @@ impl PayloadFieldIndex for FullTextIndex {
&'a self,
condition: &'a FieldCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
let parsed_query_opt = match &condition.r#match {
Some(Match::Text(MatchText { text })) => self.parse_text_query(text, hw_counter),
Some(Match::Phrase(MatchPhrase { phrase })) => {
self.parse_phrase_query(phrase, hw_counter)
}
_ => return Ok(None),
_ => return None,
};
let Some(parsed_query) = parsed_query_opt else {
return Ok(Some(Box::new(std::iter::empty())));
return Some(Box::new(std::iter::empty()));
};
Ok(Some(self.filter_query(parsed_query, hw_counter)))
Some(self.filter_query(parsed_query, hw_counter))
}
fn estimate_cardinality(
lib/segment/src/index/field_index/geo_index/mod.rs
+47 -43
View File
@@ -228,30 +228,32 @@ impl GeoMapIndex {
}
}
#[expect(clippy::unnecessary_wraps, reason = "will return Err later")] // FIXME(uio-errors)
fn iterator(
&self,
values: Vec<GeoHash>,
) -> OperationResult<Box<dyn Iterator<Item = PointOffsetType> + '_>> {
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + '_> {
match self {
GeoMapIndex::Mutable(index) => Ok(Box::new(
GeoMapIndex::Mutable(index) => Box::new(
values
.into_iter()
.flat_map(|top_geo_hash| index.stored_sub_regions(top_geo_hash))
.unique(),
)),
GeoMapIndex::Immutable(index) => Ok(Box::new(
.unique()
.map(Ok),
),
GeoMapIndex::Immutable(index) => Box::new(
values
.into_iter()
.flat_map(|top_geo_hash| index.stored_sub_regions(top_geo_hash))
.unique(),
)),
GeoMapIndex::Mmap(index) => Ok(Box::new(
.unique()
.map(Ok),
),
GeoMapIndex::Mmap(index) => Box::new(
values
.into_iter()
.flat_map(|top_geo_hash| index.stored_sub_regions(top_geo_hash))
.unique(),
)),
.unique()
.map(Ok),
),
}
}
@@ -523,51 +525,47 @@ impl PayloadFieldIndex for GeoMapIndex {
&'a self,
condition: &FieldCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
if let Some(geo_bounding_box) = &condition.geo_bounding_box {
let Some(geo_hashes) = rectangle_hashes(geo_bounding_box, GEO_QUERY_MAX_REGION).ok()
else {
return Ok(None);
};
let geo_hashes = rectangle_hashes(geo_bounding_box, GEO_QUERY_MAX_REGION).ok()?;
let geo_condition_copy = *geo_bounding_box;
return Ok(Some(Box::new(self.iterator(geo_hashes)?.filter(
return Some(Box::new(self.iterator(geo_hashes).filter_map_ok(
move |point| {
self.check_values_any(*point, hw_counter, |geo_point| {
self.check_values_any(point, hw_counter, |geo_point| {
geo_condition_copy.check_point(geo_point)
})
.then_some(point)
},
))));
)));
}
if let Some(geo_radius) = &condition.geo_radius {
let Some(geo_hashes) = circle_hashes(geo_radius, GEO_QUERY_MAX_REGION).ok() else {
return Ok(None);
};
let geo_hashes = circle_hashes(geo_radius, GEO_QUERY_MAX_REGION).ok()?;
let geo_condition_copy = *geo_radius;
return Ok(Some(Box::new(self.iterator(geo_hashes)?.filter(
return Some(Box::new(self.iterator(geo_hashes).filter_map_ok(
move |point| {
self.check_values_any(*point, hw_counter, |geo_point| {
self.check_values_any(point, hw_counter, |geo_point| {
geo_condition_copy.check_point(geo_point)
})
.then_some(point)
},
))));
)));
}
if let Some(geo_polygon) = &condition.geo_polygon {
let Some(geo_hashes) = polygon_hashes(geo_polygon, GEO_QUERY_MAX_REGION).ok() else {
return Ok(None);
};
let geo_hashes = polygon_hashes(geo_polygon, GEO_QUERY_MAX_REGION).ok()?;
let geo_condition_copy = geo_polygon.convert();
return Ok(Some(Box::new(self.iterator(geo_hashes)?.filter(
return Some(Box::new(self.iterator(geo_hashes).filter_map_ok(
move |point| {
self.check_values_any(*point, hw_counter, |geo_point| {
self.check_values_any(point, hw_counter, |geo_point| {
geo_condition_copy.check_point(geo_point)
})
.then_some(point)
},
))));
)));
}
Ok(None)
None
}
fn estimate_cardinality(
@@ -837,7 +835,10 @@ mod tests {
index_type: IndexType,
) {
let (field_index, _, _) = build_random_index(500, 20, index_type);
let exact_points_for_hashes = field_index.iterator(hashes).unwrap().collect_vec();
let exact_points_for_hashes = field_index
.iterator(hashes)
.map(|r| r.unwrap())
.collect_vec();
let real_cardinality = exact_points_for_hashes.len();
let hw_counter = HardwareCounterCell::new();
@@ -911,7 +912,10 @@ mod tests {
index_type: IndexType,
) {
let (field_index, _, _) = build_random_index(500, 20, index_type);
let exact_points_for_hashes = field_index.iterator(hashes).unwrap().collect_vec();
let exact_points_for_hashes = field_index
.iterator(hashes)
.map(|r| r.unwrap())
.collect_vec();
let real_cardinality = exact_points_for_hashes.len();
let hw_counter = HardwareCounterCell::new();
@@ -986,7 +990,7 @@ mod tests {
let mut indexed_matched_points = field_index
.filter(&field_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
matched_points.sort_unstable();
@@ -1052,7 +1056,7 @@ mod tests {
let block_points = field_index
.filter(&block.condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert_eq!(block_points.len(), block.cardinality);
});
@@ -1260,7 +1264,7 @@ mod tests {
let point_offsets = new_index
.filter(&field_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert_eq!(point_offsets, vec![1]);
@@ -1272,7 +1276,7 @@ mod tests {
let point_offsets = new_index
.filter(&field_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert_eq!(point_offsets, vec![1]);
}
@@ -1464,7 +1468,7 @@ mod tests {
let point_offsets = new_index
.filter(&field_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
// Only LOS_ANGELES is in the bounding box
assert_eq!(point_offsets, vec![2]);
@@ -1523,7 +1527,7 @@ mod tests {
let results = index
.filter(&field_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert_eq!(results, vec![1]);
@@ -1579,7 +1583,7 @@ mod tests {
let results = index
.filter(&field_condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert_eq!(results, vec![0]);
@@ -1665,11 +1669,11 @@ mod tests {
assert_eq!(
indices[0]
.iterator(hashes.clone())
.unwrap()
.map(|r| r.unwrap())
.collect::<HashSet<_>>(),
index
.iterator(hashes.clone())
.unwrap()
.map(|r| r.unwrap())
.collect::<HashSet<_>>(),
);
for point_id in 0..POINT_COUNT {
lib/segment/src/index/field_index/map_index/mod.rs
+39 -40
View File
@@ -468,7 +468,7 @@ where
&'a self,
excluded: &'a IndexSet<K, A>,
hw_counter: &'a HardwareCounterCell,
) -> Box<dyn Iterator<Item = PointOffsetType> + 'a>
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>
where
A: BuildHasher,
K: Borrow<N> + Hash + Eq,
@@ -477,7 +477,8 @@ where
self.iter_values()
.filter(|key| !excluded.contains((*key).borrow()))
.flat_map(move |key| self.get_iterator(key.borrow(), hw_counter))
.unique(),
.unique()
.map(Ok),
)
}
@@ -720,12 +721,12 @@ impl PayloadFieldIndex for MapIndex<str> {
&'a self,
condition: &'a FieldCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
Ok(match &condition.r#match {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
match &condition.r#match {
Some(Match::Value(MatchValue { value })) => match value {
ValueVariants::String(keyword) => {
Some(Box::new(self.get_iterator(keyword.as_str(), hw_counter)))
}
ValueVariants::String(keyword) => Some(Box::new(
self.get_iterator(keyword.as_str(), hw_counter).map(Ok),
)),
ValueVariants::Integer(_) => None,
ValueVariants::Bool(_) => None,
},
@@ -734,7 +735,8 @@ impl PayloadFieldIndex for MapIndex<str> {
keywords
.iter()
.flat_map(move |keyword| self.get_iterator(keyword.as_str(), hw_counter))
.unique(),
.unique()
.map(Ok),
)),
AnyVariants::Integers(integers) => {
if integers.is_empty() {
@@ -755,7 +757,7 @@ impl PayloadFieldIndex for MapIndex<str> {
}
},
_ => None,
})
}
}
fn estimate_cardinality(
@@ -869,34 +871,32 @@ impl PayloadFieldIndex for MapIndex<UuidIntType> {
&'a self,
condition: &'a FieldCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
Ok(match &condition.r#match {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
match &condition.r#match {
Some(Match::Value(MatchValue { value })) => match value {
ValueVariants::String(uuid_string) => {
let Some(uuid) = Uuid::from_str(uuid_string).ok() else {
return Ok(None);
};
Some(Box::new(self.get_iterator(&uuid.as_u128(), hw_counter)))
let uuid = Uuid::from_str(uuid_string).ok()?;
Some(Box::new(
self.get_iterator(&uuid.as_u128(), hw_counter).map(Ok),
))
}
ValueVariants::Integer(_) => None,
ValueVariants::Bool(_) => None,
},
Some(Match::Any(MatchAny { any: any_variant })) => match any_variant {
AnyVariants::Strings(uuids_string) => {
let uuids: Result<IndexSet<u128>, _> = uuids_string
let uuids: IndexSet<u128> = uuids_string
.iter()
.map(|uuid_string| Uuid::from_str(uuid_string).map(|x| x.as_u128()))
.collect();
let Some(uuids) = uuids.ok() else {
return Ok(None);
};
.collect::<Result<_, _>>()
.ok()?;
Some(Box::new(
uuids
.into_iter()
.flat_map(move |uuid| self.get_iterator(&uuid, hw_counter))
.unique(),
.unique()
.map(Ok),
))
}
AnyVariants::Integers(integers) => {
@@ -909,20 +909,18 @@ impl PayloadFieldIndex for MapIndex<UuidIntType> {
},
Some(Match::Except(MatchExcept { except })) => match except {
AnyVariants::Strings(uuids_string) => {
let uuids: Result<IndexSet<u128>, _> = uuids_string
let excluded_uuids: IndexSet<u128> = uuids_string
.iter()
.map(|uuid_string| Uuid::from_str(uuid_string).map(|x| x.as_u128()))
.collect();
let Some(excluded_uuids) = uuids.ok() else {
return Ok(None);
};
let exclude_iter = self
.iter_values()
.filter(move |key| !excluded_uuids.contains(*key))
.flat_map(move |key| self.get_iterator(key, hw_counter))
.unique();
Some(Box::new(exclude_iter))
.collect::<Result<_, _>>()
.ok()?;
Some(Box::new(
self.iter_values()
.filter(move |key| !excluded_uuids.contains(*key))
.flat_map(move |key| self.get_iterator(key, hw_counter))
.unique()
.map(Ok),
))
}
AnyVariants::Integers(other) => {
if other.is_empty() {
@@ -933,7 +931,7 @@ impl PayloadFieldIndex for MapIndex<UuidIntType> {
}
},
_ => None,
})
}
}
fn estimate_cardinality(
@@ -1071,12 +1069,12 @@ impl PayloadFieldIndex for MapIndex<IntPayloadType> {
&'a self,
condition: &'a FieldCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
Ok(match &condition.r#match {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
match &condition.r#match {
Some(Match::Value(MatchValue { value })) => match value {
ValueVariants::String(_) => None,
ValueVariants::Integer(integer) => {
Some(Box::new(self.get_iterator(integer, hw_counter)))
Some(Box::new(self.get_iterator(integer, hw_counter).map(Ok)))
}
ValueVariants::Bool(_) => None,
},
@@ -1092,7 +1090,8 @@ impl PayloadFieldIndex for MapIndex<IntPayloadType> {
integers
.iter()
.flat_map(move |integer| self.get_iterator(integer, hw_counter))
.unique(),
.unique()
.map(Ok),
)),
},
Some(Match::Except(MatchExcept { except })) => match except {
@@ -1106,7 +1105,7 @@ impl PayloadFieldIndex for MapIndex<IntPayloadType> {
AnyVariants::Integers(integers) => Some(self.except_set(integers, hw_counter)),
},
_ => None,
})
}
}
fn estimate_cardinality(
lib/segment/src/index/field_index/null_index/mutable_null_index.rs
+11 -13
View File
@@ -254,7 +254,7 @@ impl PayloadFieldIndex for MutableNullIndex {
&'a self,
condition: &'a FieldCondition,
_hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
let FieldCondition {
key: _,
r#match: _,
@@ -267,29 +267,27 @@ impl PayloadFieldIndex for MutableNullIndex {
is_null,
} = condition;
Ok(if let Some(is_empty) = is_empty {
if let Some(is_empty) = is_empty {
if *is_empty {
// Return points that don't have values
let iter = self.storage.has_values_flags.iter_falses();
Some(Box::new(iter))
Some(Box::new(
self.storage.has_values_flags.iter_falses().map(Ok),
))
} else {
// Return points that have values
let iter = self.storage.has_values_flags.iter_trues();
Some(Box::new(iter))
Some(Box::new(self.storage.has_values_flags.iter_trues().map(Ok)))
}
} else if let Some(is_null) = is_null {
if *is_null {
// Return points that have null values
let iter = self.storage.is_null_flags.iter_trues();
Some(Box::new(iter))
Some(Box::new(self.storage.is_null_flags.iter_trues().map(Ok)))
} else {
// Return points that don't have null values
let iter = self.storage.is_null_flags.iter_falses();
Some(Box::new(iter))
Some(Box::new(self.storage.is_null_flags.iter_falses().map(Ok)))
}
} else {
None
})
}
}
fn estimate_cardinality(
@@ -449,12 +447,12 @@ mod tests {
let is_null_values: Vec<_> = null_index
.filter(&filter_is_null, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect();
let not_empty_values: Vec<_> = null_index
.filter(&filter_is_not_empty, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect();
let is_empty_values: Vec<_> = (0..n)
lib/segment/src/index/field_index/numeric_index/mod.rs
+23 -17
View File
@@ -387,13 +387,19 @@ where
&'a self,
value: T,
hw_counter: &'a HardwareCounterCell,
) -> Box<dyn Iterator<Item = PointOffsetType> + 'a> {
) -> Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a> {
let start = Bound::Included(Point::new(value, PointOffsetType::MIN));
let end = Bound::Included(Point::new(value, PointOffsetType::MAX));
match &self {
NumericIndexInner::Mutable(mutable) => Box::new(mutable.values_range(start, end)),
NumericIndexInner::Immutable(immutable) => Box::new(immutable.values_range(start, end)),
NumericIndexInner::Mmap(mmap) => Box::new(mmap.values_range(start, end, hw_counter)),
NumericIndexInner::Mutable(mutable) => {
Box::new(mutable.values_range(start, end).map(Ok))
}
NumericIndexInner::Immutable(immutable) => {
Box::new(immutable.values_range(start, end).map(Ok))
}
NumericIndexInner::Mmap(mmap) => {
Box::new(mmap.values_range(start, end, hw_counter).map(Ok))
}
}
}
@@ -764,7 +770,7 @@ where
&'a self,
condition: &FieldCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
if let Some(Match::Value(MatchValue {
value: ValueVariants::String(keyword),
})) = &condition.r#match
@@ -773,13 +779,11 @@ where
if let Ok(uuid) = Uuid::from_str(keyword) {
let value = T::from_u128(uuid.as_u128());
return Ok(Some(self.point_ids_by_value(value, hw_counter)));
return Some(self.point_ids_by_value(value, hw_counter));
}
}
let Some(range_cond) = condition.range.as_ref() else {
return Ok(None);
};
let range_cond = condition.range.as_ref()?;
let (start_bound, end_bound) = match range_cond {
RangeInterface::Float(float_range) => float_range.map(|float| T::from_f64(float.0)),
@@ -792,20 +796,22 @@ where
// map.range
// Panics if range start > end. Panics if range start == end and both bounds are Excluded.
if !check_boundaries(&start_bound, &end_bound) {
return Ok(Some(Box::new(std::iter::empty())));
return Some(Box::new(std::iter::empty()));
}
Ok(Some(match self {
Some(match self {
NumericIndexInner::Mutable(index) => {
Box::new(index.values_range(start_bound, end_bound))
Box::new(index.values_range(start_bound, end_bound).map(Ok))
}
NumericIndexInner::Immutable(index) => {
Box::new(index.values_range(start_bound, end_bound))
Box::new(index.values_range(start_bound, end_bound).map(Ok))
}
NumericIndexInner::Mmap(index) => {
Box::new(index.values_range(start_bound, end_bound, hw_counter))
}
}))
NumericIndexInner::Mmap(index) => Box::new(
index
.values_range(start_bound, end_bound, hw_counter)
.map(Ok),
),
})
}
fn estimate_cardinality(
lib/segment/src/index/field_index/numeric_index/tests.rs
+2 -2
View File
@@ -127,7 +127,7 @@ fn cardinality_request(
&hw_counter,
)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.unique()
.collect_vec();
@@ -541,7 +541,7 @@ fn test_cond<
let offsets = index
.filter(&condition, &hw_counter)
.unwrap()
.unwrap()
.map(|r| r.unwrap())
.collect_vec();
assert_eq!(offsets, result);
}
lib/segment/src/index/hnsw_index/hnsw.rs
+7 -7
View File
@@ -15,7 +15,7 @@ use common::fs::clear_disk_cache;
use common::progress_tracker::ProgressTracker;
use common::types::{PointOffsetType, ScoredPointOffset, TelemetryDetail};
use fs_err as fs;
use itertools::EitherOrBoth;
use itertools::{EitherOrBoth, Itertools};
use log::{debug, trace};
use parking_lot::Mutex;
use rand::Rng;
@@ -751,7 +751,7 @@ impl HNSWIndex {
payload_index.estimate_cardinality(&filter, &disposed_hw_counter)?;
let point_mappings = id_tracker.point_mappings();
Ok(payload_index
payload_index
.iter_filtered_points(
&filter,
id_tracker,
@@ -761,8 +761,8 @@ impl HNSWIndex {
stopped,
None,
)?
.filter(|&point_id| !deleted_bitslice.get_bit(point_id as usize).unwrap_or(false))
.collect())
.filter_ok(|&point_id| !deleted_bitslice.get_bit(point_id as usize).unwrap_or(false))
.collect()
}
#[allow(clippy::too_many_arguments)]
@@ -1207,7 +1207,7 @@ impl HNSWIndex {
fn search_plain_iterator_batched(
&self,
query_vectors: &[&QueryVector],
points: impl Iterator<Item = PointOffsetType>,
points: impl Iterator<Item = OperationResult<PointOffsetType>>,
top: usize,
params: Option<&SearchParams>,
vector_query_context: &VectorQueryContext,
@@ -1252,7 +1252,7 @@ impl HNSWIndex {
fn search_plain_batched(
&self,
vectors: &[&QueryVector],
filtered_points: impl Iterator<Item = PointOffsetType>,
filtered_points: impl Iterator<Item = OperationResult<PointOffsetType>>,
top: usize,
params: Option<&SearchParams>,
vector_query_context: &VectorQueryContext,
@@ -1274,7 +1274,7 @@ impl HNSWIndex {
vector_query_context: &VectorQueryContext,
) -> OperationResult<Vec<Vec<ScoredPointOffset>>> {
let id_tracker = self.id_tracker.borrow();
let ids_iterator = id_tracker.point_mappings().iter_internal();
let ids_iterator = id_tracker.point_mappings().iter_internal().map(Ok);
self.search_plain_iterator_batched(vectors, ids_iterator, top, params, vector_query_context)
}
lib/segment/src/index/hnsw_index/point_scorer.rs
+8 -6
View File
@@ -8,7 +8,7 @@ use common::generic_consts::Random;
use common::types::{PointOffsetType, ScoreType, ScoredPointOffset};
use smallvec::SmallVec;
use crate::common::operation_error::{CancellableResult, OperationResult, check_process_stopped};
use crate::common::operation_error::{OperationResult, check_process_stopped};
use crate::data_types::vectors::QueryVector;
use crate::payload_storage::FilterContext;
use crate::vector_storage::common::VECTOR_READ_BATCH_SIZE;
@@ -351,7 +351,7 @@ impl<'a> BatchFilteredSearcher<'a> {
self,
is_stopped: &AtomicBool,
deferred_internal_id: Option<PointOffsetType>,
) -> CancellableResult<Vec<Vec<ScoredPointOffset>>> {
) -> OperationResult<Vec<Vec<ScoredPointOffset>>> {
let iter = self
.filters
.point_deleted
@@ -360,7 +360,8 @@ impl<'a> BatchFilteredSearcher<'a> {
.take_while(|&point_id| {
// Early exit if we hit the max point ID (e.g. a deferred point).
point_id < deferred_internal_id.unwrap_or(PointOffsetType::MAX)
});
})
.map(Ok);
self.peek_top_iter(iter, is_stopped)
}
@@ -368,9 +369,9 @@ impl<'a> BatchFilteredSearcher<'a> {
/// This function expects deferred points to be already filtered from the iterator.
pub fn peek_top_iter(
mut self,
mut points: impl Iterator<Item = PointOffsetType>,
mut points: impl Iterator<Item = OperationResult<PointOffsetType>>,
is_stopped: &AtomicBool,
) -> CancellableResult<Vec<Vec<ScoredPointOffset>>> {
) -> OperationResult<Vec<Vec<ScoredPointOffset>>> {
// Reuse the same buffer for all chunks, to avoid reallocation
let mut chunk = [0; VECTOR_READ_BATCH_SIZE];
let mut scores_buffer = [0.0; VECTOR_READ_BATCH_SIZE];
@@ -379,8 +380,9 @@ impl<'a> BatchFilteredSearcher<'a> {
check_process_stopped(is_stopped)?;
let mut chunk_size = 0;
for point_id in &mut points {
for point_result in &mut points {
check_process_stopped(is_stopped)?;
let point_id = point_result?;
if !self.filters.check_vector(point_id) {
continue;
lib/segment/src/index/plain_vector_index.rs
+2 -1
View File
@@ -144,7 +144,8 @@ impl VectorIndex for PlainVectorIndex {
&is_stopped,
deferred_internal_id,
)?;
batch_searcher.peek_top_iter(filtered_ids_vec.iter().copied(), &is_stopped)?
batch_searcher
.peek_top_iter(filtered_ids_vec.iter().copied().map(Ok), &is_stopped)?
}
None => batch_searcher.peek_top_all(&is_stopped, deferred_internal_id)?,
};
lib/segment/src/index/sparse_index/sparse_vector_index.rs
+2 -2
View File
@@ -320,7 +320,7 @@ impl<TInvertedIndex: InvertedIndex> SparseVectorIndex<TInvertedIndex> {
let mut results = match filter {
Some(filter) => {
let payload_index = self.payload_index.borrow();
let mut filtered_points = match prefiltered_points {
let filtered_points = match prefiltered_points {
// `prefiltered_points` always contains visible points only so we don't need additional filtering here.
Some(filtered_points) => filtered_points.iter().copied(),
None => {
@@ -334,7 +334,7 @@ impl<TInvertedIndex: InvertedIndex> SparseVectorIndex<TInvertedIndex> {
prefiltered_points.as_ref().unwrap().iter().copied()
}
};
searcher.peek_top_iter(&mut filtered_points, &is_stopped)?
searcher.peek_top_iter(filtered_points.map(Ok), &is_stopped)?
}
None => {
searcher.peek_top_all(&is_stopped, vector_query_context.deferred_internal_id())?
lib/segment/src/index/struct_payload_index.rs
+25 -28
View File
@@ -12,6 +12,7 @@ use common::either_variant::EitherVariant;
use common::iterator_ext::IteratorExt;
use common::types::PointOffsetType;
use fs_err as fs;
use itertools::Itertools;
use schemars::_serde_json::Value;
use super::field_index::facet_index::FacetIndexEnum;
@@ -100,24 +101,19 @@ impl StructPayloadIndex {
&'a self,
condition: &'a PrimaryCondition,
hw_counter: &'a HardwareCounterCell,
) -> OperationResult<Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>>> {
) -> Option<Box<dyn Iterator<Item = OperationResult<PointOffsetType>> + 'a>> {
match condition {
PrimaryCondition::Condition(field_condition) => {
let field_key = &field_condition.key;
let Some(field_indexes) = self.field_indexes.get(field_key) else {
return Ok(None);
};
let field_indexes = self.field_indexes.get(field_key)?;
field_indexes
.iter()
.find_map(|field_index| {
field_index.filter(field_condition, hw_counter).transpose()
})
.transpose()
.find_map(|field_index| field_index.filter(field_condition, hw_counter))
}
PrimaryCondition::Ids(ids) => {
Ok(Some(Box::new(ids.resolved_point_offsets.iter().copied())))
Some(Box::new(ids.resolved_point_offsets.iter().copied().map(Ok)))
}
PrimaryCondition::HasVector(_) => Ok(None),
PrimaryCondition::HasVector(_) => None,
}
}
@@ -482,7 +478,7 @@ impl StructPayloadIndex {
hw_counter: &'a HardwareCounterCell,
is_stopped: &'a AtomicBool,
deferred_internal_id: Option<PointOffsetType>,
) -> OperationResult<impl Iterator<Item = PointOffsetType> + 'a> {
) -> OperationResult<impl Iterator<Item = OperationResult<PointOffsetType>> + 'a> {
if query_cardinality.primary_clauses.is_empty() {
let full_scan_iterator = point_mappings.iter_internal_visible(deferred_internal_id);
@@ -490,7 +486,8 @@ impl StructPayloadIndex {
// Worst case: query expected to return few matches, but index can't be used
let matched_points = full_scan_iterator
.stop_if(is_stopped)
.filter(move |i| struct_filtered_context.check(*i));
.filter(move |i| struct_filtered_context.check(*i))
.map(Ok);
Ok(EitherVariant::A(matched_points))
} else {
@@ -503,7 +500,7 @@ impl StructPayloadIndex {
.primary_clauses
.iter()
.map(|clause| self.query_field(clause, hw_counter))
.collect::<OperationResult<_>>()?;
.collect();
if let Some(primary_iterators) = primary_clause_iterators {
let all_conditions_are_primary = filter
@@ -515,7 +512,7 @@ impl StructPayloadIndex {
// Filter out deferred points.
// This iterator (and each primary iterator too) can yield items in non sorted order, depending on the type of index and primary condition.
.flatten()
.filter(move |&internal_id| {
.filter_ok(move |&internal_id| {
internal_id < deferred_internal_id.unwrap_or(PointOffsetType::MAX)
})
.stop_if(is_stopped);
@@ -524,13 +521,13 @@ impl StructPayloadIndex {
// All conditions are primary clauses,
// We can avoid post-filtering
let iter = joined_primary_iterator
.filter(move |&id| !visited_list.check_and_update_visited(id));
.filter_ok(move |&id| !visited_list.check_and_update_visited(id));
EitherVariant::B(iter)
} else {
// Some conditions are primary clauses, some are not
let struct_filtered_context =
self.struct_filtered_context(filter, hw_counter)?;
let iter = joined_primary_iterator.filter(move |&id| {
let iter = joined_primary_iterator.filter_ok(move |&id| {
!visited_list.check_and_update_visited(id)
&& struct_filtered_context.check(id)
});
@@ -551,7 +548,8 @@ impl StructPayloadIndex {
})
.filter(move |&id| {
!visited_list.check_and_update_visited(id) && struct_filtered_context.check(id)
});
})
.map(Ok);
Ok(EitherVariant::D(iter))
}
@@ -781,17 +779,16 @@ impl PayloadIndex for StructPayloadIndex {
let query_cardinality = self.estimate_cardinality(filter, hw_counter)?;
let id_tracker = self.id_tracker.borrow();
let point_mappings = id_tracker.point_mappings();
Ok(self
.iter_filtered_points(
filter,
&id_tracker,
&point_mappings,
&query_cardinality,
hw_counter,
is_stopped,
deferred_internal_id,
)?
.collect())
self.iter_filtered_points(
filter,
&id_tracker,
&point_mappings,
&query_cardinality,
hw_counter,
is_stopped,
deferred_internal_id,
)?
.collect()
}
fn indexed_points(&self, field: PayloadKeyTypeRef) -> usize {
lib/segment/src/segment/facet.rs
+23 -15
View File
@@ -62,16 +62,20 @@ impl Segment {
is_stopped,
self.deferred_internal_id(),
)?
.filter(|point_id| !id_tracker.is_deleted_point(*point_id))
.fold(HashMap::new(), |mut map, point_id| {
facet_index
.get_point_values(point_id, hw_counter)
.unique()
.for_each(|value| {
*map.entry(value).or_insert(0) += 1;
});
map
})
.filter_ok(|&point_id| !id_tracker.is_deleted_point(point_id))
.try_fold(
HashMap::new(),
|mut map, point_result| -> OperationResult<_> {
let point_id = point_result?;
facet_index
.get_point_values(point_id, hw_counter)
.unique()
.for_each(|value| {
*map.entry(value).or_insert(0) += 1;
});
Ok(map)
},
)?
.into_iter()
.map(|(value, count)| FacetHit { value, count });
@@ -159,11 +163,15 @@ impl Segment {
is_stopped,
self.deferred_internal_id(),
)?
.filter(|point_id| !id_tracker.is_deleted_point(*point_id))
.fold(BTreeSet::new(), |mut set, point_id| {
set.extend(facet_index.get_point_values(point_id, hw_counter));
set
})
.filter_ok(|&point_id| !id_tracker.is_deleted_point(point_id))
.try_fold(
BTreeSet::new(),
|mut set, point_result| -> OperationResult<_> {
let point_id = point_result?;
set.extend(facet_index.get_point_values(point_id, hw_counter));
Ok(set)
},
)?
.into_iter()
.map(|value| value.to_owned())
.collect()
lib/segment/src/segment/order_by.rs
+17 -16
View File
@@ -4,7 +4,7 @@ use std::sync::atomic::AtomicBool;
use common::counter::hardware_counter::HardwareCounterCell;
use common::iterator_ext::IteratorExt;
use common::types::{DeferredBehavior, PointOffsetType};
use itertools::Either;
use itertools::{Either, Itertools};
use super::Segment;
use crate::common::operation_error::{OperationError, OperationResult};
@@ -53,24 +53,27 @@ impl Segment {
is_stopped,
effective_deferred_id,
)?
.flat_map(|internal_id| {
// Repeat a point for as many values as it has
numeric_index
.get_ordering_values(internal_id)
// But only those which start from `start_from`
.filter(|value| match order_by.direction() {
Direction::Asc => value >= &start_from,
Direction::Desc => value <= &start_from,
})
.map(move |ordering_value| (ordering_value, internal_id))
.flat_map(|point_result| match point_result {
Err(e) => Either::Left(std::iter::once(Err(e))),
Ok(internal_id) => Either::Right(
// Repeat a point for as many values as it has
numeric_index
.get_ordering_values(internal_id)
// But only those which start from `start_from`
.filter(|value| match order_by.direction() {
Direction::Asc => value >= &start_from,
Direction::Desc => value <= &start_from,
})
.map(move |ordering_value| Ok((ordering_value, internal_id))),
),
})
.filter_map(|(value, internal_id)| {
.filter_map_ok(|(value, internal_id)| {
id_tracker
.external_id(internal_id)
.map(|external_id| (value, external_id))
});
let page = match order_by.direction() {
values_ids_iterator.process_results(|values_ids_iterator| match order_by.direction() {
Direction::Asc => {
let mut page = match limit {
Some(limit) => peek_top_smallest_iterable(values_ids_iterator, limit),
@@ -87,9 +90,7 @@ impl Segment {
page.sort_unstable_by_key(|(value, _)| Reverse(*value));
page
}
};
Ok(page)
})
}
pub fn filtered_read_by_value_stream(
lib/segment/src/segment/sampling.rs
+7 -5
View File
@@ -2,6 +2,7 @@ use std::sync::atomic::AtomicBool;
use common::counter::hardware_counter::HardwareCounterCell;
use common::iterator_ext::IteratorExt;
use itertools::Itertools;
use rand::seq::{IteratorRandom, SliceRandom};
use super::Segment;
@@ -33,13 +34,14 @@ impl Segment {
is_stopped,
self.deferred_internal_id(),
)?
.filter_map(|internal_id| id_tracker.external_id(internal_id));
.filter_map_ok(|internal_id| id_tracker.external_id(internal_id));
let mut rng = rand::rng();
let mut shuffled = ids_iterator.sample(&mut rng, limit);
shuffled.shuffle(&mut rng);
Ok(shuffled)
ids_iterator.process_results(|iter| {
let mut shuffled = iter.sample(&mut rng, limit);
shuffled.shuffle(&mut rng);
shuffled
})
}
pub fn filtered_read_by_random_stream(
lib/segment/src/segment/scroll.rs
+14 -16
View File
@@ -3,6 +3,7 @@ use std::sync::atomic::AtomicBool;
use common::counter::hardware_counter::HardwareCounterCell;
use common::iterator_ext::IteratorExt;
use common::types::DeferredBehavior;
use itertools::Itertools;
use super::Segment;
use crate::common::operation_error::OperationResult;
@@ -129,24 +130,21 @@ impl Segment {
is_stopped,
effective_deferred_id,
)?
.filter_map(|internal_id| {
let external_id = id_tracker.external_id(internal_id);
match external_id {
Some(external_id) => match offset {
Some(offset) if external_id < offset => None,
_ => Some(external_id),
},
None => None,
.filter_map_ok(|internal_id| {
let external_id = id_tracker.external_id(internal_id)?;
match offset {
Some(offset) if external_id < offset => None,
_ => Some(external_id),
}
});
let mut page = match limit {
Some(limit) => peek_top_smallest_iterable(ids_iterator, limit),
None => ids_iterator.collect(),
};
page.sort_unstable();
Ok(page)
ids_iterator.process_results(|iter| {
let mut page = match limit {
Some(limit) => peek_top_smallest_iterable(iter, limit),
None => iter.collect(),
};
page.sort_unstable();
page
})
}
}
lib/segment/src/vector_storage/dense/dense_vector_storage.rs
+4 -4
View File
@@ -504,7 +504,7 @@ mod tests {
2,
);
let res = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.into_iter()
.exactly_one()
@@ -576,7 +576,7 @@ mod tests {
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.into_iter()
.exactly_one()
@@ -605,7 +605,7 @@ mod tests {
5,
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.into_iter()
.exactly_one()
@@ -694,7 +694,7 @@ mod tests {
5,
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.into_iter()
.exactly_one()
lib/segment/src/vector_storage/tests/test_appendable_dense_vector_storage.rs
+5 -5
View File
@@ -62,7 +62,7 @@ fn do_test_delete_points(storage: &mut VectorStorageEnum) {
5,
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.into_iter()
.exactly_one()
@@ -90,7 +90,7 @@ fn do_test_delete_points(storage: &mut VectorStorageEnum) {
5,
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.into_iter()
.exactly_one()
@@ -175,7 +175,7 @@ fn do_test_update_from_delete_points(storage: &mut VectorStorageEnum) {
5,
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.into_iter()
.exactly_one()
@@ -223,7 +223,7 @@ fn do_test_score_points(storage: &mut VectorStorageEnum) {
2,
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.into_iter()
.exactly_one()
@@ -262,7 +262,7 @@ fn do_test_score_points(storage: &mut VectorStorageEnum) {
)
.unwrap();
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.into_iter()
.exactly_one()
lib/segment/src/vector_storage/tests/test_appendable_multi_dense_vector_storage.rs
+3 -3
View File
@@ -129,7 +129,7 @@ fn do_test_delete_points(vector_dim: usize, vec_count: usize, storage: &mut Vect
5,
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.pop()
.unwrap();
@@ -156,7 +156,7 @@ fn do_test_delete_points(vector_dim: usize, vec_count: usize, storage: &mut Vect
5,
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.pop()
.unwrap();
@@ -244,7 +244,7 @@ fn do_test_update_from_delete_points(
5,
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.pop()
.unwrap();
lib/segment/src/vector_storage/tests/test_appendable_sparse_vector_storage.rs
+2 -2
View File
@@ -80,7 +80,7 @@ fn do_test_delete_points(storage: &mut VectorStorageEnum) {
5,
);
let closest = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap()
.into_iter()
.exactly_one()
@@ -172,7 +172,7 @@ fn do_test_update_from_delete_points(storage: &mut VectorStorageEnum) {
5,
);
let results = searcher
.peek_top_iter(&mut [0, 1, 2, 3, 4, 5].iter().cloned(), &DEFAULT_STOPPED)
.peek_top_iter([0, 1, 2, 3, 4, 5].iter().cloned().map(Ok), &DEFAULT_STOPPED)
.unwrap();
let closest = results.into_iter().exactly_one().unwrap();