From 1122aca211005f9931d989a51dcbfeab92c23cfe Mon Sep 17 00:00:00 2001 From: Evgeniya Sukhodolskaya Date: Fri, 24 Jul 2026 18:48:11 +0200 Subject: [PATCH 1/2] add incremental embedding updates part 2 (scaling with bucket digests) --- README.md | 2 +- .../sync_at_scale_with_digests.ipynb | 480 ++++++++++++++++++ 2 files changed, 481 insertions(+), 1 deletion(-) create mode 100644 temporal-data-drift/sync_at_scale_with_digests.ipynb diff --git a/README.md b/README.md index 7542b2e..e9b5cc7 100644 --- a/README.md +++ b/README.md @@ -16,4 +16,4 @@ This repo contains a collection of tutorials, demos, and how-to guides on how to | [Collaborative Filtering and MovieLens](./sparse-vectors-movies-reco) | A notebook demonstrating how to build a collaborative filtering system using Qdrant | Sparse Vectors, Qdrant | | [Use semantic search to navigate your codebase](./code-search/) | Implement semantic search application for code search task | Qdrant, Python, sentence-transformers, Jina | -| [Incremental Embedding Updates](./temporal-data-drift) | Sync embeddings with changing raw text data | Qdrant Cloud Inference, Python | +| [Incremental Embedding Updates](./temporal-data-drift) | Sync embeddings with changing text: a simple pipeline (Part 1) and a bucket-digest version that scales with change (Part 2) | Qdrant Cloud Inference, Python | diff --git a/temporal-data-drift/sync_at_scale_with_digests.ipynb b/temporal-data-drift/sync_at_scale_with_digests.ipynb new file mode 100644 index 0000000..bafa07d --- /dev/null +++ b/temporal-data-drift/sync_at_scale_with_digests.ipynb @@ -0,0 +1,480 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "d4a95357", + "metadata": {}, + "source": [ + "# Incremental Embedding Updates at Scale (Part 2)\n", + "\n", + "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/qdrant/examples/blob/master/temporal-data-drift/sync_at_scale_with_digests.ipynb)\n", + "\n", + "[Part 1](https://qdrant.tech/documentation/tutorials-operations/incremental-embedding-updates/) keeps a Qdrant collection in sync with changing documentation by reconciling the full list of current chunks against the whole Qdrant collection on every run. It is a simple setup that works, but since every run reads the whole collection, the sync cost grows linearly with the size of the corpus.\n", + "\n", + "This notebook proposes an alternative, more involved approach for the case where scale is a real factor (millions or billions of chunks). On top of the Part 1 idea it adds a small summary layer, so each run scans a fixed-size summary and touches only the parts of the collection that changed." + ] + }, + { + "cell_type": "markdown", + "id": "77909da6", + "metadata": {}, + "source": [ + "## The Idea\n", + "\n", + "The method splits the chunks into a fixed number of **buckets** and gives each bucket a **digest**: a single number that summarizes everything in it. A bucket's digest is built from a small per-chunk value, the chunk's **contribution**, which depends on the chunk's position and its text. So if any chunk's text or position changes, a new chunk lands in the bucket, or one leaves it, the bucket's digest changes and we know that bucket needs a look. If the digest is unchanged, the bucket's contents are unchanged and we skip it.\n", + "\n", + "We keep two collections:\n", + "\n", + "- the **chunks collection**, whose embeddings serve vector search: the Part 1 collection plus one extra payload field, `sync_bucket`. The point ID is unchanged from Part 1 (a stable ID derived from the chunk's address). We do not store the per-chunk contributions; they are cheap to recompute, and only the per-bucket digest is kept.\n", + "- a small **summary collection**, holding each bucket and its digest. This is the sync state. For a given bucket count it is a fixed size, and it stays a small fraction of the collection even at large scale.\n", + "\n", + "Two things make it work:\n", + "\n", + "- **Comparison instead of scanning.** We compute the digests from the current source and compare them against the digests Qdrant already stored. Only the buckets whose digest differs get reconciled; the rest of the collection is never read.\n", + "- **XOR to combine contributions into a bucket digest.** XOR because:\n", + " 1. it is order-independent, so however the chunks come back, the same set of chunks gives the same digest;\n", + " 2. it folds any number of contributions into one fixed-width number with one cheap operation, so the summary stays small no matter how big the bucket;\n", + " 3. it is reversible (XORing a value twice cancels it), so at scale a bucket's digest could be patched for a single added or removed chunk instead of rebuilt. This notebook recomputes each changed bucket from scratch for clarity, but that reversibility is why XOR is the natural choice.\n", + "\n", + "### The Math\n", + "\n", + "Say we use 2^16 = 65536 buckets. For a corpus of 1,000,000 chunks that is about 15 chunks per bucket. Each run compares the 65536 digests to find the changed buckets, then reads chunks only from those buckets. If 50 chunks changed, they sit in at most ~50 buckets, so we read on the order of 50 x 15 = 750 chunks, not 1,000,000. Part 1 reads all 1,000,000 every run.\n", + "\n", + "We also pack the 65536 digests into a handful of points rather than one point per bucket, so the whole summary is a single read. The build section below explains why and how.\n", + "\n", + "To keep the summary printable, this notebook uses **16 buckets**. All the code is written against one constant, `N_BUCKETS`; set it to 65536 for production, and higher for very large corpora so each bucket stays small." + ] + }, + { + "cell_type": "markdown", + "id": "60d49886", + "metadata": {}, + "source": [ + "## Prerequisites\n", + "\n", + "This notebook uses Qdrant Cloud and its Free Tier Inference. Create a Free Tier Qdrant Cloud cluster and paste its URL and API key into the client cell below." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "885bf3bf", + "metadata": {}, + "outputs": [], + "source": [ + "%pip install -q \"qdrant-client>=1.18\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "809c6af9", + "metadata": {}, + "outputs": [], + "source": [ + "from qdrant_client import QdrantClient, models\n", + "\n", + "# Replace url and api_key with your own from https://cloud.qdrant.io\n", + "client = QdrantClient(\n", + " url=\"https://xyz-example.qdrant.io:6333\",\n", + " api_key=\"\",\n", + " cloud_inference=True\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "9233e503", + "metadata": {}, + "source": [ + "## The Chunks\n", + "\n", + "We use the same documentation chunks as Part 1. Each chunk has an address (which page, which section, which piece of that section) and its text. Two derived values carry the method:\n", + "\n", + "- `point_id`: a stable ID computed from the address, the chunk's `url`, section `anchor`, and `chunk_num`. Same address, same ID.\n", + "- `content_hash`: a fingerprint of the text. Same text, same hash." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "6e6808ed", + "metadata": {}, + "outputs": [], + "source": [ + "CHUNKS = [\n", + " {\"url\": \"https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/\",\n", + " \"anchor\": \"prerequisites\", \"chunk_num\": 0,\n", + " \"text\": \"Prerequisites - Docker and Docker Compose installed - curl available in your terminal ...\"},\n", + " {\"url\": \"https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/\",\n", + " \"anchor\": \"step-2-enable-tls\", \"chunk_num\": 0,\n", + " \"text\": \"Step 2: Enable TLS. Generate a local self-signed certificate and point Qdrant at it ...\"},\n", + " {\"url\": \"https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/\",\n", + " \"anchor\": \"step-3-enable-an-admin-api-key\", \"chunk_num\": 0,\n", + " \"text\": \"Step 3: Enable an Admin API Key. Without authentication, anyone with network access ...\"},\n", + "]" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c0920d25", + "metadata": {}, + "outputs": [], + "source": [ + "import hashlib\n", + "import uuid\n", + "\n", + "def content_hash(text):\n", + " return hashlib.sha256(text.encode()).hexdigest()\n", + "\n", + "def point_id(url, anchor, num):\n", + " return str(uuid.uuid5(uuid.NAMESPACE_URL, f\"{url}#{anchor}::{num}\"))\n", + "\n", + "def prepare(chunks):\n", + " out = []\n", + " for c in chunks:\n", + " # here we can normalize(c[\"text\"]) as recommended in Part 1\n", + " out.append({\n", + " **c,\n", + " \"text\": c[\"text\"],\n", + " \"section_url\": f'{c[\"url\"]}#{c[\"anchor\"]}' if c[\"anchor\"] else c[\"url\"],\n", + " \"content_hash\": content_hash(c[\"text\"]),\n", + " \"point_id\": point_id(c[\"url\"], c[\"anchor\"], c[\"chunk_num\"]),\n", + " })\n", + " return out" + ] + }, + { + "cell_type": "markdown", + "id": "5b117a77", + "metadata": {}, + "source": [ + "## Buckets: Where Each Chunk Lives\n", + "\n", + "A bucket is one of `N_BUCKETS` slots. A chunk's bucket comes from its `point_id`, the stable ID derived from its address, so editing the text never moves a chunk to a different bucket:\n", + "\n", + "```\n", + "point_id = stable ID from (url, anchor, chunk_num)\n", + "bucket = sha256(point_id) mod N_BUCKETS\n", + "\n", + "example (the \"Step 2: Enable TLS\" chunk):\n", + " point_id = \"4e72de03-624c-5b8e-a3ef-93e3a2d3267b\"\n", + " bucket = sha256(point_id) mod 16 = 11\n", + "```\n", + "\n", + "The next cell prints the bucket each of our chunks lands in." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "54ac3f2f", + "metadata": {}, + "outputs": [], + "source": [ + "N_BUCKETS = 16\n", + "GROUP_SIZE = 4 # buckets packed per group; 16 / 4 = 4 groups\n", + "\n", + "def bucket(pid):\n", + " return int(hashlib.sha256(pid.encode()).hexdigest(), 16) % N_BUCKETS\n", + "\n", + "for c in prepare(CHUNKS):\n", + " print(bucket(c[\"point_id\"]), c[\"anchor\"])" + ] + }, + { + "cell_type": "markdown", + "id": "c8828f3a", + "metadata": {}, + "source": [ + "## Digests: One Number per Bucket\n", + "\n", + "Each chunk contributes one number, computed from its ID and its content hash together (the first 64 bits of their hash). A bucket's digest is the XOR of the contributions of its chunks:\n", + "\n", + "```\n", + "bucket 5 holds two chunks, each contributes a number (shown tiny, in binary):\n", + " chunk A -> 0011\n", + " chunk B -> 0101\n", + " digest = 0011 XOR 0101 = 0110\n", + "\n", + "edit chunk B's text, so its contribution changes:\n", + " chunk B -> 1001\n", + " digest = 0011 XOR 1001 = 1010 (changed: bucket 5 gets investigated)\n", + "```\n", + "\n", + "Because the contribution folds in both the ID and the text, any edit, insert, or delete in a bucket changes its digest (a collision is about 2^-64, negligible), and an untouched bucket keeps the exact same digest. That is the signal we compare on." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "fde5eed3", + "metadata": {}, + "outputs": [], + "source": [ + "def contribution(pid, chash):\n", + " return int(hashlib.sha256((pid + chash).encode()).hexdigest()[:16], 16) # first 64 bits\n", + "\n", + "def compute_digests(chunks):\n", + " digests = [0] * N_BUCKETS\n", + " for c in chunks:\n", + " digests[bucket(c[\"point_id\"])] ^= contribution(c[\"point_id\"], c[\"content_hash\"])\n", + " return digests\n", + "\n", + "compute_digests(prepare(CHUNKS))" + ] + }, + { + "cell_type": "markdown", + "id": "3d65ef26", + "metadata": {}, + "source": [ + "## Storing the Chunks\n", + "\n", + "We store the chunks in a Qdrant collection as in Part 1, with one addition: each point carries its `sync_bucket` in the payload, and we index that field so we can later read a single bucket without scanning the collection." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "cc2eae14", + "metadata": {}, + "outputs": [], + "source": [ + "MAIN = \"docs-sync-scale\"\n", + "MODEL = \"sentence-transformers/all-MiniLM-L6-v2\"\n", + "\n", + "client.delete_collection(MAIN)\n", + "client.create_collection(\n", + " MAIN,\n", + " vectors_config=models.VectorParams(size=384, distance=models.Distance.COSINE),\n", + ")\n", + "client.create_payload_index(MAIN, \"sync_bucket\", models.PayloadSchemaType.INTEGER)\n", + "\n", + "def payload(c):\n", + " return {\n", + " \"url\": c[\"url\"], \"anchor\": c[\"anchor\"], \"chunk_num\": c[\"chunk_num\"],\n", + " \"section_url\": c[\"section_url\"], \"text\": c[\"text\"],\n", + " \"content_hash\": c[\"content_hash\"], \"sync_bucket\": bucket(c[\"point_id\"]),\n", + " }\n", + "\n", + "def as_points(chunks):\n", + " return [models.PointStruct(id=c[\"point_id\"],\n", + " vector=models.Document(text=c[\"text\"], model=MODEL),\n", + " payload=payload(c)) for c in chunks]\n", + "\n", + "client.upsert(MAIN, points=as_points(prepare(CHUNKS)), wait=True)" + ] + }, + { + "cell_type": "markdown", + "id": "3e1f25dc", + "metadata": {}, + "source": [ + "## The Summary Collection of Digests\n", + "\n", + "We keep the digests in Qdrant too, in a small separate collection. We could store one point per bucket, but at scale that is 65536 points, and reading the whole summary would then mean fetching 65536 points every run, each with Qdrant's per-point overhead. That per-run cost is exactly what we are trying to avoid.\n", + "\n", + "So we pack a group of digests into each point. At scale: 256 points holding 256 digests each (256 x 256 = 65536), so the whole summary is one retrieve of 256 points, and a changed bucket rewrites only the one point its group lives in. A bucket's digest sits at `point = bucket // 256`, `slot = bucket % 256`.\n", + "\n", + "This notebook uses 16 buckets packed 4 per point, so 4 points:\n", + "\n", + "```\n", + " point 0 = digests for buckets 0, 1, 2, 3\n", + " point 1 = digests for buckets 4, 5, 6, 7\n", + " point 2 = digests for buckets 8, 9, 10, 11\n", + " point 3 = digests for buckets 12, 13, 14, 15\n", + "\n", + "a bucket's digest sits at point = bucket // 4, slot = bucket % 4\n", + "```\n", + "\n", + "Digests are stored as strings, so their full 64-bit value round-trips without hitting integer limits. The vectors are dummy 1-dimensional values; this collection is never searched." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "36e9c0cb", + "metadata": {}, + "outputs": [], + "source": [ + "META = \"docs-sync-digests\"\n", + "N_META = N_BUCKETS // GROUP_SIZE\n", + "\n", + "client.delete_collection(META)\n", + "client.create_collection(\n", + " META,\n", + " vectors_config=models.VectorParams(size=1, distance=models.Distance.COSINE),\n", + ")\n", + "\n", + "def write_meta(digests, groups=None):\n", + " \"\"\"Write digests to the summary collection. Default writes every group.\"\"\"\n", + " groups = range(N_META) if groups is None else groups\n", + " points = [\n", + " models.PointStruct(\n", + " id=g, vector=[1.0],\n", + " payload={\"group\": g, \"digests\": [str(d) for d in digests[g * GROUP_SIZE:(g + 1) * GROUP_SIZE]]})\n", + " for g in groups\n", + " ]\n", + " client.upsert(META, points=points, wait=True)\n", + "\n", + "def read_meta():\n", + " digests = [0] * N_BUCKETS\n", + " for p in client.retrieve(META, ids=list(range(N_META)), with_payload=True):\n", + " g = p.payload[\"group\"]\n", + " for i, d in enumerate(p.payload[\"digests\"]):\n", + " digests[g * GROUP_SIZE + i] = int(d)\n", + " return digests\n", + "\n", + "write_meta(compute_digests(prepare(CHUNKS)))\n", + "read_meta()" + ] + }, + { + "cell_type": "markdown", + "id": "fd1e6015", + "metadata": {}, + "source": [ + "## Syncing\n", + "\n", + "One run:\n", + "\n", + "1. Compute the digest summary from the current source chunks.\n", + "2. Read the stored summary from the summary collection.\n", + "3. The buckets where the two differ are the only ones that changed.\n", + "4. For each changed bucket: read its chunks from the collection, compare with the source by content hash, embed and upsert what is new or changed, delete what is gone.\n", + "5. Rewrite the summary for the changed groups only, after the data writes.\n", + "\n", + "Step 5 runs after the writes on purpose: if a run stops halfway, the summary still points at the unfinished bucket, so the next run redoes it. Redoing is harmless because the writes are keyed by ID and hash." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "def read_bucket(b):\n", + " \"\"\"The content_hash of every chunk currently stored in bucket b.\"\"\"\n", + " stored, offset = {}, None\n", + " while True:\n", + " points, offset = client.scroll(\n", + " MAIN,\n", + " scroll_filter=models.Filter(must=[\n", + " models.FieldCondition(key=\"sync_bucket\", match=models.MatchValue(value=b))]),\n", + " with_payload=[\"content_hash\"], with_vectors=False, limit=1000, offset=offset)\n", + " for p in points:\n", + " stored[str(p.id)] = p.payload[\"content_hash\"]\n", + " if offset is None:\n", + " return stored\n", + "\n", + "def reconcile_bucket(b, source_b):\n", + " \"\"\"Make bucket b in Qdrant match the source. Returns (added, re_embedded, deleted).\"\"\"\n", + " stored_b = read_bucket(b)\n", + "\n", + " added = [c for pid, c in source_b.items() if pid not in stored_b]\n", + " changed = [c for pid, c in source_b.items()\n", + " if pid in stored_b and stored_b[pid] != c[\"content_hash\"]]\n", + " gone = [pid for pid in stored_b if pid not in source_b]\n", + "\n", + " if added or changed:\n", + " client.upsert(MAIN, points=as_points(added + changed), wait=True)\n", + " if gone:\n", + " client.delete(MAIN, points_selector=models.PointIdsList(points=gone), wait=True)\n", + " return len(added), len(changed), len(gone)\n", + "\n", + "def sync(latest_chunks):\n", + " latest = prepare(latest_chunks)\n", + "\n", + " by_bucket = {} # group the source once, not per changed bucket\n", + " for c in latest:\n", + " by_bucket.setdefault(bucket(c[\"point_id\"]), {})[c[\"point_id\"]] = c\n", + "\n", + " source = compute_digests(latest) # digests from the current source\n", + " stored = read_meta() # digests Qdrant holds\n", + " changed_buckets = [b for b in range(N_BUCKETS) if source[b] != stored[b]]\n", + "\n", + " report = {\"changed_buckets\": changed_buckets, \"added\": 0, \"re_embedded\": 0, \"deleted\": 0}\n", + " for b in changed_buckets:\n", + " a, r, d = reconcile_bucket(b, by_bucket.get(b, {}))\n", + " report[\"added\"] += a\n", + " report[\"re_embedded\"] += r\n", + " report[\"deleted\"] += d\n", + "\n", + " changed_groups = {b // GROUP_SIZE for b in changed_buckets}\n", + " write_meta(source, changed_groups) # rewrite only the changed groups, after the data writes\n", + " return report" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## A Month of Edits\n", + "\n", + "We simulate the source after some edits: one chunk unchanged, one text edited, one removed, one new. Running sync should touch only the buckets those changes fall in." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "LATEST = [\n", + " # unchanged\n", + " {\"url\": \"https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/\",\n", + " \"anchor\": \"prerequisites\", \"chunk_num\": 0,\n", + " \"text\": \"Prerequisites - Docker and Docker Compose installed - curl available in your terminal ...\"},\n", + " # edited text\n", + " {\"url\": \"https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/\",\n", + " \"anchor\": \"step-2-enable-tls\", \"chunk_num\": 0,\n", + " \"text\": \"Step 2: Enable TLS. Generate a certificate with mkcert and set the TLS config keys ...\"},\n", + " # step-3 removed; new step-4 added\n", + " {\"url\": \"https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/\",\n", + " \"anchor\": \"step-4-restrict-access\", \"chunk_num\": 0,\n", + " \"text\": \"Step 4: Restrict access with read-only API keys for untrusted clients ...\"},\n", + "]\n", + "\n", + "sync(LATEST)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "The report's `changed_buckets` lists only the buckets holding an edit, insert, or delete, and only the edited and new chunks get re-embedded. The unchanged `prerequisites` chunk sits in another bucket, so it is never fetched.\n", + "\n", + "## Conclusion\n", + "\n", + "The method in one breath: each chunk gets a bucket from its address and a contribution from its text; one small collection holds the XOR digest of each bucket; every run compares the current digests against the stored ones and reconciles only the buckets that differ. The Qdrant reads and re-embeddings then track how much changed, not the size of the whole corpus.\n", + "\n", + "**Use Part 1** when the corpus is small, up to roughly 100k chunks: fewer moving parts, nothing extra to maintain.\n", + "\n", + "**Use this approach** when the corpus is large and each run's changes are small next to its size (around a million chunks and up): a run reads a small summary plus only the changed buckets, instead of the whole collection. Raise the bucket count as the corpus grows, so each bucket stays small." + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": ".venv-1 (3.13.2)", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.13.2" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} From 402016e33b14db6333375db33314af6211340c97 Mon Sep 17 00:00:00 2001 From: Evgeniya Sukhodolskaya Date: Fri, 24 Jul 2026 19:47:59 +0200 Subject: [PATCH 2/2] made it readable --- .../sync_at_scale_with_digests.ipynb | 621 +++++++++++++----- 1 file changed, 462 insertions(+), 159 deletions(-) diff --git a/temporal-data-drift/sync_at_scale_with_digests.ipynb b/temporal-data-drift/sync_at_scale_with_digests.ipynb index bafa07d..3f7f57c 100644 --- a/temporal-data-drift/sync_at_scale_with_digests.ipynb +++ b/temporal-data-drift/sync_at_scale_with_digests.ipynb @@ -2,52 +2,50 @@ "cells": [ { "cell_type": "markdown", - "id": "d4a95357", "metadata": {}, "source": [ "# Incremental Embedding Updates at Scale (Part 2)\n", "\n", "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/qdrant/examples/blob/master/temporal-data-drift/sync_at_scale_with_digests.ipynb)\n", "\n", - "[Part 1](https://qdrant.tech/documentation/tutorials-operations/incremental-embedding-updates/) keeps a Qdrant collection in sync with changing documentation by reconciling the full list of current chunks against the whole Qdrant collection on every run. It is a simple setup that works, but since every run reads the whole collection, the sync cost grows linearly with the size of the corpus.\n", + "[Part 1](https://qdrant.tech/documentation/tutorials-operations/incremental-embedding-updates/) keeps a Qdrant collection in sync with changing documentation by reconciling the full list of current chunks against the whole Qdrant collection on every run. It is a simple setup that works for many mid-sized corpora. However, it doesn't scale that well, since every run reads the whole collection, the sync cost grows linearly with the size of the corpus.\n", "\n", - "This notebook proposes an alternative, more involved approach for the case where scale is a real factor (millions or billions of chunks). On top of the Part 1 idea it adds a small summary layer, so each run scans a fixed-size summary and touches only the parts of the collection that changed." + "This notebook proposes an alternative, more involved approach for the case where scale is a real factor (millions or billions of chunks). On top of the Part 1 idea it adds a small summary layer, so each run scans this summary and touches only the parts of the collection that changed." ] }, { "cell_type": "markdown", - "id": "77909da6", "metadata": {}, "source": [ "## The Idea\n", "\n", - "The method splits the chunks into a fixed number of **buckets** and gives each bucket a **digest**: a single number that summarizes everything in it. A bucket's digest is built from a small per-chunk value, the chunk's **contribution**, which depends on the chunk's position and its text. So if any chunk's text or position changes, a new chunk lands in the bucket, or one leaves it, the bucket's digest changes and we know that bucket needs a look. If the digest is unchanged, the bucket's contents are unchanged and we skip it.\n", + "The method splits the chunks into a fixed number of **buckets** and gives each bucket a **digest**: a single number that summarizes everything in it, the whole bucket's state. \n", + "A bucket's digest is built from each chunk's digest. A chunk's digest depends on the current chunk's position in the docs (f.e. section of a page) the and its current text. So if any chunk's text or position in the docs changes, a new chunk lands in the bucket, or one leaves it, the bucket's digest changes and we know that bucket needs an inspection. If the digest is unchanged, the bucket's contents are unchanged and we skip it, saving sync time.\n", "\n", "We keep two collections:\n", "\n", - "- the **chunks collection**, whose embeddings serve vector search: the Part 1 collection plus one extra payload field, `sync_bucket`. The point ID is unchanged from Part 1 (a stable ID derived from the chunk's address). We do not store the per-chunk contributions; they are cheap to recompute, and only the per-bucket digest is kept.\n", - "- a small **summary collection**, holding each bucket and its digest. This is the sync state. For a given bucket count it is a fixed size, and it stays a small fraction of the collection even at large scale.\n", + "- the **chunks collection**, whose embeddings serve vector search: the Part 1 collection plus one extra payload field, `sync_bucket`. The point ID is unchanged from Part 1 (a stable ID derived from the chunk's address). We do not store the per-chunk digests; they are cheap to recompute, and only the per-bucket digest is kept.\n", + "- a small **summary collection**, holding each bucket and its digest. This is the sync state.\n", "\n", - "Two things make it work:\n", + "This approach is built on two main ideas:\n", "\n", - "- **Comparison instead of scanning.** We compute the digests from the current source and compare them against the digests Qdrant already stored. Only the buckets whose digest differs get reconciled; the rest of the collection is never read.\n", - "- **XOR to combine contributions into a bucket digest.** XOR because:\n", - " 1. it is order-independent, so however the chunks come back, the same set of chunks gives the same digest;\n", - " 2. it folds any number of contributions into one fixed-width number with one cheap operation, so the summary stays small no matter how big the bucket;\n", + "- **Comparison instead of full scanning.** We compute the digests from the current source (e.g. current state of documentation, for example) and compare them against the digests Qdrant already stored. Only the chunks in the buckets whose digest differs get reconciled; the rest of the chunks collection is never read.\n", + "- **Using XOR to combine chunk digests into a bucket digest.**:\n", + " 1. it is order-independent, so however the chunks come back, the same set of chunks' digests gives the same bucket digest;\n", + " 2. it folds any number of chunk digests into one fixed-width number with one cheap operation, so the summary stays small no matter how big the bucket;\n", " 3. it is reversible (XORing a value twice cancels it), so at scale a bucket's digest could be patched for a single added or removed chunk instead of rebuilt. This notebook recomputes each changed bucket from scratch for clarity, but that reversibility is why XOR is the natural choice.\n", "\n", - "### The Math\n", + "### The Math Behind The Idea\n", "\n", "Say we use 2^16 = 65536 buckets. For a corpus of 1,000,000 chunks that is about 15 chunks per bucket. Each run compares the 65536 digests to find the changed buckets, then reads chunks only from those buckets. If 50 chunks changed, they sit in at most ~50 buckets, so we read on the order of 50 x 15 = 750 chunks, not 1,000,000. Part 1 reads all 1,000,000 every run.\n", "\n", - "We also pack the 65536 digests into a handful of points rather than one point per bucket, so the whole summary is a single read. The build section below explains why and how.\n", + "We also could group the 65536 digests into a handful of points rather than storing one bucket in one point of the **summary collection**, to optimize even further. The build section below explains why and how.\n", "\n", - "To keep the summary printable, this notebook uses **16 buckets**. All the code is written against one constant, `N_BUCKETS`; set it to 65536 for production, and higher for very large corpora so each bucket stays small." + "For explainability, this notebook uses only **16 buckets**. All the code is written against one constant, `N_BUCKETS`; set it to something reasonable for production use cases." ] }, { "cell_type": "markdown", - "id": "60d49886", "metadata": {}, "source": [ "## Prerequisites\n", @@ -58,7 +56,7 @@ { "cell_type": "code", "execution_count": null, - "id": "885bf3bf", + "id": "4bc0ec71", "metadata": {}, "outputs": [], "source": [ @@ -67,8 +65,8 @@ }, { "cell_type": "code", - "execution_count": null, - "id": "809c6af9", + "execution_count": 2, + "id": "01e8cd3f", "metadata": {}, "outputs": [], "source": [ @@ -84,12 +82,12 @@ }, { "cell_type": "markdown", - "id": "9233e503", + "id": "7c17eb04", "metadata": {}, "source": [ "## The Chunks\n", "\n", - "We use the same documentation chunks as Part 1. Each chunk has an address (which page, which section, which piece of that section) and its text. Two derived values carry the method:\n", + "We use the same documentation chunking approach as described in Part 1. Each chunk has an address (which page (url), which section (anchor), which piece of that section (chunk_num)) and its text. Two derived values carry the method:\n", "\n", "- `point_id`: a stable ID computed from the address, the chunk's `url`, section `anchor`, and `chunk_num`. Same address, same ID.\n", "- `content_hash`: a fingerprint of the text. Same text, same hash." @@ -97,8 +95,8 @@ }, { "cell_type": "code", - "execution_count": null, - "id": "6e6808ed", + "execution_count": 3, + "id": "f9532361", "metadata": {}, "outputs": [], "source": [ @@ -117,45 +115,50 @@ }, { "cell_type": "code", - "execution_count": null, - "id": "c0920d25", + "execution_count": 4, + "id": "3952a7b1", "metadata": {}, "outputs": [], "source": [ "import hashlib\n", "import uuid\n", "\n", + "\n", "def content_hash(text):\n", " return hashlib.sha256(text.encode()).hexdigest()\n", "\n", + "\n", "def point_id(url, anchor, num):\n", " return str(uuid.uuid5(uuid.NAMESPACE_URL, f\"{url}#{anchor}::{num}\"))\n", "\n", + "\n", "def prepare(chunks):\n", - " out = []\n", + " prepared = []\n", " for c in chunks:\n", - " # here we can normalize(c[\"text\"]) as recommended in Part 1\n", - " out.append({\n", + " # section_url is the human-readable address; point_id and content_hash are derived once here.\n", + " # (Part 1 also normalizes c[\"text\"] first; left out here to stay on topic.)\n", + " section_url = f'{c[\"url\"]}#{c[\"anchor\"]}' if c[\"anchor\"] else c[\"url\"]\n", + " prepared.append({\n", " **c,\n", - " \"text\": c[\"text\"],\n", - " \"section_url\": f'{c[\"url\"]}#{c[\"anchor\"]}' if c[\"anchor\"] else c[\"url\"],\n", + " \"section_url\": section_url,\n", " \"content_hash\": content_hash(c[\"text\"]),\n", " \"point_id\": point_id(c[\"url\"], c[\"anchor\"], c[\"chunk_num\"]),\n", " })\n", - " return out" + " return prepared" ] }, { "cell_type": "markdown", - "id": "5b117a77", + "id": "0f9b1468", "metadata": {}, "source": [ "## Buckets: Where Each Chunk Lives\n", "\n", - "A bucket is one of `N_BUCKETS` slots. A chunk's bucket comes from its `point_id`, the stable ID derived from its address, so editing the text never moves a chunk to a different bucket:\n", + "A bucket is one of `N_BUCKETS` slots (16 is this notebook, something reasonably big in production). \n", + "A chunk's bucket comes from its `point_id`, so editing the text never moves a chunk to a different bucket:\n", "\n", "```\n", - "point_id = stable ID from (url, anchor, chunk_num)\n", + "point_id = ID from (url, anchor, chunk_num)\n", "bucket = sha256(point_id) mod N_BUCKETS\n", "\n", "example (the \"Step 2: Enable TLS\" chunk):\n", @@ -168,66 +171,110 @@ }, { "cell_type": "code", - "execution_count": null, - "id": "54ac3f2f", + "execution_count": 5, + "id": "2b35ef65", "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "1 2ff5204a-0353-5991-ba55-acd1995063e8 https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/#prerequisites\n", + "11 4e72de03-624c-5b8e-a3ef-93e3a2d3267b https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/#step-2-enable-tls\n", + "12 3ecb6382-a741-5704-9797-b2b8d70847c1 https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/#step-3-enable-an-admin-api-key\n" + ] + } + ], "source": [ "N_BUCKETS = 16\n", - "GROUP_SIZE = 4 # buckets packed per group; 16 / 4 = 4 groups\n", + "GROUP_SIZE = 4 # buckets packed per group; 16 / 4 = 4 groups\n", + "\n", "\n", "def bucket(pid):\n", " return int(hashlib.sha256(pid.encode()).hexdigest(), 16) % N_BUCKETS\n", "\n", + "\n", "for c in prepare(CHUNKS):\n", - " print(bucket(c[\"point_id\"]), c[\"anchor\"])" + " print(bucket(c[\"point_id\"]), c[\"point_id\"], c[\"section_url\"])" ] }, { "cell_type": "markdown", - "id": "c8828f3a", + "id": "8c395c88", "metadata": {}, "source": [ "## Digests: One Number per Bucket\n", "\n", - "Each chunk contributes one number, computed from its ID and its content hash together (the first 64 bits of their hash). A bucket's digest is the XOR of the contributions of its chunks:\n", + "Each chunk contributes to a bucket one number, chunk's digest, computed from chunk's ID and its content hash (basically, hash of 2 concatenated hashes). If chunk's position or content stay the same, chunk's digest stays the same. A bucket's digest is the XOR of the chunk digests in it:\n", "\n", "```\n", - "bucket 5 holds two chunks, each contributes a number (shown tiny, in binary):\n", + "bucket 5 holds two chunks, each contributes a digest:\n", " chunk A -> 0011\n", " chunk B -> 0101\n", " digest = 0011 XOR 0101 = 0110\n", "\n", - "edit chunk B's text, so its contribution changes:\n", + "edit chunk B's text, so its chunk digest changes:\n", " chunk B -> 1001\n", " digest = 0011 XOR 1001 = 1010 (changed: bucket 5 gets investigated)\n", "```\n", "\n", - "Because the contribution folds in both the ID and the text, any edit, insert, or delete in a bucket changes its digest (a collision is about 2^-64, negligible), and an untouched bucket keeps the exact same digest. That is the signal we compare on." + "Because the chunk digest folds in both the ID and the text, any edit, insert, or delete in a bucket changes its digest (a collision is about 2^-60, negligible), and an untouched bucket keeps the exact same digest. That is the signal we compare on." ] }, { "cell_type": "code", - "execution_count": null, - "id": "fde5eed3", + "execution_count": 6, + "id": "b19bc530", "metadata": {}, - "outputs": [], + "outputs": [ + { + "data": { + "text/plain": [ + "[0,\n", + " 1065001772501011583,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 361274710181928245,\n", + " 526478318249294445,\n", + " 0,\n", + " 0,\n", + " 0]" + ] + }, + "execution_count": 6, + "metadata": {}, + "output_type": "execute_result" + } + ], "source": [ - "def contribution(pid, chash):\n", - " return int(hashlib.sha256((pid + chash).encode()).hexdigest()[:16], 16) # first 64 bits\n", + "def chunk_digest(pid, chash):\n", + " # First 15 hex digits of the combined hash = a 60-bit number.\n", + " # 60 bits fits Qdrant's signed 64-bit integer payload, so digests can be stored as plain integers.\n", + " combined = hashlib.sha256((pid + chash).encode()).hexdigest()\n", + " return int(combined[:15], 16)\n", + "\n", "\n", "def compute_digests(chunks):\n", " digests = [0] * N_BUCKETS\n", " for c in chunks:\n", - " digests[bucket(c[\"point_id\"])] ^= contribution(c[\"point_id\"], c[\"content_hash\"])\n", + " b = bucket(c[\"point_id\"])\n", + " digests[b] ^= chunk_digest(c[\"point_id\"], c[\"content_hash\"])\n", " return digests\n", "\n", + "\n", "compute_digests(prepare(CHUNKS))" ] }, { "cell_type": "markdown", - "id": "3d65ef26", + "id": "2a27b469", "metadata": {}, "source": [ "## Storing the Chunks\n", @@ -237,48 +284,71 @@ }, { "cell_type": "code", - "execution_count": null, - "id": "cc2eae14", + "execution_count": 7, + "id": "ee3e3231", "metadata": {}, - "outputs": [], + "outputs": [ + { + "data": { + "text/plain": [ + "UpdateResult(operation_id=3, status=)" + ] + }, + "execution_count": 7, + "metadata": {}, + "output_type": "execute_result" + } + ], "source": [ "MAIN = \"docs-sync-scale\"\n", "MODEL = \"sentence-transformers/all-MiniLM-L6-v2\"\n", "\n", - "client.delete_collection(MAIN)\n", - "client.create_collection(\n", - " MAIN,\n", - " vectors_config=models.VectorParams(size=384, distance=models.Distance.COSINE),\n", - ")\n", - "client.create_payload_index(MAIN, \"sync_bucket\", models.PayloadSchemaType.INTEGER)\n", + "if not client.collection_exists(MAIN):\n", + " client.create_collection(\n", + " MAIN,\n", + " vectors_config=models.VectorParams(size=384, distance=models.Distance.COSINE),\n", + " )\n", + " client.create_payload_index(MAIN, \"sync_bucket\", models.PayloadSchemaType.INTEGER)\n", + "\n", "\n", "def payload(c):\n", " return {\n", - " \"url\": c[\"url\"], \"anchor\": c[\"anchor\"], \"chunk_num\": c[\"chunk_num\"],\n", - " \"section_url\": c[\"section_url\"], \"text\": c[\"text\"],\n", - " \"content_hash\": c[\"content_hash\"], \"sync_bucket\": bucket(c[\"point_id\"]),\n", + " \"url\": c[\"url\"],\n", + " \"anchor\": c[\"anchor\"],\n", + " \"chunk_num\": c[\"chunk_num\"],\n", + " \"section_url\": c[\"section_url\"],\n", + " \"text\": c[\"text\"],\n", + " \"content_hash\": c[\"content_hash\"],\n", + " \"sync_bucket\": bucket(c[\"point_id\"]),\n", " }\n", "\n", + "\n", "def as_points(chunks):\n", - " return [models.PointStruct(id=c[\"point_id\"],\n", - " vector=models.Document(text=c[\"text\"], model=MODEL),\n", - " payload=payload(c)) for c in chunks]\n", + " points = []\n", + " for c in chunks:\n", + " points.append(models.PointStruct(\n", + " id=c[\"point_id\"],\n", + " vector=models.Document(text=c[\"text\"], model=MODEL), # embedded by Qdrant Cloud Inference\n", + " payload=payload(c),\n", + " ))\n", + " return points\n", + "\n", "\n", "client.upsert(MAIN, points=as_points(prepare(CHUNKS)), wait=True)" ] }, { "cell_type": "markdown", - "id": "3e1f25dc", + "id": "45b0087c", "metadata": {}, "source": [ "## The Summary Collection of Digests\n", "\n", - "We keep the digests in Qdrant too, in a small separate collection. We could store one point per bucket, but at scale that is 65536 points, and reading the whole summary would then mean fetching 65536 points every run, each with Qdrant's per-point overhead. That per-run cost is exactly what we are trying to avoid.\n", + "We keep the digests in Qdrant, in a small separate collection. We could store one point per bucket, but at true scale that could be hundreds thousands of points (f.e. ~65k for 2^16 buckets), and reading the whole summary would then again mean fetching many points per run.\n", "\n", - "So we pack a group of digests into each point. At scale: 256 points holding 256 digests each (256 x 256 = 65536), so the whole summary is one retrieve of 256 points, and a changed bucket rewrites only the one point its group lives in. A bucket's digest sits at `point = bucket // 256`, `slot = bucket % 256`.\n", + "So we could also group bucket digests into each point, balancing number of reads and updates. For example, for our running 2^16 example, we could have 256 points holding 256 digests each in an array (256 x 256 = 65536). A bucket's digest sits at `point = bucket // 256`, `slot = bucket % 256`.\n", "\n", - "This notebook uses 16 buckets packed 4 per point, so 4 points:\n", + "This notebook uses 16 buckets, so for 4 groups it'll be:\n", "\n", "```\n", " point 0 = digests for buckets 0, 1, 2, 3\n", @@ -289,136 +359,135 @@ "a bucket's digest sits at point = bucket // 4, slot = bucket % 4\n", "```\n", "\n", - "Digests are stored as strings, so their full 64-bit value round-trips without hitting integer limits. The vectors are dummy 1-dimensional values; this collection is never searched." + "Each digest is a 60-bit number (see the previous cell), which fits Qdrant's signed 64-bit integer payload, so we store digests as plain integers. The vectors can be dummy 1-dimensional values, as this collection is never used for vector search, only for lookups." ] }, { "cell_type": "code", - "execution_count": null, - "id": "36e9c0cb", + "execution_count": 8, + "id": "e859a2c2", "metadata": {}, - "outputs": [], + "outputs": [ + { + "data": { + "text/plain": [ + "[0,\n", + " 1065001772501011583,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 0,\n", + " 361274710181928245,\n", + " 526478318249294445,\n", + " 0,\n", + " 0,\n", + " 0]" + ] + }, + "execution_count": 8, + "metadata": {}, + "output_type": "execute_result" + } + ], "source": [ "META = \"docs-sync-digests\"\n", "N_META = N_BUCKETS // GROUP_SIZE\n", "\n", - "client.delete_collection(META)\n", - "client.create_collection(\n", - " META,\n", - " vectors_config=models.VectorParams(size=1, distance=models.Distance.COSINE),\n", - ")\n", + "if not client.collection_exists(META):\n", + " client.create_collection(\n", + " META,\n", + " vectors_config=models.VectorParams(size=1, distance=models.Distance.COSINE),\n", + " )\n", + "\n", "\n", "def write_meta(digests, groups=None):\n", - " \"\"\"Write digests to the summary collection. Default writes every group.\"\"\"\n", - " groups = range(N_META) if groups is None else groups\n", - " points = [\n", - " models.PointStruct(\n", - " id=g, vector=[1.0],\n", - " payload={\"group\": g, \"digests\": [str(d) for d in digests[g * GROUP_SIZE:(g + 1) * GROUP_SIZE]]})\n", - " for g in groups\n", - " ]\n", + " \"\"\"Store bucket digests in the summary collection, one point per group.\n", + "\n", + " digests: the full list of N_BUCKETS digests.\n", + " groups: which group points to (re)write; defaults to all of them.\n", + " \"\"\"\n", + " if groups is None:\n", + " groups = range(N_META)\n", + "\n", + " points = []\n", + " for g in groups:\n", + " # group g holds the digests of buckets [g * GROUP_SIZE .. g * GROUP_SIZE + GROUP_SIZE - 1]\n", + " start = g * GROUP_SIZE\n", + " group_digests = digests[start:start + GROUP_SIZE]\n", + " points.append(models.PointStruct(\n", + " id=g,\n", + " vector=[1.0], # dummy: this collection is never searched\n", + " payload={\"group\": g, \"digests\": group_digests},\n", + " ))\n", " client.upsert(META, points=points, wait=True)\n", "\n", + "\n", "def read_meta():\n", + " \"\"\"Read the summary back as a flat list of N_BUCKETS digests.\n", + "\n", + " Retrieves the N_META group points and unpacks each group's digest array\n", + " back into its bucket positions.\n", + " \"\"\"\n", " digests = [0] * N_BUCKETS\n", - " for p in client.retrieve(META, ids=list(range(N_META)), with_payload=True):\n", - " g = p.payload[\"group\"]\n", - " for i, d in enumerate(p.payload[\"digests\"]):\n", - " digests[g * GROUP_SIZE + i] = int(d)\n", + " for point in client.retrieve(META, ids=list(range(N_META)), with_payload=True):\n", + " g = point.payload[\"group\"]\n", + " group_digests = point.payload[\"digests\"]\n", + " for slot, digest in enumerate(group_digests):\n", + " digests[g * GROUP_SIZE + slot] = digest\n", " return digests\n", "\n", + "\n", "write_meta(compute_digests(prepare(CHUNKS)))\n", "read_meta()" ] }, { "cell_type": "markdown", - "id": "fd1e6015", + "id": "4df97e5b", "metadata": {}, "source": [ "## Syncing\n", "\n", - "One run:\n", + "A sync run reconciles the collection with the current source in five steps. We follow the running example: since the last sync, `step-2` was edited, `step-3` was removed, and a new `step-4` was added.\n", "\n", - "1. Compute the digest summary from the current source chunks.\n", - "2. Read the stored summary from the summary collection.\n", - "3. The buckets where the two differ are the only ones that changed.\n", - "4. For each changed bucket: read its chunks from the collection, compare with the source by content hash, embed and upsert what is new or changed, delete what is gone.\n", + "1. Compute the digest summary from the current source. `source = [d0, ..., d11, d12, ...]`\n", + "2. Read the stored summary from the summary collection. `stored = [d0', ..., d11', d12', ...]`\n", + "3. Take the buckets where the two differ. `changed = [0, 11, 12]`\n", + "4. Reconcile each changed bucket against the source, comparing by content hash (see below).\n", "5. Rewrite the summary for the changed groups only, after the data writes.\n", "\n", - "Step 5 runs after the writes on purpose: if a run stops halfway, the summary still points at the unfinished bucket, so the next run redoes it. Redoing is harmless because the writes are keyed by ID and hash." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "def read_bucket(b):\n", - " \"\"\"The content_hash of every chunk currently stored in bucket b.\"\"\"\n", - " stored, offset = {}, None\n", - " while True:\n", - " points, offset = client.scroll(\n", - " MAIN,\n", - " scroll_filter=models.Filter(must=[\n", - " models.FieldCondition(key=\"sync_bucket\", match=models.MatchValue(value=b))]),\n", - " with_payload=[\"content_hash\"], with_vectors=False, limit=1000, offset=offset)\n", - " for p in points:\n", - " stored[str(p.id)] = p.payload[\"content_hash\"]\n", - " if offset is None:\n", - " return stored\n", - "\n", - "def reconcile_bucket(b, source_b):\n", - " \"\"\"Make bucket b in Qdrant match the source. Returns (added, re_embedded, deleted).\"\"\"\n", - " stored_b = read_bucket(b)\n", - "\n", - " added = [c for pid, c in source_b.items() if pid not in stored_b]\n", - " changed = [c for pid, c in source_b.items()\n", - " if pid in stored_b and stored_b[pid] != c[\"content_hash\"]]\n", - " gone = [pid for pid in stored_b if pid not in source_b]\n", - "\n", - " if added or changed:\n", - " client.upsert(MAIN, points=as_points(added + changed), wait=True)\n", - " if gone:\n", - " client.delete(MAIN, points_selector=models.PointIdsList(points=gone), wait=True)\n", - " return len(added), len(changed), len(gone)\n", - "\n", - "def sync(latest_chunks):\n", - " latest = prepare(latest_chunks)\n", - "\n", - " by_bucket = {} # group the source once, not per changed bucket\n", - " for c in latest:\n", - " by_bucket.setdefault(bucket(c[\"point_id\"]), {})[c[\"point_id\"]] = c\n", + "Reconciling one bucket is a set-diff between the source and what Qdrant stores:\n", "\n", - " source = compute_digests(latest) # digests from the current source\n", - " stored = read_meta() # digests Qdrant holds\n", - " changed_buckets = [b for b in range(N_BUCKETS) if source[b] != stored[b]]\n", + "```\n", + "bucket 11: step-2 new_hash vs step-2 old_hash -> changed -> re-embed\n", + "bucket 0: step-4 hash vs (absent) -> new -> embed + insert\n", + "bucket 12: (absent) vs step-3 hash -> gone -> delete\n", + "```\n", "\n", - " report = {\"changed_buckets\": changed_buckets, \"added\": 0, \"re_embedded\": 0, \"deleted\": 0}\n", - " for b in changed_buckets:\n", - " a, r, d = reconcile_bucket(b, by_bucket.get(b, {}))\n", - " report[\"added\"] += a\n", - " report[\"re_embedded\"] += r\n", - " report[\"deleted\"] += d\n", + "Step 5 runs after the writes on purpose: if a run stops halfway, the summary still points at the unfinished bucket, so the next run redoes it. Redoing is harmless because the writes are keyed by ID and content hash.\n", "\n", - " changed_groups = {b // GROUP_SIZE for b in changed_buckets}\n", - " write_meta(source, changed_groups) # rewrite only the changed groups, after the data writes\n", - " return report" + "We build this up one function at a time below, running each so you can see what it returns." ] }, { "cell_type": "markdown", + "id": "b8269909", "metadata": {}, "source": [ - "## A Month of Edits\n", + "### The Edited Source\n", "\n", - "We simulate the source after some edits: one chunk unchanged, one text edited, one removed, one new. Running sync should touch only the buckets those changes fall in." + "Here is the source after a month of edits: `prerequisites` is unchanged, `step-2` is edited, `step-3` is gone, and a new `step-4` appears." ] }, { "cell_type": "code", - "execution_count": null, + "execution_count": 9, + "id": "1b4a1b72", "metadata": {}, "outputs": [], "source": [ @@ -435,8 +504,242 @@ " {\"url\": \"https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/\",\n", " \"anchor\": \"step-4-restrict-access\", \"chunk_num\": 0,\n", " \"text\": \"Step 4: Restrict access with read-only API keys for untrusted clients ...\"},\n", - "]\n", + "]" + ] + }, + { + "cell_type": "markdown", + "id": "cec12571", + "metadata": {}, + "source": [ + "### Steps 1 to 3: Which Buckets Changed\n", + "\n", + "Compute the digest summary from the edited source, read the stored summary, and take the buckets where they differ. This reads only the small summary, never the chunks collection." + ] + }, + { + "cell_type": "code", + "execution_count": 10, + "id": "0c005fa0", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "[0, 11, 12]" + ] + }, + "execution_count": 10, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "latest = prepare(LATEST)\n", + "source = compute_digests(latest) # digests of the edited source\n", + "stored = read_meta() # digests Qdrant currently holds\n", + "\n", + "changed_buckets = []\n", + "for b in range(N_BUCKETS):\n", + " if source[b] != stored[b]:\n", + " changed_buckets.append(b)\n", + "\n", + "changed_buckets" + ] + }, + { + "cell_type": "markdown", + "id": "3d63576e", + "metadata": {}, + "source": [ + "### `read_bucket`: What a Bucket Holds in Qdrant\n", + "\n", + "**Purpose:** read the chunks currently stored in one bucket.\n", + "**Input:** a bucket number.\n", + "**Output:** a dict mapping each stored chunk's `point_id` to its `content_hash`.\n", + "\n", + "It filters on the indexed `sync_bucket` field, so only that one bucket is read, not the whole collection." + ] + }, + { + "cell_type": "code", + "execution_count": 11, + "id": "fceddf3c", + "metadata": {}, + "outputs": [], + "source": [ + "def read_bucket(b):\n", + " \"\"\"Return {point_id: content_hash} for every chunk stored in bucket b.\n", + "\n", + " Pages through the results so nothing is missed in a large bucket.\n", + " \"\"\"\n", + " stored = {}\n", + " offset = None\n", + " while True:\n", + " points, offset = client.scroll(\n", + " MAIN,\n", + " scroll_filter=models.Filter(\n", + " must=[models.FieldCondition(\n", + " key=\"sync_bucket\",\n", + " match=models.MatchValue(value=b),\n", + " )],\n", + " ),\n", + " with_payload=[\"content_hash\"],\n", + " with_vectors=False,\n", + " limit=1000,\n", + " offset=offset,\n", + " )\n", + " for point in points:\n", + " stored[str(point.id)] = point.payload[\"content_hash\"]\n", + " if offset is None:\n", + " return stored" + ] + }, + { + "cell_type": "code", + "execution_count": 12, + "id": "bfbf04df", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "{'4e72de03-624c-5b8e-a3ef-93e3a2d3267b': '5f10768bbae2b0a5b3c6bc53947fd9e920411897aba9b3280e12ab704fdc65fb'}" + ] + }, + "execution_count": 12, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "# bucket 11 holds the \"step-2\" chunk; here is what Qdrant stored for it\n", + "read_bucket(11)" + ] + }, + { + "cell_type": "markdown", + "id": "26a75189", + "metadata": {}, + "source": [ + "### `reconcile_bucket`: Make One Bucket Match the Source\n", + "\n", + "**Purpose:** bring a single changed bucket in line with the source.\n", + "**Input:** the bucket number, and the source chunks in that bucket as `{point_id: chunk}`.\n", + "**Output:** the counts `(added, re_embedded, deleted)`.\n", + "\n", + "It set-diffs the source against what is stored: new or content-changed chunks are embedded and upserted, and chunks the source no longer has are deleted." + ] + }, + { + "cell_type": "code", + "execution_count": 13, + "id": "b800e84c", + "metadata": {}, + "outputs": [], + "source": [ + "def reconcile_bucket(b, source_chunks):\n", + " \"\"\"Make bucket b in Qdrant match source_chunks. Returns (added, re_embedded, deleted).\"\"\"\n", + " stored = read_bucket(b) # {point_id: content_hash} currently in Qdrant\n", + "\n", + " to_write = [] # new or content-changed chunks: embed and upsert\n", + " added = 0\n", + " re_embedded = 0\n", + " for pid, chunk in source_chunks.items():\n", + " if pid not in stored:\n", + " to_write.append(chunk) # new chunk in this bucket\n", + " added += 1\n", + " elif stored[pid] != chunk[\"content_hash\"]:\n", + " to_write.append(chunk) # same chunk, changed text\n", + " re_embedded += 1\n", + "\n", + " to_delete = [] # chunks Qdrant has but the source no longer does\n", + " for pid in stored:\n", + " if pid not in source_chunks:\n", + " to_delete.append(pid)\n", + "\n", + " if to_write:\n", + " client.upsert(MAIN, points=as_points(to_write), wait=True)\n", + " if to_delete:\n", + " client.delete(MAIN, points_selector=models.PointIdsList(points=to_delete), wait=True)\n", + "\n", + " return added, re_embedded, len(to_delete)" + ] + }, + { + "cell_type": "markdown", + "id": "68a25fc8", + "metadata": {}, + "source": [ + "### `sync`: The Whole Run\n", "\n", + "**Purpose:** reconcile the collection with the full current source.\n", + "**Input:** the current chunk list.\n", + "**Output:** a report of which buckets changed and how many chunks were added, re-embedded, and deleted.\n", + "\n", + "It groups the source by bucket once, finds the changed buckets by comparing summaries (steps 1 to 3), reconciles each one (step 4), then rewrites only the changed groups of the summary, last (step 5)." + ] + }, + { + "cell_type": "code", + "execution_count": 14, + "id": "84c1d670", + "metadata": {}, + "outputs": [], + "source": [ + "def sync(latest_chunks):\n", + " latest = prepare(latest_chunks)\n", + "\n", + " # group the source chunks by bucket once, so each bucket's chunks are ready to hand\n", + " source_by_bucket = {}\n", + " for c in latest:\n", + " b = bucket(c[\"point_id\"])\n", + " source_by_bucket.setdefault(b, {})[c[\"point_id\"]] = c\n", + "\n", + " # steps 1-3: which buckets changed\n", + " source = compute_digests(latest)\n", + " stored = read_meta()\n", + " changed_buckets = []\n", + " for b in range(N_BUCKETS):\n", + " if source[b] != stored[b]:\n", + " changed_buckets.append(b)\n", + "\n", + " # step 4: reconcile each changed bucket\n", + " report = {\"changed_buckets\": changed_buckets, \"added\": 0, \"re_embedded\": 0, \"deleted\": 0}\n", + " for b in changed_buckets:\n", + " source_chunks = source_by_bucket.get(b, {})\n", + " added, re_embedded, deleted = reconcile_bucket(b, source_chunks)\n", + " report[\"added\"] += added\n", + " report[\"re_embedded\"] += re_embedded\n", + " report[\"deleted\"] += deleted\n", + "\n", + " # step 5: rewrite only the changed groups of the summary, after the data writes\n", + " changed_groups = set()\n", + " for b in changed_buckets:\n", + " changed_groups.add(b // GROUP_SIZE)\n", + " write_meta(source, changed_groups)\n", + "\n", + " return report" + ] + }, + { + "cell_type": "code", + "execution_count": 15, + "id": "5b334e98", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "{'changed_buckets': [0, 11, 12], 'added': 1, 're_embedded': 1, 'deleted': 1}" + ] + }, + "execution_count": 15, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ "sync(LATEST)" ] }, @@ -448,7 +751,7 @@ "\n", "## Conclusion\n", "\n", - "The method in one breath: each chunk gets a bucket from its address and a contribution from its text; one small collection holds the XOR digest of each bucket; every run compares the current digests against the stored ones and reconciles only the buckets that differ. The Qdrant reads and re-embeddings then track how much changed, not the size of the whole corpus.\n", + "The method in one breath: each chunk gets a bucket from its address and a chunk digest from its id and text; one small collection holds the XOR digest of each bucket; every run compares the current digests against the stored ones and reconciles only the buckets that differ. The Qdrant reads and re-embeddings then track how much changed, not the size of the whole corpus.\n", "\n", "**Use Part 1** when the corpus is small, up to roughly 100k chunks: fewer moving parts, nothing extra to maintain.\n", "\n", @@ -458,7 +761,7 @@ ], "metadata": { "kernelspec": { - "display_name": ".venv-1 (3.13.2)", + "display_name": "Python 3", "language": "python", "name": "python3" },