Reingesting without duplicating

Description

The previous three lessons built each piece separately: upsert_chunk (Lesson 03) writes a chunk only if its content changed; detect_changes (Lesson 04) classifies every document without touching ingest_document. This lesson ties them together into a single function, reingest(), which is the complete answer to the problem raised in Lesson 02: running ingestion two, ten, or a thousand times on the same corpus, without the chunk store ever ending up with a single extra row.

reingest() doesn't add any new concept — it's the orchestration of what you already have: first it asks what changed (detect_changes), then it does the expensive work (ingest_document + upsert_chunk) only for what needs it, and it purges whatever no longer exists. This lesson's result is the central piece of evidence for the whole module: reingesting the full canonical corpus, twice in a row, produces exactly the same store both times.

Connection to the module

This lesson combines content_hash/upsert_chunk (Lesson 03) with detect_changes (Lesson 04) into reingest(), the function Lessons 06, 07, and 08 use unchanged for the modified-document scenario, the deleted-document scenario, and the mini-project's final combined run.

Analogy: the clerk's full round

The clerk from earlier lessons already has their two tools: the cover label (doc_hash, for knowing which folders changed without opening them) and the per-page fingerprint notebook (content_hash, for knowing exactly which pages to refile inside a folder that did change). This lesson gives them their full morning routine: first they walk through every folder in the inbox looking only at the cover labels — a quick glance, nothing opened. For folders whose label didn't change, they do nothing else: leave them where they are. For new folders or folders with a different label, only then do they open them, scan their pages, and file only the ones that genuinely changed. And for any folder that had a record but is no longer in the tray, they pull its pages from the archive. That full routine, run end to end, is reingest().

reingest(): detect, process only what changed, purge what's gone

reingest() needs a third piece, alongside upsert_doc_chunks: what to do with a doc_id that detect_changes classified as deleted. The minimal version is simple — delete its chunks and its row in documents — and it's exactly what reingest() uses in this lesson; Lesson 07 comes back to this same function to confirm, with an explicit query, that the purge leaves no orphaned chunk.

def purge_doc(conn: sqlite3.Connection, doc_id: str) -> int:
    """Removes every chunk of `doc_id` from the store, plus its row in
    `documents`. Returns how many chunk rows were deleted."""
    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:
    """The full incremental ingestion pass: detect what changed, do the
    expensive work (ingest_document + upsert_chunk) only for new/modified
    docs, and purge chunks for anything that disappeared from current_docs."""
    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,
    }

Three steps, in the same order as the analogy:

1. detect_changes(conn, current_docs) — the full classification from Lesson 04, without touching ingest_document.
2. For every doc_id in new or modified — never for unchanged — call upsert_doc_chunks, which only then runs ingest_document (parse + chunk) and then upsert_chunk for each resulting chunk.
3. For every doc_id in deleted, call purge_doc (defined above; Lesson 07 goes deeper into it with the check that no chunk is left orphaned).

upsert_doc_chunks is the piece that was missing to connect ingest_document (Module 1) with upsert_chunk (Lesson 03):

from reservo_corpus import ingest_document


def upsert_doc_chunks(conn: sqlite3.Connection, doc_id: str, fmt: str, raw_text: str) -> int:
    """Re-ingests ONE document (parse + chunk) and upserts its chunks.
    Returns how many chunks were actually written (changed), not the total
    chunk count of the document."""
    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

upsert_doc_chunks does two things: it calls ingest_document (the only time parse_markdown/parse_html/clean_text run in this entire lesson, and only for documents detect_changes flagged as new or modified), and then it registers the updated doc_hash in documents — so the next detect_changes run sees this document as unchanged, even if today it was new or modified.

The central evidence: reingest twice, zero duplicates

from reservo_corpus import RAW_DOCS

conn = create_store(":memory:")

print("--- run 1: first ingestion ---")
r1 = reingest(conn, RAW_DOCS)
total_1 = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print(f"new={len(r1['new'])} modified={len(r1['modified'])} deleted={len(r1['deleted'])} "
      f"unchanged={r1['unchanged_count']} written={r1['chunks_written']} purged={r1['chunks_purged']}")
print("total chunks in store:", total_1)

print()
print("--- run 2: reingest the identical corpus ---")
r2 = reingest(conn, RAW_DOCS)
total_2 = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print(f"new={len(r2['new'])} modified={len(r2['modified'])} deleted={len(r2['deleted'])} "
      f"unchanged={r2['unchanged_count']} written={r2['chunks_written']} purged={r2['chunks_purged']}")
print("total chunks in store:", total_2)

distinct = conn.execute("SELECT COUNT(DISTINCT chunk_id) FROM chunks").fetchone()[0]
print()
print("distinct chunk_id after 2 runs:", distinct)
print("matches build_corpus():", distinct == len(build_corpus()))

What to expect (run):

--- run 1: first ingestion ---
new=13 modified=0 deleted=0 unchanged=0 written=57 purged=0
total chunks in store: 57

--- run 2: reingest the identical corpus ---
new=0 modified=0 deleted=0 unchanged=13 written=0 purged=0
total chunks in store: 57

distinct chunk_id after 2 runs: 57
matches build_corpus(): True

This is the result this module opened with, now achieved: run 1 ingests all 13 documents from scratch (new=13, written=57). Run 2, on exactly the same RAW_DOCS, finds nothing new, modified, or deleted (unchanged=13) and writes not a single row (written=0). The total chunk count stays at 57 both times — not 114, like the naive ingestion from Lesson 02. distinct chunk_id matches len(build_corpus()) exactly: after two runs, the store has precisely the chunks the canonical corpus produces, not one more.

Common mistakes

1. Calling ingest_document for every document in current_docs, not just new + modified. This is a subtle bug: the code "works" (the final result is correct), but it completely loses the efficiency detect_changes makes possible — it reparses documents that didn't change, every run, forever. Lesson 06 measures exactly how much work doing this right avoids.
2. Forgetting to register doc_hash in documents inside upsert_doc_chunks. If reingest() writes the chunks but never updates the documents table, the next detect_changes run will keep seeing that document as new (because documents was never updated) and will repeat ingest_document's work on every run, even though the chunks are already correctly saved.
3. Not committing (conn.commit()) at the end of reingest(). Without the commit, changes stay in the open transaction and don't actually persist in the database — an invisible problem with :memory: within the same process (where conn stays the same object), but a real one for any file-backed store if the process ends before committing.
4. Confusing chunks_written=0 with "the run did nothing useful". A reingest() that reports written=0, purged=0 on an unchanged corpus is exactly the correct behavior — it means the system cheaply confirmed the store is still in sync with the source. It's the same idea as Common Mistake 4 from Lesson 04: unchanged=13 is a signal of success, not that "nothing happened".

Exercises

Exercise 1: A third identical run (Easy)

On the same conn from run 1 and run 2 of the worked example, call reingest(conn, RAW_DOCS) a third time. What do you expect for chunks_written and for the total chunk count in the store? Confirm by running it.

r3 = reingest(conn, RAW_DOCS)
total_3 = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print(f"written={r3['chunks_written']} purged={r3['chunks_purged']} unchanged={r3['unchanged_count']}")
print("total chunks:", total_3)

written=0 purged=0 unchanged=13
total chunks: 57

Exercise 2: The two tables stay in sync (Medium)

After the worked example's runs, confirm with a SQL query that the documents table has exactly 13 rows — one per doc_id in the corpus, not one more or fewer, regardless of reingest() having been called twice.

docs_count = conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
print("documents table rows:", docs_count)

documents table rows: 13

Exercise 3: The danger of an incomplete current_docs (Hard)

Build partial_docs as a dictionary with only the first 5 documents from RAW_DOCS (for example, dict(list(RAW_DOCS.items())[:5])) — simulating a bug where the code that reads the source directory only found 5 of the 13 files. On a store that already has all 13 documents ingested (as at the end of the worked example), call reingest(conn, partial_docs). What happens to the other 8 documents? Think through the answer before running it — this is the most dangerous mistake in the entire module.

partial_docs = dict(list(RAW_DOCS.items())[:5])
print("partial_docs has", len(partial_docs), "documents:", list(partial_docs.keys()))

r_partial = reingest(conn, partial_docs)
total_partial = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
print("deleted:", r_partial["deleted"])
print("chunks_purged:", r_partial["chunks_purged"])
print("total chunks after:", total_partial)

partial_docs has 5 documents: ['cancellation-policy', 'no-show-policy', 'refund-policy', 'booking-faq', 'membership-tiers-faq']
deleted: ['boardroom-room-manual', 'focus-room-manual', 'lounge-room-manual', 'operations-manual-raw', 'payment-methods-faq', 'phonebooth-room-manual', 'studio-room-manual', 'wifi-and-equipment-faq']
chunks_purged: 37
total chunks after: 20

Summary and next step

• reingest() ties detect_changes (Lesson 04) together with upsert_chunk (Lesson 03) into a single function: it detects what changed, processes only new/modified with ingest_document + upsert_chunk, and purges deleted — never touching unchanged at all.
• Run twice on the full canonical corpus: run 1 writes 57 chunks; run 2, on the same corpus, writes zero — the store stays at 57, not 114. This is the module's central piece of evidence.
• The chunks and documents tables stay in sync with each other after any number of runs, thanks to the same primary-key UPSERT pattern on both.
• An incomplete current_docs is indistinguishable, to reingest(), from a real deletion — a genuine risk of any pipeline with automatic purging, worth knowing about even though this guide doesn't solve it.

Next lesson: 06 — Updating only what changed. A document genuinely edited (refund-policy gains a sentence), and the confirmation that reingest() rewrites a single chunk, not all 57.

Additional resources

1. Python — sqlite3 — commit() and why it's needed at the end of any function that writes, including reingest().
2. SQLite — Transactions — the model behind why a missing commit() leaves changes unpersisted.
3. production-rag-and-document-ingestion-guide — Module 1, Lesson 08 (08-mini-project-ingest-the-reservo-corpus.md): load_corpus(), the real function that, in a production system, builds the current_docs that reingest() receives — and whose correct behavior is the real safeguard against Exercise 3's risk.
4. airflow-and-declarative-orchestration-guide (Data Engineering ecosystem) — where you learn to add retries, alerts, and threshold safeguards to a pipeline like this before running it unsupervised in production.