Function test_performance

tests/test-performance.py:89–213

()

def test_performance():
    # Get password from .env file or prompt securely
    password = os.getenv("COSDATA_PASSWORD")
    if not password:
        password = getpass.getpass("Enter your database password: ")

    # Initialize client
    host = os.getenv("COSDATA_HOST", "http://127.0.0.1:8443")
    username = os.getenv("COSDATA_USERNAME", "admin")
    client = Client(host=host, username=username, password=password, verify=False)

    # Test parameters
    collection_name = "performance_test"
    dimensions = 1024
    perturbation_degree = 0.95
    batch_size = 100
    batch_count = 1000
    num_workers = 32

    try:
        # Create collection
        print("Creating collection...")
        collection = client.create_collection(
            name=collection_name,
            dimension=dimensions,
            description="Performance test collection",
        )

        # Create index
        print("Creating index...")
        collection.create_index(
            distance_metric="cosine",
            num_layers=7,
            max_cache_size=1000,
            ef_construction=512,
            ef_search=256,
            neighbors_count=32,
            level_0_neighbors_count=64,
        )

        start_time = time.time()

        # Generate vectors in parallel
        print("Generating test vectors...")
        test_vectors = []
        with ThreadPoolExecutor(max_workers=num_workers) as executor:
            futures = []
            for base_idx in range(batch_count):
                futures.append(
                    executor.submit(
                        generate_batch,
                        base_idx,
                        batch_size,
                        dimensions,
                        perturbation_degree,
                    )
                )