search_pagination_duplicate_test.py

import random
import subprocess
import uuid
from qdrant_client import QdrantClient
from qdrant_client import models

client = QdrantClient(host="127.0.0.1", port=6333)

collection_name = "test"
page_size = 20
pages = 5
searches = 50
tests = 20

def run():
    for i in range(tests):
        print(f"Test {i + 1}")

        # Create new collection with bfb
        subprocess.run(["bfb", "--collection-name", collection_name, "--indexing-threshold=10", "-d4", "-n10000"])

        for _ in range(searches):
            search_test()

    print("Done")


def search_test():
    vector = [random.random(), random.random(), random.random(), random.random()]
    ids = []

    for page in range(pages):
        results = search(vector, page=page);

        # Ensure point IDs are not in list
        for found in results:
            if found.id in ids:
                print(f"- DUPLICATE ID {found.id} PAGE {page}")
            ids.append(found.id)


def search(vector, page=0):
    return client.search(
        collection_name=collection_name,
        query_vector=vector,
        offset=page * page_size,
        limit=page_size,
    )


run()