Chapter 25 · Ingestion

Pulling from a relational database

Most of a database should never be indexed. For the free-text columns that should, how to read a consistent snapshot while the table moves underneath you.

11 min read

Chapter 12 was emphatic that database rows are queried live, not indexed, and nothing here walks that back. Atlas's orders and inventory are a tool, not an index.

This chapter is about the exception it named: free-text columns inside a relational source. A notes field on a support ticket, a description on a product, a resolution_summary on a closed case. Those are prose that happens to live in Postgres, and prose is a document-shaped problem.

Deciding what to index

A column qualifies on three tests, and it needs all three.

Test	Why
It is prose, not a value	A `status` of `shipped` has nothing to retrieve
Questions ask about its meaning	"What did we do about damaged pallets last year?"
It is not reachable by a query	If `where status = 'x'` answers it, that is a tool

-- Before indexing a column, look at it. Length distribution answers the first
-- test faster than reading a schema does.
select
  count(*)                                             as rows,
  count(*) filter (where resolution_notes is not null) as populated,
  round(avg(length(resolution_notes)))                 as avg_chars,
  percentile_cont(0.5)  within group (order by length(resolution_notes)) as p50,
  percentile_cont(0.95) within group (order by length(resolution_notes)) as p95,
  max(length(resolution_notes))                        as max_chars
from tickets
where closed_at is not null;

A column with a p50 of 14 characters is a status field people type into. A column with a p50 of 400 and a p95 of 3,000 is prose. The p95 also tells you whether chunking applies at all: if almost everything fits in one chunk, the chunker from chapter 30 is not needed for this source and a row is a chunk.

For Atlas, one column qualifies: tickets.resolution_notes, 41,000 closed tickets, p50 of 380 characters. It answers questions no document can — "has anyone seen this fault before?" — and the orders and inventory tables remain tools.

A row is not a document

The naive version embeds resolution_notes alone and retrieves fragments with no referent: "Replaced the seal and re-tested. Customer accepted." Which customer, which part, when?

The row becomes a document by assembling its context, exactly as chapter 15 serialised a spreadsheet row.

-- One query produces whole documents, with the joins that give the prose
-- its referent. Doing the joins here rather than per row avoids 41,000
-- round trips, which is the difference between four minutes and two hours.
select
  t.id,
  t.closed_at,
  t.account_id,                       -- ⚠️ the permission key. Chapter 13.
  t.subject,
  t.category,
  p.sku,
  p.name          as product_name,
  t.resolution_notes,
  t.updated_at
from tickets t
left join products p on p.id = t.product_id
where t.closed_at is not null
  and t.resolution_notes is not null
  and length(t.resolution_notes) >= 80          -- below this there is no content
  and t.updated_at > $1                         -- the watermark. Chapter 34.
order by t.updated_at, t.id                     -- stable, total ordering
limit $2;

// Assemble the document. The structure is blocks, as chapter 23 requires,
// so the chunker and the heading path work unchanged. The block contract is [chapter 23](/engineering/ai-agents-company-data/the-ingestion-pipeline)'s.
function ticketToDocument(row: TicketRow): ParsedDocument {
  return {
    sourceId: 'ticket_notes',
    sourceRef: String(row.id),
    title: `Ticket ${row.id}: ${row.subject}`,
    updatedAt: row.updated_at,
    contentType: 'text/plain',
    // Classification from the row, never from the text.
    sourceMetadata: { accountId: row.account_id, category: row.category },
    parseWarnings: [],
    blocks: [
      { type: 'heading', depth: 1, text: `Ticket ${row.id}: ${row.subject}` },
      {
        type: 'table',
        headers: ['Field', 'Value'],
        rows: [
          ['Closed', fmtDate(row.closed_at)],
          ['Category', row.category],
          ['Product', row.sku ? `${row.sku} — ${row.product_name}` : 'n/a'],
        ],
      },
      { type: 'heading', depth: 2, text: 'Resolution' },
      { type: 'paragraph', text: row.resolution_notes },
    ],
  }
}

That retrieves far better than the bare notes, for the reason chapter 9 gave: the question's words — a SKU, a category, a product name — now appear in the chunk.

Reading a consistent snapshot

The table changes while you read it. Over a 41,000-row ingest taking several minutes, that is guaranteed, and the two obvious approaches are both wrong.

The third is what you want, and "consistent enough" is precise rather than hand-waving: rows modified during the run are either caught by this run or by the next one, because the watermark only advances to a point you have definitely passed.

// Keyset pagination with a safety margin on the watermark.
export async function* discoverTickets(since: Date): AsyncIterable<DiscoveredDocument> {
  // ⚠️ Rewind the watermark. A row committed at T but with updated_at slightly
  // earlier — clock skew, or a long transaction — is invisible to a strict
  // `> since`. Re-reading a few minutes of rows is cheap; missing them is not.
  const from = new Date(+since - 5 * 60_000)

  let cursor = { updatedAt: from, id: 0 }
  for (;;) {
    const { rows } = await db.query(
      `select id, updated_at from tickets
        where closed_at is not null
          and resolution_notes is not null
          and (updated_at, id) > ($1, $2)
        order by updated_at, id
        limit 500`,
      [cursor.updatedAt, cursor.id],
    )
    if (rows.length === 0) return

    for (const row of rows) {
      yield { sourceId: 'ticket_notes', sourceRef: String(row.id),
              title: '', updatedAt: row.updated_at }
    }
    const last = rows[rows.length - 1]
    cursor = { updatedAt: last.updated_at, id: last.id }
  }
}

The row-value comparison (updated_at, id) > ($1, $2) is the part that makes this correct. Thousands of rows can share an updated_at after a bulk update, and paginating on the timestamp alone either loops forever on the same page or skips the rest of the group. The composite key is total.

-- And the index that makes it a range scan rather than a sort of 41,000 rows.
create index tickets_sync on tickets (updated_at, id)
  where closed_at is not null and resolution_notes is not null;

A partial index matching the query's own predicates is worth the thirty seconds: it is smaller, and it cannot be ignored by the planner in favour of a sequential scan when the predicate is selective.

Read from a replica, and say so

An ingest that reads 41,000 rows competes with the application. Point it at a read replica.

// Two pools, named for what they are. The ingest never touches primary.
export const replica = new Pool({ connectionString: secrets.replicaUrl, max: 4 })
export const primary = new Pool({ connectionString: secrets.databaseUrl, max: 10 })

Replica lag then becomes part of the freshness number from chapter 19, and it should be measured rather than assumed:

-- Lag, in seconds. Record it with every run; a growing number is a warning
-- long before it is an incident.
select extract(epoch from (now() - pg_last_xact_replay_timestamp())) as lag_seconds;

For an indexed source with a tolerance measured in days, a few seconds of lag is irrelevant. The reason to record it anyway is that it is the first thing you will want when a document that was definitely updated is definitely not in the index.

Deletes, and rows that stop qualifying

A document store tells you a file is gone. A database usually does not, and there are two ways a row leaves the corpus.

-- 1. The row is deleted. Reconcile ids rather than detecting events.
select d.source_ref
  from documents d
 where d.source = 'ticket_notes'
   and d.deleted_at is null
   and not exists (select 1 from tickets t where t.id::text = d.source_ref);

-- 2. The row still exists but no longer qualifies — reopened, or the notes
--    were cleared. Easy to miss, because the row is right there.
select d.source_ref
  from documents d
  join tickets t on t.id::text = d.source_ref
 where d.source = 'ticket_notes'
   and d.deleted_at is null
   and (t.closed_at is null or t.resolution_notes is null
        or length(t.resolution_notes) < 80);

The second query is the one people forget. A reopened ticket whose resolution was wrong keeps serving its old resolution notes as a cited answer, because nothing deleted anything. Chapter 35 handles the removal; recognising that qualification is a condition, not an event, belongs here.

Run both against the full id set periodically rather than per run — reconciliation is cheap relative to a full ingest and it is the only thing that catches these.

The failure mode: indexing the column that is a tool

The failure worth guarding against is applying this chapter too enthusiastically.

Someone notices that orders.special_instructions is free text, applies the three tests loosely, and indexes it. The column is prose, questions do mention it, and it is not obviously reachable by a query. It passes.

What goes wrong is the third test, which was applied to the wrong question. "What special instructions does Henderson usually give?" is a genuine retrieval question. But the question people actually ask is "what are the special instructions on Henderson's current order?" — and that has an exact answer that changes hourly, which makes it a tool, and the index answers it from last night's snapshot.

Both questions exist, and the right answer is both routes, with the indexed version scoped to make its nature obvious:

// Indexed: historical, aggregate, explicitly past tense.
{ sourceId: 'order_instructions_history',
  filter: 'orders completed more than 30 days ago',
  title: (row) => `Order ${row.id} (${fmtDate(row.completed_at)})` }

// Tool: anything current.
{ name: 'get_order_instructions', description: 'Special instructions on a ' +
  'specific current or recent order. Use for any question about an order ' +
  'that is not yet complete.' }

Excluding the recent window from the index is what makes the split safe. With it, the indexed copy cannot answer a question about a live order, because the live orders are not in it — and a boundary enforced by a where clause beats one enforced by a tool description the model may or may not follow.

Key points

Most of a database is a tool. Index only free-text columns, and only when they pass all three tests.
Look at the length distribution before deciding. A p50 of 14 characters is a status field people type into.
Assemble a row into a document with its joined context, as blocks. The bare column retrieves fragments with no referent.
Paginate on a composite (updated_at, id) keyset, with a partial index matching the predicates. Timestamps alone are not a total order.
Rewind the watermark a few minutes. Clock skew and long transactions hide rows from a strict comparison.
Read from a replica, and record replica lag with every run.
Reconcile deletions and, separately, rows that stopped qualifying. Qualification is a condition, not an event.
When a column serves both a historical question and a current one, index the historical window only and enforce the boundary in SQL.

Next: pulling from a third-party API — rate limits, pagination you do not control, and the backfill that takes four days.