qdrant_edge/shard/operations/

point_ops.rs

use std::collections::{HashMap, HashSet};
use std::fmt::{Debug, Formatter};
use std::hash::{Hash, Hasher};
use std::mem;

use crate::common::validation::validate_multi_vector;
use itertools::Itertools as _;
use ordered_float::OrderedFloat;
use schemars::JsonSchema;
use crate::segment::common::operation_error::OperationError;
use crate::segment::common::utils::unordered_hash_unique;
use crate::segment::data_types::named_vectors::NamedVectors;
use crate::segment::data_types::segment_record::SegmentRecord;
use crate::segment::data_types::vectors::{
    BatchVectorStructInternal, DEFAULT_VECTOR_NAME, DenseVector, MultiDenseVector,
    MultiDenseVectorInternal, VectorInternal, VectorStructInternal,
};
use crate::segment::types::{Filter, Payload, PointIdType, VectorNameBuf};
use serde::{Deserialize, Serialize};
use crate::sparse::common::types::{DimId, DimWeight};
use strum::{EnumDiscriminants, EnumIter};
use validator::{Validate, ValidationErrors};

/// Defines the mode of the upsert operation
///
/// * `Upsert` - default mode, insert new points, update existing points
/// * `InsertOnly` - only insert new points, do not update existing points
/// * `UpdateOnly` - only update existing points, do not insert new points
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Deserialize, Serialize, Hash)]
#[serde(rename_all = "snake_case")]
pub enum UpdateMode {
    // Default mode - insert new points, update existing points
    #[default]
    Upsert,
    // Only insert new points, do not update existing points
    InsertOnly,
    // Only update existing points, do not insert new points
    UpdateOnly,
}

#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, JsonSchema, Validate, Hash)]
#[serde(rename_all = "snake_case")]
pub struct PointIdsList {
    pub points: Vec<PointIdType>,
    #[cfg(feature = "api")]
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub shard_key: Option<api::rest::ShardKeySelector>,
}

impl From<Vec<PointIdType>> for PointIdsList {
    fn from(points: Vec<PointIdType>) -> Self {
        Self {
            points,
            #[cfg(feature = "api")]
            shard_key: None,
        }
    }
}

// General idea of having an extra layer of data structures after REST and gRPC
// is to ensure that all vectors are inferenced and validated before they are persisted.
//
// This separation allows to have a single point, enforced by the type system,
// where all Documents and other inference-able objects are resolved into raw vectors.
//
// Separation between VectorStructPersisted and VectorStructInternal is only needed
// for legacy reasons, as the previous implementations wrote VectorStruct to WAL,
// so we need an ability to read it back. VectorStructPersisted reproduces the same
// structure as VectorStruct had in the previous versions.
//
//
//        gRPC              REST API           ┌───┐              WAL
//          │                  │               │ I │               ▲
//          │                  │               │ n │               │
//          │                  │               │ f │               │
//  ┌───────▼───────┐    ┌─────▼──────┐        │ e │     ┌─────────┴───────────┐
//  │ grpc::Vectors ├───►│VectorStruct├───────►│ r ├────►│VectorStructPersisted├─────┐
//  └───────────────┘    └────────────┘        │ e │     └─────────────────────┘     │
//                        Vectors              │ n │      Only Vectors               │
//                        + Documents          │ c │                                 │
//                        + Images             │ e │                                 │
//                        + Other inference    └───┘                                 │
//                        Implement JsonSchema                                       │
//                                                       ┌─────────────────────┐     │
//                                                       │                     ◄─────┘
//                                                       │   Storage           │
//                                                       │                     │
//                        REST API Response              └────────┬────────────┘
//                             ▲                                  │
//                             │                                  │
//                      ┌──────┴──────────────┐         ┌─────────▼───────────┐
//                      │ VectorStructOutput  ◄───┬─────┤VectorStructInternal │
//                      └─────────────────────┘   │     └─────────────────────┘
//                       Only Vectors             │      Only Vectors
//                       Implement JsonSchema     │      Optimized for search
//                                                │
//                                                │
//                      ┌─────────────────────┐   │
//                      │ grpc::VectorsOutput ◄───┘
//                      └───────────┬─────────┘
//                                  │
//                                  ▼
//                              gPRC Response

#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, EnumDiscriminants, Hash)]
#[strum_discriminants(derive(EnumIter))]
#[serde(rename_all = "snake_case")]
pub enum PointOperations {
    /// Insert or update points
    UpsertPoints(PointInsertOperationsInternal),
    /// Insert points, or update existing points if condition matches
    UpsertPointsConditional(ConditionalInsertOperationInternal),
    /// Delete point if exists
    DeletePoints { ids: Vec<PointIdType> },
    /// Delete points by given filter criteria
    DeletePointsByFilter(Filter),
    /// Points Sync
    SyncPoints(PointSyncOperation),
}

impl PointOperations {
    pub fn point_ids(&self) -> Option<Vec<PointIdType>> {
        match self {
            Self::UpsertPoints(op) => Some(op.point_ids()),
            Self::UpsertPointsConditional(op) => Some(op.points_op.point_ids()),
            Self::DeletePoints { ids } => Some(ids.clone()),
            Self::DeletePointsByFilter(_) => None,
            Self::SyncPoints(op) => Some(op.points.iter().map(|point| point.id).collect()),
        }
    }

    pub fn retain_point_ids<F>(&mut self, filter: F)
    where
        F: Fn(&PointIdType) -> bool,
    {
        match self {
            Self::UpsertPoints(op) => op.retain_point_ids(filter),
            Self::UpsertPointsConditional(op) => {
                op.points_op.retain_point_ids(filter);
            }
            Self::DeletePoints { ids } => ids.retain(filter),
            Self::DeletePointsByFilter(_) => (),
            Self::SyncPoints(op) => op.points.retain(|point| filter(&point.id)),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, EnumDiscriminants, Hash)]
#[strum_discriminants(derive(EnumIter))]
#[serde(rename_all = "snake_case")]
pub enum PointInsertOperationsInternal {
    /// Inset points from a batch.
    #[serde(rename = "batch")]
    PointsBatch(BatchPersisted),
    /// Insert points from a list
    #[serde(rename = "points")]
    PointsList(Vec<PointStructPersisted>),
}

impl PointInsertOperationsInternal {
    pub fn point_ids(&self) -> Vec<PointIdType> {
        match self {
            Self::PointsBatch(batch) => batch.ids.clone(),
            Self::PointsList(points) => points.iter().map(|point| point.id).collect(),
        }
    }

    pub fn into_point_vec(self) -> Vec<PointStructPersisted> {
        match self {
            PointInsertOperationsInternal::PointsBatch(batch) => {
                let batch_vectors = BatchVectorStructInternal::from(batch.vectors);
                let all_vectors = batch_vectors.into_all_vectors(batch.ids.len());
                let vectors_iter = batch.ids.into_iter().zip(all_vectors);
                match batch.payloads {
                    None => vectors_iter
                        .map(|(id, vectors)| PointStructPersisted {
                            id,
                            vector: VectorStructInternal::from(vectors).into(),
                            payload: None,
                        })
                        .collect(),
                    Some(payloads) => vectors_iter
                        .zip(payloads)
                        .map(|((id, vectors), payload)| PointStructPersisted {
                            id,
                            vector: VectorStructInternal::from(vectors).into(),
                            payload,
                        })
                        .collect(),
                }
            }
            PointInsertOperationsInternal::PointsList(points) => points,
        }
    }

    pub fn retain_point_ids<F>(&mut self, filter: F)
    where
        F: Fn(&PointIdType) -> bool,
    {
        match self {
            Self::PointsBatch(batch) => {
                let mut retain_indices = HashSet::new();

                retain_with_index(&mut batch.ids, |index, id| {
                    if filter(id) {
                        retain_indices.insert(index);
                        true
                    } else {
                        false
                    }
                });

                match &mut batch.vectors {
                    BatchVectorStructPersisted::Single(vectors) => {
                        retain_with_index(vectors, |index, _| retain_indices.contains(&index));
                    }

                    BatchVectorStructPersisted::MultiDense(vectors) => {
                        retain_with_index(vectors, |index, _| retain_indices.contains(&index));
                    }

                    BatchVectorStructPersisted::Named(vectors) => {
                        for (_, vectors) in vectors.iter_mut() {
                            retain_with_index(vectors, |index, _| retain_indices.contains(&index));
                        }
                    }
                }

                if let Some(payload) = &mut batch.payloads {
                    retain_with_index(payload, |index, _| retain_indices.contains(&index));
                }
            }

            Self::PointsList(points) => points.retain(|point| filter(&point.id)),
        }
    }
}

impl From<BatchPersisted> for PointInsertOperationsInternal {
    fn from(batch: BatchPersisted) -> Self {
        PointInsertOperationsInternal::PointsBatch(batch)
    }
}

impl From<Vec<PointStructPersisted>> for PointInsertOperationsInternal {
    fn from(points: Vec<PointStructPersisted>) -> Self {
        PointInsertOperationsInternal::PointsList(points)
    }
}

#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, Hash)]
pub struct ConditionalInsertOperationInternal {
    pub points_op: PointInsertOperationsInternal,
    /// Condition to check, if the point already exists
    pub condition: Filter,
    /// Mode of the upsert operation. If None, defaults to Upsert behavior.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub update_mode: Option<UpdateMode>,
}

#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, Hash)]
pub struct PointSyncOperation {
    /// Minimal id of the sync range
    pub from_id: Option<PointIdType>,
    /// Maximal id og
    pub to_id: Option<PointIdType>,
    pub points: Vec<PointStructPersisted>,
}

#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, Hash)]
#[serde(rename_all = "snake_case")]
pub struct BatchPersisted {
    pub ids: Vec<PointIdType>,
    pub vectors: BatchVectorStructPersisted,
    pub payloads: Option<Vec<Option<Payload>>>,
}

#[cfg(feature = "api")]
impl TryFrom<BatchPersisted> for Vec<api::grpc::qdrant::PointStruct> {
    type Error = tonic::Status;

    fn try_from(batch: BatchPersisted) -> Result<Self, Self::Error> {
        let BatchPersisted {
            ids,
            vectors,
            payloads,
        } = batch;
        let mut points = Vec::with_capacity(ids.len());
        let batch_vectors = BatchVectorStructInternal::from(vectors);
        let all_vectors = batch_vectors.into_all_vectors(ids.len());
        for (i, p_id) in ids.into_iter().enumerate() {
            let id = Some(p_id.into());
            let vector = all_vectors.get(i).cloned();
            let payload = payloads.as_ref().and_then(|payloads| {
                payloads.get(i).map(|payload| match payload {
                    None => HashMap::new(),
                    Some(payload) => api::conversions::json::payload_to_proto(payload.clone()),
                })
            });
            let vectors: Option<VectorStructInternal> = vector.map(|v| v.into());

            let point = api::grpc::qdrant::PointStruct {
                id,
                vectors: vectors.map(api::grpc::qdrant::Vectors::from),
                payload: payload.unwrap_or_default(),
            };
            points.push(point);
        }

        Ok(points)
    }
}

#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
#[serde(untagged, rename_all = "snake_case")]
pub enum BatchVectorStructPersisted {
    Single(Vec<DenseVector>),
    MultiDense(Vec<MultiDenseVector>),
    Named(HashMap<VectorNameBuf, Vec<VectorPersisted>>),
}

impl Hash for BatchVectorStructPersisted {
    fn hash<H: Hasher>(&self, state: &mut H) {
        mem::discriminant(self).hash(state);
        match self {
            BatchVectorStructPersisted::Single(dense) => {
                for vector in dense {
                    for v in vector {
                        OrderedFloat(*v).hash(state);
                    }
                }
            }
            BatchVectorStructPersisted::MultiDense(multidense) => {
                for vector in multidense {
                    for v in vector {
                        for element in v {
                            OrderedFloat(*element).hash(state);
                        }
                    }
                }
            }
            BatchVectorStructPersisted::Named(named) => unordered_hash_unique(state, named.iter()),
        }
    }
}

impl From<BatchVectorStructPersisted> for BatchVectorStructInternal {
    fn from(value: BatchVectorStructPersisted) -> Self {
        match value {
            BatchVectorStructPersisted::Single(vector) => BatchVectorStructInternal::Single(vector),
            BatchVectorStructPersisted::MultiDense(vectors) => {
                BatchVectorStructInternal::MultiDense(
                    vectors
                        .into_iter()
                        .map(MultiDenseVectorInternal::new_unchecked)
                        .collect(),
                )
            }
            BatchVectorStructPersisted::Named(vectors) => BatchVectorStructInternal::Named(
                vectors
                    .into_iter()
                    .map(|(k, v)| (k, v.into_iter().map(VectorInternal::from).collect()))
                    .collect(),
            ),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, Validate, Hash)]
#[serde(rename_all = "snake_case")]
pub struct PointStructPersisted {
    /// Point id
    pub id: PointIdType,
    /// Vectors
    pub vector: VectorStructPersisted,
    /// Payload values (optional)
    pub payload: Option<Payload>,
}

impl PointStructPersisted {
    pub fn get_vectors(&self) -> NamedVectors<'_> {
        let mut named_vectors = NamedVectors::default();
        match &self.vector {
            VectorStructPersisted::Single(vector) => named_vectors.insert(
                DEFAULT_VECTOR_NAME.to_owned(),
                VectorInternal::from(vector.clone()),
            ),
            VectorStructPersisted::MultiDense(vector) => named_vectors.insert(
                DEFAULT_VECTOR_NAME.to_owned(),
                VectorInternal::from(MultiDenseVectorInternal::new_unchecked(vector.clone())),
            ),
            VectorStructPersisted::Named(vectors) => {
                for (name, vector) in vectors {
                    named_vectors.insert(name.clone(), VectorInternal::from(vector.clone()));
                }
            }
        }
        named_vectors
    }

    pub fn is_equal_to(&self, segment_record: &SegmentRecord) -> bool {
        let SegmentRecord {
            id,
            vectors,
            payload,
        } = segment_record;

        if &self.id != id {
            return false;
        }

        let self_vectors = self.get_vectors().into_owned_map();

        if let Some(segment_vectors) = vectors {
            if self_vectors.len() != segment_vectors.len() {
                return false;
            }
            for (name, vec) in segment_vectors {
                if self_vectors.get(name) != Some(vec) {
                    return false;
                }
            }
        } else if !self_vectors.is_empty() {
            return false;
        }

        // Check if payloads are equal, empty and non-existent payloads are considered equal
        let self_payload = self.payload.as_ref().filter(|p| !p.is_empty());
        let segment_payload = payload.as_ref().filter(|p| !p.is_empty());
        self_payload == segment_payload
    }
}

#[cfg(feature = "api")]
impl TryFrom<api::rest::schema::Record> for PointStructPersisted {
    type Error = String;

    fn try_from(record: api::rest::schema::Record) -> Result<Self, Self::Error> {
        let api::rest::schema::Record {
            id,
            payload,
            vector,
            shard_key: _,
            order_value: _,
        } = record;

        if vector.is_none() {
            return Err("Vector is empty".to_string());
        }

        Ok(Self {
            id,
            payload,
            vector: VectorStructPersisted::from(vector.unwrap()),
        })
    }
}

#[cfg(feature = "api")]
impl TryFrom<PointStructPersisted> for api::grpc::qdrant::PointStruct {
    type Error = tonic::Status;

    fn try_from(value: PointStructPersisted) -> Result<Self, Self::Error> {
        let PointStructPersisted {
            id,
            vector,
            payload,
        } = value;

        let vectors_internal = VectorStructInternal::try_from(vector).map_err(|e| {
            tonic::Status::invalid_argument(format!("Failed to convert vectors: {e}"))
        })?;

        let vectors = api::grpc::qdrant::Vectors::from(vectors_internal);
        let converted_payload = match payload {
            None => HashMap::new(),
            Some(payload) => api::conversions::json::payload_to_proto(payload),
        };

        Ok(Self {
            id: Some(id.into()),
            vectors: Some(vectors),
            payload: converted_payload,
        })
    }
}

/// Data structure for point vectors, as it is persisted in WAL
#[derive(Clone, PartialEq, Deserialize, Serialize)]
#[serde(untagged, rename_all = "snake_case")]
pub enum VectorStructPersisted {
    Single(DenseVector),
    MultiDense(MultiDenseVector),
    Named(HashMap<VectorNameBuf, VectorPersisted>),
}

impl std::hash::Hash for VectorStructPersisted {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        mem::discriminant(self).hash(state);
        match self {
            VectorStructPersisted::Single(vec) => {
                for v in vec {
                    OrderedFloat(*v).hash(state);
                }
            }
            VectorStructPersisted::MultiDense(multi_vec) => {
                for vec in multi_vec {
                    for v in vec {
                        OrderedFloat(*v).hash(state);
                    }
                }
            }
            VectorStructPersisted::Named(map) => {
                unordered_hash_unique(state, map.iter());
            }
        }
    }
}

impl Debug for VectorStructPersisted {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            VectorStructPersisted::Single(vector) => {
                let first_elements = vector.iter().take(4).join(", ");
                write!(f, "Single([{}, ... x {}])", first_elements, vector.len())
            }
            VectorStructPersisted::MultiDense(vector) => {
                let first_vectors = vector
                    .iter()
                    .take(4)
                    .map(|v| {
                        let first_elements = v.iter().take(4).join(", ");
                        format!("[{}, ... x {}]", first_elements, v.len())
                    })
                    .join(", ");
                write!(f, "MultiDense([{}, ... x {})", first_vectors, vector.len())
            }
            VectorStructPersisted::Named(vectors) => write!(f, "Named(( ")
                .and_then(|_| {
                    for (name, vector) in vectors {
                        write!(f, "{name}: {vector:?}, ")?;
                    }
                    Ok(())
                })
                .and_then(|_| write!(f, "))")),
        }
    }
}

impl VectorStructPersisted {
    /// Check if this vector struct is empty.
    pub fn is_empty(&self) -> bool {
        match self {
            VectorStructPersisted::Single(vector) => vector.is_empty(),
            VectorStructPersisted::MultiDense(vector) => vector.is_empty(),
            VectorStructPersisted::Named(vectors) => vectors.values().all(|v| match v {
                VectorPersisted::Dense(vector) => vector.is_empty(),
                VectorPersisted::Sparse(vector) => vector.indices.is_empty(),
                VectorPersisted::MultiDense(vector) => vector.is_empty(),
            }),
        }
    }
}

impl Validate for VectorStructPersisted {
    fn validate(&self) -> Result<(), ValidationErrors> {
        match self {
            VectorStructPersisted::Single(_) => Ok(()),
            VectorStructPersisted::MultiDense(v) => validate_multi_vector(v),
            VectorStructPersisted::Named(v) => crate::common::validation::validate_iter(v.values()),
        }
    }
}

impl From<DenseVector> for VectorStructPersisted {
    fn from(value: DenseVector) -> Self {
        VectorStructPersisted::Single(value)
    }
}

impl From<VectorStructInternal> for VectorStructPersisted {
    fn from(value: VectorStructInternal) -> Self {
        match value {
            VectorStructInternal::Single(vector) => VectorStructPersisted::Single(vector),
            VectorStructInternal::MultiDense(vector) => {
                VectorStructPersisted::MultiDense(vector.into_multi_vectors())
            }
            VectorStructInternal::Named(vectors) => VectorStructPersisted::Named(
                vectors
                    .into_iter()
                    .map(|(k, v)| (k, VectorPersisted::from(v)))
                    .collect(),
            ),
        }
    }
}

#[cfg(feature = "api")]
impl From<api::rest::VectorStructOutput> for VectorStructPersisted {
    fn from(value: api::rest::VectorStructOutput) -> Self {
        match value {
            api::rest::VectorStructOutput::Single(vector) => VectorStructPersisted::Single(vector),
            api::rest::VectorStructOutput::MultiDense(vector) => {
                VectorStructPersisted::MultiDense(vector)
            }
            api::rest::VectorStructOutput::Named(vectors) => VectorStructPersisted::Named(
                vectors
                    .into_iter()
                    .map(|(k, v)| (k, VectorPersisted::from(v)))
                    .collect(),
            ),
        }
    }
}

impl TryFrom<VectorStructPersisted> for VectorStructInternal {
    type Error = OperationError;
    fn try_from(value: VectorStructPersisted) -> Result<Self, Self::Error> {
        let vector_struct = match value {
            VectorStructPersisted::Single(vector) => VectorStructInternal::Single(vector),
            VectorStructPersisted::MultiDense(vector) => {
                VectorStructInternal::MultiDense(MultiDenseVectorInternal::try_from(vector)?)
            }
            VectorStructPersisted::Named(vectors) => VectorStructInternal::Named(
                vectors
                    .into_iter()
                    .map(|(k, v)| (k, VectorInternal::from(v)))
                    .collect(),
            ),
        };
        Ok(vector_struct)
    }
}

impl From<VectorStructPersisted> for NamedVectors<'_> {
    fn from(value: VectorStructPersisted) -> Self {
        match value {
            VectorStructPersisted::Single(vector) => {
                NamedVectors::from_pairs([(DEFAULT_VECTOR_NAME.to_owned(), vector)])
            }
            VectorStructPersisted::MultiDense(vector) => {
                let mut named_vector = NamedVectors::default();
                let multivec = MultiDenseVectorInternal::new_unchecked(vector);

                named_vector.insert(
                    DEFAULT_VECTOR_NAME.to_owned(),
                    crate::segment::data_types::vectors::VectorInternal::from(multivec),
                );
                named_vector
            }
            VectorStructPersisted::Named(vectors) => {
                let mut named_vector = NamedVectors::default();
                for (name, vector) in vectors {
                    named_vector.insert(
                        name,
                        crate::segment::data_types::vectors::VectorInternal::from(vector),
                    );
                }
                named_vector
            }
        }
    }
}

/// Single vector data, as it is persisted in WAL
/// Unlike [`api::rest::Vector`], this struct only stores raw vectors, inferenced or resolved.
/// Unlike [`VectorInternal`], is not optimized for search
#[derive(Clone, PartialEq, Deserialize, Serialize)]
#[serde(untagged, rename_all = "snake_case")]
pub enum VectorPersisted {
    Dense(DenseVector),
    Sparse(crate::sparse::common::sparse_vector::SparseVector),
    MultiDense(MultiDenseVector),
}

impl Hash for VectorPersisted {
    fn hash<H: Hasher>(&self, state: &mut H) {
        mem::discriminant(self).hash(state);
        match self {
            VectorPersisted::Dense(vec) => {
                for v in vec {
                    OrderedFloat(*v).hash(state);
                }
            }
            VectorPersisted::Sparse(sparse) => {
                sparse.hash(state);
            }
            VectorPersisted::MultiDense(multi_vec) => {
                for vec in multi_vec {
                    for v in vec {
                        OrderedFloat(*v).hash(state);
                    }
                }
            }
        }
    }
}

impl VectorPersisted {
    pub fn new_sparse(indices: Vec<DimId>, values: Vec<DimWeight>) -> Self {
        Self::Sparse(crate::sparse::common::sparse_vector::SparseVector { indices, values })
    }

    pub fn empty_sparse() -> Self {
        Self::new_sparse(vec![], vec![])
    }
}

impl Debug for VectorPersisted {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            VectorPersisted::Dense(vector) => {
                let first_elements = vector.iter().take(4).join(", ");
                write!(f, "Dense([{}, ... x {}])", first_elements, vector.len())
            }
            VectorPersisted::Sparse(vector) => {
                let first_elements = vector
                    .indices
                    .iter()
                    .zip(vector.values.iter())
                    .take(4)
                    .map(|(k, v)| format!("{k}->{v}"))
                    .join(", ");
                write!(
                    f,
                    "Sparse([{}, ... x {})",
                    first_elements,
                    vector.indices.len()
                )
            }
            VectorPersisted::MultiDense(vector) => {
                let first_vectors = vector
                    .iter()
                    .take(4)
                    .map(|v| {
                        let first_elements = v.iter().take(4).join(", ");
                        format!("[{}, ... x {}]", first_elements, v.len())
                    })
                    .join(", ");
                write!(f, "MultiDense([{}, ... x {})", first_vectors, vector.len())
            }
        }
    }
}

impl Validate for VectorPersisted {
    fn validate(&self) -> Result<(), ValidationErrors> {
        match self {
            VectorPersisted::Dense(_) => Ok(()),
            VectorPersisted::Sparse(v) => v.validate(),
            VectorPersisted::MultiDense(m) => validate_multi_vector(m),
        }
    }
}

impl From<VectorInternal> for VectorPersisted {
    fn from(value: VectorInternal) -> Self {
        match value {
            VectorInternal::Dense(vector) => VectorPersisted::Dense(vector),
            VectorInternal::Sparse(vector) => VectorPersisted::Sparse(vector),
            VectorInternal::MultiDense(vector) => {
                VectorPersisted::MultiDense(vector.into_multi_vectors())
            }
        }
    }
}

#[cfg(feature = "api")]
impl From<api::rest::VectorOutput> for VectorPersisted {
    fn from(value: api::rest::VectorOutput) -> Self {
        match value {
            api::rest::VectorOutput::Dense(vector) => VectorPersisted::Dense(vector),
            api::rest::VectorOutput::Sparse(vector) => VectorPersisted::Sparse(vector),
            api::rest::VectorOutput::MultiDense(vector) => VectorPersisted::MultiDense(vector),
        }
    }
}

impl From<VectorPersisted> for VectorInternal {
    fn from(value: VectorPersisted) -> Self {
        match value {
            VectorPersisted::Dense(vector) => VectorInternal::Dense(vector),
            VectorPersisted::Sparse(vector) => VectorInternal::Sparse(vector),
            VectorPersisted::MultiDense(vector) => {
                // the REST vectors have been validated already
                // we can use an internal constructor
                VectorInternal::MultiDense(MultiDenseVectorInternal::new_unchecked(vector))
            }
        }
    }
}

fn retain_with_index<T, F>(vec: &mut Vec<T>, mut filter: F)
where
    F: FnMut(usize, &T) -> bool,
{
    let mut index = 0;

    vec.retain(|item| {
        let retain = filter(index, item);
        index += 1;
        retain
    });
}