upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-07-23 11:11:00 -05:00
Code Issues Packages Projects Releases Wiki Activity

Callback-based FacetIndex (#8764)

Browse Source 
* Segment::approximate_facet: don't use iter

* Callback-based FacetIndex

* take deferred internal id out of the loop

---------

Co-authored-by: Luis Cossío <luis.cossio@outlook.com>
This commit is contained in:
xzfc
2026-04-24 14:06:48 +00:00
committed by timvisee
parent a39b2b98d9
commit 58083c694a
10 changed files with 420 additions and 306 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
lib/segment/src/index/field_index/bool_index/immutable_bool_index.rs
View File

@@ -4,7 +4,6 @@ use common::bitvec::BitSlice;
use common::counter::hardware_counter::HardwareCounterCell;
use common::types::PointOffsetType;
use super::IdIter;
use super::mutable_bool_index::MutableBoolIndex;
use crate::common::operation_error::{OperationError, OperationResult};
use crate::index::field_index::{
@@ -50,11 +49,12 @@ impl ImmutableBoolIndex {
}
#[inline]
pub fn iter_values_map<'a>(
&'a self,
hw_acc: &'a HardwareCounterCell,
) -> impl Iterator<Item = (bool, IdIter<'a>)> + 'a {
self.0.iter_values_map(hw_acc)
pub fn for_each_value_map(
&self,
hw_counter: &HardwareCounterCell,
f: impl FnMut(bool, &mut dyn Iterator<Item = PointOffsetType>) -> OperationResult<()>,
) -> OperationResult<()> {
self.0.for_each_value_map(hw_counter, f)
}
#[inline]
@@ -63,11 +63,12 @@ impl ImmutableBoolIndex {
}
#[inline]
pub fn iter_counts_per_value(
pub fn for_each_count_per_value(
&self,
deferred_internal_id: Option<PointOffsetType>,
) -> impl Iterator<Item = (bool, usize)> + '_ {
self.0.iter_counts_per_value(deferred_internal_id)
f: impl FnMut(bool, usize) -> OperationResult<()>,
) -> OperationResult<()> {
self.0.for_each_count_per_value(deferred_internal_id, f)
}
#[inline]

75
lib/segment/src/index/field_index/bool_index/mod.rs
View File

@@ -4,7 +4,6 @@ use immutable_bool_index::ImmutableBoolIndex;
use mutable_bool_index::MutableBoolIndex;
use super::facet_index::FacetIndex;
use super::map_index::IdIter;
use super::{PayloadFieldIndex, ValueIndexer};
use crate::common::operation_error::{OperationError, OperationResult};
use crate::data_types::facets::{FacetHit, FacetValueRef};
@@ -27,13 +26,14 @@ impl BoolIndex {
}
}
pub fn iter_values_map<'a>(
&'a self,
hw_acc: &'a HardwareCounterCell,
) -> Box<dyn Iterator<Item = (bool, IdIter<'a>)> + 'a> {
pub fn for_each_value_map(
&self,
hw_acc: &HardwareCounterCell,
f: impl FnMut(bool, &mut dyn Iterator<Item = PointOffsetType>) -> OperationResult<()>,
) -> OperationResult<()> {
match self {
BoolIndex::Mmap(index) => Box::new(index.iter_values_map(hw_acc)),
BoolIndex::Immutable(index) => Box::new(index.iter_values_map(hw_acc)),
BoolIndex::Mmap(index) => index.for_each_value_map(hw_acc, f),
BoolIndex::Immutable(index) => index.for_each_value_map(hw_acc, f),
}
}
@@ -44,15 +44,14 @@ impl BoolIndex {
}
}
pub fn iter_counts_per_value(
pub fn for_each_count_per_value(
&self,
deferred_internal_id: Option<PointOffsetType>,
) -> Box<dyn Iterator<Item = (bool, usize)> + '_> {
f: impl FnMut(bool, usize) -> OperationResult<()>,
) -> OperationResult<()> {
match self {
BoolIndex::Mmap(index) => Box::new(index.iter_counts_per_value(deferred_internal_id)),
BoolIndex::Immutable(index) => {
Box::new(index.iter_counts_per_value(deferred_internal_id))
}
BoolIndex::Mmap(index) => index.for_each_count_per_value(deferred_internal_id, f),
BoolIndex::Immutable(index) => index.for_each_count_per_value(deferred_internal_id, f),
}
}
@@ -227,37 +226,51 @@ impl PayloadFieldIndex for BoolIndex {
}
impl FacetIndex for BoolIndex {
fn get_point_values(
fn for_points_values(
&self,
point_id: PointOffsetType,
points: impl Iterator<Item = PointOffsetType>,
_hw_counter: &HardwareCounterCell,
) -> impl Iterator<Item = FacetValueRef<'_>> + '_ {
self.get_point_values(point_id)
.into_iter()
.map(FacetValueRef::Bool)
mut f: impl FnMut(PointOffsetType, &mut dyn Iterator<Item = FacetValueRef<'_>>),
) -> OperationResult<()> {
points.for_each(|point_id| {
let values = self.get_point_values(point_id);
f(point_id, &mut values.into_iter().map(FacetValueRef::Bool));
});
Ok(())
}
fn iter_values(&self) -> impl Iterator<Item = FacetValueRef<'_>> + '_ {
self.iter_values().map(FacetValueRef::Bool)
fn for_each_value(
&self,
mut f: impl FnMut(FacetValueRef<'_>) -> OperationResult<()>,
) -> OperationResult<()> {
self.iter_values()
.try_for_each(|v| f(FacetValueRef::Bool(v)))
}
fn iter_values_map<'a>(
&'a self,
hw_counter: &'a HardwareCounterCell,
) -> impl Iterator<Item = (FacetValueRef<'a>, IdIter<'a>)> + 'a {
self.iter_values_map(hw_counter)
.map(|(value, iter)| (FacetValueRef::Bool(value), iter))
fn for_each_value_map(
&self,
hw_counter: &HardwareCounterCell,
mut f: impl FnMut(
FacetValueRef<'_>,
&mut dyn Iterator<Item = PointOffsetType>,
) -> OperationResult<()>,
) -> OperationResult<()> {
self.for_each_value_map(hw_counter, |value, iter| {
f(FacetValueRef::Bool(value), iter)
})
}
fn iter_counts_per_value(
fn for_each_count_per_value(
&self,
deferred_internal_id: Option<PointOffsetType>,
) -> impl Iterator<Item = FacetHit<FacetValueRef<'_>>> + '_ {
self.iter_counts_per_value(deferred_internal_id)
.map(|(value, count)| FacetHit {
mut f: impl FnMut(FacetHit<FacetValueRef<'_>>) -> OperationResult<()>,
) -> OperationResult<()> {
self.for_each_count_per_value(deferred_internal_id, |value, count| {
f(FacetHit {
value: FacetValueRef::Bool(value),
count,
})
})
}
}

41
lib/segment/src/index/field_index/bool_index/mutable_bool_index.rs
View File

@@ -10,7 +10,6 @@ use roaring::RoaringBitmap;
use crate::common::flags::dynamic_mmap_flags::DynamicMmapFlags;
use crate::common::flags::roaring_flags::RoaringFlags;
use crate::common::operation_error::{OperationError, OperationResult};
use crate::index::field_index::map_index::IdIter;
use crate::index::field_index::{
CardinalityEstimation, FieldIndexBuilderTrait, PayloadBlockCondition, PayloadFieldIndex,
PrimaryCondition, ValueIndexer,
@@ -240,24 +239,20 @@ impl MutableBoolIndex {
!self.storage.trues_flags.get(point_id) && !self.storage.falses_flags.get(point_id)
}
pub fn iter_values_map<'a>(
&'a self,
hw_counter: &'a HardwareCounterCell,
) -> impl Iterator<Item = (bool, IdIter<'a>)> + 'a {
[
(
false,
Box::new(self.storage.falses_flags.iter_trues()) as IdIter,
),
(
true,
Box::new(self.storage.trues_flags.iter_trues()) as IdIter,
),
]
.into_iter()
.measure_hw_with_acc(hw_counter.new_accumulator(), u8::BITS as usize, |i| {
i.payload_index_io_read_counter()
})
pub fn for_each_value_map(
&self,
hw_counter: &HardwareCounterCell,
mut f: impl FnMut(bool, &mut dyn Iterator<Item = PointOffsetType>) -> OperationResult<()>,
) -> OperationResult<()> {
f(false, &mut self.storage.falses_flags.iter_trues())?;
hw_counter
.payload_index_io_read_counter()
.incr_delta(u8::BITS as usize);
f(true, &mut self.storage.trues_flags.iter_trues())?;
hw_counter
.payload_index_io_read_counter()
.incr_delta(u8::BITS as usize);
Ok(())
}
pub fn iter_values(&self) -> impl Iterator<Item = bool> + '_ {
@@ -269,10 +264,11 @@ impl MutableBoolIndex {
.flatten()
}
pub fn iter_counts_per_value(
pub fn for_each_count_per_value(
&self,
deferred_internal_id: Option<PointOffsetType>,
) -> impl Iterator<Item = (bool, usize)> + '_ {
mut f: impl FnMut(bool, usize) -> OperationResult<()>,
) -> OperationResult<()> {
let (false_count, true_count) = match deferred_internal_id {
Some(deferred_internal_id) => {
let false_count =
@@ -295,7 +291,8 @@ impl MutableBoolIndex {
),
};
[(false, false_count), (true, true_count)].into_iter()
f(false, false_count)?;
f(true, true_count)
}
pub(crate) fn get_point_values(&self, point_id: u32) -> Vec<bool> {

146
lib/segment/src/index/field_index/facet_index.rs
View File

@@ -2,32 +2,42 @@ use common::counter::hardware_counter::HardwareCounterCell;
use common::types::PointOffsetType;
use super::bool_index::BoolIndex;
use super::map_index::{IdIter, MapIndex};
use super::map_index::MapIndex;
use crate::common::operation_error::OperationResult;
use crate::data_types::facets::{FacetHit, FacetValueRef};
use crate::types::{IntPayloadType, UuidIntType};
pub trait FacetIndex {
/// Get all values for a point
fn get_point_values(
/// Call a closure on value->point_ids mapping for specified `points`.
fn for_points_values(
&self,
point_id: PointOffsetType,
points: impl Iterator<Item = PointOffsetType>,
hw_counter: &HardwareCounterCell,
) -> impl Iterator<Item = FacetValueRef<'_>> + '_;
f: impl FnMut(PointOffsetType, &mut dyn Iterator<Item = FacetValueRef<'_>>),
) -> OperationResult<()>;
/// Get all values in the index
fn iter_values(&self) -> impl Iterator<Item = FacetValueRef<'_>> + '_;
/// Call a closure on each value in the index.
fn for_each_value(
&self,
f: impl FnMut(FacetValueRef<'_>) -> OperationResult<()>,
) -> OperationResult<()>;
/// Get all value->point_ids mappings
fn iter_values_map<'a>(
&'a self,
hw_acc: &'a HardwareCounterCell,
) -> impl Iterator<Item = (FacetValueRef<'a>, IdIter<'a>)> + 'a;
/// Call a closure on each value->point_ids mapping.
fn for_each_value_map(
&self,
hw_acc: &HardwareCounterCell,
f: impl FnMut(
FacetValueRef<'_>,
&mut dyn Iterator<Item = PointOffsetType>,
) -> OperationResult<()>,
) -> OperationResult<()>;
/// Get all value->count mappings
fn iter_counts_per_value(
/// Call a closure on each value->count mapping.
fn for_each_count_per_value(
&self,
deferred_internal_id: Option<PointOffsetType>,
) -> impl Iterator<Item = FacetHit<FacetValueRef<'_>>> + '_;
f: impl FnMut(FacetHit<FacetValueRef<'_>>) -> OperationResult<()>,
) -> OperationResult<()>;
}
pub enum FacetIndexEnum<'a> {
@@ -38,91 +48,91 @@ pub enum FacetIndexEnum<'a> {
}
impl<'a> FacetIndexEnum<'a> {
pub fn get_point_values(
pub fn for_points_values(
&self,
point_id: PointOffsetType,
points: impl Iterator<Item = PointOffsetType>,
hw_counter: &HardwareCounterCell,
) -> Box<dyn Iterator<Item = FacetValueRef<'a>> + 'a> {
f: impl FnMut(PointOffsetType, &mut dyn Iterator<Item = FacetValueRef<'_>>),
) -> OperationResult<()> {
match self {
FacetIndexEnum::Keyword(index) => {
Box::new(FacetIndex::get_point_values(*index, point_id, hw_counter))
FacetIndex::for_points_values(*index, points, hw_counter, f)
}
FacetIndexEnum::Int(index) => {
Box::new(FacetIndex::get_point_values(*index, point_id, hw_counter))
FacetIndex::for_points_values(*index, points, hw_counter, f)
}
FacetIndexEnum::Uuid(index) => {
Box::new(FacetIndex::get_point_values(*index, point_id, hw_counter))
FacetIndex::for_points_values(*index, points, hw_counter, f)
}
FacetIndexEnum::Bool(index) => {
Box::new(FacetIndex::get_point_values(*index, point_id, hw_counter))
FacetIndex::for_points_values(*index, points, hw_counter, f)
}
}
}
pub fn iter_values(
&'a self,
hw_counter: &'a HardwareCounterCell,
pub fn for_each_value(
&self,
hw_counter: &HardwareCounterCell,
deferred_internal_id: Option<PointOffsetType>,
) -> Box<dyn Iterator<Item = FacetValueRef<'a>> + 'a> {
mut f: impl FnMut(FacetValueRef<'_>) -> OperationResult<()>,
) -> OperationResult<()> {
match deferred_internal_id {
Some(deferred_internal_id) => Box::new(self.iter_values_map(hw_counter).filter_map(
move |(facet_value, id_iter)| {
Some(deferred_internal_id) => {
self.for_each_value_map(hw_counter, |facet_value, id_iter| {
let has_visible_point = id_iter
.take_while(|&id| id < deferred_internal_id)
.next()
.is_some();
has_visible_point.then_some(facet_value)
},
)),
if has_visible_point {
f(facet_value)?;
}
Ok(())
})
}
None => match self {
FacetIndexEnum::Keyword(index) => Box::new(FacetIndex::iter_values(*index)),
FacetIndexEnum::Int(index) => Box::new(FacetIndex::iter_values(*index)),
FacetIndexEnum::Uuid(index) => Box::new(FacetIndex::iter_values(*index)),
FacetIndexEnum::Bool(index) => Box::new(FacetIndex::iter_values(*index)),
FacetIndexEnum::Keyword(index) => FacetIndex::for_each_value(*index, f),
FacetIndexEnum::Int(index) => FacetIndex::for_each_value(*index, f),
FacetIndexEnum::Uuid(index) => FacetIndex::for_each_value(*index, f),
FacetIndexEnum::Bool(index) => FacetIndex::for_each_value(*index, f),
},
}
}
pub fn iter_values_map<'b>(
&'b self,
hw_counter: &'b HardwareCounterCell,
) -> Box<dyn Iterator<Item = (FacetValueRef<'b>, IdIter<'b>)> + 'b> {
pub fn for_each_value_map(
&self,
hw_counter: &HardwareCounterCell,
f: impl FnMut(
FacetValueRef<'_>,
&mut dyn Iterator<Item = PointOffsetType>,
) -> OperationResult<()>,
) -> OperationResult<()> {
match self {
FacetIndexEnum::Keyword(index) => {
Box::new(FacetIndex::iter_values_map(*index, hw_counter))
}
FacetIndexEnum::Int(index) => Box::new(FacetIndex::iter_values_map(*index, hw_counter)),
FacetIndexEnum::Uuid(index) => {
Box::new(FacetIndex::iter_values_map(*index, hw_counter))
}
FacetIndexEnum::Bool(index) => {
Box::new(FacetIndex::iter_values_map(*index, hw_counter))
}
FacetIndexEnum::Keyword(index) => FacetIndex::for_each_value_map(*index, hw_counter, f),
FacetIndexEnum::Int(index) => FacetIndex::for_each_value_map(*index, hw_counter, f),
FacetIndexEnum::Uuid(index) => FacetIndex::for_each_value_map(*index, hw_counter, f),
FacetIndexEnum::Bool(index) => FacetIndex::for_each_value_map(*index, hw_counter, f),
}
}
pub fn iter_counts_per_value(
&'a self,
pub fn for_each_count_per_value(
&self,
deferred_internal_id: Option<PointOffsetType>,
) -> Box<dyn Iterator<Item = FacetHit<FacetValueRef<'a>>> + 'a> {
f: impl FnMut(FacetHit<FacetValueRef<'_>>) -> OperationResult<()>,
) -> OperationResult<()> {
match self {
FacetIndexEnum::Keyword(index) => Box::new(FacetIndex::iter_counts_per_value(
*index,
deferred_internal_id,
)),
FacetIndexEnum::Int(index) => Box::new(FacetIndex::iter_counts_per_value(
*index,
deferred_internal_id,
)),
FacetIndexEnum::Uuid(index) => Box::new(FacetIndex::iter_counts_per_value(
*index,
deferred_internal_id,
)),
FacetIndexEnum::Bool(index) => Box::new(FacetIndex::iter_counts_per_value(
*index,
deferred_internal_id,
)),
FacetIndexEnum::Keyword(index) => {
FacetIndex::for_each_count_per_value(*index, deferred_internal_id, f)
}
FacetIndexEnum::Int(index) => {
FacetIndex::for_each_count_per_value(*index, deferred_internal_id, f)
}
FacetIndexEnum::Uuid(index) => {
FacetIndex::for_each_count_per_value(*index, deferred_internal_id, f)
}
FacetIndexEnum::Bool(index) => {
FacetIndex::for_each_count_per_value(*index, deferred_internal_id, f)
}
}
}
}

8
lib/segment/src/index/field_index/immutable_point_to_values.rs
View File

@@ -115,12 +115,16 @@ impl<N: Default> ImmutablePointToValues<N> {
}
pub fn get_values(&self, idx: PointOffsetType) -> Option<impl Iterator<Item = &N> + '_> {
Some(self.get_values_slice(idx)?.iter())
}
pub fn get_values_slice(&self, idx: PointOffsetType) -> Option<&[N]> {
let entry = self.point_entries.get(idx as usize)?;
match entry {
PointValueEntry::Single(v) => Some(std::slice::from_ref(v).iter()),
PointValueEntry::Single(v) => Some(std::slice::from_ref(v)),
PointValueEntry::Slice { start, count } => {
let range = *start as usize..(*start + *count) as usize;
Some(self.values_container[range].iter())
Some(&self.values_container[range])
}
}
}

81
lib/segment/src/index/field_index/map_index/immutable_map_index.rs
View File

@@ -336,48 +336,75 @@ where
.map(|entry| entry.count as usize)
}
pub fn iter_counts_per_value(&self) -> impl Iterator<Item = (&N, usize)> + '_ {
self.value_to_points
.iter()
.map(|(k, entry)| (k.borrow(), entry.count as usize))
pub fn for_points_values(
&self,
points: impl Iterator<Item = PointOffsetType>,
mut f: impl FnMut(PointOffsetType, &[<N as MapIndexKey>::Owned]),
) {
points.for_each(|idx| {
if let Some(values) = self.point_to_values.get_values_slice(idx) {
f(idx, values);
}
});
}
pub fn iter_values_map(&self) -> impl Iterator<Item = (&N, IdIter<'_>)> {
self.value_to_points.keys().map(move |k| {
(
k.borrow(),
Box::new(self.get_iterator(k.borrow())) as IdIter,
)
})
pub fn for_each_count_per_value(
&self,
mut f: impl FnMut(&N, usize) -> OperationResult<()>,
) -> OperationResult<()> {
self.value_to_points
.iter()
.try_for_each(|(k, entry)| f(k.borrow(), entry.count as usize))
}
pub fn for_each_value_map(
&self,
mut f: impl FnMut(&N, &mut dyn Iterator<Item = PointOffsetType>) -> OperationResult<()>,
) -> OperationResult<()> {
self.value_to_points
.iter()
.try_for_each(|(k, entry)| f(k.borrow(), &mut self.get_entry_iterator(entry)))
}
pub fn get_iterator(&self, value: &N) -> IdIter<'_> {
if let Some(entry) = self.value_to_points.get(value) {
let range = entry.range.start as usize..entry.range.end as usize;
let deleted_flags = self
.deleted_value_to_points_container
.iter()
.by_vals()
.skip(range.start)
.chain(std::iter::repeat(false));
let values = self.value_to_points_container[range]
.iter()
.zip(deleted_flags)
.filter(|(_, is_deleted)| !is_deleted)
.map(|(idx, _)| *idx);
Box::new(values)
Box::new(self.get_entry_iterator(entry))
} else {
Box::new(iter::empty::<PointOffsetType>())
}
}
fn get_entry_iterator(
&self,
entry: &ContainerSegment,
) -> impl Iterator<Item = PointOffsetType> {
let range = entry.range.start as usize..entry.range.end as usize;
let deleted_flags = self
.deleted_value_to_points_container
.iter()
.by_vals()
.skip(range.start)
.chain(std::iter::repeat(false));
self.value_to_points_container[range]
.iter()
.zip(deleted_flags)
.filter(|(_, is_deleted)| !is_deleted)
.map(|(idx, _)| *idx)
}
pub fn iter_values(&self) -> Box<dyn Iterator<Item = &N> + '_> {
Box::new(self.value_to_points.keys().map(|v| v.borrow()))
}
pub fn for_each_value(
&self,
mut f: impl FnMut(&N) -> OperationResult<()>,
) -> OperationResult<()> {
self.value_to_points.keys().try_for_each(|v| f(v.borrow()))
}
pub fn storage_type(&self) -> StorageType {
match &self.storage {
Storage::Mmap(index) => StorageType::Mmap {

79
lib/segment/src/index/field_index/map_index/mmap_map_index.rs
View File

@@ -21,7 +21,7 @@ use super::{IdIter, MapIndexKey};
use crate::common::Flusher;
use crate::common::buffered_update_bitslice::BufferedUpdateBitSlice;
use crate::common::operation_error::{OperationError, OperationResult};
use crate::index::field_index::stored_point_to_values::StoredPointToValues;
use crate::index::field_index::stored_point_to_values::{StoredPointToValues, ValuesIter};
const DELETED_PATH: &str = "deleted.bin";
const HASHMAP_PATH: &str = "values_to_points.bin";
@@ -235,6 +235,25 @@ impl<N: MapIndexKey + Key + ?Sized> MmapMapIndex<N> {
})
}
pub fn for_points_values(
&self,
mut points: impl Iterator<Item = PointOffsetType>,
hw_counter: &HardwareCounterCell,
mut f: impl FnMut(PointOffsetType, ValuesIter<'_, N>),
) -> OperationResult<()> {
let hw_counter = self.make_conditioned_counter(hw_counter);
points.try_for_each(|idx| {
if self.storage.deleted.get(idx as usize) != Some(false) {
return Ok(());
}
if let Some(iter) = self.storage.point_to_values.values_iter(idx, hw_counter)? {
f(idx, iter);
}
Ok(())
})
}
pub fn values_count(&self, idx: PointOffsetType) -> Option<usize> {
self.storage
.deleted
@@ -326,11 +345,19 @@ impl<N: MapIndexKey + Key + ?Sized> MmapMapIndex<N> {
self.storage.value_to_points.keys()
}
pub fn iter_counts_per_value(
pub fn for_each_value(
&self,
mut f: impl FnMut(&N) -> OperationResult<()>,
) -> OperationResult<()> {
self.storage.value_to_points.keys().try_for_each(&mut f)
}
pub fn for_each_count_per_value(
&self,
deferred_internal_id: Option<PointOffsetType>,
) -> impl Iterator<Item = (&N, usize)> + '_ {
self.storage.value_to_points.iter().map(move |(k, v)| {
mut f: impl FnMut(&N, usize) -> OperationResult<()>,
) -> OperationResult<()> {
self.storage.value_to_points.iter().try_for_each(|(k, v)| {
let count = v
.iter()
.filter(|&&idx| {
@@ -342,35 +369,37 @@ impl<N: MapIndexKey + Key + ?Sized> MmapMapIndex<N> {
})
.unique()
.count();
(k, count)
f(k, count)
})
}
pub fn iter_values_map<'a>(
pub fn for_each_value_map<'a>(
&'a self,
hw_counter: &'a HardwareCounterCell,
) -> impl Iterator<Item = (&'a N, IdIter<'a>)> + 'a {
mut f: impl FnMut(&'a N, &mut dyn Iterator<Item = PointOffsetType>) -> OperationResult<()> + 'a,
) -> OperationResult<()> {
let hw_counter = self.make_conditioned_counter(hw_counter);
self.storage.value_to_points.iter().map(move |(k, v)| {
hw_counter
.payload_index_io_read_counter()
.incr_delta(k.write_bytes());
self.storage
.value_to_points
.iter()
.try_for_each(move |(k, v)| {
hw_counter
.payload_index_io_read_counter()
.incr_delta(k.write_bytes());
(
k,
Box::new(
v.iter()
.copied()
.filter(|idx| !self.storage.deleted.get(*idx as usize).unwrap_or(true))
.measure_hw_with_acc(
hw_counter.new_accumulator(),
size_of::<PointOffsetType>(),
|i| i.payload_index_io_read_counter(),
),
) as IdIter,
)
})
let mut iter = v
.iter()
.copied()
.filter(|idx| !self.storage.deleted.get(*idx as usize).unwrap_or(true))
.measure_hw_with_acc(
hw_counter.new_accumulator(),
size_of::<PointOffsetType>(),
|i| i.payload_index_io_read_counter(),
);
f(k, &mut iter)
})
}
fn make_conditioned_counter<'a>(

89
lib/segment/src/index/field_index/map_index/mod.rs
View File

@@ -245,12 +245,21 @@ where
}
}
pub fn iter_counts_per_value(
pub fn for_each_value(&self, f: impl FnMut(&N) -> OperationResult<()>) -> OperationResult<()> {
match self {
MapIndex::Mutable(index) => index.for_each_value(f),
MapIndex::Immutable(index) => index.for_each_value(f),
MapIndex::Mmap(index) => index.for_each_value(f),
}
}
pub fn for_each_count_per_value(
&self,
deferred_internal_id: Option<PointOffsetType>,
) -> Box<dyn Iterator<Item = (&N, usize)> + '_> {
f: impl FnMut(&N, usize) -> OperationResult<()>,
) -> OperationResult<()> {
match self {
MapIndex::Mutable(index) => Box::new(index.iter_counts_per_value(deferred_internal_id)),
MapIndex::Mutable(index) => index.for_each_count_per_value(deferred_internal_id, f),
// Two reasons we don't implement deferred filtering here:
// - We don't have both deferred points and an immutable index.
@@ -258,21 +267,22 @@ where
// it doesn't work well in combination with the way it handles deletions.
MapIndex::Immutable(index) => {
debug_assert!(deferred_internal_id.is_none());
Box::new(index.iter_counts_per_value())
index.for_each_count_per_value(f)
}
MapIndex::Mmap(index) => Box::new(index.iter_counts_per_value(deferred_internal_id)),
MapIndex::Mmap(index) => index.for_each_count_per_value(deferred_internal_id, f),
}
}
pub fn iter_values_map<'a>(
&'a self,
hw_cell: &'a HardwareCounterCell,
) -> Box<dyn Iterator<Item = (&'a N, IdIter<'a>)> + 'a> {
pub fn for_each_value_map(
&self,
hw_cell: &HardwareCounterCell,
f: impl FnMut(&N, &mut dyn Iterator<Item = PointOffsetType>) -> OperationResult<()>,
) -> OperationResult<()> {
match self {
MapIndex::Mutable(index) => Box::new(index.iter_values_map()),
MapIndex::Immutable(index) => Box::new(index.iter_values_map()),
MapIndex::Mmap(index) => Box::new(index.iter_values_map(hw_cell)),
MapIndex::Mutable(index) => index.for_each_value_map(f),
MapIndex::Immutable(index) => index.for_each_value_map(f),
MapIndex::Mmap(index) => index.for_each_value_map(hw_cell, f),
}
}
@@ -1214,38 +1224,55 @@ where
for<'a> Cow<'a, N>: Into<FacetValueRef<'a>>,
for<'a> &'a N: Into<FacetValueRef<'a>>,
{
fn get_point_values(
fn for_points_values(
&self,
point_id: PointOffsetType,
points: impl Iterator<Item = PointOffsetType>,
hw_counter: &HardwareCounterCell,
) -> impl Iterator<Item = FacetValueRef<'_>> + '_ {
MapIndex::get_values(self, point_id, hw_counter)
.into_iter()
.flatten()
.map(Into::into)
mut f: impl FnMut(PointOffsetType, &mut dyn Iterator<Item = FacetValueRef<'_>>),
) -> OperationResult<()> {
match self {
MapIndex::Mutable(index) => index.for_points_values(points, |idx, slice| {
f(idx, &mut slice.iter().map(|v| v.borrow().into()));
}),
MapIndex::Immutable(index) => index.for_points_values(points, |idx, slice| {
f(idx, &mut slice.iter().map(|v| v.borrow().into()));
}),
MapIndex::Mmap(index) => index.for_points_values(points, hw_counter, |idx, vals| {
f(idx, &mut vals.map(|v| v.into()));
})?,
}
Ok(())
}
fn iter_values(&self) -> impl Iterator<Item = FacetValueRef<'_>> + '_ {
self.iter_values().map(Into::into)
fn for_each_value(
&self,
mut f: impl FnMut(FacetValueRef<'_>) -> OperationResult<()>,
) -> OperationResult<()> {
self.for_each_value(|v| f(v.into()))
}
fn iter_values_map<'a>(
&'a self,
hw_counter: &'a HardwareCounterCell,
) -> impl Iterator<Item = (FacetValueRef<'a>, IdIter<'a>)> + 'a {
self.iter_values_map(hw_counter)
.map(|(k, iter)| (k.into(), iter))
fn for_each_value_map(
&self,
hw_counter: &HardwareCounterCell,
mut f: impl FnMut(
FacetValueRef<'_>,
&mut dyn Iterator<Item = PointOffsetType>,
) -> OperationResult<()>,
) -> OperationResult<()> {
self.for_each_value_map(hw_counter, |value, iter| f(value.into(), iter))
}
fn iter_counts_per_value(
fn for_each_count_per_value(
&self,
deferred_internal_id: Option<PointOffsetType>,
) -> impl Iterator<Item = FacetHit<FacetValueRef<'_>>> + '_ {
self.iter_counts_per_value(deferred_internal_id)
.map(|(value, count)| FacetHit {
mut f: impl FnMut(FacetHit<FacetValueRef<'_>>) -> OperationResult<()>,
) -> OperationResult<()> {
self.for_each_count_per_value(deferred_internal_id, |value, count| {
f(FacetHit {
value: value.into(),
count,
})
})
}
}

49
lib/segment/src/index/field_index/map_index/mutable_map_index.rs
View File

@@ -275,25 +275,39 @@ where
self.map.get(value).map(|p| p.len() as usize)
}
pub fn iter_counts_per_value(
pub fn for_points_values(
&self,
deferred_internal_id: Option<PointOffsetType>,
) -> impl Iterator<Item = (&N, usize)> + '_ {
self.map
.iter()
.map(move |(k, v)| match deferred_internal_id {
Some(deferred_internal_id) => {
let count = v.range_cardinality(..deferred_internal_id) as usize;
(k.borrow(), count)
}
None => (k.borrow(), v.len() as usize),
})
points: impl Iterator<Item = PointOffsetType>,
mut f: impl FnMut(PointOffsetType, &[<N as MapIndexKey>::Owned]),
) {
points.for_each(|idx| {
if let Some(values) = self.point_to_values.get(idx as usize) {
f(idx, values);
}
});
}
pub fn iter_values_map(&self) -> impl Iterator<Item = (&N, IdIter<'_>)> {
pub fn for_each_count_per_value(
&self,
deferred_internal_id: Option<PointOffsetType>,
mut f: impl FnMut(&N, usize) -> OperationResult<()>,
) -> OperationResult<()> {
self.map.iter().try_for_each(|(k, v)| {
let count = match deferred_internal_id {
Some(deferred_internal_id) => v.range_cardinality(..deferred_internal_id) as usize,
None => v.len() as usize,
};
f(k.borrow(), count)
})
}
pub fn for_each_value_map(
&self,
mut f: impl FnMut(&N, &mut dyn Iterator<Item = PointOffsetType>) -> OperationResult<()>,
) -> OperationResult<()> {
self.map
.iter()
.map(move |(k, v)| (k.borrow(), Box::new(v.iter()) as IdIter))
.try_for_each(|(k, v)| f(k.borrow(), &mut v.iter()))
}
pub fn get_iterator(&self, value: &N) -> IdIter<'_> {
@@ -307,6 +321,13 @@ where
Box::new(self.map.keys().map(|v| v.borrow()))
}
pub fn for_each_value(
&self,
mut f: impl FnMut(&N) -> OperationResult<()>,
) -> OperationResult<()> {
self.map.keys().try_for_each(|v| f(v.borrow()))
}
pub fn storage_type(&self) -> StorageType {
match &self.storage {
Storage::Gridstore(_) => StorageType::Gridstore,

139
lib/segment/src/segment/facet.rs
View File

@@ -2,13 +2,12 @@ use std::collections::{BTreeSet, HashMap};
use std::sync::atomic::AtomicBool;
use common::counter::hardware_counter::HardwareCounterCell;
use common::iterator_ext::IteratorExt;
use common::types::PointOffsetType;
use itertools::{Either, Itertools};
use itertools::Itertools;
use super::Segment;
use crate::common::operation_error::OperationResult;
use crate::data_types::facets::{FacetHit, FacetParams, FacetValue};
use crate::common::operation_error::{OperationResult, check_process_stopped};
use crate::data_types::facets::{FacetParams, FacetValue};
use crate::entry::ReadSegmentEntry;
use crate::id_tracker::IdTracker;
use crate::index::PayloadIndex;
@@ -34,7 +33,13 @@ impl Segment {
let facet_index = payload_index.get_facet_index(&request.key)?;
let context;
let hits_iter = if let Some(filter) = &request.filter {
// We can't just select top values, because we need to aggregate across segments,
// which we can't assume to select the same best top.
//
// We need all values to be able to aggregate correctly across segments
let mut hits = HashMap::new();
if let Some(filter) = &request.filter {
let id_tracker = self.id_tracker.borrow();
let filter_cardinality = payload_index.estimate_cardinality(filter, hw_counter)?;
@@ -48,11 +53,11 @@ impl Segment {
// - a collection with almost a unique key per point
let use_iterative_approach = percentage_filtered < 0.3;
let iter = if use_iterative_approach {
if use_iterative_approach {
// go over the filtered points and aggregate the values
// aka. read from other indexes
let point_mappings = id_tracker.point_mappings();
let iter = payload_index
let points = payload_index
.iter_filtered_points(
filter,
&id_tracker,
@@ -62,20 +67,12 @@ 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
})
.into_iter()
.map(|(value, count)| FacetHit { value, count });
Either::Left(iter)
.filter(|&point_id| !id_tracker.is_deleted_point(point_id));
facet_index.for_points_values(points, hw_counter, |_point_id, iter| {
iter.unique().for_each(|value| {
*hits.entry(value.to_owned()).or_insert(0) += 1;
});
})?;
} else {
// go over the values and filter the points
// aka. read from facet index
@@ -83,52 +80,43 @@ impl Segment {
// This is more similar to a full-scan, but we won't be hashing so many times.
context = payload_index.struct_filtered_context(filter, hw_counter)?;
let iter = facet_index
.iter_values_map(hw_counter)
.stop_if(is_stopped)
.filter_map(|(value, iter)| {
#[cfg(debug_assertions)]
let iter = {
let mut prev_id = None;
iter.inspect(move |&id| {
let previous = prev_id.get_or_insert(id);
debug_assert!(*previous <= id, "Sorted iter assertion broken");
*previous = id;
})
};
let max_id = self.deferred_internal_id().unwrap_or(PointOffsetType::MAX);
let count = iter
.dedup()
.take_while(|&point_id| {
point_id
< self.deferred_internal_id().unwrap_or(PointOffsetType::MAX)
})
.filter(|&point_id| context.check(point_id))
.count();
facet_index.for_each_value_map(hw_counter, |value, iter| {
check_process_stopped(is_stopped)?;
(count > 0).then_some(FacetHit { value, count })
});
#[cfg(debug_assertions)]
let iter = {
let mut prev_id = None;
iter.inspect(move |&id| {
let previous = prev_id.get_or_insert(id);
debug_assert!(*previous <= id, "Sorted iter assertion broken");
*previous = id;
})
};
Either::Right(iter)
};
Either::Left(iter)
let count = iter
.dedup()
.take_while(|&point_id| point_id < max_id)
.filter(|&point_id| context.check(point_id))
.count();
if count > 0 {
hits.insert(value.to_owned(), count);
}
Ok(())
})?;
}
} else {
// just count how many points each value has
let iter = facet_index
.iter_counts_per_value(self.deferred_internal_id())
.stop_if(is_stopped)
.filter(|hit| hit.count > 0);
Either::Right(iter)
};
// We can't just select top values, because we need to aggregate across segments,
// which we can't assume to select the same best top.
//
// We need all values to be able to aggregate correctly across segments
let hits: HashMap<_, _> = hits_iter
.map(|hit| (hit.value.to_owned(), hit.count))
.collect();
facet_index.for_each_count_per_value(self.deferred_internal_id(), |hit| {
check_process_stopped(is_stopped)?;
if hit.count > 0 {
hits.insert(hit.value.to_owned(), hit.count);
}
Ok(())
})?;
}
Ok(hits)
}
@@ -143,13 +131,14 @@ impl Segment {
let payload_index = self.payload_index.borrow();
let facet_index = payload_index.get_facet_index(key)?;
let mut values = BTreeSet::new();
let values = if let Some(filter) = filter {
if let Some(filter) = filter {
let id_tracker = self.id_tracker.borrow();
let filter_cardinality = payload_index.estimate_cardinality(filter, hw_counter)?;
let point_mappings = id_tracker.point_mappings();
payload_index
let points = payload_index
.iter_filtered_points(
filter,
&id_tracker,
@@ -159,20 +148,16 @@ 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
})
.into_iter()
.map(|value| value.to_owned())
.collect()
.filter(|&point_id| !id_tracker.is_deleted_point(point_id));
facet_index.for_points_values(points, hw_counter, |_point_id, iter| {
values.extend(iter.map(|v| v.to_owned()));
})?;
} else {
facet_index
.iter_values(hw_counter, self.deferred_internal_id())
.stop_if(is_stopped)
.map(|value_ref| value_ref.to_owned())
.collect()
facet_index.for_each_value(hw_counter, self.deferred_internal_id(), |value_ref| {
check_process_stopped(is_stopped)?;
values.insert(value_ref.to_owned());
Ok(())
})?;
};
Ok(values)