qdrant_local.py

import importlib.metadata
import itertools
import json
import logging
import os
import shutil
from copy import deepcopy
from io import TextIOWrapper
from typing import (
    Any,
    Dict,
    Generator,
    Iterable,
    List,
    Mapping,
    Optional,
    Sequence,
    Tuple,
    Union,
    get_args,
    Set,
)
from uuid import uuid4

import numpy as np
import portalocker

from qdrant_client._pydantic_compat import to_dict
from qdrant_client.client_base import QdrantBase
from qdrant_client.conversions import common_types as types
from qdrant_client.http import models as rest_models
from qdrant_client.http.models.models import RecommendExample
from qdrant_client.local.local_collection import (
    LocalCollection,
    DEFAULT_VECTOR_NAME,
    ignore_mentioned_ids_filter,
)

META_INFO_FILENAME = "meta.json"


class QdrantLocal(QdrantBase):
    """
    Everything Qdrant server can do, but locally.

    Use this implementation to run vector search without running a Qdrant server.
    Everything that works with local Qdrant will work with server Qdrant as well.

    Use for small-scale data, demos, and tests.
    If you need more speed or size, use Qdrant server.
    """

    def __init__(self, location: str, force_disable_check_same_thread: bool = False) -> None:
        """
        Initialize local Qdrant.

        Args:
            location: Where to store data. Can be a path to a directory or `:memory:` for in-memory storage.
            force_disable_check_same_thread: Disable SQLite check_same_thread check. Use only if you know what you are doing.
        """
        super().__init__()
        self.force_disable_check_same_thread = force_disable_check_same_thread
        self.location = location
        self.persistent = location != ":memory:"
        self.collections: Dict[str, LocalCollection] = {}
        self.aliases: Dict[str, str] = {}
        self._flock_file: Optional[TextIOWrapper] = None
        self._load()
        self._closed: bool = False

    @property
    def closed(self) -> bool:
        return self._closed

    def close(self, **kwargs: Any) -> None:
        self._closed = True
        for collection in self.collections.values():
            if collection is not None:
                collection.close()
            else:
                logging.warning(
                    f"Collection appears to be None before closing. The existing collections are: "
                    f"{list(self.collections.keys())}"
                )

        try:
            if self._flock_file is not None and not self._flock_file.closed:
                portalocker.unlock(self._flock_file)
                self._flock_file.close()
        except TypeError:  # sometimes portalocker module can be garbage collected before
            # QdrantLocal instance
            pass

    def _load(self) -> None:
        if not self.persistent:
            return
        meta_path = os.path.join(self.location, META_INFO_FILENAME)
        if not os.path.exists(meta_path):
            os.makedirs(self.location, exist_ok=True)
            with open(meta_path, "w") as f:
                f.write(json.dumps({"collections": {}, "aliases": {}}))
        else:
            with open(meta_path, "r") as f:
                meta = json.load(f)
                for collection_name, config_json in meta["collections"].items():
                    config = rest_models.CreateCollection(**config_json)
                    collection_path = self._collection_path(collection_name)
                    self.collections[collection_name] = LocalCollection(
                        config,
                        collection_path,
                        force_disable_check_same_thread=self.force_disable_check_same_thread,
                    )
                self.aliases = meta["aliases"]

        lock_file_path = os.path.join(self.location, ".lock")
        if not os.path.exists(lock_file_path):
            os.makedirs(self.location, exist_ok=True)
            with open(lock_file_path, "w") as f:
                f.write("tmp lock file")
        self._flock_file = open(lock_file_path, "r+")
        try:
            portalocker.lock(
                self._flock_file,
                portalocker.LockFlags.EXCLUSIVE | portalocker.LockFlags.NON_BLOCKING,
            )
        except portalocker.exceptions.LockException:
            raise RuntimeError(
                f"Storage folder {self.location} is already accessed by another instance of Qdrant client."
                f" If you require concurrent access, use Qdrant server instead."
            )

    def _save(self) -> None:
        if not self.persistent:
            return

        if self.closed:
            raise RuntimeError("QdrantLocal instance is closed. Please create a new instance.")

        meta_path = os.path.join(self.location, META_INFO_FILENAME)
        with open(meta_path, "w") as f:
            f.write(
                json.dumps(
                    {
                        "collections": {
                            collection_name: to_dict(collection.config)
                            for collection_name, collection in self.collections.items()
                        },
                        "aliases": self.aliases,
                    }
                )
            )

    def _get_collection(self, collection_name: str) -> LocalCollection:
        if self.closed:
            raise RuntimeError("QdrantLocal instance is closed. Please create a new instance.")

        if collection_name in self.collections:
            return self.collections[collection_name]
        if collection_name in self.aliases:
            return self.collections[self.aliases[collection_name]]
        raise ValueError(f"Collection {collection_name} not found")

    def search_batch(
        self,
        collection_name: str,
        requests: Sequence[types.SearchRequest],
        **kwargs: Any,
    ) -> List[List[types.ScoredPoint]]:
        collection = self._get_collection(collection_name)

        return [
            collection.search(
                query_vector=request.vector,
                query_filter=request.filter,
                limit=request.limit,
                offset=request.offset,
                with_payload=request.with_payload,
                with_vectors=request.with_vector,
                score_threshold=request.score_threshold,
            )
            for request in requests
        ]

    def search(
        self,
        collection_name: str,
        query_vector: Union[
            types.NumpyArray,
            Sequence[float],
            Tuple[str, List[float]],
            types.NamedVector,
            types.NamedSparseVector,
        ],
        query_filter: Optional[types.Filter] = None,
        search_params: Optional[types.SearchParams] = None,
        limit: int = 10,
        offset: Optional[int] = None,
        with_payload: Union[bool, Sequence[str], types.PayloadSelector] = True,
        with_vectors: Union[bool, Sequence[str]] = False,
        score_threshold: Optional[float] = None,
        **kwargs: Any,
    ) -> List[types.ScoredPoint]:
        collection = self._get_collection(collection_name)
        return collection.search(
            query_vector=query_vector,
            query_filter=query_filter,
            limit=limit,
            offset=offset,
            with_payload=with_payload,
            with_vectors=with_vectors,
            score_threshold=score_threshold,
        )

    def search_matrix_offsets(
        self,
        collection_name: str,
        query_filter: Optional[types.Filter] = None,
        limit: int = 3,
        sample: int = 10,
        using: Optional[str] = None,
        **kwargs: Any,
    ) -> types.SearchMatrixOffsetsResponse:
        collection = self._get_collection(collection_name)
        return collection.search_matrix_offsets(
            query_filter=query_filter, limit=limit, sample=sample, using=using
        )

    def search_matrix_pairs(
        self,
        collection_name: str,
        query_filter: Optional[types.Filter] = None,
        limit: int = 3,
        sample: int = 10,
        using: Optional[str] = None,
        **kwargs: Any,
    ) -> types.SearchMatrixPairsResponse:
        collection = self._get_collection(collection_name)
        return collection.search_matrix_pairs(
            query_filter=query_filter, limit=limit, sample=sample, using=using
        )

    def search_groups(
        self,
        collection_name: str,
        query_vector: Union[
            types.NumpyArray,
            Sequence[float],
            Tuple[str, List[float]],
            types.NamedVector,
        ],
        group_by: str,
        query_filter: Optional[rest_models.Filter] = None,
        search_params: Optional[rest_models.SearchParams] = None,
        limit: int = 10,
        group_size: int = 1,
        with_payload: Union[bool, Sequence[str], rest_models.PayloadSelector] = True,
        with_vectors: Union[bool, Sequence[str]] = False,
        score_threshold: Optional[float] = None,
        with_lookup: Optional[types.WithLookupInterface] = None,
        **kwargs: Any,
    ) -> types.GroupsResult:
        collection = self._get_collection(collection_name)
        with_lookup_collection = None
        if with_lookup is not None:
            if isinstance(with_lookup, str):
                with_lookup_collection = self._get_collection(with_lookup)
            else:
                with_lookup_collection = self._get_collection(with_lookup.collection)

        return collection.search_groups(
            query_vector=query_vector,
            query_filter=query_filter,
            limit=limit,
            group_by=group_by,
            group_size=group_size,
            with_payload=with_payload,
            with_vectors=with_vectors,
            score_threshold=score_threshold,
            with_lookup=with_lookup,
            with_lookup_collection=with_lookup_collection,
        )

    def _resolve_query_input(
        self,
        collection_name: str,
        query: Optional[types.Query],
        using: Optional[str],
        lookup_from: Optional[types.LookupLocation],
    ) -> Tuple[types.Query, Set[types.PointId]]:
        """
        Resolves any possible ids into vectors and returns a new query object, along with a set of the mentioned
        point ids that should be filtered when searching.
        """

        lookup_collection_name = lookup_from.collection if lookup_from else collection_name
        collection = self._get_collection(lookup_collection_name)

        search_in_vector_name = using if using is not None else DEFAULT_VECTOR_NAME
        vector_name = (
            lookup_from.vector
            if lookup_from is not None and lookup_from.vector is not None
            else search_in_vector_name
        )

        sparse = vector_name in collection.sparse_vectors
        multi = vector_name in collection.multivectors
        if sparse:
            collection_vectors = collection.sparse_vectors
        elif multi:
            collection_vectors = collection.multivectors
        else:
            collection_vectors = collection.vectors

        # mentioned ids in the search collection which should be excluded from search
        mentioned_ids: Set[types.PointId] = set()

        def input_into_vector(
            vector_input: types.VectorInput,
        ) -> types.VectorInput:
            if isinstance(vector_input, get_args(types.PointId)):
                point_id = vector_input  # rename for clarity
                if point_id not in collection.ids:
                    raise ValueError(f"Point {point_id} is not found in the collection")

                idx = collection.ids[point_id]
                if vector_name in collection_vectors:
                    vec = collection_vectors[vector_name][idx]
                else:
                    raise ValueError(f"Vector {vector_name} not found")
                if isinstance(vec, np.ndarray):
                    vec = vec.tolist()
                if collection_name == lookup_collection_name:
                    mentioned_ids.add(point_id)
                return vec
            else:
                return vector_input

        query = deepcopy(query)
        if isinstance(query, rest_models.NearestQuery):
            query.nearest = input_into_vector(query.nearest)

        elif isinstance(query, rest_models.RecommendQuery):
            if query.recommend.negative is not None:
                query.recommend.negative = [
                    input_into_vector(vector_input) for vector_input in query.recommend.negative
                ]
            if query.recommend.positive is not None:
                query.recommend.positive = [
                    input_into_vector(vector_input) for vector_input in query.recommend.positive
                ]

        elif isinstance(query, rest_models.DiscoverQuery):
            query.discover.target = input_into_vector(query.discover.target)
            pairs = (
                query.discover.context
                if isinstance(query.discover.context, list)
                else [query.discover.context]
            )
            query.discover.context = [
                rest_models.ContextPair(
                    positive=input_into_vector(pair.positive),
                    negative=input_into_vector(pair.negative),
                )
                for pair in pairs
            ]
        elif isinstance(query, rest_models.ContextQuery):
            pairs = query.context if isinstance(query.context, list) else [query.context]
            query.context = [
                rest_models.ContextPair(
                    positive=input_into_vector(pair.positive),
                    negative=input_into_vector(pair.negative),
                )
                for pair in pairs
            ]
        elif isinstance(query, rest_models.OrderByQuery):
            pass
        elif isinstance(query, rest_models.FusionQuery):
            pass

        return query, mentioned_ids

    def _resolve_prefetches_input(
        self,
        prefetch: Optional[Union[Sequence[types.Prefetch], types.Prefetch]],
        collection_name: str,
    ) -> List[types.Prefetch]:
        if prefetch is None:
            return []

        if isinstance(prefetch, list) and len(prefetch) == 0:
            return []

        prefetches = []
        if isinstance(prefetch, types.Prefetch):
            prefetches = [prefetch]
            prefetches.extend(
                prefetch.prefetch if isinstance(prefetch.prefetch, list) else [prefetch.prefetch]
            )
        elif isinstance(prefetch, Sequence):
            prefetches = list(prefetch)

        return [
            self._resolve_prefetch_input(prefetch, collection_name)
            for prefetch in prefetches
            if prefetch is not None
        ]

    def _resolve_prefetch_input(
        self, prefetch: types.Prefetch, collection_name: str
    ) -> types.Prefetch:
        if prefetch.query is None:
            return prefetch

        prefetch = deepcopy(prefetch)
        query, mentioned_ids = self._resolve_query_input(
            collection_name,
            prefetch.query,
            prefetch.using,
            prefetch.lookup_from,
        )
        prefetch.query = query

        prefetch.filter = ignore_mentioned_ids_filter(prefetch.filter, list(mentioned_ids))

        prefetch.prefetch = self._resolve_prefetches_input(prefetch.prefetch, collection_name)

        return prefetch

    def query_points(
        self,
        collection_name: str,
        query: Optional[types.Query] = None,
        using: Optional[str] = None,
        prefetch: Union[types.Prefetch, List[types.Prefetch], None] = None,
        query_filter: Optional[types.Filter] = None,
        search_params: Optional[types.SearchParams] = None,
        limit: int = 10,
        offset: Optional[int] = None,
        with_payload: Union[bool, Sequence[str], types.PayloadSelector] = True,
        with_vectors: Union[bool, Sequence[str]] = False,
        score_threshold: Optional[float] = None,
        lookup_from: Optional[types.LookupLocation] = None,
        **kwargs: Any,
    ) -> types.QueryResponse:
        collection = self._get_collection(collection_name)

        if query is not None:
            query, mentioned_ids = self._resolve_query_input(
                collection_name, query, using, lookup_from
            )
            query_filter = ignore_mentioned_ids_filter(query_filter, list(mentioned_ids))

        prefetch = self._resolve_prefetches_input(prefetch, collection_name)
        return collection.query_points(
            query=query,
            prefetch=prefetch,
            query_filter=query_filter,
            using=using,
            score_threshold=score_threshold,
            limit=limit,
            offset=offset or 0,
            with_payload=with_payload,
            with_vectors=with_vectors,
        )

    def query_batch_points(
        self,
        collection_name: str,
        requests: Sequence[types.QueryRequest],
        **kwargs: Any,
    ) -> List[types.QueryResponse]:
        collection = self._get_collection(collection_name)

        return [
            collection.query_points(
                query=request.query,
                prefetch=request.prefetch,
                query_filter=request.filter,
                limit=request.limit,
                offset=request.offset or 0,
                with_payload=request.with_payload,
                with_vectors=request.with_vector,
                score_threshold=request.score_threshold,
                using=request.using,
                lookup_from_collection=self._get_collection(request.lookup_from.collection)
                if request.lookup_from
                else None,
                lookup_from_vector_name=request.lookup_from.vector
                if request.lookup_from
                else None,
            )
            for request in requests
        ]

    def query_points_groups(
        self,
        collection_name: str,
        group_by: str,
        query: Union[
            types.PointId,
            List[float],
            List[List[float]],
            types.SparseVector,
            types.Query,
            types.NumpyArray,
            types.Document,
            None,
        ] = None,
        using: Optional[str] = None,
        prefetch: Union[types.Prefetch, List[types.Prefetch], None] = None,
        query_filter: Optional[types.Filter] = None,
        search_params: Optional[types.SearchParams] = None,
        limit: int = 10,
        group_size: int = 3,
        with_payload: Union[bool, Sequence[str], types.PayloadSelector] = True,
        with_vectors: Union[bool, Sequence[str]] = False,
        score_threshold: Optional[float] = None,
        with_lookup: Optional[types.WithLookupInterface] = None,
        lookup_from: Optional[types.LookupLocation] = None,
        **kwargs: Any,
    ) -> types.GroupsResult:
        collection = self._get_collection(collection_name)
        if query is not None:
            query, mentioned_ids = self._resolve_query_input(
                collection_name, query, using, lookup_from
            )
            query_filter = ignore_mentioned_ids_filter(query_filter, list(mentioned_ids))
        with_lookup_collection = None
        if with_lookup is not None:
            if isinstance(with_lookup, str):
                with_lookup_collection = self._get_collection(with_lookup)
            else:
                with_lookup_collection = self._get_collection(with_lookup.collection)

        return collection.query_groups(
            query=query,
            query_filter=query_filter,
            using=using,
            prefetch=prefetch,
            limit=limit,
            group_by=group_by,
            group_size=group_size,
            with_payload=with_payload,
            with_vectors=with_vectors,
            score_threshold=score_threshold,
            with_lookup=with_lookup,
            with_lookup_collection=with_lookup_collection,
        )

    def recommend_batch(
        self,
        collection_name: str,
        requests: Sequence[types.RecommendRequest],
        **kwargs: Any,
    ) -> List[List[types.ScoredPoint]]:
        collection = self._get_collection(collection_name)

        return [
            collection.recommend(
                positive=request.positive,
                negative=request.negative,
                query_filter=request.filter,
                limit=request.limit,
                offset=request.offset,
                with_payload=request.with_payload,
                with_vectors=request.with_vector,
                score_threshold=request.score_threshold,
                using=request.using,
                lookup_from_collection=self._get_collection(request.lookup_from.collection)
                if request.lookup_from
                else None,
                lookup_from_vector_name=request.lookup_from.vector
                if request.lookup_from
                else None,
                strategy=request.strategy,
            )
            for request in requests
        ]

    def recommend(
        self,
        collection_name: str,
        positive: Optional[Sequence[RecommendExample]] = None,
        negative: Optional[Sequence[RecommendExample]] = None,
        query_filter: Optional[types.Filter] = None,
        search_params: Optional[types.SearchParams] = None,
        limit: int = 10,
        offset: int = 0,
        with_payload: Union[bool, List[str], types.PayloadSelector] = True,
        with_vectors: Union[bool, List[str]] = False,
        score_threshold: Optional[float] = None,
        using: Optional[str] = None,
        lookup_from: Optional[types.LookupLocation] = None,
        strategy: Optional[types.RecommendStrategy] = None,
        **kwargs: Any,
    ) -> List[types.ScoredPoint]:
        collection = self._get_collection(collection_name)
        return collection.recommend(
            positive=positive,
            negative=negative,
            query_filter=query_filter,
            limit=limit,
            offset=offset,
            with_payload=with_payload,
            with_vectors=with_vectors,
            score_threshold=score_threshold,
            using=using,
            lookup_from_collection=self._get_collection(lookup_from.collection)
            if lookup_from
            else None,
            lookup_from_vector_name=lookup_from.vector if lookup_from else None,
            strategy=strategy,
        )

    def recommend_groups(
        self,
        collection_name: str,
        group_by: str,
        positive: Optional[Sequence[Union[types.PointId, List[float]]]] = None,
        negative: Optional[Sequence[Union[types.PointId, List[float]]]] = None,
        query_filter: Optional[types.Filter] = None,
        search_params: Optional[types.SearchParams] = None,
        limit: int = 10,
        group_size: int = 1,
        score_threshold: Optional[float] = None,
        with_payload: Union[bool, Sequence[str], types.PayloadSelector] = True,
        with_vectors: Union[bool, Sequence[str]] = False,
        using: Optional[str] = None,
        lookup_from: Optional[types.LookupLocation] = None,
        with_lookup: Optional[types.WithLookupInterface] = None,
        strategy: Optional[types.RecommendStrategy] = None,
        **kwargs: Any,
    ) -> types.GroupsResult:
        collection = self._get_collection(collection_name)
        with_lookup_collection = None
        if with_lookup is not None:
            if isinstance(with_lookup, str):
                with_lookup_collection = self._get_collection(with_lookup)
            else:
                with_lookup_collection = self._get_collection(with_lookup.collection)

        return collection.recommend_groups(
            positive=positive,
            negative=negative,
            group_by=group_by,
            group_size=group_size,
            query_filter=query_filter,
            limit=limit,
            with_payload=with_payload,
            with_vectors=with_vectors,
            score_threshold=score_threshold,
            using=using,
            lookup_from_collection=self._get_collection(lookup_from.collection)
            if lookup_from
            else None,
            lookup_from_vector_name=lookup_from.vector if lookup_from else None,
            with_lookup=with_lookup,
            with_lookup_collection=with_lookup_collection,
            strategy=strategy,
        )

    def discover(
        self,
        collection_name: str,
        target: Optional[types.TargetVector] = None,
        context: Optional[Sequence[types.ContextExamplePair]] = None,
        query_filter: Optional[types.Filter] = None,
        search_params: Optional[types.SearchParams] = None,
        limit: int = 10,
        offset: int = 0,
        with_payload: Union[bool, List[str], types.PayloadSelector] = True,
        with_vectors: Union[bool, List[str]] = False,
        using: Optional[str] = None,
        lookup_from: Optional[types.LookupLocation] = None,
        consistency: Optional[types.ReadConsistency] = None,
        timeout: Optional[int] = None,
        **kwargs: Any,
    ) -> List[types.ScoredPoint]:
        collection = self._get_collection(collection_name)
        return collection.discover(
            target=target,
            context=context,
            query_filter=query_filter,
            limit=limit,
            offset=offset,
            with_payload=with_payload,
            with_vectors=with_vectors,
            using=using,
            lookup_from_collection=self._get_collection(lookup_from.collection)
            if lookup_from
            else None,
            lookup_from_vector_name=lookup_from.vector if lookup_from else None,
        )

    def discover_batch(
        self,
        collection_name: str,
        requests: Sequence[types.DiscoverRequest],
        **kwargs: Any,
    ) -> List[List[types.ScoredPoint]]:
        collection = self._get_collection(collection_name)

        return [
            collection.discover(
                target=request.target,
                context=request.context,
                query_filter=request.filter,
                limit=request.limit,
                offset=request.offset,
                with_payload=request.with_payload,
                with_vectors=request.with_vector,
                using=request.using,
                lookup_from_collection=self._get_collection(request.lookup_from.collection)
                if request.lookup_from
                else None,
                lookup_from_vector_name=request.lookup_from.vector
                if request.lookup_from
                else None,
            )
            for request in requests
        ]

    def scroll(
        self,
        collection_name: str,
        scroll_filter: Optional[types.Filter] = None,
        limit: int = 10,
        order_by: Optional[types.OrderBy] = None,
        offset: Optional[types.PointId] = None,
        with_payload: Union[bool, Sequence[str], types.PayloadSelector] = True,
        with_vectors: Union[bool, Sequence[str]] = False,
        **kwargs: Any,
    ) -> Tuple[List[types.Record], Optional[types.PointId]]:
        collection = self._get_collection(collection_name)
        return collection.scroll(
            scroll_filter=scroll_filter,
            limit=limit,
            order_by=order_by,
            offset=offset,
            with_payload=with_payload,
            with_vectors=with_vectors,
        )

    def count(
        self,
        collection_name: str,
        count_filter: Optional[types.Filter] = None,
        exact: bool = True,
        **kwargs: Any,
    ) -> types.CountResult:
        collection = self._get_collection(collection_name)
        return collection.count(count_filter=count_filter)

    def facet(
        self,
        collection_name: str,
        key: str,
        facet_filter: Optional[types.Filter] = None,
        limit: int = 10,
        exact: bool = False,
        **kwargs: Any,
    ) -> types.FacetResponse:
        collection = self._get_collection(collection_name)
        return collection.facet(key=key, facet_filter=facet_filter, limit=limit)

    def upsert(
        self, collection_name: str, points: types.Points, **kwargs: Any
    ) -> types.UpdateResult:
        collection = self._get_collection(collection_name)
        collection.upsert(points)
        return self._default_update_result()

    def update_vectors(
        self,
        collection_name: str,
        points: Sequence[types.PointVectors],
        **kwargs: Any,
    ) -> types.UpdateResult:
        collection = self._get_collection(collection_name)
        collection.update_vectors(points)
        return self._default_update_result()

    def delete_vectors(
        self,
        collection_name: str,
        vectors: Sequence[str],
        points: types.PointsSelector,
        **kwargs: Any,
    ) -> types.UpdateResult:
        collection = self._get_collection(collection_name)
        collection.delete_vectors(vectors, points)
        return self._default_update_result()

    def retrieve(
        self,
        collection_name: str,
        ids: Sequence[types.PointId],
        with_payload: Union[bool, Sequence[str], types.PayloadSelector] = True,
        with_vectors: Union[bool, Sequence[str]] = False,
        **kwargs: Any,
    ) -> List[types.Record]:
        collection = self._get_collection(collection_name)
        return collection.retrieve(ids, with_payload, with_vectors)

    @classmethod
    def _default_update_result(cls, operation_id: int = 0) -> types.UpdateResult:
        return types.UpdateResult(
            operation_id=operation_id,
            status=rest_models.UpdateStatus.COMPLETED,
        )

    def delete(
        self, collection_name: str, points_selector: types.PointsSelector, **kwargs: Any
    ) -> types.UpdateResult:
        collection = self._get_collection(collection_name)
        collection.delete(points_selector)
        return self._default_update_result()

    def set_payload(
        self,
        collection_name: str,
        payload: types.Payload,
        points: types.PointsSelector,
        key: Optional[str] = None,
        **kwargs: Any,
    ) -> types.UpdateResult:
        collection = self._get_collection(collection_name)
        collection.set_payload(payload=payload, selector=points, key=key)
        return self._default_update_result()

    def overwrite_payload(
        self,
        collection_name: str,
        payload: types.Payload,
        points: types.PointsSelector,
        **kwargs: Any,
    ) -> types.UpdateResult:
        collection = self._get_collection(collection_name)
        collection.overwrite_payload(payload=payload, selector=points)
        return self._default_update_result()

    def delete_payload(
        self,
        collection_name: str,
        keys: Sequence[str],
        points: types.PointsSelector,
        **kwargs: Any,
    ) -> types.UpdateResult:
        collection = self._get_collection(collection_name)
        collection.delete_payload(keys=keys, selector=points)
        return self._default_update_result()

    def clear_payload(
        self, collection_name: str, points_selector: types.PointsSelector, **kwargs: Any
    ) -> types.UpdateResult:
        collection = self._get_collection(collection_name)
        collection.clear_payload(selector=points_selector)
        return self._default_update_result()

    def batch_update_points(
        self,
        collection_name: str,
        update_operations: Sequence[types.UpdateOperation],
        **kwargs: Any,
    ) -> List[types.UpdateResult]:
        collection = self._get_collection(collection_name)
        collection.batch_update_points(update_operations)
        return [self._default_update_result()] * len(update_operations)

    def update_collection_aliases(
        self, change_aliases_operations: Sequence[types.AliasOperations], **kwargs: Any
    ) -> bool:
        for operation in change_aliases_operations:
            if isinstance(operation, rest_models.CreateAliasOperation):
                self._get_collection(operation.create_alias.collection_name)
                self.aliases[operation.create_alias.alias_name] = (
                    operation.create_alias.collection_name
                )
            elif isinstance(operation, rest_models.DeleteAliasOperation):
                self.aliases.pop(operation.delete_alias.alias_name, None)
            elif isinstance(operation, rest_models.RenameAliasOperation):
                new_name = operation.rename_alias.new_alias_name
                old_name = operation.rename_alias.old_alias_name
                self.aliases[new_name] = self.aliases.pop(old_name)
            else:
                raise ValueError(f"Unknown operation: {operation}")
        self._save()
        return True

    def get_collection_aliases(
        self, collection_name: str, **kwargs: Any
    ) -> types.CollectionsAliasesResponse:
        if self.closed:
            raise RuntimeError("QdrantLocal instance is closed. Please create a new instance.")

        return types.CollectionsAliasesResponse(
            aliases=[
                rest_models.AliasDescription(
                    alias_name=alias_name,
                    collection_name=name,
                )
                for alias_name, name in self.aliases.items()
                if name == collection_name
            ]
        )

    def get_aliases(self, **kwargs: Any) -> types.CollectionsAliasesResponse:
        if self.closed:
            raise RuntimeError("QdrantLocal instance is closed. Please create a new instance.")

        return types.CollectionsAliasesResponse(
            aliases=[
                rest_models.AliasDescription(
                    alias_name=alias_name,
                    collection_name=name,
                )
                for alias_name, name in self.aliases.items()
            ]
        )

    def get_collections(self, **kwargs: Any) -> types.CollectionsResponse:
        if self.closed:
            raise RuntimeError("QdrantLocal instance is closed. Please create a new instance.")

        return types.CollectionsResponse(
            collections=[
                rest_models.CollectionDescription(name=name)
                for name, _ in self.collections.items()
            ]
        )

    def get_collection(self, collection_name: str, **kwargs: Any) -> types.CollectionInfo:
        collection = self._get_collection(collection_name)
        return collection.info()

    def collection_exists(self, collection_name: str, **kwargs: Any) -> bool:
        try:
            self._get_collection(collection_name)
            return True
        except ValueError:
            return False

    def update_collection(
        self,
        collection_name: str,
        sparse_vectors_config: Optional[Mapping[str, types.SparseVectorParams]] = None,
        **kwargs: Any,
    ) -> bool:
        _collection = self._get_collection(collection_name)

        if sparse_vectors_config is not None:
            for vector_name, vector_params in sparse_vectors_config.items():
                _collection.update_sparse_vectors_config(vector_name, vector_params)

            return True
        return False

    def _collection_path(self, collection_name: str) -> Optional[str]:
        if self.persistent:
            return os.path.join(self.location, "collection", collection_name)
        else:
            return None

    def delete_collection(self, collection_name: str, **kwargs: Any) -> bool:
        if self.closed:
            raise RuntimeError("QdrantLocal instance is closed. Please create a new instance.")

        _collection = self.collections.pop(collection_name, None)
        del _collection
        self.aliases = {
            alias_name: name
            for alias_name, name in self.aliases.items()
            if name != collection_name
        }
        collection_path = self._collection_path(collection_name)
        if collection_path is not None:
            shutil.rmtree(collection_path, ignore_errors=True)
        self._save()
        return True

    def create_collection(
        self,
        collection_name: str,
        vectors_config: Union[types.VectorParams, Mapping[str, types.VectorParams]],
        init_from: Optional[types.InitFrom] = None,
        sparse_vectors_config: Optional[Mapping[str, types.SparseVectorParams]] = None,
        **kwargs: Any,
    ) -> bool:
        if self.closed:
            raise RuntimeError("QdrantLocal instance is closed. Please create a new instance.")

        src_collection = None
        from_collection_name = None
        if init_from is not None:
            from_collection_name = (
                init_from if isinstance(init_from, str) else init_from.collection
            )
            src_collection = self._get_collection(from_collection_name)

        if collection_name in self.collections:
            raise ValueError(f"Collection {collection_name} already exists")
        collection_path = self._collection_path(collection_name)
        if collection_path is not None:
            os.makedirs(collection_path, exist_ok=True)

        collection = LocalCollection(
            rest_models.CreateCollection(
                vectors=vectors_config,
                sparse_vectors=sparse_vectors_config,
            ),
            location=collection_path,
            force_disable_check_same_thread=self.force_disable_check_same_thread,
        )
        self.collections[collection_name] = collection

        if src_collection and from_collection_name:
            batch_size = 100
            records, next_offset = self.scroll(from_collection_name, limit=2, with_vectors=True)
            self.upload_records(
                collection_name, records
            )  # it is not crucial to replace upload_records here
            # since it is an internal usage, and we don't have custom shard keys in qdrant local
            while next_offset is not None:
                records, next_offset = self.scroll(
                    from_collection_name, offset=next_offset, limit=batch_size, with_vectors=True
                )
                self.upload_records(collection_name, records)

        self._save()
        return True

    def recreate_collection(
        self,
        collection_name: str,
        vectors_config: Union[types.VectorParams, Mapping[str, types.VectorParams]],
        init_from: Optional[types.InitFrom] = None,
        sparse_vectors_config: Optional[Mapping[str, types.SparseVectorParams]] = None,
        **kwargs: Any,
    ) -> bool:
        self.delete_collection(collection_name)
        return self.create_collection(
            collection_name, vectors_config, init_from, sparse_vectors_config
        )

    def upload_points(
        self, collection_name: str, points: Iterable[types.PointStruct], **kwargs: Any
    ) -> None:
        self._upload_points(collection_name, points)

    def upload_records(
        self, collection_name: str, records: Iterable[types.Record], **kwargs: Any
    ) -> None:
        # upload_records in local mode behaves like upload_records with wait=True in server mode
        self._upload_points(collection_name, records)

    def _upload_points(
        self,
        collection_name: str,
        points: Iterable[Union[types.PointStruct, types.Record]],
    ) -> None:
        collection = self._get_collection(collection_name)
        collection.upsert(
            [
                rest_models.PointStruct(
                    id=point.id,
                    vector=point.vector or {},
                    payload=point.payload or {},
                )
                for point in points
            ]
        )

    def upload_collection(
        self,
        collection_name: str,
        vectors: Union[
            Dict[str, types.NumpyArray], types.NumpyArray, Iterable[types.VectorStruct]
        ],
        payload: Optional[Iterable[Dict[Any, Any]]] = None,
        ids: Optional[Iterable[types.PointId]] = None,
        **kwargs: Any,
    ) -> None:
        # upload_collection in local mode behaves like upload_collection with wait=True in server mode
        def uuid_generator() -> Generator[str, None, None]:
            while True:
                yield str(uuid4())

        collection = self._get_collection(collection_name)
        if isinstance(vectors, dict) and any(isinstance(v, np.ndarray) for v in vectors.values()):
            assert (
                len(set([arr.shape[0] for arr in vectors.values()])) == 1
            ), "Each named vector should have the same number of vectors"

            num_vectors = next(iter(vectors.values())).shape[0]
            # convert Dict[str, np.ndarray] to List[Dict[str, List[float]]]
            vectors = [
                {name: vectors[name][i].tolist() for name in vectors.keys()}
                for i in range(num_vectors)
            ]

        collection.upsert(
            [
                rest_models.PointStruct(
                    id=point_id,
                    vector=(vector.tolist() if isinstance(vector, np.ndarray) else vector) or {},
                    payload=payload or {},
                )
                for (point_id, vector, payload) in zip(
                    ids or uuid_generator(),
                    iter(vectors),
                    payload or itertools.cycle([{}]),
                )
            ]
        )

    def create_payload_index(
        self,
        collection_name: str,
        field_name: str,
        field_schema: Optional[types.PayloadSchemaType] = None,
        field_type: Optional[types.PayloadSchemaType] = None,
        **kwargs: Any,
    ) -> types.UpdateResult:
        logging.warning(
            "Payload indexes have no effect in the local Qdrant. Please use server Qdrant if you need payload indexes."
        )
        return self._default_update_result()

    def delete_payload_index(
        self, collection_name: str, field_name: str, **kwargs: Any
    ) -> types.UpdateResult:
        logging.warning(
            "Payload indexes have no effect in the local Qdrant. Please use server Qdrant if you need payload indexes."
        )
        return self._default_update_result()

    def list_snapshots(
        self, collection_name: str, **kwargs: Any
    ) -> List[types.SnapshotDescription]:
        return []

    def create_snapshot(
        self, collection_name: str, **kwargs: Any
    ) -> Optional[types.SnapshotDescription]:
        raise NotImplementedError(
            "Snapshots are not supported in the local Qdrant. Please use server Qdrant if you need full snapshots."
        )

    def delete_snapshot(self, collection_name: str, snapshot_name: str, **kwargs: Any) -> bool:
        raise NotImplementedError(
            "Snapshots are not supported in the local Qdrant. Please use server Qdrant if you need full snapshots."
        )

    def list_full_snapshots(self, **kwargs: Any) -> List[types.SnapshotDescription]:
        return []

    def create_full_snapshot(self, **kwargs: Any) -> types.SnapshotDescription:
        raise NotImplementedError(
            "Snapshots are not supported in the local Qdrant. Please use server Qdrant if you need full snapshots."
        )

    def delete_full_snapshot(self, snapshot_name: str, **kwargs: Any) -> bool:
        raise NotImplementedError(
            "Snapshots are not supported in the local Qdrant. Please use server Qdrant if you need full snapshots."
        )

    def recover_snapshot(self, collection_name: str, location: str, **kwargs: Any) -> bool:
        raise NotImplementedError(
            "Snapshots are not supported in the local Qdrant. Please use server Qdrant if you need full snapshots."
        )

    def list_shard_snapshots(
        self, collection_name: str, shard_id: int, **kwargs: Any
    ) -> List[types.SnapshotDescription]:
        return []

    def create_shard_snapshot(
        self, collection_name: str, shard_id: int, **kwargs: Any
    ) -> Optional[types.SnapshotDescription]:
        raise NotImplementedError(
            "Snapshots are not supported in the local Qdrant. Please use server Qdrant if you need snapshots."
        )

    def delete_shard_snapshot(
        self, collection_name: str, shard_id: int, snapshot_name: str, **kwargs: Any
    ) -> bool:
        raise NotImplementedError(
            "Snapshots are not supported in the local Qdrant. Please use server Qdrant if you need snapshots."
        )

    def recover_shard_snapshot(
        self,
        collection_name: str,
        shard_id: int,
        location: str,
        **kwargs: Any,
    ) -> bool:
        raise NotImplementedError(
            "Snapshots are not supported in the local Qdrant. Please use server Qdrant if you need snapshots."
        )

    def lock_storage(self, reason: str, **kwargs: Any) -> types.LocksOption:
        raise NotImplementedError(
            "Locks are not supported in the local Qdrant. Please use server Qdrant if you need full snapshots."
        )

    def unlock_storage(self, **kwargs: Any) -> types.LocksOption:
        raise NotImplementedError(
            "Locks are not supported in the local Qdrant. Please use server Qdrant if you need full snapshots."
        )

    def get_locks(self, **kwargs: Any) -> types.LocksOption:
        return types.LocksOption(
            error_message=None,
            write=False,
        )

    def create_shard_key(
        self,
        collection_name: str,
        shard_key: types.ShardKey,
        shards_number: Optional[int] = None,
        replication_factor: Optional[int] = None,
        placement: Optional[List[int]] = None,
        **kwargs: Any,
    ) -> bool:
        raise NotImplementedError(
            "Sharding is not supported in the local Qdrant. Please use server Qdrant if you need sharding."
        )

    def delete_shard_key(
        self,
        collection_name: str,
        shard_key: types.ShardKey,
        **kwargs: Any,
    ) -> bool:
        raise NotImplementedError(
            "Sharding is not supported in the local Qdrant. Please use server Qdrant if you need sharding."
        )

    def info(self) -> types.VersionInfo:
        version = importlib.metadata.version("qdrant-client")
        return rest_models.VersionInfo(
            title="qdrant - vector search engine", version=version, commit=None
        )