Capsule 04: The ChromaDB → Pinecone migration — without losing data or breaking queries

Capsule description

The actual migration is where most projects get complicated. It isn't "export from ChromaDB and dump into Pinecone" — that fails 50% of the time. The difference between a smooth migration and an incident is in the details: integrity validation, per-batch error handling, dual-write during the transition, a rollback plan, and A/B testing before the full switchover.

This capsule teaches you the right migration: an incremental strategy with minimal downtime, robust scripts with exponential retry, count and integrity validation after every step, and a rollback plan if something goes wrong in production.

By the end of this capsule you'll be able to:

• ✅ Implement a batched migration script with retry and logging
• ✅ Validate integrity: counts, preserved metadata, consistent embeddings
• ✅ Design an incremental strategy (not big-bang) with dual-write during the transition
• ✅ Implement a rollback plan with a feature flag
• ✅ Validate retrieval quality post-migration with an eval set
• ✅ Anticipate the four most common errors: payload size, metadata types, the wrong namespace, a dimension mismatch

Estimated time: 35-40 minutes

The strategy: incremental, not big-bang

                            BIG-BANG (dangerous)

   Day 1: Stop the app
   Day 1-3: Migrate everything to Pinecone
   Day 3: Switch the app to Pinecone
   Day 3+: Hope it works

   If something fails: the rollback takes hours, extended downtime


                          INCREMENTAL (recommended)

   Week 1: Pinecone setup + the migration script + tests
   Week 2: Dual-write (write to ChromaDB AND Pinecone for every new doc)
   Week 3: Backfill (migrate the ChromaDB history into Pinecone)
   Week 4: Dual-read with an A/B test (query both, compare)
   Week 5: Switch the primary to Pinecone, ChromaDB as the fallback
   Week 6: Decommission ChromaDB

   If something fails: rollback in minutes via a feature flag

Incremental migration is the rule. Big-bang is only justified for a small corpus (<100K docs) or systems with planned downtime.

The base script: exporting from ChromaDB

# migration_script.py
from typing import Iterable
import chromadb
from chromadb.utils import embedding_functions
import os


def export_from_chromadb(collection_name: str, batch_size: int = 1000) -> Iterable[dict]:
    """
    Exports documents in batches so you don't load everything into RAM.
    Yields batches of dicts with id, embedding, metadata.
    """
    chroma_client = chromadb.PersistentClient(path="./chroma_db")
    collection = chroma_client.get_collection(collection_name)

    total = collection.count()
    print(f"Exporting {total} docs from '{collection_name}'")

    offset = 0
    while offset < total:
        batch = collection.get(
            limit=batch_size,
            offset=offset,
            include=["embeddings", "metadatas", "documents"],
        )

        # Yield the batch as a list of dicts
        items = []
        for i in range(len(batch["ids"])):
            items.append({
                "id": batch["ids"][i],
                "embedding": batch["embeddings"][i],
                "metadata": batch["metadatas"][i] or {},
                "document": batch["documents"][i],
            })

        yield items
        offset += batch_size
        print(f"  Exported {min(offset, total)}/{total}")

Why batches: large collections (5M+ docs) don't fit in RAM. A batch of 1000 keeps RAM at ~500MB.

Upserting to Pinecone with retry

import time
from pinecone import Pinecone


def upsert_to_pinecone(
    index,
    items: list,
    namespace: str = "",
    max_retries: int = 3,
) -> int:
    """
    Upserts a batch to Pinecone with exponential retry.
    Returns the count of successfully inserted items.
    """
    # Build the vectors in Pinecone's format
    vectors = []
    for item in items:
        # Pinecone metadata must be serializable and flat
        metadata = clean_metadata_for_pinecone(item["metadata"])

        # Add the document's content as metadata for later retrieval
        metadata["content"] = item["document"][:40000]  # Pinecone limit

        vectors.append({
            "id": item["id"],
            "values": item["embedding"],
            "metadata": metadata,
        })

    # Upsert with exponential retry
    for attempt in range(max_retries):
        try:
            response = index.upsert(vectors=vectors, namespace=namespace)
            return response.upserted_count
        except Exception as e:
            if attempt == max_retries - 1:
                # The last attempt, re-raise
                raise
            wait_time = 2 ** attempt  # 1, 2, 4 seconds
            print(f"  Error on attempt {attempt + 1}: {e}. Retrying in {wait_time}s...")
            time.sleep(wait_time)


def clean_metadata_for_pinecone(metadata: dict) -> dict:
    """
    Pinecone restricts metadata to: strings, numbers, booleans, lists of strings.
    No nested dicts or other types.
    """
    cleaned = {}
    for key, value in metadata.items():
        if isinstance(value, (str, int, float, bool)):
            cleaned[key] = value
        elif isinstance(value, list):
            # Convert everything to strings
            cleaned[key] = [str(v) for v in value]
        elif value is None:
            # Skip None values
            continue
        elif isinstance(value, dict):
            # Flatten the dict: {a: {b: 1}} → {a_b: 1}
            for sub_key, sub_value in value.items():
                if isinstance(sub_value, (str, int, float, bool)):
                    cleaned[f"{key}_{sub_key}"] = sub_value
        else:
            # Convert other types to a string
            cleaned[key] = str(value)
    return cleaned

The important details:

• Pinecone has a limit of 40,000 characters per metadata field. Truncate content.
• Metadata must be flat (no nested dicts). Flatten it if necessary.
• Only primitive types in metadata: str, int, float, bool, list[str].

Integrity validation

After migrating, validate that NOTHING was lost:

def validate_migration(
    chroma_collection,
    pinecone_index,
    namespace: str = "",
) -> dict:
    """
    Post-migration validations:
    1. The counts match
    2. A sample of random IDs exists in both
    3. The metadata is preserved
    """
    # 1. The counts
    chroma_count = chroma_collection.count()
    pinecone_stats = pinecone_index.describe_index_stats()
    pinecone_count = pinecone_stats["namespaces"].get(namespace, {}).get("vector_count", 0)

    print(f"Counts: ChromaDB={chroma_count}, Pinecone={pinecone_count}")
    if chroma_count != pinecone_count:
        return {
            "valid": False,
            "error": f"Count mismatch: chroma={chroma_count} != pinecone={pinecone_count}",
        }

    # 2. A sample of random IDs
    import random
    chroma_data = chroma_collection.get(limit=20)  # the first 20 as a sample
    sample_ids = random.sample(chroma_data["ids"], min(20, len(chroma_data["ids"])))

    pinecone_fetched = pinecone_index.fetch(ids=sample_ids, namespace=namespace)
    missing_ids = [
        sid for sid in sample_ids
        if sid not in pinecone_fetched.get("vectors", {})
    ]

    if missing_ids:
        return {
            "valid": False,
            "error": f"Missing IDs in Pinecone: {missing_ids[:5]}",
        }

    # 3. Validate that the metadata is preserved (a sample)
    for sid in sample_ids[:5]:
        chroma_meta = chroma_collection.get(ids=[sid], include=["metadatas"])["metadatas"][0]
        pinecone_meta = pinecone_fetched["vectors"][sid]["metadata"]

        # Compare the critical fields
        for critical_field in ["workspace_id", "doc_id", "type"]:
            if chroma_meta.get(critical_field) != pinecone_meta.get(critical_field):
                return {
                    "valid": False,
                    "error": f"Metadata mismatch for {sid} field {critical_field}",
                }

    return {
        "valid": True,
        "chroma_count": chroma_count,
        "pinecone_count": pinecone_count,
        "sample_validated": len(sample_ids),
    }


# Usage
result = validate_migration(chroma_collection, pinecone_index, namespace="acme_corp")
if result["valid"]:
    print("✅ Migration validated")
else:
    print(f"❌ Migration FAILED: {result['error']}")

Quality validation: a comparative eval set

Matching counts aren't enough. Validate that the retrieval gives similar results:

def validate_quality_post_migration(
    chroma_collection,
    pinecone_index,
    eval_set: list,
    threshold_recall: float = 0.85,
):
    """
    Compares the retrieval's recall between ChromaDB and Pinecone.
    If Pinecone has significantly lower recall, the migration has a problem.
    """
    chroma_recalls = []
    pinecone_recalls = []

    for item in eval_set:
        # The ChromaDB query
        chroma_results = chroma_collection.query(
            query_texts=[item["query"]],
            n_results=5,
            where={"workspace_id": item["workspace_id"]},
        )
        chroma_top_5 = set(chroma_results["ids"][0])

        # The Pinecone query
        from openai import OpenAI
        client = OpenAI()
        query_emb = client.embeddings.create(
            input=item["query"],
            model="text-embedding-3-small",
        ).data[0].embedding

        pinecone_results = pinecone_index.query(
            vector=query_emb,
            top_k=5,
            namespace=item["workspace_id"],
            include_metadata=True,
        )
        pinecone_top_5 = set(m["id"] for m in pinecone_results["matches"])

        # Recall vs the ground truth
        relevant = set(item["expected_doc_ids"])
        chroma_recall = len(chroma_top_5 & relevant) / max(len(relevant), 1)
        pinecone_recall = len(pinecone_top_5 & relevant) / max(len(relevant), 1)

        chroma_recalls.append(chroma_recall)
        pinecone_recalls.append(pinecone_recall)

    avg_chroma = sum(chroma_recalls) / len(chroma_recalls)
    avg_pinecone = sum(pinecone_recalls) / len(pinecone_recalls)

    delta = avg_pinecone - avg_chroma
    print(f"Recall ChromaDB: {avg_chroma:.2%}")
    print(f"Recall Pinecone: {avg_pinecone:.2%}")
    print(f"Delta: {delta:+.2%}")

    if avg_pinecone < threshold_recall:
        return {
            "valid": False,
            "error": f"Pinecone recall {avg_pinecone:.2%} < threshold {threshold_recall:.2%}",
        }

    if delta < -0.03:  # more than 3% worse
        return {
            "valid": False,
            "error": f"Pinecone recall is {delta:+.2%} worse than ChromaDB",
        }

    return {"valid": True, "chroma_recall": avg_chroma, "pinecone_recall": avg_pinecone}

The incremental strategy with a feature flag

During the transition, we want to be able to switch instantly between ChromaDB and Pinecone:

# retrieval.py
import os


VECTOR_DB_BACKEND = os.getenv("VECTOR_DB_BACKEND", "chromadb")  # chromadb | pinecone | dual


def retrieve(query: str, tenant: TenantContext, n_results: int = 5):
    """Retrieval that respects the VECTOR_DB_BACKEND feature flag."""

    if VECTOR_DB_BACKEND == "chromadb":
        return retrieve_chromadb(query, tenant, n_results)

    elif VECTOR_DB_BACKEND == "pinecone":
        return retrieve_pinecone(query, tenant, n_results)

    elif VECTOR_DB_BACKEND == "dual":
        # For A/B testing: run both, return chromadb's but log both
        chroma_results = retrieve_chromadb(query, tenant, n_results)
        try:
            pinecone_results = retrieve_pinecone(query, tenant, n_results)
            log_comparison(query, chroma_results, pinecone_results)
        except Exception as e:
            print(f"Pinecone failed (logging only): {e}")
        return chroma_results

    else:
        raise ValueError(f"Unknown backend: {VECTOR_DB_BACKEND}")

The rollout plan:

1. Deploy with VECTOR_DB_BACKEND=chromadb (no change).
2. Set VECTOR_DB_BACKEND=dual in production → it generates comparison logs.
3. If after 1 week the logs show consistent results, switch to pinecone.
4. If something's wrong, roll back to chromadb with an env var change (5 minutes, no redeploy).

Traps and common mistakes

Trap 1: a payload that's too large

The mistake: a batch of 5000 vectors with 3072-dim embeddings each.

The symptom: RequestError: Request body too large. Pinecone has a ~2MB limit per request.

How to prevent it: a batch_size of 100-200 with 1536d embeddings. If the dim is larger, a smaller batch_size.

Trap 2: metadata with unsupported types

The mistake:

metadata = {
    "tags": ["a", "b"],
    "config": {"nested": "dict"},  # ← Pinecone doesn't support a nested dict
    "score": np.float64(0.5),       # ← a numpy type
}

The symptom: ValidationError: metadata field "config" type not supported.

How to prevent it: clean_metadata_for_pinecone (shown above) flattens the dicts and converts the types.

Trap 3: a silent dimension mismatch

The mistake: ChromaDB has 1536d embeddings. The Pinecone index has dim=384 (left over from a test).

The symptom: Vector dimension 1536 does not match index dimension 384.

How to prevent it: validate the first batch's dimension before processing everything.

def validate_dimensions(items, expected_dim):
    if items and len(items[0]["embedding"]) != expected_dim:
        raise ValueError(
            f"Dimension mismatch: items={len(items[0]['embedding'])}, expected={expected_dim}"
        )

Trap 4: the wrong namespace

The mistake: forgetting the namespace in the upsert. Everything ends up in the default namespace "".

The symptom: queries filtering by namespace come back empty.

How to prevent it: the namespace is always mandatory in the wrappers:

def safe_upsert(index, vectors, workspace_id):
    if not workspace_id:
        raise ValueError("workspace_id is mandatory")
    namespace = f"ws_{workspace_id}"
    index.upsert(vectors=vectors, namespace=namespace)

Trap 5: not rate-limiting the migration script

The mistake: the script parallelizes the upsert with 50 concurrent workers.

The symptom: Pinecone rate limits, 429 errors, and the migration fails halfway.

How to prevent it: Pinecone Standard allows ~1000 upserts/second. With a batch of 100, that's 10 batches/second. Anything more requires a higher plan.

# Throttling
time.sleep(0.05)  # 50ms between batches → 20 batches/second

Trap 6: the migration cutting out halfway

The mistake: the VM runs out of memory halfway through. You have 3M of 5M docs migrated.

The symptom: a mixed state, and you can't tell what's missing.

How to prevent it: track the progress by offset in a file. If the script fails, resume from the last offset.

def migrate_with_checkpoint(checkpoint_file: str = "./migration_progress.txt"):
    last_offset = 0
    if os.path.exists(checkpoint_file):
        with open(checkpoint_file) as f:
            last_offset = int(f.read().strip())
        print(f"Resuming from offset {last_offset}")

    # ... migrate from last_offset onwards ...

    # Save progress periodically
    with open(checkpoint_file, "w") as f:
        f.write(str(current_offset))

Applied exercise

Scenario: you're an AI Engineer and you have to migrate to Pinecone. The data:

• ChromaDB with 3M chunks from 50 tenants
• Model: OpenAI text-embedding-3-small (1536 dim)
• The system is in production 24/7, there's no maintenance window
• Team: 3 people
• Deadline: 2 weeks

Your job:

1. Design the incremental migration plan.
2. Implement pseudo-code for the main script.
3. Define the rollback plan.

Days 1-2: Setup + tests
  - Create the Pinecone index (capsule 03)
  - Implement export_from_chromadb + upsert_to_pinecone
  - Tests with a small corpus (10K docs in staging)
  - Validate dimension/metric/quality

Days 3-5: The historical backfill (3M docs)
  - A migration script with checkpointing (resumable)
  - Throttling so we don't hit the rate limits
  - Logging of every batch + exponential retry
  - Estimated time: 30-45 minutes of pure work, ~3-4 hours with throttling

Days 6-7: Post-backfill validation
  - The counts match (ChromaDB vs Pinecone)
  - Sample 100 random IDs, validate the metadata is preserved
  - A comparative eval set (recall must be within 2% of the baseline)

Days 8-9: Dual-write
  - The app writes new docs into BOTH systems
  - Logs catch the inconsistencies
  - VECTOR_DB_BACKEND=chromadb (the queries still go there)

Days 10-12: Dual-read with an A/B
  - VECTOR_DB_BACKEND=dual
  - Queries go to chromadb (primary) and pinecone (shadow)
  - The logs compare the results
  - If the quality delta is <2%, we're OK to switch

Day 13: Switch the primary
  - VECTOR_DB_BACKEND=pinecone
  - ChromaDB stays as the fallback in case of errors
  - Extra monitoring for 24-48 hours

Day 14: Decommission
  - If Pinecone is stable for 48 hours, deprecate ChromaDB
  - A final ChromaDB backup to retain
  - Clean up the legacy code

# scripts/migrate_to_pinecone.py
import os
import time
from pathlib import Path

CHECKPOINT_FILE = Path("./migration_checkpoint.txt")
BATCH_SIZE = 100
THROTTLE_MS = 50


def migrate():
    chroma_collection = get_chroma_collection()
    pinecone_index = get_pinecone_index()

    # Resume from the checkpoint if it exists
    last_offset = 0
    if CHECKPOINT_FILE.exists():
        last_offset = int(CHECKPOINT_FILE.read_text().strip())
        print(f"Resuming from offset {last_offset}")

    total = chroma_collection.count()
    failed_batches = []

    for batch_start in range(last_offset, total, BATCH_SIZE):
        # Export
        batch = chroma_collection.get(
            limit=BATCH_SIZE,
            offset=batch_start,
            include=["embeddings", "metadatas", "documents"],
        )

        # Group by namespace (workspace_id)
        by_namespace = {}
        for i in range(len(batch["ids"])):
            ws_id = batch["metadatas"][i].get("workspace_id", "default")
            namespace = f"ws_{ws_id}"
            by_namespace.setdefault(namespace, []).append({
                "id": batch["ids"][i],
                "values": batch["embeddings"][i],
                "metadata": clean_metadata({**batch["metadatas"][i], "content": batch["documents"][i][:40000]}),
            })

        # Upsert per namespace
        for namespace, vectors in by_namespace.items():
            try:
                upsert_with_retry(pinecone_index, vectors, namespace)
            except Exception as e:
                print(f"FAILED batch {batch_start} namespace {namespace}: {e}")
                failed_batches.append((batch_start, namespace))

        # Save the checkpoint
        CHECKPOINT_FILE.write_text(str(batch_start + BATCH_SIZE))

        # Progress log
        progress = (batch_start + BATCH_SIZE) / total * 100
        print(f"  [{progress:.1f}%] Migrated up to offset {batch_start + BATCH_SIZE}")

        # Throttle
        time.sleep(THROTTLE_MS / 1000)

    # The final report
    print(f"\nMigration complete.")
    print(f"Failed batches: {failed_batches}")
    if failed_batches:
        # Save the list for a manual retry
        Path("./failed_batches.json").write_text(json.dumps(failed_batches))


if __name__ == "__main__":
    migrate()

# Change the env var in production
export VECTOR_DB_BACKEND=chromadb

# Restart the app (rolling, no downtime)
kubectl rollout restart deployment/rag-api

# 1. Keep ChromaDB as the primary (don't change it)
# 2. Delete the Pinecone index (the migrated data is invalid)
pinecone delete-index --name production-rag

# 3. Root cause analysis
# 4. Re-plan the migration with the problem fixed

# Identify which docs failed
failed = json.loads(Path("./failed_batches.json").read_text())

# Migrate only the failed ones with the adjusted script
for batch_start, namespace in failed:
    # ... retry ...

Summary and next step

What you learned:

• Incremental migration >> big-bang. Dual-write and dual-read allow a rollback in 5 minutes.
• Export in batches (don't load everything into RAM). A file checkpoint lets you resume.
• Pinecone metadata: only str, int, float, bool, list[str]. Nested dicts have to be flattened.
• A batch size of 100-200 for the upserts. Throttling so you don't hit the rate limits.
• Mandatory post-migration validation: counts + a sample of IDs + quality with the eval set.
• A feature flag with VECTOR_DB_BACKEND allows an instant switch in production.
• A rollback plan on three levels: immediate, the migrated data, a redo.

Checkpoint: before moving on, you should be able to:

• Implement a migration script with retry, throttling, checkpointing.
• Validate post-migration integrity with counts + a sample + the eval set.
• Design a rollback plan appropriate to your system.

Next capsule: 05 — Namespaces as native multi-tenant isolation.

In ChromaDB you used workspace_id as metadata to isolate tenants. Pinecone has a stronger mechanism: namespaces. Capsule 05 covers how to use them for isolation guaranteed at the engine level.

Resources

1. Pinecone — Upsert Data — The official API
2. Pinecone — Migration Guide — Migration patterns
3. Exponential Backoff (AWS Best Practices) — For the retry
4. Feature Flag Best Practices (LaunchDarkly) — For the rollout
5. ChromaDB — Get Operation — For the export
6. Pinecone — Rate Limits — For the throttling

Estimated time: 35-40 minutes Next: 05-namespaces-and-multi-tenancy.md