Mini-project: an incremental ingest

Description

Six lessons gave you, one at a time, the pieces of an ingestion pipeline that survives a corpus that changes: content_hash and the chunk store in sqlite3 (Lesson 03), doc_hash and detect_changes (Lesson 04), reingest() tying it all together (Lesson 05), and the confirmation, with real instrumentation, that a modified or deleted document gets handled surgically (Lessons 06 and 07). This mini-project adds no new concept — it brings the seven pieces together into a single chunk_store.py, and runs them against the full end-to-end scenario: initial ingestion, an identical reingest (zero duplicates), and a final run where one document gets modified and another gets deleted at the same time.

The deliverable is the evidence that closes out the module: a store that, after four reingest() runs across three different versions of the corpus, reflects exactly the correct state at every point — not one chunk extra, not one missing, not a single orphan.

Connection to the module

From Lesson 03 you use content_hash, create_store, and upsert_chunk. From Lesson 04, doc_hash, ChangeSet, and detect_changes. From Lessons 05, 06, and 07, upsert_doc_chunks, purge_doc, and reingest() — the full orchestration, already tested separately against a modified document and against a deleted one. The only thing this mini-project adds is the missing layer: running the four runs in sequence, on a single store, and checking each one's result against what it should be.

The assignment

Reservo asks you to finish the full incremental ingestion pipeline, with four deliverables:

a) chunk_store.py, the single file with the module's seven pieces. Everything you built in Lessons 03-07, together in one place, ready to import from any future ingestion run.

b) The initial run and the identical reingest, with evidence of zero duplicates. Reingest the canonical corpus twice in a row, with no real change in between, and confirm with a COUNT(*) that the store has 57 chunks both times — never 114.

c) A final run with one modified document and one deleted document at the same time. refund-policy gains the same clarifying sentence from Lesson 06, and phonebooth-room-manual disappears as in Lesson 07 — but this time both things happen in the same reingest() run, not separately.

d) Confirm, with an orphan query, that the final store is consistent. After the four runs, zero chunks pointing at a doc_id that no longer exists in documents.

The boundary of what this project does not ask of you: you're not going to schedule when this ingestion fires (that's airflow-and-declarative-orchestration-guide, out of scope for this guide), and you're not going to evaluate whether the search index retrieves a given chunk correctly (that's Module 6). If at any point you feel tempted to "while I'm at it, wire up a cron" or "while I'm at it, try a search query", note it as another module's job and move on: today's deliverable is the correct chunk store after four runs, not the system that schedules it or the one that queries it.

Before you start: what you already have built

Everything that follows assumes you already have, from previous lessons, these pieces working:

content_hash(chunk)                        -> str (sha256, 64 hex chars)          -- L03
create_store(path)                         -> sqlite3.Connection (2 tables)        -- L03/L04
upsert_chunk(conn, chunk)                  -> bool (True if it wrote)              -- L03
doc_hash(raw_text)                         -> str (sha256, 64 hex chars)           -- L04
ChangeSet(new, modified, deleted, unchanged) -- frozen dataclass                   -- L04
detect_changes(conn, current_docs)         -> ChangeSet                            -- L04
upsert_doc_chunks(conn, doc_id, fmt, raw)  -> int (chunks written)                 -- L05
purge_doc(conn, doc_id)                    -> int (chunks purged)                  -- L05/L07
reingest(conn, current_docs)               -> dict (run summary)                   -- L05

And reservo_corpus.py, with the full canonical corpus (RAW_DOCS, Chunk, ingest_document, build_corpus, INGESTION_DATE), exactly as it stood at the end of Module 1 and reproduced verbatim in Lesson 02 of this module.

Step 1 — chunk_store.py: the seven pieces, together

import hashlib
import sqlite3
from dataclasses import dataclass

from reservo_corpus import RAW_DOCS, Chunk, ingest_document, INGESTION_DATE


def content_hash(chunk: Chunk) -> str:
    payload = f"{chunk.doc_id}\x1f{chunk.section}\x1f{chunk.position}\x1f{chunk.text}"
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def doc_hash(raw_text: str) -> str:
    return hashlib.sha256(raw_text.encode("utf-8")).hexdigest()


def create_store(path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS documents (
            doc_id TEXT PRIMARY KEY,
            fmt TEXT NOT NULL,
            raw_hash TEXT NOT NULL,
            ingested_at TEXT NOT NULL
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS chunks (
            chunk_id TEXT PRIMARY KEY,
            doc_id TEXT NOT NULL,
            title TEXT NOT NULL,
            section TEXT NOT NULL,
            position INTEGER NOT NULL,
            text TEXT NOT NULL,
            content_hash TEXT NOT NULL,
            ingested_at TEXT NOT NULL
        )
    """)
    conn.commit()
    return conn


@dataclass(frozen=True)
class ChangeSet:
    new: list[str]
    modified: list[str]
    deleted: list[str]
    unchanged: list[str]


def detect_changes(conn: sqlite3.Connection, current_docs: dict[str, tuple[str, str]]) -> ChangeSet:
    stored = dict(conn.execute("SELECT doc_id, raw_hash FROM documents").fetchall())
    new, modified, unchanged = [], [], []
    for doc_id, (fmt, raw_text) in current_docs.items():
        h = doc_hash(raw_text)
        if doc_id not in stored:
            new.append(doc_id)
        elif stored[doc_id] != h:
            modified.append(doc_id)
        else:
            unchanged.append(doc_id)
    deleted = [doc_id for doc_id in stored if doc_id not in current_docs]
    return ChangeSet(sorted(new), sorted(modified), sorted(deleted), sorted(unchanged))


def upsert_chunk(conn: sqlite3.Connection, chunk: Chunk) -> bool:
    h = content_hash(chunk)
    row = conn.execute(
        "SELECT content_hash FROM chunks WHERE chunk_id = ?", (chunk.chunk_id,)
    ).fetchone()
    if row is not None and row[0] == h:
        return False
    conn.execute("""
        INSERT INTO chunks (chunk_id, doc_id, title, section, position, text, content_hash, ingested_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(chunk_id) DO UPDATE SET
            title=excluded.title, section=excluded.section, position=excluded.position,
            text=excluded.text, content_hash=excluded.content_hash, ingested_at=excluded.ingested_at
    """, (chunk.chunk_id, chunk.doc_id, chunk.title, chunk.section, chunk.position,
          chunk.text, h, INGESTION_DATE))
    return True


def upsert_doc_chunks(conn: sqlite3.Connection, doc_id: str, fmt: str, raw_text: str) -> int:
    chunks = ingest_document(doc_id, fmt, raw_text)
    written = sum(upsert_chunk(conn, c) for c in chunks)
    conn.execute("""
        INSERT INTO documents (doc_id, fmt, raw_hash, ingested_at)
        VALUES (?, ?, ?, ?)
        ON CONFLICT(doc_id) DO UPDATE SET
            fmt=excluded.fmt, raw_hash=excluded.raw_hash, ingested_at=excluded.ingested_at
    """, (doc_id, fmt, doc_hash(raw_text), INGESTION_DATE))
    return written


def purge_doc(conn: sqlite3.Connection, doc_id: str) -> int:
    cur = conn.execute("DELETE FROM chunks WHERE doc_id = ?", (doc_id,))
    conn.execute("DELETE FROM documents WHERE doc_id = ?", (doc_id,))
    return cur.rowcount


def reingest(conn: sqlite3.Connection, current_docs: dict[str, tuple[str, str]]) -> dict:
    changes = detect_changes(conn, current_docs)
    chunks_written = 0
    for doc_id in changes.new + changes.modified:
        fmt, raw_text = current_docs[doc_id]
        chunks_written += upsert_doc_chunks(conn, doc_id, fmt, raw_text)
    chunks_purged = 0
    for doc_id in changes.deleted:
        chunks_purged += purge_doc(conn, doc_id)
    conn.commit()
    return {
        "new": changes.new, "modified": changes.modified, "deleted": changes.deleted,
        "unchanged_count": len(changes.unchanged),
        "chunks_written": chunks_written, "chunks_purged": chunks_purged,
    }

None of this is new — it's exactly what Lessons 03-07 built, copied into a single importable file. Before moving on to Step 2, confirm this file, run on its own (without calling any function yet), raises no import error.

Step 2 — The initial run and the identical reingest

from chunk_store import create_store, reingest
from reservo_corpus import RAW_DOCS

conn = create_store(":memory:")

print("--- run 1: initial ingestion ---")
r1 = reingest(conn, RAW_DOCS)
total_1 = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print(f"new={len(r1['new'])} written={r1['chunks_written']} total={total_1}")

print("--- run 2: reingest the same corpus, no changes ---")
r2 = reingest(conn, RAW_DOCS)
total_2 = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print(f"unchanged={r2['unchanged_count']} written={r2['chunks_written']} total={total_2}")

What to expect (run):

--- run 1: initial ingestion ---
new=13 written=57 total=57
--- run 2: reingest the same corpus, no changes ---
unchanged=13 written=0 total=57

Before moving on to Step 3, confirm it yourself: the total stays at 57 both times, run 2 writes nothing. This is deliverable b) of the assignment.

Step 3 — Modified and deleted, in the same run

RAW_DOCS_FINAL combines, in a single dictionary, the two changes Lessons 06 and 07 tested separately: refund-policy gains the same clarifying sentence from Lesson 06, and phonebooth-room-manual disappears as in Lesson 07 — both at once, simulating a single real run where more than one document changed since yesterday.

RAW_DOCS_FINAL = dict(RAW_DOCS)

fmt, raw = RAW_DOCS_FINAL["refund-policy"]
old_section = (
    "## Refund Amount and Timing\n\n"
    "Eligible refunds return the full amount charged for the booking to the "
    "original payment method. Processing takes up to 5 business days once "
    "the cancellation is confirmed in the booking system."
)
new_section = old_section + (
    " For refunds tied to a facility issue rather than a cancellation, "
    "processing begins from the date Reservo confirms the outage, not the "
    "date of the original booking."
)
RAW_DOCS_FINAL["refund-policy"] = (fmt, raw.replace(old_section, new_section))

del RAW_DOCS_FINAL["phonebooth-room-manual"]

print("documents in RAW_DOCS_FINAL:", len(RAW_DOCS_FINAL))

print("--- run 3: refund-policy modified + phonebooth-room-manual deleted ---")
r3 = reingest(conn, RAW_DOCS_FINAL)
total_3 = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print("modified:", r3["modified"])
print("deleted:", r3["deleted"])
print(f"unchanged={r3['unchanged_count']} written={r3['chunks_written']} purged={r3['chunks_purged']} total={total_3}")

print("--- run 4: reingest RAW_DOCS_FINAL again, no changes ---")
r4 = reingest(conn, RAW_DOCS_FINAL)
total_4 = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print(f"unchanged={r4['unchanged_count']} written={r4['chunks_written']} purged={r4['chunks_purged']} total={total_4}")

What to expect (run):

documents in RAW_DOCS_FINAL: 12
--- run 3: refund-policy modified + phonebooth-room-manual deleted ---
modified: ['refund-policy']
deleted: ['phonebooth-room-manual']
unchanged=11 written=1 purged=5 total=52
--- run 4: reingest RAW_DOCS_FINAL again, no changes ---
unchanged=12 written=0 purged=0 total=52

Run 3 detects both things correctly in a single detect_changes pass: refund-policy as modified (the same single chunk from Lesson 06 gets rewritten, written=1) and phonebooth-room-manual as deleted (its 5 chunks get purged, purged=5) — and the remaining 11 documents aren't touched (unchanged=11). The total drops from 57 to 52. Run 4, on the same RAW_DOCS_FINAL with no further changes, again gives written=0, purged=0 — idempotency holds in this new state too, not just in the corpus's original state. This is deliverable c) of the assignment.

Step 4 — The final orphan check

orphans = conn.execute("""
    SELECT COUNT(*) FROM chunks
    WHERE doc_id NOT IN (SELECT doc_id FROM documents)
""").fetchone()[0]

distinct_chunks = conn.execute("SELECT COUNT(DISTINCT chunk_id) FROM chunks").fetchone()[0]
total_docs = conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0]

print("orphaned chunks:", orphans)
print("distinct chunk_id:", distinct_chunks)
print("documents in the store:", total_docs)

What to expect (run):

orphaned chunks: 0
distinct chunk_id: 52
documents in the store: 12

Zero orphans, 52 distinct chunks — none duplicated, matches run 4's total — and 12 documents, exactly len(RAW_DOCS_FINAL). This is deliverable d) of the assignment.

Final check: the four runs, summarized

print("=== Mini-project final verification ===")
print(f"1. Run 1 (initial):            57 chunks written, total=57")
print(f"2. Run 2 (identical):          0 chunks written, total=57 -- zero duplicates")
print(f"3. Run 3 (modif.+deleted):     1 written, 5 purged, total=52")
print(f"4. Run 4 (identical to v3):    0 written, 0 purged, total=52")
print(f"5. Orphaned chunks at the end: {orphans} (expected: 0)")

What to expect:

=== Mini-project final verification ===
1. Run 1 (initial):            57 chunks written, total=57
2. Run 2 (identical):          0 chunks written, total=57 -- zero duplicates
3. Run 3 (modif.+deleted):     1 written, 5 purged, total=52
4. Run 4 (identical to v3):    0 written, 0 purged, total=52
5. Orphaned chunks at the end: 0 (expected: 0)

The five lines close out the module: the initial ingestion, the reingest that duplicates nothing, the run that updates and purges at once without touching what didn't change, the reingest of the new state that also duplicates nothing, and a final store with no orphaned chunk at all.

Common mistakes

1. Rebuilding RAW_DOCS_FINAL from scratch instead of starting from RAW_DOCS_V2/V3 in Lessons 06/07. If the text of the sentence added to refund-policy isn't exactly the same, character for character, as in Lesson 06, refund-policy-001's content_hash will be different — not an error, but an unnecessary discrepancy with what was already verified in that lesson. Copying the same exact old_section/new_section avoids that drift.
2. Forgetting run 3 combines two kinds of change, and only checking one of the two. After Lessons 06 and 07 separately, it's easy to look only at r3["modified"] or only at r3["deleted"] and not both — but this mini-project's deliverable c) is specifically that both things happen together, in the same reingest(), not in separate runs.
3. Running the four runs on different conn objects instead of the same one. The point of this mini-project is that the four runs share a single store, accumulating real state from one run to the next — as would happen with a persistent chunk store in production. Creating a new create_store(":memory:") for every run completely breaks the narrative (every run would see an empty store, and everything would come out as new).
4. Not checking for orphans until the end, instead of after every run. This mini-project does it once at the end for brevity, but nothing stops you from running the same query after every reingest() — in fact, that's what's recommended in a real pipeline, to catch an inconsistency in the run where it happened, not four runs later.

Exercises

Exercise 1: A fifth run, original corpus (Easy)

On the mini-project's final conn (after the 4 runs), reingest RAW_DOCS — the original corpus, with phonebooth-room-manual back and refund-policy without the added sentence. Does phonebooth-room-manual show up as new? Does refund-policy show up as modified? What's the final total?

r5 = reingest(conn, RAW_DOCS)
total_5 = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print("new:", r5["new"])
print("modified:", r5["modified"])
print("total:", total_5)

new: ['phonebooth-room-manual']
modified: ['refund-policy']
total: 57

Exercise 2: Measure the store's size in bytes (Medium)

sqlite3 lets you write to a real file instead of ":memory:". Using the project's standard scratch pattern (mktemp -d), create the mini-project's final store (after the 4 runs) in a real file, and report its size in bytes with pathlib.Path.stat().st_size.

import subprocess
from pathlib import Path

scratch_dir = subprocess.run(["mktemp", "-d"], capture_output=True, text=True).stdout.strip()
db_path = str(Path(scratch_dir) / "chunk_store.db")

conn_file = create_store(db_path)
reingest(conn_file, RAW_DOCS)
reingest(conn_file, RAW_DOCS)
reingest(conn_file, RAW_DOCS_FINAL)
reingest(conn_file, RAW_DOCS_FINAL)
conn_file.close()

size_bytes = Path(db_path).stat().st_size
print("size of the .db file:", size_bytes, "bytes")

size of the .db file: 45056 bytes

Exercise 3: What would happen with a doc_id that reappears with content different from what it had before being deleted? (Hard)

Imagine a four-run scenario: (1) the full original corpus, (2) phonebooth-room-manual gets deleted, (3) the corpus has phonebooth-room-manual again, but with text different from the original (say, the room reopens with reduced capacity, "Capacity: 1 person" changes to "Capacity: 1 person (temporary reduced layout)"). Does the store end up with the old chunks, the new ones, or a mix of the two? Build the scenario and confirm.

conn3 = create_store(":memory:")
reingest(conn3, RAW_DOCS)  # run 1: full corpus

docs_sin_phonebooth = dict(RAW_DOCS)
del docs_sin_phonebooth["phonebooth-room-manual"]
reingest(conn3, docs_sin_phonebooth)  # run 2: gets deleted

docs_phonebooth_reducido = dict(RAW_DOCS)
fmt, raw = docs_phonebooth_reducido["phonebooth-room-manual"]
new_raw = raw.replace("Capacity: 1 person.", "Capacity: 1 person (temporary reduced layout).")
docs_phonebooth_reducido["phonebooth-room-manual"] = (fmt, new_raw)
r3 = reingest(conn3, docs_phonebooth_reducido)  # run 3: reappears, with different text

print("new in run 3:", r3["new"])
rows = conn3.execute(
    "SELECT text FROM chunks WHERE chunk_id = 'phonebooth-room-manual-001'"
).fetchall()
print("stored text for phonebooth-room-manual-001:", rows[0][0][:80], "...")

new in run 3: ['phonebooth-room-manual']
stored text for phonebooth-room-manual-001: Capacity: 1 person (temporary reduced layout). The room has a narrow shelf-desk  ...

Summary and next step

• chunk_store.py brings the seven pieces from Lessons 03-07 together into a single file: content_hash, doc_hash, create_store, ChangeSet, detect_changes, upsert_chunk, upsert_doc_chunks, purge_doc, and reingest().
• Four runs on the same store, three versions of the corpus: the initial ingestion (57 chunks), the identical reingest (0 duplicates), the run that modifies refund-policy and deletes phonebooth-room-manual at once (1 chunk rewritten, 5 purged, total 52), and the reingest of that new state (0 duplicates again).
• The final store has no orphaned chunk at all — the explicit check, not just the assumption that the code "should" be correct.
• The boundary held on purpose: no periodic run was scheduled (airflow-and-declarative-orchestration-guide covers that), and nothing about whether the search index retrieves any of this correctly was evaluated (Module 6).

This closes out Module 5. You have an ingestion pipeline that survives exactly the scenario every real production RAG system opens with: documents that get added, edited, and removed, without the chunk store accumulating garbage or losing sync with the source. In Module 6 we take the BM25 index from Module 2 — which, in an end-to-end connected system, would need to be rebuilt on top of this same store every time it changes — and build it an evaluation harness: a fixed set of queries with an expected doc_id, and the recall@k and precision@k metrics that confirm, with real numbers, whether what's indexed retrieves what it should.

Additional resources

1. Python — sqlite3 — the full API behind chunk_store.py, including the difference between ":memory:" and a real file that Exercise 2 explores.
2. Python — hashlib — sha256, the foundation of content_hash and doc_hash across this module's seven pieces.
3. SQLite — Upsert (INSERT ... ON CONFLICT) — the clause behind upsert_chunk and document registration in upsert_doc_chunks.
4. production-rag-and-document-ingestion-guide — Module 1, module-01-parsing-and-chunking-documents: the origin of reservo_corpus.py, the foundation this entire module runs on.
5. production-rag-and-document-ingestion-guide — Module 6, module-06-evaluating-retrieval-quality: the exact point where this guide picks back up, evaluating whether the index built on top of this same chunk store retrieves the correct results.
6. airflow-and-declarative-orchestration-guide (Data Engineering ecosystem) — where you learn to schedule this same reingest() logic as a recurring production pipeline, with retries and dependencies — the exact boundary of where this module ends.