use std::{future::Future, sync::Arc};
use diskann_utils::{Reborrow, future::SendFuture};
use diskann_vector::DistanceFunction;
use futures_util::FutureExt;
use crate::{
ANNError, ANNResult,
error::StandardError,
graph::{SearchOutputBuffer, workingset},
neighbor::Neighbor,
provider::{self, DataProvider, HasId},
};
pub trait SearchAccessor: HasId + Send + Sync {
fn starting_points(&self)
-> impl std::future::Future<Output = ANNResult<Vec<Self::Id>>> + Send;
fn start_point_distances<F>(
&mut self,
f: F,
) -> impl std::future::Future<Output = ANNResult<()>> + Send
where
F: FnMut(Self::Id, f32) + Send;
fn expand_beam<Itr, P, F>(
&mut self,
ids: Itr,
pred: P,
on_neighbors: F,
) -> impl std::future::Future<Output = ANNResult<()>> + Send
where
Itr: Iterator<Item = Self::Id> + Send,
P: HybridPredicate<Self::Id> + Send + Sync,
F: FnMut(Self::Id, f32) + Send;
fn terminate_early(&mut self) -> bool {
false
}
fn is_not_start_point(
&self,
) -> impl std::future::Future<
Output = ANNResult<impl Fn(Self::Id) -> bool + Send + Sync + 'static>,
> + Send {
async move {
let set: std::collections::HashSet<_> =
self.starting_points().await?.into_iter().collect();
Ok(move |id| !set.contains(&id))
}
}
fn num_starting_points(&self) -> impl std::future::Future<Output = ANNResult<usize>> + Send {
self.starting_points()
.map(|result: ANNResult<_>| result.map(|v: Vec<_>| v.len()))
}
}
pub trait Predicate<T> {
fn eval(&self, item: &T) -> bool;
}
pub trait PredicateMut<T> {
fn eval_mut(&mut self, item: &T) -> bool;
}
pub trait HybridPredicate<T>: Predicate<T> + PredicateMut<T> {}
pub struct NotInMut<'a, K>(&'a mut hashbrown::HashSet<K>);
impl<'a, K> NotInMut<'a, K> {
pub fn new(set: &'a mut hashbrown::HashSet<K>) -> Self {
Self(set)
}
}
impl<T> Predicate<T> for NotInMut<'_, T>
where
T: Eq + std::hash::Hash,
{
fn eval(&self, item: &T) -> bool {
!self.0.contains(item)
}
}
impl<T> PredicateMut<T> for NotInMut<'_, T>
where
T: Clone + Eq + std::hash::Hash,
{
fn eval_mut(&mut self, item: &T) -> bool {
self.0.insert(item.clone())
}
}
impl<T> HybridPredicate<T> for NotInMut<'_, T> where T: Clone + Eq + std::hash::Hash {}
pub trait SearchStrategy<'a, Provider, T>: Send + Sync
where
Provider: DataProvider,
{
type SearchAccessorError: StandardError;
type SearchAccessor: SearchAccessor<Id = Provider::InternalId>;
fn search_accessor(
&'a self,
provider: &'a Provider,
context: &'a Provider::Context,
query: T,
) -> Result<Self::SearchAccessor, Self::SearchAccessorError>;
}
pub trait DefaultPostProcessor<'a, Provider, T, O = <Provider as DataProvider>::InternalId>:
SearchStrategy<'a, Provider, T>
where
Provider: DataProvider,
O: Send,
{
type Processor: SearchPostProcess<Self::SearchAccessor, T, O> + Send + Sync;
fn default_post_processor(&'a self) -> Self::Processor;
}
pub trait DefaultSearchStrategy<'a, Provider, T, O = <Provider as DataProvider>::InternalId>:
SearchStrategy<'a, Provider, T> + DefaultPostProcessor<'a, Provider, T, O>
where
Provider: DataProvider,
O: Send,
{
}
impl<'a, S, Provider, T, O> DefaultSearchStrategy<'a, Provider, T, O> for S
where
S: SearchStrategy<'a, Provider, T> + DefaultPostProcessor<'a, Provider, T, O>,
Provider: DataProvider,
O: Send,
{
}
#[macro_export]
macro_rules! default_post_processor {
($Processor:ty) => {
type Processor = $Processor;
fn default_post_processor(&self) -> Self::Processor {
Default::default()
}
};
}
pub trait SearchPostProcess<A, T, O = <A as HasId>::Id>
where
A: HasId,
{
type Error: StandardError;
fn post_process<I, B>(
&self,
accessor: &mut A,
query: T,
candidates: I,
output: &mut B,
) -> impl std::future::Future<Output = Result<usize, Self::Error>> + Send
where
I: Iterator<Item = Neighbor<A::Id>> + Send,
B: SearchOutputBuffer<O> + Send + ?Sized;
}
#[derive(Debug, Default, Clone, Copy)]
pub struct CopyIds;
impl<A, T> SearchPostProcess<A, T> for CopyIds
where
A: HasId,
{
type Error = std::convert::Infallible;
fn post_process<I, B>(
&self,
_accessor: &mut A,
_query: T,
candidates: I,
output: &mut B,
) -> impl std::future::Future<Output = Result<usize, Self::Error>> + Send
where
I: Iterator<Item = Neighbor<A::Id>> + Send,
B: SearchOutputBuffer<A::Id> + Send + ?Sized,
{
let count = output.extend(candidates.map(|n| (n.id, n.distance)));
std::future::ready(Ok(count))
}
}
pub trait SearchPostProcessStep<A, T, O = <A as HasId>::Id>
where
A: HasId,
{
type Error<NextError>: StandardError
where
NextError: StandardError;
type NextAccessor: HasId<Id = A::Id>;
fn post_process_step<I, B, Next>(
&self,
next: &Next,
accessor: &mut A,
query: T,
candidates: I,
output: &mut B,
) -> impl std::future::Future<Output = Result<usize, Self::Error<Next::Error>>> + Send
where
I: Iterator<Item = Neighbor<A::Id>> + Send,
B: SearchOutputBuffer<O> + Send + ?Sized,
Next: SearchPostProcess<Self::NextAccessor, T, O> + Sync;
}
#[derive(Debug, Default, Clone, Copy)]
pub struct FilterStartPoints;
impl<A, T, O> SearchPostProcessStep<A, T, O> for FilterStartPoints
where
A: SearchAccessor,
T: Copy + Send + Sync,
{
type Error<NextError>
= ANNError
where
NextError: StandardError;
type NextAccessor = A;
async fn post_process_step<I, B, Next>(
&self,
next: &Next,
accessor: &mut A,
query: T,
candidates: I,
output: &mut B,
) -> ANNResult<usize>
where
I: Iterator<Item = Neighbor<A::Id>> + Send,
B: SearchOutputBuffer<O> + Send + ?Sized,
Next: SearchPostProcess<A, T, O> + Sync,
{
let filter = accessor.is_not_start_point().await?;
next.post_process(accessor, query, candidates.filter(|n| filter(n.id)), output)
.await
.map_err(|err| {
let err = err.into();
err.context("after filtering start points")
})
}
}
#[derive(Debug, Default, Clone, Copy)]
pub struct Pipeline<Head, Tail> {
head: Head,
tail: Tail,
}
impl<Head, Tail> Pipeline<Head, Tail> {
pub fn new(head: Head, tail: Tail) -> Self {
Self { head, tail }
}
}
impl<A, T, O, Head, Tail> SearchPostProcess<A, T, O> for Pipeline<Head, Tail>
where
A: HasId,
Head: SearchPostProcessStep<A, T, O>,
Tail: SearchPostProcess<Head::NextAccessor, T, O> + Sync,
{
type Error = Head::Error<Tail::Error>;
fn post_process<I, B>(
&self,
accessor: &mut A,
query: T,
candidates: I,
output: &mut B,
) -> impl std::future::Future<Output = Result<usize, Self::Error>> + Send
where
I: Iterator<Item = Neighbor<A::Id>> + Send,
B: SearchOutputBuffer<O> + Send + ?Sized,
{
self.head
.post_process_step(&self.tail, accessor, query, candidates, output)
}
}
pub trait PruneAccessor: HasId + Send + Sync {
type Neighbors<'a>: provider::NeighborAccessorMut<Id = Self::Id>
where
Self: 'a;
type ElementRef<'a>;
type View<'a>: for<'x> workingset::View<Self::Id, ElementRef<'x> = Self::ElementRef<'x>>
+ Send
+ Sync
where
Self: 'a;
type Distance<'a>: for<'x, 'y> DistanceFunction<Self::ElementRef<'x>, Self::ElementRef<'y>, f32>
+ Send
+ Sync
where
Self: 'a;
fn neighbors(&mut self) -> Self::Neighbors<'_>;
fn fill<Itr>(
&mut self,
itr: Itr,
) -> impl SendFuture<ANNResult<(Self::View<'_>, Self::Distance<'_>)>>
where
Itr: ExactSizeIterator<Item = Self::Id> + Clone + Send + Sync;
}
pub trait InsertStrategy<'a, Provider, T>: SearchStrategy<'a, Provider, T> + 'static
where
Provider: DataProvider,
{
type PruneStrategy: PruneStrategy<Provider>;
fn prune_strategy(&self) -> Self::PruneStrategy;
fn insert_search_accessor(
&'a self,
provider: &'a Provider,
context: &'a Provider::Context,
vector: T,
) -> Result<Self::SearchAccessor, Self::SearchAccessorError> {
self.search_accessor(provider, context, vector)
}
}
pub trait PruneStrategy<Provider>: Send + Sync + 'static
where
Provider: DataProvider,
{
type PruneAccessor<'a>: PruneAccessor<Id = Provider::InternalId>;
type PruneAccessorError: StandardError;
fn prune_accessor<'a>(
&'a self,
provider: &'a Provider,
context: &'a Provider::Context,
capacity: usize,
) -> Result<Self::PruneAccessor<'a>, Self::PruneAccessorError>;
}
pub trait MultiInsertStrategy<Provider, B>: Send + Sync + 'static
where
Provider: DataProvider,
B: Batch,
{
type Seed: Send + Sync + 'static;
type FinishError: Into<ANNError> + std::fmt::Debug + Send + Sync;
type PruneStrategy: PruneStrategy<Provider>;
type InsertStrategy: for<'a> InsertStrategy<'a, Provider, B::Element<'a>, PruneStrategy = Self::PruneStrategy>;
fn insert_strategy(&self) -> Self::InsertStrategy;
fn finish<Itr>(
&self,
provider: &Provider,
context: &Provider::Context,
batch: &Arc<B>,
ids: Itr,
) -> impl std::future::Future<Output = Result<Self::Seed, Self::FinishError>> + Send
where
Itr: ExactSizeIterator<Item = Provider::InternalId> + Send;
fn seeded_prune_accessor<'a>(
&'a self,
provider: &'a Provider,
context: &'a Provider::Context,
seed: &'a Self::Seed,
capacity: usize,
) -> ANNResult<<Self::PruneStrategy as PruneStrategy<Provider>>::PruneAccessor<'a>>;
}
pub trait Batch: Send + Sync + 'static {
type Element<'a>: Copy;
fn len(&self) -> usize;
fn get(&self, i: usize) -> Self::Element<'_>;
fn is_empty(&self) -> bool {
self.len() == 0
}
}
impl<T: Send + Sync + 'static> Batch for diskann_utils::views::Matrix<T> {
type Element<'a> = &'a [T];
fn len(&self) -> usize {
self.nrows()
}
fn get(&self, i: usize) -> Self::Element<'_> {
self.row(i)
}
}
pub trait InplaceDeleteStrategy<Provider>: Send + Sync + 'static
where
Provider: DataProvider,
{
type DeleteElement<'a>: Copy + Send + Sync;
type DeleteElementGuard: Send
+ Sync
+ for<'a> Reborrow<'a, Target = Self::DeleteElement<'a>>
+ 'static;
type DeleteElementError: StandardError;
type PruneStrategy: PruneStrategy<Provider>;
type DeleteSearchAccessor<'a>: SearchAccessor<Id = Provider::InternalId>;
type SearchPostProcessor: for<'a> SearchPostProcess<Self::DeleteSearchAccessor<'a>, Self::DeleteElement<'a>>
+ Send
+ Sync;
type SearchStrategy: for<'a> SearchStrategy<
'a,
Provider,
Self::DeleteElement<'a>,
SearchAccessor = Self::DeleteSearchAccessor<'a>,
>;
fn prune_strategy(&self) -> Self::PruneStrategy;
fn search_strategy(&self) -> Self::SearchStrategy;
fn search_post_processor(&self) -> Self::SearchPostProcessor;
fn get_delete_element<'a>(
&'a self,
provider: &'a Provider,
context: &'a Provider::Context,
id: Provider::InternalId,
) -> impl Future<Output = Result<Self::DeleteElementGuard, Self::DeleteElementError>> + Send;
}