# Apache Avro — Automation: 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. # 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: ```json { "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: ```json { "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`: ```json { "autoMatchByMapsTo": true, "drops": ["CheckoutEvent.riskScore", "CheckoutEvent.operatorNotes"] } ``` ```bash 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 `Drop`s, a `TypeMapping`, and three `ElementMapping`s, each stamped `"origin": "Explicit"`. The drops carry their own reason inside the plan: ```json { "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: ```bash 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: ```json { "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: ```bash 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: ```json [ { "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: ```bash 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.