Ship the Plan, Not the Script: Avro Pipelines That Replay

Here is a job every streaming platform eventually has. An internal topic carries a rich record; a partner, a public catalog, or another business unit gets a redacted projection of it. Somebody writes a script. The script knows which fields to drop and which to rename, and that knowledge lives nowhere else. Six months later the source schema gains a field, the script silently passes it through, and the first person to notice is on the other side of the boundary.

CoreModels takes a different position: the mapping is a first-class artifact. Every transform returns the plan it executed as JSON — reviewable, committable, replayable — and execution is deterministic, so the same plan over the same source produces the same bytes every run. This article builds an Avro pipeline on those two properties.

Three mapping kinds, applied to Avro

The engine surface is POST graph/transform/schema/map/{projectId} (and the identical engine behind the transform_schema MCP tool). Its mapping object takes one of three kinds:

• inferred (default) — label and type matching against a target hint. caseInsensitive defaults to true, which is what you want when camelCase Avro fields meet an upper-case or snake_case target: against a hint whose fields are BASKETID and GRANDTOTAL, the default produces three operations; flip it to false and only the record itself matches, leaving one. Every operation is stamped "origin": "Inferred".
• explicit — you author a mapping guide: autoMatchByMapsTo, fieldMappings (each {sourceElementIds, targetElementIds, transformName}), taxonomyDirectives, drops. This is the kind that suits Avro best, for a reason specific to the format — see below.
• ai — a server-side Claude proposer drafts the plan and the same validation gate accepts or rejects it, with at most one repair attempt. It needs a server-configured Anthropic key and Editor/Admin membership, and it sends schema content to the Anthropic API. Reserve it for the first draft of a gnarly mapping — then freeze the returned plan and never call the proposer again.

Whichever kind produced it, the plan is what you version. Strategies are for authoring; pipelines replay.

Why explicit fits Avro: x-maps-to

Avro carries no IRIs of its own, but the specification tolerates unknown attributes in schema JSON, and CoreModels reads four of them (x-sia-role, x-sia-priority, x-sia-instruction, and x-maps-to) on records and on fields. x-maps-to is the one that earns its keep: with autoMatchByMapsTo: true, source and target constructs that share an IRI are aligned even when their labels differ — so a mapping survives a rename on either side.

Here is the redaction job. CheckoutEvent.avsc, the internal record:

{
  "type": "record",
  "name": "CheckoutEvent",
  "namespace": "com.acme.checkout",
  "doc": "One completed checkout, as published to the payments topic.",
  "x-maps-to": { "acme": "https://data.acme.example/checkout-event" },
  "fields": [
    { "name": "basketId", "type": "string",
      "x-maps-to": { "acme": "https://data.acme.example/checkout-event/basket" } },
    { "name": "grandTotal", "type": "double",
      "x-maps-to": { "acme": "https://data.acme.example/checkout-event/total" } },
    { "name": "riskScore", "type": ["null", "double"] },
    { "name": "operatorNotes", "type": ["null", "string"] },
    { "name": "completedAt", "type": { "type": "long", "logicalType": "timestamp-millis" },
      "x-maps-to": { "acme": "https://data.acme.example/checkout-event/completed-at" } }
  ]
}

CheckoutEvent.public.avsc, the shape partners are promised — same record name, same IRIs, fewer fields:

{
  "type": "record",
  "name": "CheckoutEvent",
  "namespace": "com.acme.public",
  "x-maps-to": { "acme": "https://data.acme.example/checkout-event" },
  "fields": [
    { "name": "basketId", "type": "string",
      "x-maps-to": { "acme": "https://data.acme.example/checkout-event/basket" } },
    { "name": "grandTotal", "type": "double",
      "x-maps-to": { "acme": "https://data.acme.example/checkout-event/total" } },
    { "name": "completedAt", "type": { "type": "long", "logicalType": "timestamp-millis" },
      "x-maps-to": { "acme": "https://data.acme.example/checkout-event/completed-at" } }
  ]
}

Keep the record name the same on both sides when the target format is Avro. The encoder roots the output document on the record the source was rooted on; rename it in the mapping and you get The schema has no root type to encode as an Avro record.

And redact.guide.json:

{
  "autoMatchByMapsTo": true,
  "drops": ["CheckoutEvent.riskScore", "CheckoutEvent.operatorNotes"]
}

jq -n --rawfile s CheckoutEvent.avsc \
      --rawfile h CheckoutEvent.public.avsc \
      --rawfile g redact.guide.json '{
  sourceFormat: "avro", sourceSchema: $s,
  targetFormat: "avro",
  targetHintFormat: "avro", targetHintSchema: $h,
  mapping: { kind: "explicit", guide: $g }
}' \
| curl -sS -X POST \
    "https://coremodels.example.com/graph/transform/schema/map/$PROJECT_ID" \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    --data-binary @- > redact-result.json

The schema is the published record: basketId, grandTotal, completedAt, with the namespace, doc string, logical type, and every x-maps-to intact — and no riskScore, no operatorNotes. The plan is six operations: two Drops, a TypeMapping, and three ElementMappings, each stamped "origin": "Explicit". The drops carry their own reason inside the plan:

{ "kind": "Drop", "origin": "Explicit",
  "sourceNodeId": "CheckoutEvent.riskScore", "sourceKind": "Element",
  "reason": "Dropped by the SIA guide.",
  "declaredLossiness": [
    { "kind": "StructuralDrop", "path": "Element[CheckoutEvent.riskScore]",
      "explanation": "Dropped by the SIA guide." } ] }

A reviewer reading redact.plan.json in a pull request sees exactly what will happen to every field before it happens to any of them. Commit it next to the guide.

One counting note while you are here: the response ledger aggregates decode, plan, gate, execute, and encode lossiness, and the gate and the executor each carry the plan's declared drops — so a single dropped field can appear twice. Deduplicate on (kind, path, explanation) before you report a number.

Replay is the steady state

From here the pipeline never re-derives the mapping:

jq '.plan' redact-result.json > redact.plan.json

jq -n --rawfile s CheckoutEvent.avsc --rawfile p redact.plan.json '{
  sourceFormat: "avro", sourceSchema: $s,
  plan: $p,
  targetFormat: "avro"
}' \
| curl -sS -X POST \
    "https://coremodels.example.com/graph/transform/plan/execute/$PROJECT_ID" \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    --data-binary @-

The plan travels as a string — hence --rawfile. The hint is optional on replay, because the plan already names its targets, and the output is identical either way. The stored plan still passes the same universal gate on every run: it earns no shortcut for having been approved once.

That gate is what makes replay safe in CI. Rename basketId to basket_id upstream and the next replay fails loudly instead of quietly emitting a two-field record:

{ "path": "plan.Operations[3](ElementMapping).SourceElementIds",
  "message": "Source element 'CheckoutEvent.basketId' does not resolve." }

Operation index, operation kind, and the id that vanished. That is a drift alarm, delivered by the same call that does the work.

Batch conversion, and a gate on the ledger

Deterministic output means diffs are meaningful, which means a directory of registry schemas can be converted on every change and reviewed like code:

mkdir -p out
for f in schemas/*.avsc; do
  name=$(basename "$f" .avsc)
  jq -n --rawfile s "$f" '{
    sourceFormat: "avro", sourceSchema: $s,
    targetFormat: "jsonschema",
    targetHintFormat: "avro", targetHintSchema: $s,
    mapping: { kind: "inferred" }
  }' \
  | curl -sS -X POST \
      "https://coremodels.example.com/graph/transform/schema/map/$PROJECT_ID" \
      -H "Authorization: Bearer $TOKEN" \
      -H "Content-Type: application/json" \
      --data-binary @- > "out/$name.result.json"

  jq '.schema'    "out/$name.result.json" > "out/$name.schema.json"
  jq '.lossiness' "out/$name.result.json" > "out/$name.ledger.json"
done

Run that twice over the same inputs and the outputs are byte-identical, which is the property that makes git diff a real review surface. Then gate on the ledger rather than on success alone — success: true only means the transform ran. A refund schema whose amount is an Avro decimal produces this, every time:

[
  { "kind": "TypeApproximation", "path": "Element[Refund.amount]",
    "explanation": "Avro decimal precision/scale not modelled; approximated as Double." }
]

Money losing its scale is exactly the kind of thing a build should say out loud:

fail=0
for r in out/*.result.json; do
  bad=$(jq '[.lossiness[] | select(.kind == "StructuralDrop" or .kind == "TypeApproximation")] | length' "$r")
  if [ "$bad" -gt 0 ]; then
    echo "REVIEW: $r"
    jq -r '.lossiness[] | "  [\(.kind)] \(.path): \(.explanation)"' "$r"
    fail=1
  fi
done
exit $fail

For the redaction pipeline you would allow precisely the drops the committed plan declares and fail on anything else — the point being that the permitted loss is written down in a reviewed artifact, and everything beyond it stops the line.

The gate also reviews your guide

Three failures worth knowing before they happen in CI, each returned inside the normal envelope with a path:

• A typo in the guide is rejected, not ignored: guide.dropz: Unknown mapping-guide key 'dropz'. Known keys: autoMatchByMapsTo, fieldMappings, taxonomyDirectives, drops.
• A drop that does not name a real source construct is rejected: guide.Drops: Drop target 'CheckoutEvent.nosuchfield' is not a source construct.
• A multi-field fieldMappings entry (split/join) needs a registered transformName; leave it out and the guide is rejected — A multi-field mapping (split/join) must name a field transform from the registry. — and a name the registry does not know reaches the gate as Field transform '…' is not registered. The built-in vocabulary is SplitName, JoinName, CoerceStringToDateTime, CoerceDateTimeToString, CoerceIntegerToDouble, and CoerceDoubleToInteger. For plain 1:1 alignment, prefer autoMatchByMapsTo — that is what the annotations are for.

And one thing the engine does for you, on inferred mappings: a source field that the plan neither maps nor drops is declared rather than dropped in silence — StructuralDrop at Type[CheckoutEvent].CheckoutEvent.riskScore, "Element is a member of the mapped type but no operation maps or drops it." Run the redaction with inferred instead of a guide and you get the same published record, with the two omissions in the ledger instead of in the plan. Both are honest; the guide is what makes them intentional.

The shape of the whole pipeline

1. Author once. Derive the plan with inferred, explicit, or ai, whichever matches how well you can state the mapping. Read it: every heuristic is marked by origin, every loss by declaredLossiness.
2. Review and commit. Plan and guide go into the repo. Human judgement happens once, visibly, instead of on every run, invisibly.
3. Replay forever. CI calls plan/execute on each source change. The gate catches drift; determinism makes diffs trustworthy; the ledger keeps residual loss auditable.
4. Regenerate deliberately. When the source genuinely evolves, re-derive and diff the plan — a far smaller and far more honest object than a diff of generated schemas.

Conversions stop being something your pipeline does and become something your pipeline has. The transform API documentation specifies the request shapes for schema/map and plan/execute, the guide grammar, and the validation gate's contract.