Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Decode & Codec Learnings — prior-art hazards to design out

Hard-won lessons from a prior production streaming engine in this same domain — the system thalweg is being built to replace and hot-swap under. Each hazard here caused a real silver-lane outage or a sustained steady-state cost on that engine. They are recorded so the decode / codec / UDF crates bake the fix in as an invariant before they move from scaffold to implementation — not after the identical outage recurs on a thalweg-powered deploy.

Status. Design guidance, not shipped code. thalweg is pre-implementation; the codec/decode/UDF crates named below are scaffolds. Where a hazard is already latent in a thalweg scaffold (the same code lineage was carried forward), the file is named so it can be fixed in place. Capabilities that are intent rather than build are marked (planned). The sink/ack side of these lessons is already carried by how-a-sink-should-work.md and ADR-0038; this document is the wire → Arrow decode boundary counterpart, and cross-references rather than duplicates the sink side (§8).

The examples use neutral names (orders, events, bronze.example) per the repository’s no-sensitive-information rule. Nothing here is domain-specific.

0. The meta-lesson: reconcile shape at the boundary, and make the boundary inspectable

Nearly every incident below has one shape:

A value’s type or shape at a boundary (decode → transform, transform → sink, helper → destination column) was decided implicitly, and only became visible when a batch was rejected in production.

The feedback loop on the predecessor was: deploy → observe rejection storm → roll back → hotfix → repeat — one recovery train burned four sequential same-day hotfix PRs, each relocating the previous failure. The cheap fix in every case was the same: surface the boundary decision before it moves data. thalweg’s existing “loud, never silent” posture (how-a-sink-should-work §11) must extend back to the decode boundary.

Design rule: if a fix would have been prevented by an inspectable signal, the missing signal is the real defect — build the signal, not just the patch.

Two concrete surfaces every codec should expose (planned):

  • twg … describe-schema <descriptor> — a twg subcommand (ADR-0045) that prints the exact Arrow schema each source field will decode to, so a transform author writes col['field'] vs a JSON-extract with certainty before deploy. twg-type-map (ADR-0019, the describe-only leaf) already owns describe-vs-execute; this is its read side extended to the decode schema.
  • Startup INFO log of the resolved decode schema, once per descriptor at register time, so the shape is in the logs before the first record moves.

1. Decode shape must be deterministic and inspectable (Struct vs Utf8)

The hazard. A descriptor-driven decoder maps a sub-message to either a native Arrow Struct or a stringified Utf8-JSON, and which one a given field gets is opaque at authoring time (it depends on nesting depth, wrapper rules, map handling, and fallback caps). Both tails failed on the predecessor, on the same descriptor, in a single deploy:

  • Tail A — author expected Struct, got Utf8. Native access (party['note']) fails at query-plan time: “type Utf8 is not Struct, Map, or Null.” The node errors, downstream starves, everything routes to DLQ, the DLQ writer is overwhelmed — full lane outage.
  • Tail B — author expected Utf8, got Struct. Access succeeds, the batch carries a Struct, and the STRING-typed destination column rejects it at write: “expected LargeUtf8 but got Struct with N fields.” Worse than A — it fails at emit, after config validation passed and every row was enriched, so the blast radius is every record for the life of the deploy.

The real defect is not either mapping — it is that the mapping was invisible until a batch was rejected.

thalweg placement. twg-proto-decode (Arrow output stage, ADR-0046), twg-proto-schema (descriptor → schema), twg-type-map (target-type mapping). The decode-shape rules must be documented, stable per descriptor, and dumpable via §0’s describe-schema surface. twg-proto-decode/src/decode.rs today mirrors the predecessor’s typing pass “matching real behavior exactly” — including the Utf8 fallbacks — so it inherits this hazard verbatim until §0 lands.

2. Native container types, never a stringified JSON scalar

The hazard. Proto map<K,V> fields decoded to a single Utf8 JSON-object string ('{"a":{…},"b":{…}}'). Downstream could not index by key, lost the value type entirely (a map<string,double> became opaque STRING), and was forced back to re-parsing the raw payload N times per row (§7). The same applied to any container the decoder chose to stringify.

The fix. Emit native Arrow containers:

  • Proto map<K,V> → Arrow Map<Utf8, V> (value type preserved — that’s the win; the key is Utf8 because the JSON intermediate stringifies keys). Downstream then uses element_at(m, key) / map_keys(m) / map_values(m), and it matches the target’s native MAP<K,V> type in twg-type-map.
  • Repeated messages → List<Struct<…>>; singular sub-messages → Struct<…>.

A stringified container is acceptable only as an explicit, logged fallback (depth cap, §3), never as the default shape for a type the engine can model natively.

thalweg placement. twg-proto-decode (build the MapArray/ListArray), twg-type-map (proto/Arrow Map ↔ target MAP<K,V>; a map_fields.proto fixture already exists under twg-codec-protobuf). Latent hazard: twg-proto-decode/src/decode.rs stringifies map entries to Utf8 today.

Hot-swap note. Making a map column land as native Map instead of a JSON string is a behavior change for any config that writes that column straight into a STRING destination — it would then hit Tail B (§1). The escape hatch is §4’s to_json. Sequence the swap so both are available together.

3. The nesting-depth cap is a configured, observable knob — not a hidden constant

The hazard. Struct recursion was bounded by a hard-coded constant (= 12 on the predecessor). A schema deeper than the cap silently collapsed its whole sub-tree to Utf8, with no telemetry beyond a generic null-patch counter firing at write time — far from the cause. Operators could not raise it without a source patch and a redeploy.

The fix.

  1. Expose the cap as config — twg-config (ADR-0028 TOML), overridable per deployment; default generously (well above realistic schema depth) so the fallback is genuinely exceptional.
  2. Name the loss at the point it happens — a startup WARN listing every message that will trip the fallback under the current cap, and a per-type (deduped) WARN when it fires, so it is visible before the pipeline moves data. This is the decode-side face of how-a-sink-should-work §11 (“loud, not silent”).

Rule that generalises: any hard-coded limit that changes fidelity — a depth cap, a byte budget, a truncation threshold — is a config knob and emits a signal when it bites. A magic constant that silently drops type information is an incident waiting for the one schema that exceeds it.

thalweg placement. twg-proto-decode (the cap), twg-config (the knob), twg-observability (the WARN + a counter).

4. Symmetric serialise escape hatch: to_json alongside the JSON reader

The hazard. The predecessor shipped a reader for the stringified case (get_json_object(payload, '$.path')) but no writer — no way to serialise a native Struct/List/Map back to a string at emit time when the destination column is STRING (Tail B, §1). The only author-side unblock was CAST(NULL AS VARCHAR) — data loss disguised as a hotfix.

The fix. Register a to_json(<any>) -> STRING UDF as a first-class family member of get_json_object. If a boundary can be crossed one way, an author will need to cross it the other way.

Implementation notes that carry over:

  • Signature::any(1, Immutable) — accept any Arrow input without coercion (coercion cannot handle Struct/List/Map).
  • Delegate to the ecosystem JSON writer (arrow-json make_encoder) rather than a hand-walked value tree — full type coverage for free, no gaps on exotic types.
  • SQL NULL round-trips as SQL NULL, not the string "null".
  • Dependency pin: the arrow-json version must track DataFusion’s re-exported Arrow major (ADR-0003), or the FieldRef/dyn Array instances mismatch and it won’t compile.

End-state (planned, defer with a trigger): sink-side auto-projection — the sink already knows the destination’s declared type (twg-type-map), so nested-into-STRING can be reconciled transparently at emit. It touches the write hot path, retry-slice, and byte-budget accounting, so ship the UDF now and gate the auto-projection on a real trigger (e.g. “> N pipelines use to_json in a quarter” or “the same mismatch recurs on a second sink type”). Record the deferral with its trigger — a deferred item without a trigger is a wish, not a plan.

thalweg placement. twg-udf. Latent hazard: twg-udf registers the JSON reader but has no to_json — the escape hatch is absent today.

5. Type-producing helpers must not hide their output type

The hazard. A convenience UDF converting epoch-millis → timestamp was hard-coded to return a timezone-aware type (Timestamp(µs, Some("UTC"))). The tag is invisible at the SQL call site and propagates to everything derived from it (date_trunc('day', …)). Destination columns were timezone-naive (TIMESTAMP_NTZ, Timestamp(µs, None)), and the write layer rejects tz-aware-into-tz-naive:

Field 'event_ts_day' is incompatible - expected
  Timestamp(Microsecond, None) but got Timestamp(Microsecond, Some("UTC"))

A sibling pipeline that used the engine’s built-in millis→timestamp (tz=None) wrote fine — the two helpers looked interchangeable at the call site; the tag was the only difference, and it was hidden.

The fix.

  • Prefer the query engine’s built-in conversion (tz-naive) for tz-naive targets.
  • If a Spark-compat helper is provided, either return tz=None by default, or offer both variants with names that state the tag (…_ntz vs the UTC form), and doc-guard the tz-aware one: “timezone-aware — do NOT use for TIMESTAMP_NTZ destinations.”
  • General rule: a type-producing helper makes its output type — tz tag, decimal precision/scale, nullability — obvious at the call site. Two helpers that differ only in a hidden type attribute guarantee that one of them causes an outage.

thalweg placement. twg-udf (the helpers), twg-type-map (tz-aware vs tz-naive target mapping). Latent hazard: twg-udf/src/local/spark_compat.rs hard-codes with_timezone("UTC") and a Some("UTC") return type for both millis_to_ts and to_timestamp_ms, with no tz-naive variant — the exact footgun, already carried forward.

6. Preserve wire binary as Arrow Binary end-to-end

The hazard. On a JSON-intermediate decode path, Avro bytes fields (and bytes + logicalType: decimal) were base64-encoded into Utf8 strings ("AA==") and stayed strings through expand and flatten. The Arrow Binary type was lost, so a downstream decode_avro_decimal UDF — which requires Binary — failed, and any future raw-bytes (non-decimal) use case was blocked. JSON has no binary type, so a JSON hop forces base64; the fault is routing binary through JSON at all.

The fix.

  • Keep an Arrow-native decode path for binary. Decode Avro bytes/fixed directly to Arrow Binary/LargeBinary (and decimal logicalType to Decimal128), and preserve that type through expand and flatten rather than round-tripping through a JSON Value. thalweg’s three-co-equal-representations model (ADR-0001) makes this the natural path — Arrow is not a second-class hop.
  • Where a JSON view is genuinely needed, base64 is the correct JSON representation for that view only — it must not become the column’s Arrow type.
  • Make the consumer tolerant too: decode_avro_decimal should accept a base64-Utf8 input in addition to Binary/FixedSizeBinary, as a defensive fallback for any residual JSON-path value.
  • Alternatively/additionally, thread the Avro logicalType into decode so decimal lands as Decimal128 in the bronze layer and no post-hoc UDF is needed.

thalweg placement. twg-codec-avro (Arrow-native bytes/decimal decode; ADR-0027 shares this for the OCF batch source), twg-proto-flatten / expand-columns (preserve Binary, don’t base64 nested bytes), twg-udf (decode_avro_decimal input tolerance). Latent hazard: twg-udf/src/local/avro_decimal.rs rejects anything but Binary/FixedSizeBinary today, and any JSON-intermediate expand will hand it a base64 string.

7. Native typed columns beat re-parsing a blob N times per row

When decode leaves values stringified (§1, §2, §6), transforms fall back to get_json_object(raw_payload, '$.…') — once per referenced path, per row. The query engine does not memoise across different paths, so a payload referenced 20 times per row is parsed 20 times per row (tens of thousands of redundant parses per second per pod on the predecessor), and it forces the raw blob to be retained through the transform graph purely as a re-parse source.

This is the steady-state tax of getting §1/§2/§6 wrong. Every field decoded as a native typed column is one the transform reads for free and the sink writes without a round-trip through string. Re-parse count per row is a good proxy metric for how much decode fidelity is being left on the floor — worth a twg-observability counter.

8. Streaming-sink concurrency & pipelined ack (cross-reference)

The predecessor’s largest single throughput loss was on the write side, and thalweg’s sink design already addresses it — this section only pins the failure signatures so they are recognisable, and points at the existing invariants.

Observed on the predecessor:

  • Per-batch ack serialisation. Ingest awaited “wait until the server durably acked offset N” inline after every submit, serialising the sink to one in-flight batch — a ~20,000× loss against a pipelined SDK (~4 batches/sec against a 100 MB/s pipe). It looked like everything else: ack-timeout false-positives, backpressure trips at trivial throughput, batchers that never fill a batch.
  • Stream churn. Opening a new sink stream per batch/worker tripped a “too many concurrent streams” limit (thousands of rejects/day); recovery felt slow because most flushes were single-record commits.

thalweg’s design already prevents these:

  • Async, non-blocking ack releasing admission budget on arrival, off the clean path — how-a-sink-should-work §10, ADR-0038 (advance offset only on durable ack), ADR-0060 (pipelined fire-and-forget ack on a background lane).
  • Advance on the delivered offset, never the submitted offset — the durable watermark is what recovery reads (ADR-0038, ADR-0053 “coverage may lag, never lead”).
  • Few, long-lived streams per destination and honour server backoff on a concurrency reject (send on an existing stream; back off before reopening) — a self-protective path must not amplify the condition it detects.
  • Size streams to bandwidth, not record count — large records hit a per-stream MB/s ceiling at low record rates; spread across a small fixed pool.

No new work here beyond keeping these signatures in the sink soak tests (docs/testing/SCENARIOS.md).

9. Decode/codec invariants (summary checklist)

A codec/decode/UDF crate built to this blueprint should satisfy:

  • The Arrow type of every decoded field is inspectable before deploy (describe-schema surface) and logged at register time. (§0, §1)
  • Modellable containers decode to native Arrow (Map/List/Struct), never a stringified JSON scalar; stringification is an explicit, logged fallback only. (§2)
  • The nesting-depth cap is a config knob, defaulted high, and its fallback names the offending message loudly, at decode time not write time. (§3)
  • Every JSON extractor has a matching serialiser (to_json(<any>)); no author is ever forced to CAST(NULL …) to unblock a STRING target. (§4)
  • Type-producing helpers make their output type obvious at the call site (tz tag, precision, nullability); tz-naive is available and named. (§5)
  • Wire binary stays Arrow Binary through decode/expand/flatten; base64 is a JSON-view detail, never the column’s type; binary consumers tolerate the base64 fallback. (§6)
  • Native typed columns are preferred over repeated get_json_object re-parse; re-parse pressure is observable. (§7)
  • Sink write path keeps the async non-blocking ack, few long-lived streams, delivered-offset watermark, and server-backoff honouring. (§8, ADR-0038)

Source grounding

Derived from production incidents on the predecessor engine thalweg replaces, abstracted to remove all environment- and domain-specific detail per the repository’s no-sensitive-information rule. Mapped onto thalweg’s current design artefacts — chiefly how-a-sink-should-work.md, AGENTS.md, and the doc-comment scaffolds of twg-proto-decode, twg-proto-flatten, twg-proto-schema, twg-codec-avro, twg-udf, and the implemented twg-type-map. Relevant ADRs: 0001 (three representations), 0003 (Arrow version pin), 0004 (unnest/offset cardinality), 0017 (unified DLQ), 0019 (type-map describe-only), 0027 (shared Avro decode), 0038 (source-ack invariant), 0045 (single twg binary/subcommands), 0046 (protobuf decode strategy), 0053 (coverage may lag, never lead). As the codec crates move from scaffold to implementation, retire the (planned) markers and fold each fixed hazard into the relevant crate README or a dedicated ADR.