Comparison: Open Source vs. Proprietary Vector DBs 2026: Pinecone vs. Weaviate vs. Qdrant

import uuid
import random
import time
from qdrant_client import QdrantClient
from qdrant_client.http import models
from qdrant_client.http.exceptions import UnexpectedResponse, TimeoutException
from typing import List, Dict, Any

def batch_upsert_qdrant(
    collection_name: str,
    vectors: List[List[float]],
    payloads: List[Dict[str, Any]],
    batch_size: int = 1000,
    host: str = "localhost",
    port: int = 6333,
    timeout: int = 30
) -> int:
    """
    Batch upserts vectors to Qdrant 1.8.0 with error handling and retry logic.
    Returns total number of successfully upserted vectors.
    """
    client = QdrantClient(host=host, port=port, timeout=timeout)
    total_upserted = 0

    try:
        # Check if collection exists, create if not
        collections = client.get_collections().collections
        existing_names = [c.name for c in collections]

        if collection_name not in existing_names:
            print(f"Creating collection {collection_name}...")
            client.create_collection(
                collection_name=collection_name,
                vectors_config=models.VectorParams(
                    size=len(vectors[0]) if vectors else 768,
                    distance=models.Distance.COSINE
                ),
                optimizers_config=models.OptimizersConfigDiff(
                    default_segment_number=2,
                    max_segment_size=100000
                )
            )
            print(f"Collection {collection_name} created successfully.")

        # Batch upsert with retry logic
        for i in range(0, len(vectors), batch_size):
            batch_vectors = vectors[i:i+batch_size]
            batch_payloads = payloads[i:i+batch_size]
            point_ids = [str(uuid.uuid4()) for _ in range(len(batch_vectors))]

            retry_count = 0
            max_retries = 3
            while retry_count < max_retries:
                try:
                    upsert_result = client.upsert(
                        collection_name=collection_name,
                        points=models.Batch(
                            ids=point_ids,
                            vectors=batch_vectors,
                            payloads=batch_payloads
                        ),
                        wait=True
                    )
                    total_upserted += len(batch_vectors)
                    print(f"Upserted batch {i//batch_size + 1}: {len(batch_vectors)} vectors")
                    break
                except (UnexpectedResponse, TimeoutException) as e:
                    retry_count += 1
                    print(f"Retry {retry_count}/{max_retries} for batch {i//batch_size + 1}: {str(e)}")
                    time.sleep(2 ** retry_count)  # Exponential backoff
                    if retry_count == max_retries:
                        print(f"Failed to upsert batch {i//batch_size + 1} after {max_retries} retries")
                        raise

        print(f"Total upserted: {total_upserted} vectors")
        return total_upserted

    except Exception as e:
        print(f"Fatal error during Qdrant upsert: {str(e)}")
        raise
    finally:
        client.close()

# Example usage (valid, runnable code)
if __name__ == "__main__":
    # Generate 10,000 768-dim test vectors (simulated embeddings)
    test_vectors = [[random.random() for _ in range(768)] for _ in range(10000)]
    test_payloads = [
        {"doc_id": f"doc_{i}", "source": "benchmark", "timestamp": time.time()}
        for i in range(10000)
    ]

    try:
        upserted = batch_upsert_qdrant(
            collection_name="benchmark_768d_10k",
            vectors=test_vectors,
            payloads=test_payloads,
            batch_size=500,
            host="localhost",
            port=6333
        )
        print(f"Successfully upserted {upserted} vectors to Qdrant")
    except Exception as e:
        print(f"Upsert failed: {str(e)}")

import uuid
import random
import time
from weaviate import Client, Batch
from weaviate.exceptions import UnexpectedStatusCodeException, RequestsConnectionError
from weaviate.auth import AuthApiKey
from typing import List, Dict, Any

def weaviate_hybrid_search(
    class_name: str,
    query_vector: List[float],
    query_text: str,
    limit: int = 10,
    alpha: float = 0.5,
    weaviate_url: str = "http://localhost:8080",
    api_key: str = None
) -> List[Dict[str, Any]]:
    """
    Performs hybrid BM25 + vector search on Weaviate 1.24.3 with error handling.
    Alpha balances vector (1.0) vs keyword (0.0) search weight.
    """
    auth = AuthApiKey(api_key) if api_key else None
    client = Client(
        url=weaviate_url,
        auth_client_secret=auth,
        timeout=(30, 60)  # Connect timeout, read timeout
    )

    try:
        # Verify Weaviate connection
        client.schema.get()
        print("Connected to Weaviate successfully")
    except RequestsConnectionError as e:
        print(f"Failed to connect to Weaviate: {str(e)}")
        raise

    try:
        # Check if class exists, create if not
        schema = client.schema.get()
        existing_classes = [c["class"] for c in schema["classes"]] if "classes" in schema else []

        if class_name not in existing_classes:
            print(f"Creating Weaviate class {class_name}...")
            class_schema = {
                "class": class_name,
                "vectorizer": "none",  # We provide our own vectors
                "properties": [
                    {"name": "doc_id", "dataType": ["string"]},
                    {"name": "source", "dataType": ["string"]},
                    {"name": "timestamp", "dataType": ["number"]}
                ],
                "vectorIndexConfig": {
                    "distance": "cosine",
                    "maxConnections": 64,
                    "efConstruction": 128
                }
            }
            client.schema.create_class(class_schema)
            print(f"Class {class_name} created successfully")

        # Batch upsert 10k test vectors
        batch = client.batch(
            batch_size=1000,
            dynamic=True,
            timeout_retries=3
        )

        print("Starting batch upsert to Weaviate...")
        with batch as b:
            for i in range(10000):
                vector = [random.random() for _ in range(768)]
                payload = {
                    "doc_id": f"doc_{i}",
                    "source": "benchmark",
                    "timestamp": time.time()
                }
                b.add_data_object(
                    data_object=payload,
                    class_name=class_name,
                    uuid=str(uuid.uuid4()),
                    vector=vector
                )
        print("Batch upsert completed")

        # Perform hybrid search
        print(f"Running hybrid search with alpha={alpha}...")
        result = (
            client.query
            .get(class_name, ["doc_id", "source", "timestamp", "_additional {certainty}"])
            .with_hybrid(
                query=query_text,
                vector=query_vector,
                alpha=alpha
            )
            .with_limit(limit)
            .do()
        )

        if "errors" in result:
            raise UnexpectedStatusCodeException(f"Hybrid search failed: {result['errors']}")

        return result["data"]["Get"][class_name]

    except UnexpectedStatusCodeException as e:
        print(f"Weaviate API error: {str(e)}")
        raise
    finally:
        client.close()

# Example usage
if __name__ == "__main__":
    test_query_vector = [random.random() for _ in range(768)]
    test_query_text = "vector database benchmark"

    try:
        results = weaviate_hybrid_search(
            class_name="Benchmark768d",
            query_vector=test_query_vector,
            query_text=test_query_text,
            limit=5,
            alpha=0.7,
            weaviate_url="http://localhost:8080"
        )
        print(f"Hybrid search returned {len(results)} results:")
        for res in results:
            print(f"Doc ID: {res['doc_id']}, Certainty: {res['_additional']['certainty']}")
    except Exception as e:
        print(f"Hybrid search failed: {str(e)}")

import os
import random
import time
from pinecone import Pinecone, ServerlessSpec, PodSpec
from pinecone.core.exceptions import PineconeException, TimeoutException
from typing import List, Dict, Any

def pinecone_serverless_workflow(
    index_name: str,
    vectors: List[List[float]],
    payloads: List[Dict[str, Any]],
    cloud: str = "aws",
    region: str = "us-east-1",
    metric: str = "cosine"
) -> Dict[str, Any]:
    """
    Manages Pinecone serverless index lifecycle: create, upsert, query, delete.
    Uses Pinecone Python client v3.1.0 (2026 stable release).
    """
    # Initialize Pinecone client with API key from env
    api_key = os.getenv("PINECONE_API_KEY")
    if not api_key:
        raise ValueError("PINECONE_API_KEY environment variable not set")

    pc = Pinecone(api_key=api_key)
    results = {"upserted": 0, "query_time_ms": 0, "index_name": index_name}

    try:
        # Check if index exists, create serverless index if not
        existing_indexes = [idx.name for idx in pc.list_indexes()]

        if index_name not in existing_indexes:
            print(f"Creating serverless index {index_name}...")
            pc.create_index(
                name=index_name,
                dimension=len(vectors[0]) if vectors else 768,
                metric=metric,
                spec=ServerlessSpec(
                    cloud=cloud,
                    region=region
                )
            )
            # Wait for index to be ready
            while not pc.describe_index(index_name).status["ready"]:
                print("Waiting for index to initialize...")
                time.sleep(5)
            print(f"Index {index_name} created and ready")

        # Connect to index
        idx = pc.Index(index_name)

        # Batch upsert with 200-vector batches (Pinecone max batch size)
        batch_size = 200
        total_upserted = 0
        print(f"Upserting {len(vectors)} vectors in batches of {batch_size}...")

        for i in range(0, len(vectors), batch_size):
            batch_vectors = vectors[i:i+batch_size]
            batch_payloads = payloads[i:i+batch_size]
            # Generate Pinecone-compatible IDs (string, max 512 chars)
            batch_ids = [f"vec_{i+j}" for j in range(len(batch_vectors))]

            retry_count = 0
            max_retries = 3
            while retry_count < max_retries:
                try:
                    upsert_response = idx.upsert(
                        vectors=zip(batch_ids, batch_vectors, batch_payloads),
                        namespace="benchmark_namespace"
                    )
                    total_upserted += upsert_response["upserted_count"]
                    print(f"Upserted batch {i//batch_size + 1}: {upsert_response['upserted_count']} vectors")
                    break
                except (PineconeException, TimeoutException) as e:
                    retry_count += 1
                    print(f"Retry {retry_count}/{max_retries} for batch {i//batch_size + 1}: {str(e)}")
                    time.sleep(2 ** retry_count)
                    if retry_count == max_retries:
                        print(f"Failed to upsert batch {i//batch_size + 1}")
                        raise

        results["upserted"] = total_upserted

        # Query example
        print("Running sample query...")
        query_start = time.time()
        query_response = idx.query(
            vector=[random.random() for _ in range(768)],
            top_k=10,
            namespace="benchmark_namespace",
            include_metadata=True
        )
        query_time = (time.time() - query_start) * 1000
        results["query_time_ms"] = round(query_time, 2)
        results["sample_results"] = len(query_response["matches"])
        print(f"Query returned {len(query_response['matches'])} matches in {query_time:.2f}ms")

        return results

    except PineconeException as e:
        print(f"Pinecone API error: {str(e)}")
        raise
    finally:
        # Cleanup: delete index if it's a test
        # pc.delete_index(index_name)
        # print(f"Deleted test index {index_name}")
        pass

# Example usage
if __name__ == "__main__":
    # Generate 5000 test vectors (Pinecone serverless free tier limit is 100k vectors)
    test_vectors = [[random.random() for _ in range(768)] for _ in range(5000)]
    test_payloads = [
        {"doc_id": f"pinecone_doc_{i}", "source": "benchmark", "timestamp": time.time()}
        for i in range(5000)
    ]

    try:
        workflow_results = pinecone_serverless_workflow(
            index_name="pinecone-benchmark-768d",
            vectors=test_vectors,
            payloads=test_payloads,
            cloud="aws",
            region="us-east-1"
        )
        print(f"Workflow completed: {workflow_results}")
    except Exception as e:
        print(f"Pinecone workflow failed: {str(e)}")

# Qdrant HNSW tuning example
from qdrant_client import QdrantClient
from qdrant_client.http import models

client = QdrantClient(host="localhost", port=6333)
client.create_collection(
    collection_name="tuned_hnsw_collection",
    vectors_config=models.VectorParams(size=768, distance=models.Distance.COSINE),
    optimizers_config=models.OptimizersConfigDiff(
        default_segment_number=4,
        max_segment_size=200000
    ),
    hnsw_config=models.HnswConfigDiff(
        ef_construct=256,  # Tune this for recall
        m=64,  # Tune this for latency
        ef=128  # Default efSearch value
    )
)

# Pinecone namespace query example
import os
from pinecone import Pinecone

pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
idx = pc.Index("saas-multi-tenant-index")

# Query tenant_123's isolated namespace
results = idx.query(
    vector=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] + [0.1]*758,  # 768-dim vector
    top_k=10,
    namespace="tenant_123",  # Isolated namespace for tenant 123
    include_metadata=True
)

# Query without namespace hits all tenants (avoid for multi-tenant)
# results = idx.query(vector=..., top_k=10)  # DON'T DO THIS

# Simple benchmark snippet for all three DBs
import time

def benchmark_qps(query_func, num_queries=10000):
    start = time.time()
    for _ in range(num_queries):
        query_func()
    elapsed = time.time() - start
    return num_queries / elapsed

# Example for Qdrant
from qdrant_client import QdrantClient
qdrant_client = QdrantClient(host="localhost", port=6333)
qdrant_qps = benchmark_qps(
    query_func=lambda: qdrant_client.search(
        collection_name="benchmark",
        query_vector=[0.1]*768,
        limit=10
    )
)
print(f"Qdrant QPS: {qdrant_qps:.0f}")