Hierarchical unit load from sorted S3 dumps

Hierarchy owned by the import service

Which models are forest roots, which child attaches via which immediate parent, and in which order units load are import-service logic. Sorted dumps and this walk are only a streaming implementation of that forest — not a separate data model invented for the loader.

TL;DR. Annotate each Zod model with a primary key and optional immediate parent ref. Build SCHEMA_MAP once. Export one UTF-8 JSONL file per entity on S3, sorted by pk or parentFk, pk, so the loader can open concurrent streams (one cursor per entity). Walk ROOT_PASS_ORDER: for each root row, drain matching children via cursors that sit on the current row until advance(). No forward-skip. Enforce parent FK not_null in dbt before export. Resume broken S3 streams with byte-range GETs from the last fully parsed line.

Companion: units, buckets, and pillars (semantic unit of work), firm import service (stage-4 writer). Linear: DM-14.

Goal

Describe how to stream BigQuery dumps (sorted files on S3) into an existing system by inserting meaningful units: each root row plus its owning descendants, using one forward linear scan per table, and keeping control of the rate of insertion.

Data layout

Each import-service Zod model has a clear primary key and an optional immediate parent. Models with no parent are roots of the forest. A unit is one instance of that forest: a root row plus all its descendants under the immediate-parent edges.

When a model has more than one possible parent in the domain, the schema still records a single immediate parent. Which parent that is — and therefore which unit the row belongs to — is decided by the import service; other relationships are satisfied because the import service loads units in an order that makes prerequisites already present.

That annotation drives dump sort order:

if M has owning parent P:
  ORDER BY owning_fk, primary_key
else:  # root
  ORDER BY primary_key

Dump format on S3

Each entity model is exported as its own UTF-8 JSONL file (newline-delimited JSON: one JSON object per line). That layout is the source format the loader streams from — not a single bundled archive of all tables.

One file per entity — e.g. matters.jsonl, matter_notes.jsonl, contacts.jsonl. Each file is independently readable.
Sorted — roots by pk; children by parentFk, pk (see above).
Concurrent streams — because files are separate, the loader opens one S3 stream (cursor) per entity involved in the active unit pass (~≤30 at a time for a typical forest), and drains them in lockstep without multiplexing rows from different models in one stream.

Lines are UTF-8 JSON objects terminated by \n (prefer \n-only; see resume risk for \r\n).

Zod field annotations

Graph metadata lives on scalar fields via Zod’s .meta(...) (not nested parent objects):

Primary key — on the id field: .meta({ primaryKey: true })
Immediate parent — optional; on at most one FK field, ref is the parent Zod schema itself: .meta({ ref: ParentSchema }). No separate owning flag — if ref is present, that field is the immediate parent.

const FirmSchema = z.object({
  id: z.string().meta({ primaryKey: true }),
});

const ContactSchema = z.object({
  id: z.string().meta({ primaryKey: true }),
  firmId: z.string().meta({ ref: FirmSchema }),
});

Walk uses primaryKey + the optional ref schema for sort keys and forest membership.

Pre-computed schema map

At initialization the loader iterates over all models in the catalog to generate a runtime SCHEMA_MAP:

SCHEMA_MAP[Model] = { pk, parentFk, children }

pk — field name marked as primaryKey
parentFk — field name marked with ref pointing to an immediate parent, or absent for a unit root
children — array of { childModel, fk, mustHaveChildren } tuples from child schemas whose ref points at this model

The walk reads SCHEMA_MAP only — no per-row Zod .meta lookups. ROOT_PASS_ORDER is manually defined by the import service (load order for prerequisite unit types with no parentFk).

Pseudocode

// SCHEMA_MAP pre-parsed from Zod models at startup
// ROOT_PASS_ORDER manually defined by the import service
//
// Cursor contract:
//   getCursor(model) — existing cursor, or open S3 stream lazily and
//                      position on the first row (row = null if empty).
//                      Does not advance.
//   cursor.row       — current row (inspect without consuming)
//   cursor.advance() — move to the next row (or null at EOF)
//
// Why current-row (not next()+pushBack): a child stream may have zero
// rows for this parent. getCursor returns the same unconsumed row until
// we advance. Inspect → match → advance; if fk != parentPk, leave the
// row in place for the next parent. No skip, no pushBack.

function importForest():
  for each rootModel in ROOT_PASS_ORDER:
    cursor = getCursor(rootModel)

    while cursor.row != null:
      rootRow = cursor.row
      unit = scanAndBuildUnit(rootModel, rootRow, [])
      insertUnit(unit)
      cursor.advance()

  closeAllCursors()


function scanAndBuildUnit(model, parentRow, unit):
  unit.append({ model, row: parentRow })
  meta = SCHEMA_MAP[model]
  parentPk = parentRow[meta.pk]

  for each { childModel, fk, mustHaveChildren } in meta.children:
    cursor = getCursor(childModel)

    if mustHaveChildren and (cursor.row == null or cursor.row[fk] != parentPk):
      raise MissingChildError("Expected at least 1 child")

    while cursor.row != null and cursor.row[fk] == parentPk:
      childRow = cursor.row
      unit = scanAndBuildUnit(childModel, childRow, unit)
      cursor.advance()

  return unit

Example (one Firm and its descendants)

A migration run is scoped to one Firm. The Firm is the run input, not a row in a loop over multiple firms. Matter attaches to that Firm via firmId, so the walker starts once at the selected Firm and drains every downstream branch recursively.

# Run input: one selected firm
#   contacts ORDER BY firm_id, id
#   firm_members ORDER BY firm_id, id
#   matters ORDER BY firm_id, id
#   matter_notes ORDER BY matter_id, id
#   comments ORDER BY matter_id, id
firm = selectedFirm
drain contacts where firm_id == firm.id
drain members where firm_id == firm.id
drain matters where firm_id == firm.id
  for each matter:
    drain notes where matter_id == matter.id
      for each note: drain note-owned children...
    drain comments where matter_id == matter.id
    drain every other matter-owned branch...
insert firm unit

Invariants

Each cursor sits on a current row (first row on open). getCursor never advances; only advance() does. Zero-child parents leave the stream where it is for the next parent — no rewind, skip, or pushBack.
Walk edges = the single immediate parent ref only; unit membership and drain keys come from that.
Multi-parent domain models still pick one immediate parent (import-service decision); other relationships are handled by unit load order, not by this walk.
Immediate parent is optional (roots have none). If a model declares one, its S3 dump must not contain null parent ids — enforced by Zod→dbt not_null on that FK. The walk does not forward-skip orphans.

Code generation from Zod schemas

The same .meta({ primaryKey }) / .meta({ ref: ParentSchema }) annotations drive codegen (extend tooling/zod-to-dbt or entity-graph generators) so dumps match the walk contract.

dbt data tests (`schema.yml`)

Annotation	Generated tests
`primaryKey: true`	`not_null` + `unique` on that column
`ref: ParentSchema` (optional; only if present)	`not_null` on that FK column
no `ref` (unit root)	no parent-FK test

Every dump has a real PK. A model that declares an immediate parent must not have a null parent id in the S3 dump — enforced by dbt not_null before export. Roots have no parent column.

dbt dump models (`ORDER BY`)

-- root (no parentFk):
select * from {{ ref('mart_…') }}
order by <pk>

-- child (has parentFk):
select * from {{ ref('mart_…') }}
order by <parentFk>, <pk>

Export to S3 as UTF-8 JSONL in that order.

Zod entity schemas
  → SCHEMA_MAP { pk, parentFk, children }
  → dbt schema.yml tests (pk unique+not_null; parentFk not_null only when declared)
  → dbt dump models ORDER BY …
  → BigQuery/dbt run + export → S3 JSONL
  → hierarchical loader (this walk)

Cost (S3 access)

S3 charges for requests and data transfer, not for “open stream” duration.

Concurrent streams (~≤30 per root-type pass): one open GetObject per entity JSONL file in the active forest — no per-open fee. Client connection/memory limits matter more than S3 cost.
GET requests: one full-object GetObject per table dump per pass is typical; occasional range GETs on reconnect add a few requests. At ~$0.0004 / 1k GETs (region-dependent), even hundreds of GETs are fractions of a cent.
Data transfer: S3 bucket and loader run in the same region — transfer cost is free/negligible. No cross-region or public-internet egress for the dump read path.

Risks

Models with multiple parents

Risk: Many domain models have more than one real parent (e.g. a row that points at both a firm and a matter). The walk allows only one immediate parent ref. Choosing the wrong one puts the row in the wrong unit; ignoring the other parent can insert before that prerequisite exists.

Mitigation:

Pick one immediate parent on the Zod schema (.meta({ ref: ParentSchema })) — that field drives dump sort order, forest membership, and the linear drain.
Load non-immediate parents first — the import service’s ROOT_PASS_ORDER / unit load order must insert the other parent’s unit before this unit runs, so those relationships already exist when the row is written.
Multi-parent picks are import-service decisions, not inferred by the walker.

Skipping records

Risk: A linear hierarchical drain can silently drop rows if the cursor advances past data that does not match the current parent (orphans, mis-sorted dumps, or an eager next()/skip that consumes a later parent’s children). Skipped rows would never appear in any unit and would not be inserted.

Mitigation:

No forward-skip in the walk — cursors sit on a current row; getCursor does not advance. Non-matching rows stay in place for the next parent.
Advance only on consume — move past a row only after it is taken into a unit (or after a successful unit insert for roots).
Sorted dumps — Zod→dbt dump models ORDER BY pk or ORDER BY parentFk, pk so the next current row is the next candidate; no seeking.
dbt not_null on immediate parent FK (when declared) — null parent ids cannot sit on the cursor and block or be skipped; they fail before S3 export.
mustHaveChildren — fail fast when a mandatory child run is missing instead of continuing and leaving that subtree empty.
Idempotent inserts + byte-offset resume — reconnects re-read from the last committed line rather than restarting the file and risking inconsistent skip/duplicate behavior.

Resume stream after disconnection

Risk: Long-lived S3 GetObject streams (JSONL dumps, ~≤30 concurrent per root-type pass) can die on idle timeout, NAT/LB drop, or SDK abort while the loader is inserting a unit. Without a resume strategy, the forward-only cursor would have to restart that table from byte 0 and break the linear scan invariant (or force a full job restart).

Mitigation:

Dump format is UTF-8 JSONL (one JSON object per line). Prefer \n-only delimiters. If a dump uses \r\n and lines are split on \n only, the line string still ends with \r — JSON.parse accepts that (CR is JSON whitespace), and byteOffset += Buffer.byteLength(line, 'utf8') + 1 still counts both CR and LF. Do not use APIs that strip \r\n while only adding +1 for LF (e.g. naive use of Node readline), or offsets drift by one byte per line.
Per open cursor, checkpoint a byte offset only after each fully parsed line: byteOffset += Buffer.byteLength(line, 'utf8') + 1 (+ 1 = the \n that terminated the line; any trailing \r left on line is included in byteLength).
On disconnect: close the broken stream, reopen with S3 Range: bytes={offset}-, continue parsing.
Keep inserts idempotent (upsert / skip-if-exists) so a rare re-read of the last line cannot double-insert.
S3 and loader are same-region; request cost of occasional range GETs is negligible.

Goal

Data layout

Dump format on S3

Zod field annotations

Pre-computed schema map

Pseudocode

Example (one Firm and its descendants)

Invariants

Code generation from Zod schemas

dbt data tests (schema.yml)

dbt dump models (ORDER BY)

Cost (S3 access)

Risks

Models with multiple parents

Skipping records

Resume stream after disconnection

dbt data tests (`schema.yml`)

dbt dump models (`ORDER BY`)