SQL and Data Modeling

Your retriever finds the right paragraph. Then the process restarts, and its document dictionary is empty. The Data Structures for AI lesson made lookup efficient inside one process; now the documents need a home that outlives it.

Structured Query Language (SQL) lets you read and change data in a relational database. The database stores durable state in tables. Named columns describe fields, and each stored fact is a row. Constraints turn assumptions into rules the database can enforce, even when another process writes the data.

Use a small, fictional support-policy corpus for Acme Cloud. Its current API-key policy says "rotate within seven days." An older version says ninety days. Both paragraphs can match a search, but only one should support a current answer. This policy is sample data, not security advice.

SQLite and Python's standard-library sqlite3 interface keep the lab local and runnable. Run the Python cells in order in one session; they share the connection and tables. The later PostgreSQL sections are extensions, not additional software you need for the SQLite lab.

From Python objects to rows

In memory, you might keep documents in a dictionary keyed by document_id. A database makes those identities and their relationships explicit. A workspace separates one customer's data from another's; a chunk is a paragraph we might retrieve; a principal is the user or service whose access we're checking:

Now the links are data rather than a convention in application code. In-memory Python dictionaries let you nest chunks inside documents, but they leave integrity to developer discipline. Separate writes to a persistent store can leave a document without its chunks if an ingestion worker fails halfway through. A relational schema turns relationship assumptions into database-enforced rules: a primary key uniquely identifies each row, while a foreign key ensures references point to existing records. Our references specify NOT NULL, so a chunk can't omit its source document or point into the void. A transaction will later keep the related writes together.

Keep identity, ownership, and permission links in these relational columns. Optional parser metadata, such as detected headings, can live in a separate JSON field. Putting a document_id inside JSON doesn't make it a foreign key; the declared column constraint is what enforces that relationship.^{[1]Reference 1Constraintshttps://www.postgresql.org/docs/current/ddl-constraints.html}

View larger

Create the smallest useful schema

The first example creates four tables and inserts a tiny corpus. CREATE TABLE declares columns and constraints; INSERT INTO adds rows. TEXT holds strings, NOT NULL requires a value, and UNIQUE rejects a repeated combination of values. A CHECK rejects a row when its condition is false. A check that evaluates to NULL can pass, so it doesn't replace NOT NULL.^{[2]Reference 2CREATE TABLEhttps://www.sqlite.org/lang_createtable.html}

The database file lives in a fresh temporary directory so rerunning the lesson won't overwrite earlier work. A file-backed connection can be closed and reopened without losing committed rows. sqlite3.connect(":memory:") would be convenient for scratch work, but it would lose the database when that connection closes.^{[3]Reference 3sqlite3: DB-API interface for SQLite databaseshttps://docs.python.org/3/library/sqlite3.html}

Its validity interval is half-open: active at valid_from, inactive at valid_to. A NULL valid_to means the version has no known end yet. This local SQLite lab stores canonical UTC timestamps as text; the PostgreSQL version later uses TIMESTAMPTZ.

PRAGMA foreign_keys = ON enables SQLite's foreign-key rules on this connection. Don't depend on the build's default setting.^{[4]Reference 4SQLite Foreign Key Supporthttps://www.sqlite.org/foreignkeys.html} SQLite also preserves a historical exception: an ordinary TEXT PRIMARY KEY can accept NULL unless it explicitly says NOT NULL. PostgreSQL primary keys reject NULL automatically, but this lab declares both constraints.^{[2]Reference 2CREATE TABLEhttps://www.sqlite.org/lang_createtable.html}

This lab requires Python 3.12 or later and explicitly selects its legacy transaction mode. In that mode, data-changing statements can open a transaction, while SELECT doesn't automatically start a multi-statement snapshot. Pinning the mode keeps the examples independent of a future change to Python's default.^{[3]Reference 3sqlite3: DB-API interface for SQLite databaseshttps://docs.python.org/3/library/sqlite3.html}

We commit the seed data so later examples can open and roll back new transactions on purpose.

import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory

lab_directory = TemporaryDirectory(prefix="leetllm-sql-")
db_path = Path(lab_directory.name) / "support.sqlite"
conn = sqlite3.connect(db_path, autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL)
conn.execute("PRAGMA foreign_keys = ON")

conn.executescript(
    """
    CREATE TABLE workspaces (
        id TEXT PRIMARY KEY NOT NULL,
        name TEXT NOT NULL
    );

    CREATE TABLE documents (
        id TEXT PRIMARY KEY NOT NULL,
        workspace_id TEXT NOT NULL REFERENCES workspaces(id),
        source_key TEXT NOT NULL,
        title TEXT NOT NULL,
        valid_from TEXT NOT NULL,
        valid_to TEXT,
        CHECK (valid_to IS NULL OR valid_to > valid_from),
        UNIQUE (workspace_id, source_key, valid_from)
    );

    CREATE TABLE chunks (
        id TEXT PRIMARY KEY NOT NULL,
        document_id TEXT NOT NULL REFERENCES documents(id),
        chunk_text TEXT NOT NULL,
        UNIQUE (document_id, chunk_text)
    );

    CREATE TABLE chunk_permissions (
        chunk_id TEXT NOT NULL REFERENCES chunks(id),
        principal TEXT NOT NULL,
        PRIMARY KEY (chunk_id, principal)
    );
    """
)

conn.executemany(
    "INSERT INTO workspaces (id, name) VALUES (?, ?)",
    [("w01", "Acme Cloud"), ("w02", "Other Workspace")],
)
conn.executemany(
    """
    INSERT INTO documents
        (id, workspace_id, source_key, title, valid_from, valid_to)
    VALUES (?, ?, ?, ?, ?, ?)
    """,
    [
        (
            "d7-old",
            "w01",
            "api-key-policy",
            "API Key Policy",
            "2026-01-01T00:00:00Z",
            "2026-07-01T00:00:00Z",
        ),
        (
            "d7",
            "w01",
            "api-key-policy",
            "API Key Policy",
            "2026-07-01T00:00:00Z",
            None,
        ),
        (
            "d9",
            "w01",
            "account-help",
            "Account Help",
            "2026-01-01T00:00:00Z",
            None,
        ),
        (
            "d99",
            "w02",
            "private-token-policy",
            "Private Token Policy",
            "2026-01-01T00:00:00Z",
            None,
        ),
    ],
)
conn.executemany(
    "INSERT INTO chunks (id, document_id, chunk_text) VALUES (?, ?, ?)",
    [
        ("c7-old", "d7-old", "API keys rotate within ninety days."),
        ("c7a", "d7", "API keys rotate within seven days."),
        ("c7b", "d7", "Email security with your workspace ID for key rotation."),
        ("c9a", "d9", "Reset passwords from account settings."),
        ("c99", "d99", "Private key exception for another workspace."),
    ],
)
conn.executemany(
    "INSERT INTO chunk_permissions (chunk_id, principal) VALUES (?, ?)",
    [
        ("c7-old", "agent:u3"),
        ("c7a", "agent:u1"),
        ("c7a", "agent:u3"),
        ("c7b", "agent:u1"),
        ("c7b", "agent:u3"),
        ("c9a", "agent:u1"),
        ("c99", "agent:u3"),
    ],
)
conn.commit()

print(
    "rows",
    [
        conn.execute("SELECT COUNT(*) FROM workspaces").fetchone()[0],
        conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0],
        conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0],
        conn.execute("SELECT COUNT(*) FROM chunk_permissions").fetchone()[0],
    ],
)

rows [2, 4, 5, 7]

Those counts describe two workspaces, four document versions, five chunks, and seven permission rows. executemany repeats one parameterized insert for each Python tuple. commit() makes those inserts a completed unit rather than leaving them pending on this connection.

d7-old and d7 share source_key = "api-key-policy", but their validity intervals don't overlap. Ownership and version time live on the document; visibility lives on the chunk. Two paragraphs in one document can have different visibility, and every retrieved paragraph still points at the exact source version.

The (document_id, chunk_text) uniqueness rule treats identical text in one version as a duplicate for this tiny corpus. Real chunkers may produce identical paragraphs at different positions; use a stable chunk position or source span if those occurrences must remain distinct. A unique text value also doesn't prove that a chunk or document is immutable.

Select rows with parameters

SELECT chooses columns. WHERE keeps some rows and drops the rest. ORDER BY makes the output order something you can test. This query keeps Acme versions that have no recorded end time.

Keep the query shape and its values separate. Put a placeholder in the statement and pass the value beside it. That split prevents untrusted data from changing the statement's structure through SQL injection.^{[3]Reference 3sqlite3: DB-API interface for SQLite databaseshttps://docs.python.org/3/library/sqlite3.html}

workspace_id = "w01"

rows = conn.execute(
    """
    SELECT id, title
    FROM documents
    WHERE workspace_id = ?
      AND valid_to IS NULL
    ORDER BY id
    """,
    (workspace_id,),
).fetchall()

print(rows)

[('d7', 'API Key Policy'), ('d9', 'Account Help')]

The SQL statement has a ? where a value belongs, and the tuple (workspace_id,) holds that value. execute runs the query; fetchall collects its result rows as Python tuples. These rows have no end time, which isn't yet the same as proving they're active now: a future version could also have valid_to IS NULL.

Placeholder spelling depends on the driver. Python's SQLite interface uses ?; PostgreSQL prepared statements use $1 and $2, as shown in the PostgreSQL PREPARE reference. Use the parameter API your driver gives you. Keep values out of SQL text.

NULL means missing or unknown, not an ordinary value. Comparisons with it yield an unknown truth value, so valid_to = NULL doesn't select open-ended rows. Write valid_to IS NULL. PostgreSQL follows the same three-valued rule in its comparison documentation.

Binding keeps a value from becoming SQL syntax. A chat message is still untrusted input, even when another model wrote it. It must not rewrite the statement shape.

Join a chunk to its source

Matching text isn't enough for an answer generator. The UI also needs the title, version, or document ID that makes the evidence citable. A join combines rows according to a condition. Ours matches a chunk's document reference to the corresponding document identity.

Here, chunks.document_id refers to documents.id. Follow that link and one result set can return both the paragraph and its source title:

term = "%key%"

citations = conn.execute(
    """
    SELECT c.id, d.title, d.valid_from, d.valid_to, c.chunk_text
    FROM chunks AS c
    JOIN documents AS d ON d.id = c.document_id
    WHERE d.workspace_id = ?
      AND LOWER(c.chunk_text) LIKE ?
    ORDER BY d.valid_from, c.id
    """,
    ("w01", term),
).fetchall()

for chunk_id, title, valid_from, valid_to, text in citations:
    print(f"{chunk_id} [{title}] {valid_from} -> {valid_to or 'open'} | {text}")

c7-old [API Key Policy] 2026-01-01T00:00:00Z -> 2026-07-01T00:00:00Z | API keys rotate within ninety days.
c7a [API Key Policy] 2026-07-01T00:00:00Z -> open | API keys rotate within seven days.
c7b [API Key Policy] 2026-07-01T00:00:00Z -> open | Email security with your workspace ID for key rotation.

For c7a, document_id is d7, so the join attaches the title and dates from d7. It doesn't attach d7-old merely because that row has the same title. AS c and AS d are short table aliases, and c.id identifies which table supplies the column. The pattern %key% was bound as a value, not pasted into SQL text.

In LIKE, % matches any sequence of characters and _ matches one character. Binding doesn't make either wildcard literal. This exploratory query intentionally uses %key% to find key anywhere in the text. Before exposing search to a caller, we'll escape user-supplied wildcards and reject blank input.

It also showed a new problem. Both the retired ninety-day rule and the current seven-day rule match key, so a correct join alone doesn't decide which fact was valid for this request.

Query the version that was valid at a specific time

Now ask a time-bound question: which policy was true at a particular instant? The lab uses a fixed timestamp so every rerun has the same answer. A version is valid when valid_from <= as_of and either its end is unknown or as_of < valid_to:

AS_OF = "2026-08-12T12:00:00Z"

current_citations = conn.execute(
    """
    SELECT c.id, d.title, c.chunk_text
    FROM chunks AS c
    JOIN documents AS d ON d.id = c.document_id
    WHERE d.workspace_id = ?
      AND d.valid_from <= ?
      AND (d.valid_to IS NULL OR ? < d.valid_to)
      AND LOWER(c.chunk_text) LIKE ?
    ORDER BY c.id
    """,
    ("w01", AS_OF, AS_OF, "%key%"),
).fetchall()

for chunk_id, title, text in current_citations:
    print(f"{chunk_id} [{title}] {text}")

c7a [API Key Policy] API keys rotate within seven days.
c7b [API Key Policy] Email security with your workspace ID for key rotation.

The half-open interval [valid_from, valid_to) gives a clean boundary: at exactly 2026-07-01T00:00:00Z, d7-old is inactive and d7 is active.

SQLite's text comparison works here because every timestamp uses the same fixed-width UTC format. Mixing offsets or formats would break that ordering.

View larger

The schema prevents duplicate start times for a source, but it doesn't reject overlapping intervals. For example, a second row beginning June 15 and ending July 15 could overlap both versions without violating any current constraint. An ingestion transaction must check the existing intervals under appropriate concurrency control, or PostgreSQL can enforce a suitable exclusion constraint.^{[1]Reference 1Constraintshttps://www.postgresql.org/docs/current/ddl-constraints.html}

PostgreSQL should store these instants as TIMESTAMPTZ; it normalizes zoned input internally and converts it for display in the active session time zone.^{[5]Reference 5Date/Time Typeshttps://www.postgresql.org/docs/current/datatype-datetime.html}

LIKE is enough for this tiny vocabulary demonstration. It isn't semantic retrieval: it will miss paraphrases such as "rotate credentials" when no row contains "key." The PostgreSQL extension later adds vector similarity without discarding time, tenant, or permission checks.

Group rows to inspect the corpus

Retrieval returns individual chunks. Operators often need a summary instead: how many current chunks does each Acme source contribute? GROUP BY collects input rows that share the named keys, and COUNT(*) reduces each group to one number.

At AS_OF, predict two chunks for d7 and one for d9; retired d7-old should contribute none. The query makes that prediction testable:

summary = conn.execute(
    """
    SELECT d.id, d.title, COUNT(*) AS chunk_count
    FROM documents AS d
    JOIN chunks AS c ON c.document_id = d.id
    WHERE d.workspace_id = ?
      AND d.valid_from <= ?
      AND (d.valid_to IS NULL OR ? < d.valid_to)
    GROUP BY d.id, d.title
    ORDER BY d.id
    """,
    ("w01", AS_OF, AS_OF),
).fetchall()

for document_id, title, chunk_count in summary:
    print(document_id, title, chunk_count)

d7 API Key Policy 2
d9 Account Help 1

WHERE filters rows before grouping, so retired d7-old never contributes to these counts. To filter completed groups, such as keeping sources with at least two chunks, add HAVING COUNT(*) >= 2 after GROUP BY.^{[6]Reference 6Aggregate Functionshttps://www.postgresql.org/docs/current/tutorial-agg.html}

An ordinary JOIN also drops documents with no matching chunks. For an ingestion audit that must show zero-chunk documents, use LEFT JOIN chunks AS c ON c.document_id = d.id and COUNT(c.id). A left join preserves the document with null chunk fields; counting c.id gives zero, whereas COUNT(*) would count that placeholder row as one.

Constraints catch broken relationships

Code can be buggy. An ingestion job can create a chunk for a document ID that never arrived. Without a constraint, that orphan may show up in retrieval with no trustworthy source.

Predict the next failure: inserting a chunk for missing-document should stop before it becomes data. The foreign key on chunks turns that silent corruption into a visible failure.

Under the lab's pinned sqlite3 transaction mode, a data-changing statement opens a transaction implicitly. These immediate constraint failures revert the failing statement; earlier successful statements in that transaction can still be pending. rollback() discards the whole pending unit before the lab continues:

try:
    conn.execute(
        "INSERT INTO chunks (id, document_id, chunk_text) VALUES (?, ?, ?)",
        ("broken", "missing-document", "This row has no source."),
    )
except sqlite3.IntegrityError:
    print("blocked orphan chunk")
    conn.rollback()

try:
    conn.execute(
        "INSERT INTO chunks (id, document_id, chunk_text) VALUES (?, ?, ?)",
        (None, "d7", "A chunk can't have a missing identity."),
    )
except sqlite3.IntegrityError:
    print("blocked null primary key")
    conn.rollback()

missing = conn.execute(
    "SELECT COUNT(*) FROM chunks WHERE id = ?",
    ("broken",),
).fetchone()[0]
print("orphan rows stored", missing)

blocked orphan chunk
blocked null primary key
orphan rows stored 0

Primary keys catch duplicate identities. The explicit NOT NULL catches SQLite's otherwise-accepted missing text identity. Foreign keys catch broken references.

These rules belong in the schema because every writer must obey them, including the current Python script. Remove NOT NULL from chunks.id and the second check stops printing blocked null primary key: SQLite accepts the missing identity even though the column says PRIMARY KEY.

Deleting a parent doesn't silently erase its children

The current schema doesn't declare ON DELETE CASCADE. Deleting d7 while c7a and c7b still reference it should fail. Deleting those chunks first also requires handling their permission rows. Predict which table must be cleared first:

try:
    conn.execute("DELETE FROM documents WHERE id = ?", ("d7",))
except sqlite3.IntegrityError:
    print("blocked parent deletion")
    conn.rollback()

try:
    conn.execute(
        "DELETE FROM chunks WHERE document_id = ?", ("d7",)
    )
except sqlite3.IntegrityError:
    print("blocked chunk deletion: permissions still reference it")
    conn.rollback()

# Rehearse dependency order without deleting the lesson's corpus permanently.
conn.execute("BEGIN")
conn.execute(
    """DELETE FROM chunk_permissions WHERE chunk_id IN
       (SELECT id FROM chunks WHERE document_id = ?)""", ("d7",)
)
conn.execute("DELETE FROM chunks WHERE document_id = ?", ("d7",))
conn.execute("DELETE FROM documents WHERE id = ?", ("d7",))
print("d7 exists inside deletion", bool(conn.execute(
    "SELECT 1 FROM documents WHERE id = ?", ("d7",)
).fetchone()))
conn.rollback()
print("d7 restored after rollback", bool(conn.execute(
    "SELECT 1 FROM documents WHERE id = ?", ("d7",)
).fetchone()))

blocked parent deletion
blocked chunk deletion: permissions still reference it
d7 exists inside deletion False
d7 restored after rollback True

Choose deletion behavior deliberately: reject deletion while references exist, delete dependent rows together, or declare cascades where that matches the product. A cascade follows declared foreign keys; it doesn't delete remote vectors or cached answers. Historical evidence needed by evaluations may need retention rather than cascading deletion.^{[4]Reference 4SQLite Foreign Key Supporthttps://www.sqlite.org/foreignkeys.html}

Make permissions queryable data

Integrity tells us which rows connect. It doesn't tell us who may read one. An access-control list says which principals may view a resource. Instead of hiding that list in application logic, this schema stores one permitted (chunk_id, principal) pair per row. The retrieval rule is:

Return a chunk only when its document belongs to the active workspace and a permission row exists for the active principal.

Alice (agent:u1) may read key and account help. Bob (agent:u3) may read key paragraphs for Acme, but not Acme account-help content. The query below turns that distinction into rows we can inspect:

In an application, the authenticated session supplies principal, and the server verifies the requested workspace against that session. They aren't claims a chat message or model may choose. The function accepts them as arguments here only so the lab can exercise several identities explicitly.

def permitted_matches(
    workspace_id: str,
    principal: str,
    word: str,
    as_of: str,
) -> list[str]:
    return [
        row[0]
        for row in conn.execute(
            """
            SELECT c.id
            FROM chunks AS c
            JOIN documents AS d ON d.id = c.document_id
            JOIN chunk_permissions AS p ON p.chunk_id = c.id
            WHERE d.workspace_id = ?
              AND d.valid_from <= ?
              AND (d.valid_to IS NULL OR ? < d.valid_to)
              AND p.principal = ?
              AND LOWER(c.chunk_text) LIKE ?
            ORDER BY c.id
            """,
            (workspace_id, as_of, as_of, principal, f"%{word.lower()}%"),
        ).fetchall()
    ]

print("Bob key", permitted_matches("w01", "agent:u3", "key", AS_OF))
print("Bob password", permitted_matches("w01", "agent:u3", "password", AS_OF))
print("Alice password", permitted_matches("w01", "agent:u1", "password", AS_OF))

Bob key ['c7a', 'c7b']
Bob password []
Alice password ['c9a']

A missing workspace predicate leaks a row

Bob also has permission to read one row in w02. That's useful for showing the failure rather than assuming the rule is correct.

Suppose the active support screen is displaying Acme (w01), but a programmer filters only by user and query word. Bob's key search can now pull a private rule from w02 even though the UI is scoped to Acme:

unsafe = conn.execute(
    """
    SELECT c.id
    FROM chunks AS c
    JOIN chunk_permissions AS p ON p.chunk_id = c.id
    JOIN documents AS d ON d.id = c.document_id
    WHERE d.valid_from <= ?
      AND (d.valid_to IS NULL OR ? < d.valid_to)
      AND p.principal = ?
      AND LOWER(c.chunk_text) LIKE ?
    ORDER BY c.id
    """,
    (AS_OF, AS_OF, "agent:u3", "%key%"),
).fetchall()

safe = permitted_matches("w01", "agent:u3", "key", AS_OF)

print("unsafe", [row[0] for row in unsafe])
print("safe", safe)

unsafe ['c7a', 'c7b', 'c99']
safe ['c7a', 'c7b']

The scoped query checks both dimensions: current workspace and current principal. Bob can read c99 in a different request, but that doesn't make it valid context for an Acme answer. This is a cross-workspace disclosure in the active request, not evidence that Bob lacks all access to c99.

View larger

For a production PostgreSQL application, row-level security (RLS) can make a database policy enforce an additional boundary. RLS isn't automatic: once it's enabled for a table, normal access must satisfy a policy, and an enabled table with no policy uses default deny.^{[7]Reference 7PostgreSQL Row Security Policieshttps://www.postgresql.org/docs/current/ddl-rowsecurity.html}

Superusers and roles with BYPASSRLS always bypass those policies. Table owners normally bypass them too unless the table uses FORCE ROW LEVEL SECURITY. Run application queries with a role that's subject to the policy, then test that role directly.

Use a transaction for all-or-nothing ingestion

A database transaction groups changes into one unit. A commit makes that unit permanent; a rollback discards it. That matters when a new document and its chunks must appear together. A visible document with missing chunks is confusing; visible chunks with no source are worse.

Use with conn: here so Python commits an open transaction on success or rolls it back when an exception escapes the block. Catch the exception outside the block: swallowing it inside could let earlier writes commit. The context manager doesn't open a transaction by itself.^{[3]Reference 3sqlite3: DB-API interface for SQLite databaseshttps://docs.python.org/3/library/sqlite3.html}

With this lab's pinned transaction mode, the first INSERT inside the block opens one. The second INSERT intentionally points at a missing source, so its foreign-key error rolls back the document insertion too:

try:
    with conn:
        conn.execute(
            """
            INSERT INTO documents
                (id, workspace_id, source_key, title, valid_from, valid_to)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                "d20",
                "w01",
                "incident-policy",
                "Incident Policy",
                "2026-08-01T00:00:00Z",
                None,
            ),
        )
        conn.execute(
            "INSERT INTO chunks (id, document_id, chunk_text) VALUES (?, ?, ?)",
            ("c20", "not-d20", "Incidents are reviewed in two days."),
        )
except sqlite3.IntegrityError:
    print("transaction rolled back")

visible = conn.execute(
    "SELECT COUNT(*) FROM documents WHERE id = ?",
    ("d20",),
).fetchone()[0]
print("partial document visible", bool(visible))

transaction rolled back
partial document visible False

The transaction protects database rows, not a remote embedding service. You have two reasonable designs: compute embeddings before the database transaction and commit document plus chunks together, or write a pending ingestion record and publish a new complete version only after all chunks are ready. Don't expose half an index to retrieval without making that state deliberate.

Publication must also retire the old version in that same unit. Committing its end time before the replacement is ready creates a gap; publishing the replacement without ending the old version creates overlapping facts. The earlier interval CHECK doesn't prevent either mistake.

Rehearse replacing d7 at AS_OF. BEGIN IMMEDIATE claims SQLite's one writer slot before inspecting or changing publication state. A second connection can still read the committed policy, but its competing writer attempt fails immediately because this fixture sets timeout=0. A real caller would wait within a bound or retry the whole publication transaction.^{[8]Reference 8Transactionhttps://www.sqlite.org/lang_transaction.html}

other = sqlite3.connect(
    db_path, timeout=0, autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL
)
other.execute("PRAGMA foreign_keys = ON")

def active_policy_ids(db):
    return [row[0] for row in db.execute(
        """SELECT id FROM documents
           WHERE workspace_id = 'w01' AND source_key = 'api-key-policy'
             AND valid_from <= ? AND (valid_to IS NULL OR ? < valid_to)
           ORDER BY id""", (AS_OF, AS_OF)
    ).fetchall()]

try:
    conn.execute("BEGIN IMMEDIATE")
    try:
        other.execute("BEGIN IMMEDIATE")
    except sqlite3.OperationalError as error:
        assert error.sqlite_errorcode == sqlite3.SQLITE_BUSY
        print("competing writer must wait or retry")
    else:
        raise AssertionError("two writers must not own publication together")

    ended = conn.execute(
        "UPDATE documents SET valid_to = ? WHERE id = ? AND valid_to IS NULL",
        (AS_OF, "d7"),
    ).rowcount
    if ended != 1:
        raise ValueError("expected open version changed; reload publication state")
    print("writer before replacement", active_policy_ids(conn))
    print("reader during publication", active_policy_ids(other))

    conn.execute(
        "INSERT INTO documents VALUES (?, ?, ?, ?, ?, ?)",
        ("d7-next", "w01", "api-key-policy", "API Key Policy", AS_OF, None),
    )
    conn.execute(
        "INSERT INTO chunks VALUES (?, ?, ?)",
        ("c7-next", "d7-next", "API keys rotate within five days."),
    )
    conn.execute(
        "INSERT INTO chunk_permissions VALUES (?, ?)", ("c7-next", "agent:u3")
    )
    print("writer's complete replacement", active_policy_ids(conn))
    print("writer's authorized result", permitted_matches("w01", "agent:u3", "key", AS_OF))
finally:
    conn.rollback()  # Rehearsal: keep later cells on the original seven-day rule.
    other.rollback()
    other.close()
print("after rehearsal rollback", active_policy_ids(conn))

competing writer must wait or retry
writer before replacement []
reader during publication ['d7']
writer's complete replacement ['d7-next']
writer's authorized result ['c7-next']
after rehearsal rollback ['d7']

The writer temporarily sees a gap inside its own unfinished transaction; the other connection continues to see committed d7. In production, commit only after the replacement chunks and intended grants are complete. Here rollback restores the original rows for later exercises. This serializes cooperating SQLite publishers; it doesn't add a global overlap constraint, validate remote embeddings, or prevent another code path from publishing invalid intervals. PostgreSQL needs its own locking, exclusion, or serializable-transaction design for that invariant.

Atomicity doesn't choose what concurrent readers see

Rollback demonstrates atomicity: one transaction's writes appear together or not at all. Isolation answers a different question: what can one transaction observe while another transaction commits changes?

PostgreSQL defaults to Read Committed isolation. Each ordinary SELECT sees a snapshot taken when that statement begins, so two SELECT statements in one transaction can observe different committed states.

That behavior is fine for many request paths. It can undermine an evaluation export that first counts current chunks and later reads them, because a concurrent ingestion may commit between those statements.^{[9]Reference 9Transaction Isolationhttps://www.postgresql.org/docs/current/transaction-iso.html}

When several reads must describe one stable corpus snapshot, use a read-only Repeatable Read transaction. This is PostgreSQL syntax, not part of the SQLite lab:

BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY;

SELECT d.id, COUNT(*)
FROM documents AS d
JOIN chunks AS c ON c.document_id = d.id
WHERE d.workspace_id = 'w01'
  AND d.valid_from <= TIMESTAMPTZ '2026-08-12 12:00:00+00'
  AND (
      d.valid_to IS NULL
      OR TIMESTAMPTZ '2026-08-12 12:00:00+00' < d.valid_to
  )
GROUP BY d.id;

-- Later reads in this transaction keep the same snapshot.
SELECT c.id, c.chunk_text
FROM chunks AS c
JOIN documents AS d ON d.id = c.document_id
WHERE d.workspace_id = 'w01'
  AND d.valid_from <= TIMESTAMPTZ '2026-08-12 12:00:00+00'
  AND (
      d.valid_to IS NULL
      OR TIMESTAMPTZ '2026-08-12 12:00:00+00' < d.valid_to
  );

COMMIT;

Repeatable Read fixes the snapshot after the first non-control statement, so the count and detail rows agree even if another transaction commits later.

Serializable isolation is stricter when concurrent transactions enforce a cross-row business rule, but PostgreSQL may abort one transaction with a serialization failure. Applications using that level must retry the whole transaction rather than the final statement alone.

View larger

Keep retries from duplicating completed work

A transaction makes one attempt all-or-nothing. It doesn't tell a worker whether the request already succeeded before a timeout. If the worker commits new rows, crashes before sending its response, and retries the same request, a second write attempt can duplicate work.

For retried writes, store a stable idempotency key in the same transaction as the rows created for that request. A uniqueness constraint lets the database accept the first attempt and recognize later repeats even after the original process disappears:

The same key must also mean the same input. Store a fingerprint of all input fields alongside it. Here, json.dumps gives a fixed field list a consistent representation, and SHA-256 turns those bytes into a compact comparison value. A changed title or paragraph must raise an error, not be mistaken for a successful retry.

import hashlib
import json

conn.execute(
    """
    CREATE TABLE ingestion_requests (
        request_id TEXT PRIMARY KEY NOT NULL,
        document_id TEXT NOT NULL,
        payload_hash TEXT NOT NULL
    )
    """
)
conn.commit()

def ingest_document_once(
    request_id: str,
    document_id: str,
    source_key: str,
    title: str,
    valid_from: str,
    chunk_id: str,
    chunk_text: str,
) -> str:
    payload = json.dumps(
        ["w01", document_id, source_key, title, valid_from, chunk_id, chunk_text],
        ensure_ascii=False,
        separators=(",", ":"),
    )
    payload_hash = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    with conn:
        created = conn.execute(
            """
            INSERT INTO ingestion_requests (request_id, document_id, payload_hash)
            VALUES (?, ?, ?)
            ON CONFLICT(request_id) DO NOTHING
            """,
            (request_id, document_id, payload_hash),
        ).rowcount
        if created == 0:
            existing_hash = conn.execute(
                "SELECT payload_hash FROM ingestion_requests WHERE request_id = ?",
                (request_id,),
            ).fetchone()[0]
            if existing_hash != payload_hash:
                raise ValueError("idempotency key reused with changed input")
            return "skipped retry"

        conn.execute(
            """
            INSERT INTO documents
                (id, workspace_id, source_key, title, valid_from, valid_to)
            VALUES (?, ?, ?, ?, ?, NULL)
            """,
            (document_id, "w01", source_key, title, valid_from),
        )
        conn.execute(
            "INSERT INTO chunks (id, document_id, chunk_text) VALUES (?, ?, ?)",
            (chunk_id, document_id, chunk_text),
        )
    return "inserted"

print(
    "first",
    ingest_document_once(
        "req-30",
        "d30",
        "key-rotation-faq",
        "Key Rotation FAQ",
        "2026-08-10T00:00:00Z",
        "c30",
        "Temporary keys expire after 30 days.",
    ),
)
print(
    "retry",
    ingest_document_once(
        "req-30",
        "d30",
        "key-rotation-faq",
        "Key Rotation FAQ",
        "2026-08-10T00:00:00Z",
        "c30",
        "Temporary keys expire after 30 days.",
    ),
)
stored = conn.execute(
    "SELECT COUNT(*) FROM documents WHERE id = ?",
    ("d30",),
).fetchone()[0]
print("d30 rows", stored)

try:
    ingest_document_once(
        "req-30", "d30", "key-rotation-faq", "Key Rotation FAQ",
        "2026-08-10T00:00:00Z", "c30", "Changed policy text.",
    )
except ValueError as exc:
    print("changed retry", exc)
else:
    raise AssertionError("changed input must not reuse an idempotency key")

first inserted
retry skipped retry
d30 rows 1
changed retry idempotency key reused with changed input

The first call stores the request marker, document, and chunk together. An identical retry skips the write. Reusing req-30 with different text raises an error and leaves the original policy untouched. The fingerprint is an equality check, not authentication; the server still owns workspace and user identity.

This function leaves the new chunk ungranted, so retrieval won't expose it yet. A complete ingestion operation would also write the intended permission rows before publishing the document. Keeping that default closed is safer than treating a missing permission as public access.

This pattern works when the protected side effect lives in the same database transaction. A local marker can't atomically cover a deployment API or another remote service. For an external side effect, pass the same key to a downstream API that enforces it, or use a workflow such as an outbox with an idempotent consumer.

Index the predicates you run often

Without an index, a database may need to inspect many rows to answer a filter. An index is an additional data structure that lets the engine reach selected rows more directly, at the cost of storage and write work.

Our authorized query repeatedly filters permissions by principal and follows document ownership. Add indexes for those paths, then ask SQLite for its query plan. The plan describes the database's chosen access strategy; measure runtime separately.^{[10]Reference 10EXPLAIN QUERY PLANhttps://www.sqlite.org/eqp.html}

conn.executescript(
    """
    CREATE INDEX permission_principal_idx
        ON chunk_permissions (principal, chunk_id);
    CREATE INDEX document_workspace_idx
        ON documents (workspace_id, valid_from, valid_to, id);
    """
)

authorized_sql = """
SELECT c.id, d.title, c.chunk_text
FROM chunks AS c
JOIN documents AS d ON d.id = c.document_id
JOIN chunk_permissions AS p ON p.chunk_id = c.id
WHERE d.workspace_id = ?
  AND d.valid_from <= ?
  AND (d.valid_to IS NULL OR ? < d.valid_to)
  AND p.principal = ?
  AND LOWER(c.chunk_text) LIKE ? ESCAPE '!'
ORDER BY c.id
"""

plan = conn.execute(
    "EXPLAIN QUERY PLAN " + authorized_sql,
    ("w01", AS_OF, AS_OF, "agent:u3", "%key%"),
).fetchall()
for _node_id, _parent_id, _unused, detail in plan:
    print(detail)

SEARCH p USING COVERING INDEX permission_principal_idx (principal=?)
SEARCH c USING INDEX sqlite_autoindex_chunks_1 (id=?)
SEARCH d USING INDEX sqlite_autoindex_documents_1 (id=?)
USE TEMP B-TREE FOR ORDER BY

Plan details vary as engines and statistics change. Inspect the production query plan with realistic tenant and permission selectivity, then measure request latency on the intended workload.

SEARCH visits a selected subset through the named access path; SCAN visits a full table or index. COVERING means the selected fields can come from that index without an additional table lookup for those fields. USE TEMP B-TREE FOR ORDER BY identifies sorting work that remains after indexed lookups. Seeing one index name doesn't establish that every join or filter became cheap.^{[10]Reference 10EXPLAIN QUERY PLANhttps://www.sqlite.org/eqp.html}

In this corpus, a sequential path inspects all seven chunk_permissions rows. An index on (principal, chunk_id) can seek the four rows for agent:u3 (c7-old, c7a, c7b, c99). Workspace and validity predicates then drop the retired version and the other-workspace chunk, leaving c7a and c7b.

Those numbers count candidate rows. They do not describe storage operations or runtime. An index still has to traverse its own structure, and the planner can choose a different join order. Whatever path it picks, the authorization rule must stay unchanged.

Query-plan check: PostgreSQL uses EXPLAIN for estimated plans and EXPLAIN ANALYZE for actual row counts and timing. The latter executes the statement, including writes. Use a read query here. A surrounding rollback can undo ordinary database writes, but it can't undo every possible effect, such as a function calling an external service. Plans from tiny tables also don't predict plans for large ones.^{[11]Reference 11Using EXPLAINhttps://www.postgresql.org/docs/current/using-explain.html}

Indexes on principal, workspace, and validity time give the planner candidate access paths for authorization and version filters. They don't make LOWER(c.chunk_text) LIKE ? a fast B-tree lookup on chunk_text. Substring LIKE at scale needs full-text search, a trigram index, or a vector path. This demo corpus is small enough that a scan is fine.

View larger

Build a small authorized retriever

The schema, filters, and indexes now fit one function a caller can test. It returns permitted matching paragraphs beside their document titles. The query above uses ESCAPE '!': !% matches a literal %, !_ matches a literal _, and !! matches a literal !. Escape those characters in the user's text before adding our own surrounding % wildcards.

def search_support(
    workspace_id: str,
    principal: str,
    query: str,
    as_of: str,
) -> list[tuple[str, str, str]]:
    text = query.strip().lower()
    if not text:
        return []
    literal = text.replace("!", "!!").replace("%", "!%").replace("_", "!_")
    pattern = f"%{literal}%"
    return conn.execute(
        authorized_sql,
        (workspace_id, as_of, as_of, principal, pattern),
    ).fetchall()

for chunk_id, title, text in search_support("w01", "agent:u3", "key", AS_OF):
    print(f"{chunk_id} | {title} | {text}")

c7a | API Key Policy | API keys rotate within seven days.
c7b | API Key Policy | Email security with your workspace ID for key rotation.

This is lexical search: it looks for a literal text fragment. A query of % now searches for an actual percent sign rather than matching every row. Blank queries return no rows. The SQL parameter still protects statement structure; escaping protects the product's substring-search behavior. Neither one supplies authorization.

Passing as_of keeps a retired policy out of current answers and still lets a historical audit reproduce an older one. A retrieval-augmented generation (RAG) system usually also retrieves semantically related text before a large language model (LLM) drafts an answer. That path should reuse the same source, time, and permission model.

Test forbidden rows too

Test forbidden results as well as matches. Make the forbidden fixture match the query first: a cross-workspace test using token would be weak here because c99 contains key, not token. It would return nothing even if the workspace predicate were missing.

bob_key = [row[0] for row in search_support("w01", "agent:u3", "key", AS_OF)]
bob_password = search_support("w01", "agent:u3", "password", AS_OF)
other_workspace_key = [
    row[0] for row in search_support("w02", "agent:u3", "key", AS_OF)
]
unknown_user = search_support("w01", "agent:unknown", "key", AS_OF)
retired_key = [
    row[0]
    for row in search_support("w01", "agent:u3", "key", "2026-06-01T00:00:00Z")
]

assert bob_key == ["c7a", "c7b"]
assert bob_password == []
assert other_workspace_key == ["c99"]
assert "c99" not in bob_key
assert unknown_user == []
assert retired_key == ["c7-old"]
assert search_support("w01", "agent:u3", "%", AS_OF) == []
assert search_support("w01", "agent:u3", "   ", AS_OF) == []
assert [
    row[0]
    for row in search_support("w01", "agent:u3", "key", "2026-07-01T00:00:00Z")
] == ["c7a", "c7b"]

print("retrieval boundary tests passed")

retrieval boundary tests passed

The tests guard the central retrieval boundary: context must be relevant, authorized, and valid at the requested time before it reaches an LLM prompt.

The historical assertion also proves that keeping old versions isn't the same as leaking them into current answers.

Close the connection and read the file again

So far, every query has used the same open connection. Test the promise that motivated the database: after committing and closing it, can another connection recover the same authorized rows?

The global conn name is rebound below, so search_support now queries the reopened file. No schema creation or seeding runs a second time:

conn.commit()
conn.close()

conn = sqlite3.connect(db_path, autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL)
conn.execute("PRAGMA foreign_keys = ON")
reopened = [row[0] for row in search_support("w01", "agent:u3", "key", AS_OF)]
assert reopened == ["c7a", "c7b"]
print("reopened", reopened)

conn.close()
lab_directory.cleanup()  # Remove only this lab's temporary files.

reopened ['c7a', 'c7b']

The file, not the connection object, preserved the rows. Cleanup removes this lab's file only after the check. An application would use a stable path on persistent storage and keep it across runs. Don't rerun the seed script against that existing database; open it and query it. A file inside an ephemeral container filesystem still disappears when that container is removed, so the Docker storage distinction applies here too.^{[3]Reference 3sqlite3: DB-API interface for SQLite databaseshttps://docs.python.org/3/library/sqlite3.html}

Extend the model with PostgreSQL and pgvector

SQLite gave us a fully runnable lab without a separate database server. A PostgreSQL server becomes useful when application processes on several machines need a shared database and its concurrency controls. The table relationships stay useful, but adapt column types, driver placeholders, and transaction behavior rather than assuming the files are interchangeable.

When keyword matching isn't enough, the pgvector extension adds a vector column and distance operators. A vector is a list of numbers representing an embedding. Nearby embeddings can capture semantic similarity even when the exact words differ.

This PostgreSQL sketch assumes you've installed pgvector on the server and created the equivalent base tables, using TIMESTAMPTZ for the validity columns. $1 through $5 are server-side query parameters, not literals to paste into an ordinary SQL console. A Python driver may use different placeholder syntax.

Three vector dimensions keep the declaration readable. Choose the dimension required by your embedding model in a real system, and embed the query with the same model and preprocessing as the stored vectors. Matching dimensions alone doesn't make two models' vector spaces compatible. Embeddings live in their own table so each derived feature keeps its source chunk, model identifier, and creation time.

CREATE EXTENSION vector;

CREATE TABLE chunk_embeddings (
    id TEXT PRIMARY KEY,
    chunk_id TEXT NOT NULL REFERENCES chunks(id),
    embedding_model TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    embedding vector(3) NOT NULL,
    UNIQUE (chunk_id, embedding_model)
);

-- Optional once exact scans no longer meet the latency target:
CREATE INDEX chunks_embedding_hnsw
    ON chunk_embeddings USING hnsw (embedding vector_cosine_ops);

SELECT e.id AS embedding_id,
       e.embedding_model,
       c.id AS chunk_id,
       d.title,
       c.chunk_text,
       1 - (e.embedding <=> $1) AS cosine_similarity
FROM chunk_embeddings AS e
JOIN chunks AS c ON c.id = e.chunk_id
JOIN documents AS d ON d.id = c.document_id
WHERE d.workspace_id = $2
  AND d.valid_from <= $3
  AND (d.valid_to IS NULL OR $3 < d.valid_to)
  AND e.embedding_model = $4
  AND EXISTS (
      SELECT 1
      FROM chunk_permissions AS p
      WHERE p.chunk_id = c.id
        AND p.principal = $5
  )
ORDER BY e.embedding <=> $1
LIMIT 3;

In pgvector, <=> means cosine distance, so 1 - distance is cosine similarity. The query works without an approximate index: pgvector performs exact nearest-neighbor search by default. HNSW (Hierarchical Navigable Small World) builds a graph of nearby vectors to search a promising subset instead of exhaustively comparing every vector.

Add an HNSW index when exact scans no longer meet the latency target, then measure the recall tradeoff introduced by approximate search.^{[12]Reference 12pgvectorhttps://github.com/pgvector/pgvector}

Filtered search has one caveat. With an approximate vector index, filtering conditions can be applied after the index scans candidates. If workspace or permission filters reject many candidates, a query may return fewer rows than requested even when allowed matches exist.

Iterative index scans continue searching when filters remove too many candidates. Available since pgvector 0.8.0 and off by default, they can be enabled for this HNSW example with SET hnsw.iterative_scan = strict_order. Configured limits can stop the scan before it finds every exact neighbor. Re-measure recall with realistic workspace, model, and permission filters.^{[12]Reference 12pgvectorhttps://github.com/pgvector/pgvector[13]Reference 13pgvector 0.8.0 releasedhttps://www.postgresql.org/about/news/pgvector-080-released-2952/}

Vector check: Four checks carry into vector retrieval:

1. Authorization predicates belong in every retrieval query.
2. Version and embedding-model predicates make current answers reproducible.
3. Approximate vector retrieval still needs plan inspection and recall testing under restrictive filters.
4. A database policy such as PostgreSQL RLS can reinforce authorization, but it doesn't replace designing and testing retrieval behavior.

Keep feature and evaluation lineage in the schema

An embedding is a derived feature. If an evaluation improves after changing the embedding model, you need to know which feature row produced each ranked result. Store that identity instead of keeping only a floating-point score.

These PostgreSQL tables keep one evaluation run beside the embedding rows it ranked, so a later reader can point at the feature behind each score:

CREATE TABLE retrieval_eval_runs (
    id TEXT PRIMARY KEY,
    dataset_version TEXT NOT NULL,
    query_embedding_model TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE retrieval_eval_results (
    eval_run_id TEXT NOT NULL REFERENCES retrieval_eval_runs(id),
    query_id TEXT NOT NULL,
    rank INTEGER NOT NULL CHECK (rank > 0),
    embedding_id TEXT NOT NULL REFERENCES chunk_embeddings(id),
    score DOUBLE PRECISION NOT NULL,
    PRIMARY KEY (eval_run_id, query_id, rank),
    UNIQUE (eval_run_id, query_id, embedding_id)
);

embedding_id leads back to the chunk and embedding model. eval_run_id records dataset and query-model versions. Preserve those rows as immutable versions: overwriting the vector or source text in place would change what an old result points to. Use a versioned model identifier and a new chunk identity when the source text changes.

For a reproducible evaluation, also record the retrieval query/configuration, requested policy time, and corpus snapshot. This minimal schema still needs those fields. A stored foreign key identifies a row; preserving its version and the ranking inputs makes the comparison reproducible.

A timestamp alone can't answer that question because two runs may use different models at the same time.

This schema stores ranked evidence alongside scores. Compute recall or hit rate by adding relevance labels for each evaluation query. Store those labels in the versioned dataset named by the run.

Database skills used in retrieval

The process can restart now without emptying the corpus. These checks keep a retrieved paragraph tied to its version, tenant, and principal: