From 1b87fd2df1e0e615056999c27bd06eee6524ae60 Mon Sep 17 00:00:00 2001 From: "turbolytics.io" Date: Sat, 12 Sep 2026 21:26:34 -0400 Subject: [PATCH] Specs / ideas --- .../plans/2026-09-12-sink-encode-failure.md | 577 +++++++++++++++++ .../2026-09-12-run-enforces-schema-design.md | 172 ++++++ .../2026-09-12-schema-registry-mvp-design.md | 581 ++++++++++++++++++ .../2026-09-12-sink-encode-failure-design.md | 212 +++++++ 4 files changed, 1542 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-12-sink-encode-failure.md create mode 100644 docs/superpowers/specs/2026-09-12-run-enforces-schema-design.md create mode 100644 docs/superpowers/specs/2026-09-12-schema-registry-mvp-design.md create mode 100644 docs/superpowers/specs/2026-09-12-sink-encode-failure-design.md diff --git a/docs/superpowers/plans/2026-09-12-sink-encode-failure.md b/docs/superpowers/plans/2026-09-12-sink-encode-failure.md new file mode 100644 index 0000000..c0abc19 --- /dev/null +++ b/docs/superpowers/plans/2026-09-12-sink-encode-failure.md @@ -0,0 +1,577 @@ +# Sink encode failure is permanent Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** A value the sink's driver cannot encode fails once, carries a `user` code, and exits 10, instead of being retried and reported as `system.sink.unreachable`. + +**Architecture:** Three layers, each a task. The error registry gains `user.sink.encode_failed`. The retry ladder in `internal/sinks/retry.go` stops deciding retryability by listing codes and decides by class: any `user` error and `system.sink.write_failed` are permanent, everything else is retried. The ClickHouse sink wraps the driver's `Append` error with the new code. A fourth task writes the retry rule into the README and the invariant manifest. + +**Tech Stack:** Go 1.2x, `internal/errs` coded errors, `github.com/zeebo/assert`, `github.com/ClickHouse/clickhouse-go/v2` v2.48.0 (`lib/driver.Batch` is an interface, so the sink test needs no server). + +**Spec:** `docs/superpowers/specs/2026-09-12-sink-encode-failure-design.md` + +## Global Constraints + +- The error registry is append-only. `TestErrorTaxonomy_RegistryIsAppendOnly` compares `internal/errs/registry.go` to `internal/errs/testdata/codes.golden`. Add, never rename or remove. +- An uncoded error stays retryable. A driver's timeout or reset arrives without a code, and that is what the ladder exists for. +- Every test calls `coverage.Covers(t, "")` first, with an id the coverage registry already knows. Use `sink.retry`, `sink.clickhouse` and `error.taxonomy`. Do not invent an id. +- Prose follows Google Technical Writing One, per the repo's `CLAUDE.md`: active voice, one idea per sentence, no hedging. Code comments explain why, not what. +- Commit messages name the defect, the fix, and the evidence, and say what breaks if the change is wrong. No attribution trailers. +- Run tests with `go test ./internal// -run '' -v`. The Go build cache under `~/Library/Caches/go-build` is outside the sandbox; if a build fails with `operation not permitted` on that path, rerun outside the sandbox. + +--- + +### Task 1: Register `user.sink.encode_failed` + +**Files:** +- Modify: `internal/errs/registry.go:44-45` (constants) and `internal/errs/registry.go:142-146` (definitions) +- Modify: `internal/errs/testdata/codes.golden` +- Test: `internal/errs/errs_test.go` + +**Interfaces:** +- Produces: `errs.CodeSinkEncodeFailed errs.Code = "user.sink.encode_failed"`. Tasks 2 and 3 reference it by that name. + +- [ ] **Step 1: Write the failing test** + +Append to `internal/errs/errs_test.go`: + +```go +// A value the sink's driver refuses is the user's to fix: the value is in the +// topic and a restart re-reads it. The exit code has to say terminal, or a +// supervisor loops on it. +func TestErrorTaxonomy_EncodeFailedIsAUserErrorThatExitsTerminal(t *testing.T) { + coverage.Covers(t, "error.taxonomy") + err := New(CodeSinkEncodeFailed, "dt_plain parsing time") + + assert.Equal(t, ClassUser, ClassOf(err)) + assert.Equal(t, "sink", CodeSinkEncodeFailed.Domain()) + assert.Equal(t, ExitUserError, ExitCode(err)) + assert.False(t, Retryable(ExitCode(err))) + + def, ok := Lookup(CodeSinkEncodeFailed) + assert.True(t, ok) + assert.True(t, strings.Contains(def.Action, "handler SQL")) +} +``` + +- [ ] **Step 2: Run the test to verify it fails** + +Run: `go test ./internal/errs/ -run 'TestErrorTaxonomy_EncodeFailedIsAUserErrorThatExitsTerminal' -v` +Expected: build failure, `undefined: CodeSinkEncodeFailed`. + +- [ ] **Step 3: Add the constant and the definition** + +In `internal/errs/registry.go`, after the `CodeSinkTypeUnsupported` constant (line 45): + +```go + // A value the sink's client could not encode for the destination column. + // The value fails identically on every attempt, so the retry ladder does + // not retry it (#233). + CodeSinkEncodeFailed Code = "user.sink.encode_failed" +``` + +In the `registry` map, after the `CodeSinkTypeUnsupported` entry (line 146): + +```go + CodeSinkEncodeFailed: { + CodeSinkEncodeFailed, + "The sink's client could not encode a result value for the destination column. It fails the same way on every attempt and is not retried.", + "Cast or format the column in the handler SQL to match the destination column's type. The message names the column and the value.", + }, +``` + +- [ ] **Step 4: Regenerate the golden file** + +Run: `UPDATE_GOLDEN=1 go test ./internal/errs/ -run 'TestErrorTaxonomy_RegistryIsAppendOnly' -v` +Expected: `golden updated`. Then `git diff internal/errs/testdata/codes.golden` shows exactly one added line, `user.sink.encode_failed`, between `user.sink.invalid` and `user.sink.type_unsupported`. If any line was removed, stop: the registry lost a code. + +- [ ] **Step 5: Run the whole package** + +Run: `go test ./internal/errs/ -v` +Expected: PASS. `TestErrorTaxonomy_ExitCodeMapsEveryCode` and `TestErrorTaxonomy_EveryCodeIsWellFormed` cover the new entry without changes. + +- [ ] **Step 6: Commit** + +```bash +git add internal/errs/registry.go internal/errs/testdata/codes.golden internal/errs/errs_test.go +git commit -m "errs: a code for a value the sink cannot encode + +A ClickHouse value the driver refuses in Append was returned uncoded, so +the retry ladder retried it and reported system.sink.unreachable (#233). +user.sink.encode_failed names the fault: a payload value the user's SQL +shaped, terminal, exit 10. + +type_unsupported stays for a column whose Arrow type the sink cannot +convert. That is a schema property and fails for every row. encode_failed +is one value in a supported column. An operator reading type unsupported +for a badly formatted string looks at the wrong thing. + +The registry is append-only; codes.golden gains one line. + +What breaks if this is wrong: a supervisor restarts a pipeline into the +same bad value forever, because the exit code said retryable." +``` + +--- + +### Task 2: The retry ladder decides by class + +**Files:** +- Modify: `internal/sinks/retry.go:161-175` (`retryable` and its comment) +- Test: `internal/sinks/retry_test.go` + +**Interfaces:** +- Consumes: `errs.CodeSinkEncodeFailed` from Task 1. `errs.ClassOf(err) errs.Class` and `errs.ClassUser` already exist in `internal/errs`. +- Produces: `retryable(err error) bool` keeps its signature. Task 3 relies on the rule: any `user` code makes one attempt. + +- [ ] **Step 1: Write the failing tests** + +Append to `internal/sinks/retry_test.go`. The helpers `flakySink`, `newTestRetry` and `testPolicy` are defined at the top of that file. + +```go +// A column type the sink cannot convert fails identically every attempt. +// Before the class rule, this code was not on retryable's list, so the ladder +// re-encoded the same batch four times and reported the destination as +// unreachable (#233). +func TestSinkRetry_DoesNotRetryAnUnsupportedType(t *testing.T) { + coverage.Covers(t, "sink.retry") + inner := &flakySink{failures: 99, err: errs.New(errs.CodeSinkTypeUnsupported, "time64 is not supported")} + r, slept := newTestRetry(inner, testPolicy()) + + err := r.Flush(context.Background()) + + assert.Error(t, err) + assert.Equal(t, 1, inner.attempts) + assert.Equal(t, 0, len(*slept)) + assert.Equal(t, errs.CodeSinkTypeUnsupported, errs.CodeOf(err)) +} + +// A value the driver refused while building the batch never reached the +// destination, and sending it again changes nothing. +func TestSinkRetry_DoesNotRetryAnEncodeFailure(t *testing.T) { + coverage.Covers(t, "sink.retry") + inner := &flakySink{failures: 99, err: errs.New(errs.CodeSinkEncodeFailed, "dt_plain parsing time")} + r, slept := newTestRetry(inner, testPolicy()) + + err := r.Flush(context.Background()) + + assert.Error(t, err) + assert.Equal(t, 1, inner.attempts) + assert.Equal(t, 0, len(*slept)) + assert.Equal(t, errs.CodeSinkEncodeFailed, errs.CodeOf(err)) +} + +// The rule is the class, not a list. Every user code the registry holds +// makes exactly one attempt, including ones added after this test. +func TestSinkRetry_NoUserCodeIsRetried(t *testing.T) { + coverage.Covers(t, "sink.retry") + for _, d := range errs.All() { + if !d.Code.IsUser() { + continue + } + inner := &flakySink{failures: 99, err: errs.New(d.Code, "x")} + r, _ := newTestRetry(inner, testPolicy()) + + err := r.Flush(context.Background()) + + assert.Error(t, err) + if inner.attempts != 1 { + t.Errorf("%s: made %d attempts on a user error", d.Code, inner.attempts) + } + assert.Equal(t, d.Code, errs.CodeOf(err)) + } +} + +// An uncoded error that never clears runs the whole ladder and ends as +// unreachable. Pinned so a tidy-up of the class rule cannot turn a driver +// timeout nobody classified into a terminal failure. +func TestSinkRetry_UncodedErrorRunsTheWholeLadder(t *testing.T) { + coverage.Covers(t, "sink.retry") + inner := &flakySink{failures: 99, err: errors.New("i/o timeout")} + p := testPolicy() + r, _ := newTestRetry(inner, p) + + err := r.Flush(context.Background()) + + assert.Error(t, err) + assert.Equal(t, p.MaxAttempts, inner.attempts) + assert.Equal(t, errs.CodeSinkUnreachable, errs.CodeOf(err)) +} +``` + +- [ ] **Step 2: Run the tests to verify the first three fail** + +Run: `go test ./internal/sinks/ -run 'TestSinkRetry_(DoesNotRetryAnUnsupportedType|DoesNotRetryAnEncodeFailure|NoUserCodeIsRetried|UncodedErrorRunsTheWholeLadder)' -v` +Expected: `DoesNotRetryAnUnsupportedType` fails with attempts 5, not 1, and code `system.sink.unreachable`. `DoesNotRetryAnEncodeFailure` fails the same way. `NoUserCodeIsRetried` reports `user.sink.type_unsupported` and `user.sink.encode_failed` among others made 5 attempts. `UncodedErrorRunsTheWholeLadder` passes already. + +- [ ] **Step 3: Replace `retryable`** + +Replace lines 161-175 of `internal/sinks/retry.go` with: + +```go +// retryable reports whether another attempt could plausibly succeed. +// +// The rule is by class, not by a list of codes. A list retries whatever it +// did not anticipate: user.sink.type_unsupported was never on it, so a column +// the sink could not convert ran the whole ladder and was reported as +// unreachable (#233). +// +// Class user never retried the config, SQL or data is wrong, +// and fails identically every time +// system.sink.write_failed never retried the destination answered and refused +// system.sink.unreachable retried the destination may come back +// uncoded retried a driver's timeout or reset arrives +// unclassified; the deadline bounds +// the cost of guessing wrong +// any other system code retried +// +// README "Sink retries" states the same table for operators. Change both. +func retryable(err error) bool { + if errs.ClassOf(err) == errs.ClassUser { + return false + } + return !errs.HasCode(err, errs.CodeSinkWriteFailed) +} +``` + +`errs.CodeOf` returns `system.internal.unexpected` for an uncoded error, so the uncoded case falls through to `true` without a special branch. + +- [ ] **Step 4: Run the retry suites** + +Run: `go test ./internal/sinks/ -run 'TestSinkRetry_' -v` +Expected: PASS, including the four new tests and the existing `DoesNotRetryARejectedWrite`, `DoesNotRetryAConfigError`, `BoundsNonRetryableStopsUnderAnyPolicy` and `RetriesAnUncodedError`. + +- [ ] **Step 5: Run the sinks package** + +Run: `go test ./internal/sinks/` +Expected: PASS. Live-integration tests skip when no ClickHouse or Kafka is running; a skip is fine, a FAIL is not. + +- [ ] **Step 6: Commit** + +```bash +git add internal/sinks/retry.go internal/sinks/retry_test.go +git commit -m "sinks: the retry ladder decides by error class, not by a list + +retryable listed three codes it would not retry and retried everything +else. user.sink.type_unsupported was not on the list, so a column the +ClickHouse sink could not convert was re-encoded four times and then +reported as system.sink.unreachable. Nothing was unreachable, and the +exit code told a supervisor to restart into the same failure (#233). + +Any user-class error now makes one attempt, and so does write_failed. +Unreachable and uncoded errors are retried as before; a driver timeout +arrives without a code and is exactly what the ladder exists for. + +TestSinkRetry_NoUserCodeIsRetried walks the registry, so a user code +added later cannot fall back into the ladder. UncodedErrorRunsTheWholeLadder +pins the other direction. + +What breaks if this is wrong: a transient fault some sink coded as user +stops being retried. No sink does that today, and the tests would show it." +``` + +--- + +### Task 3: The ClickHouse sink codes its encode failures + +**Files:** +- Modify: `internal/sinks/clickhouse.go:255-303` (`appendTables`) +- Test: `internal/sinks/clickhouse_test.go` + +**Interfaces:** +- Consumes: `errs.CodeSinkEncodeFailed` from Task 1; `appendTables(batch driver.Batch, types []column.Type, tables []arrow.Table) error` and `temporalFromString(colType column.Type, s string) (time.Time, bool)`, both already in `clickhouse.go`. +- Produces: nothing new. The `append row` error now carries `user.sink.encode_failed`. + +- [ ] **Step 1: Write the failing tests** + +Append to `internal/sinks/clickhouse_test.go`. The file already imports `context`, `fmt`, `testing`, `time`, `arrow`, `array`, `memory`, `config`, `coverage`, `errs` and `assert`. Add `"errors"` and `"github.com/ClickHouse/clickhouse-go/v2/lib/column"` and `"github.com/ClickHouse/clickhouse-go/v2/lib/driver"` to the import block. + +```go +// stubBatch is a driver.Batch that refuses every Append. The driver's real +// batch does the same thing in memory and touches no connection until Send, +// so a fake is a faithful stand-in for the failure this tests. +type stubBatch struct { + appendErr error + aborted bool +} + +func (b *stubBatch) Abort() error { b.aborted = true; return nil } +func (b *stubBatch) Append(v ...any) error { return b.appendErr } +func (b *stubBatch) AppendStruct(v any) error { return b.appendErr } +func (b *stubBatch) Column(int) driver.BatchColumn { return nil } +func (b *stubBatch) Flush() error { return nil } +func (b *stubBatch) Send() error { return nil } +func (b *stubBatch) IsSent() bool { return false } +func (b *stubBatch) Rows() int { return 0 } +func (b *stubBatch) Columns() []column.Interface { return nil } +func (b *stubBatch) Close() error { return nil } + +// A value the driver refuses while building the batch is a user fault that +// fails the same way every attempt. Returned uncoded, the retry ladder +// re-encoded it four times and reported the destination unreachable (#233). +func TestSinkClickhouse_ADriverRefusedValueCarriesEncodeFailed(t *testing.T) { + coverage.Covers(t, "sink.clickhouse") + + driverErr := errors.New(`clickhouse [AppendRow]: dt_plain parsing time "2026-09-01T12:00:00Z" as "2006-01-02 15:04:05": cannot parse "T12:00:00Z" as " "`) + batch := &stubBatch{appendErr: driverErr} + + schema := arrow.NewSchema([]arrow.Field{{Name: "dt_plain", Type: arrow.BinaryTypes.String}}, nil) + b := array.NewRecordBuilder(memory.NewGoAllocator(), schema) + defer b.Release() + b.Field(0).(*array.StringBuilder).Append("2026-09-01T12:00:00Z") + rec := b.NewRecord() + defer rec.Release() + table := array.NewTableFromRecords(schema, []arrow.Record{rec}) + defer table.Release() + + err := appendTables(batch, []column.Type{"DateTime"}, []arrow.Table{table}) + + assert.Error(t, err) + if !errs.HasCode(err, errs.CodeSinkEncodeFailed) { + t.Fatalf("code = %s, want %s", errs.CodeOf(err), errs.CodeSinkEncodeFailed) + } + // The driver names the column and quotes the value. That text is the + // operator's only pointer to the bad row and has to survive the wrap. + assert.True(t, errors.Is(err, driverErr)) +} + +// The value from #233 misses every layout the sink accepts, so it reaches the +// driver unchanged. This documents the precondition for the encode path; it +// is not a claim the value should fail. Accepting RFC 3339 is a separate +// type-matrix change. +func TestSinkClickhouse_ISO8601WithTAndZReachesTheDriver(t *testing.T) { + coverage.Covers(t, "sink.clickhouse") + _, ok := temporalFromString(column.Type("DateTime"), "2026-09-01T12:00:00Z") + assert.False(t, ok) +} +``` + +- [ ] **Step 2: Run the tests to verify the first fails** + +Run: `go test ./internal/sinks/ -run 'TestSinkClickhouse_(ADriverRefusedValueCarriesEncodeFailed|ISO8601WithTAndZReachesTheDriver)' -v` +Expected: `ADriverRefusedValueCarriesEncodeFailed` fails with `code = system.internal.unexpected, want user.sink.encode_failed`. `ISO8601WithTAndZReachesTheDriver` passes. + +The table construction follows `clickhouseFixtureTable` at `clickhouse_test.go:326`, which uses the same `NewRecordBuilder` and `NewTableFromRecords` calls. + +- [ ] **Step 3: Wrap the driver's error** + +In `internal/sinks/clickhouse.go` `appendTables`, replace: + +```go + if err := batch.Append(row...); err != nil { + reader.Release() + return fmt.Errorf("clickhouse sink: append row: %w", err) + } +``` + +with: + +```go + if err := batch.Append(row...); err != nil { + reader.Release() + // Append validates and buffers in memory; the driver sends + // nothing until Send. A value it refuses here fails the + // same way on every attempt, so this is coded permanent + // rather than left for the ladder to guess at (#233). + return errs.Wrap(errs.CodeSinkEncodeFailed, err, "clickhouse sink: encode row") + } +``` + +`errs` is already imported in `clickhouse.go`. The `arrowValue` branch above it keeps `user.sink.type_unsupported` and its `fmt.Errorf` wrap, because `fmt.Errorf` with `%w` preserves the inner code and `errs.CodeOf` walks the chain. + +- [ ] **Step 4: Run the ClickHouse unit tests** + +Run: `go test ./internal/sinks/ -run 'TestSinkClickhouse_' -v` +Expected: PASS for the unit tests. The live tests (`InsertsRows`, `InsertsArrays`, `StringTemporalsAreNotShiftedByHostZone` and others that call `newLiveClickhouseSink`) skip with `clickhouse unavailable` unless a ClickHouse is up. + +- [ ] **Step 5: Run the end-to-end check against a live ClickHouse, if one is available** + +Start the dev stack: `make start-backing-services`. If it fails on port 5432, ignore it; ClickHouse and Kafka still come up. Then run the live sink tests: + +Run: `go test ./internal/sinks/ -run 'TestSinkClickhouse_' -v 2>&1 | grep -E '^(--- |ok|FAIL)'` +Expected: every `TestSinkClickhouse_` test reports PASS, none SKIP. If they still skip, note it in the commit message as "live ClickHouse tests not run" and continue. Do not claim the live path passed. + +- [ ] **Step 6: Run the sinks package** + +Run: `go test ./internal/sinks/` +Expected: PASS. + +- [ ] **Step 7: Commit** + +```bash +git add internal/sinks/clickhouse.go internal/sinks/clickhouse_test.go +git commit -m "clickhouse sink: a value the driver refuses is coded permanent + +batch.Append failed on an ISO 8601 timestamp bound for a DateTime column +and the sink returned the error with no code. The ladder retried it four +times and reported system.sink.unreachable for a server that was up (#233). + +Append validates and buffers in memory. clickhouse-go v2.48.0's native and +HTTP batches both call block.Append and return; neither touches the +connection until Send. The value cannot succeed on a later attempt, so it +is wrapped with user.sink.encode_failed at the driver boundary. The +driver's message, which names the column and quotes the value, survives +the wrap. + +PrepareBatch and Send keep sinkError, which is where bytes cross the +network and where unreachable is a real answer. + +What breaks if this is wrong: Append had a failure mode a retry could fix, +and that batch is now reported terminal after one attempt." +``` + +--- + +### Task 4: Document the retry rule + +**Files:** +- Modify: `README.md` (insert a `### Sink retries` subsection at the end of `## Sinks`, before `## Error policies` at line 537) +- Modify: `docs/coverage/invariants.yml:117-126` (`sink.error.classifies` claim) +- Modify: `docs/superpowers/specs/2026-09-12-sink-encode-failure-design.md` (Docs section) + +**Interfaces:** +- Consumes: the rule from Task 2 and the code from Task 1. The README table must match the `retryable` doc comment line for line in meaning. + +- [ ] **Step 1: Add the README subsection** + +Insert immediately before `## Error policies` in `README.md`: + +````markdown +### Sink retries + +ClickHouse and Iceberg flushes retry when the destination is not answering. +Omit the block to accept the defaults. Set `max_attempts: 1` to turn retrying +off. The Kafka sink ignores this block: franz-go already retries a produce +with its own backoff. + +```yaml +sink: + type: clickhouse + retry: + max_attempts: 4 # total attempts, including the first + initial_backoff_ms: 100 # doubles each attempt + max_backoff_ms: 2000 # ceiling on the backoff + deadline_seconds: 10 # bounds the whole ladder, not one attempt +``` + +The values shown are the defaults, from `internal/sinks/policy.go`. + +Keep `deadline_seconds` below `pipeline.flush_interval_seconds`. The retry +runs inside the open state transaction, and a ladder that outlives the flush +interval freezes the window clock. + +The ladder retries only what another attempt could change. The error code +decides: + +| Error | Retried | Why | +| --- | --- | --- | +| Any `user.*` code, including `user.sink.encode_failed` and `user.sink.type_unsupported` | No | The config, SQL or a value is wrong. It fails identically every time. | +| `system.sink.write_failed` | No | The destination answered and refused the write. | +| `system.sink.unreachable` | Yes | The destination may come back. | +| An error with no code | Yes | A driver's timeout or reset arrives unclassified. The deadline bounds the cost. | +| Any other `system.*` code | Yes | | + +A failure that is not retried keeps its own code and exit code. A retried +failure that outlasts the ladder is reported as `system.sink.unreachable`, +exit 12, and `sink_retry_count_total` counts each attempt after the first. + +`user.sink.encode_failed` is the sink's client refusing a value before +anything reaches the network, such as a timestamp string the ClickHouse +driver's `DateTime` layout cannot parse. Cast or format the column in the +handler SQL. The message names the column and quotes the value. +```` + +- [ ] **Step 2: Update the invariant claim** + +In `docs/coverage/invariants.yml`, replace the `sink.error.classifies` claim: + +```yaml + claim: > + The sink's errors classify as unreachable, rejected, or a user fault, + so the retry ladder retries only the first. A value the sink's client + cannot encode is a user fault: it never reaches the network and fails + the same way every attempt. +``` + +Run: `go test ./internal/coverage/ ./internal/sinks/ -run 'Coverage|Invariant|Conformance' 2>&1 | tail -5` +Expected: PASS or no matching tests. If a test parses `invariants.yml` and fails on the edit, the YAML is malformed; fix the indentation. + +- [ ] **Step 3: Correct the spec's Docs section** + +In `docs/superpowers/specs/2026-09-12-sink-encode-failure-design.md`, replace the Docs section with: + +```markdown +### Docs + +README gains a "Sink retries" subsection under Sinks. It documents the retry +block, which the README did not cover, and states the retry rule as a table +keyed by error code. The `retryable` doc comment carries the same table and +names the README section, so a change to one points at the other. + +The `sink.error.classifies` invariant claim names the third class. + +Release notes live in the annotated tag, not in `CHANGELOG.md`, which stops +at v1.0.0. The v1.2.1 tag message names the defect: an encode failure was +retried and then reported as unreachable, and any user-class sink error took +the same path. It names the new code and the retry rule. + +Nothing renders `errs.All()` into documentation today, so the registry entry +is the code's only published description. +``` + +- [ ] **Step 4: Read the README section once as an operator** + +Open `README.md` at the new subsection and check: every code named in the table exists in `internal/errs/testdata/codes.golden`; the exit codes 10 and 12 match `internal/errs/exit.go`; the metric name matches the `## Metrics` table at README line 716. + +- [ ] **Step 5: Commit** + +```bash +git add README.md docs/coverage/invariants.yml docs/superpowers/specs/2026-09-12-sink-encode-failure-design.md +git commit -m "docs: the sink retry ladder, keyed by error code + +The README did not document sink.retry at all, and nothing stated which +failures the ladder retries. An operator reading system.sink.unreachable +after four attempts on a bad value had no way to learn the rule (#233). + +Sink retries is a new README subsection: the config block, the deadline's +relation to the flush interval, and a table of what is retried by error +code. The retryable doc comment carries the same table and names the +section. The sink.error.classifies invariant names the user-fault class. + +What breaks if this is wrong: the README and the code disagree about what +is retried, and the operator trusts the README." +``` + +--- + +### Task 5: Verify the whole change + +**Files:** none modified. + +- [ ] **Step 1: Run the full test suite** + +Run: `go test ./... 2>&1 | grep -v '^ok' | grep -v 'no test files'` +Expected: no output. Anything printed is a FAIL or a build error to fix before finishing. + +- [ ] **Step 2: Run vet** + +Run: `go vet ./internal/errs/ ./internal/sinks/` +Expected: no output. + +- [ ] **Step 3: Confirm the spec's acceptance by hand, if a ClickHouse is up** + +Build: `make build` (or `go build -o bin/sqlflow ./cmd/sqlflow`; check the Makefile for the binary name). Create a table `sf_tz` with a `DateTime` column named `dt_plain` and a `String` column `label`. Run the issue's config from the spec with one message `{"label":"isoz","ts":"2026-09-01T12:00:00Z"}` on the topic. + +Expected on stderr, once, with no `sink retry` log lines before it: + +``` +Error: [user.sink.encode_failed] clickhouse sink: encode row: clickhouse [AppendRow]: dt_plain parsing time "2026-09-01T12:00:00Z" as "2006-01-02 15:04:05": cannot parse "T12:00:00Z" as " " +``` + +And `echo $?` prints `10`. + +If no ClickHouse is available, say so in the final report. Do not report the acceptance as passed. + +- [ ] **Step 4: Report** + +State which tests ran, which skipped, and whether the live acceptance ran. Paste the `go test ./...` summary line for `internal/sinks` and `internal/errs`. diff --git a/docs/superpowers/specs/2026-09-12-run-enforces-schema-design.md b/docs/superpowers/specs/2026-09-12-run-enforces-schema-design.md new file mode 100644 index 0000000..7d9a38f --- /dev/null +++ b/docs/superpowers/specs/2026-09-12-run-enforces-schema-design.md @@ -0,0 +1,172 @@ +# `run` enforces the schema that `validate` enforces + +Issue #231. Verified against `main` at a3da41e on 2026-09-12. + +## The problem + +`sqlflow run` loads a config without checking it against the config schema. +`sqlflow validate` and `sqlflow config validate` both do. A config that +`validate` rejects still runs. + +The issue's example is a missing `batch_size`. The engine passes the parsed +value straight through, so `Turbine.batchSize` is 0. The consume loop closes a +batch with `numBatchMessages == t.batchSize`, and the count is at least 1 when +that runs, so the count-based flush never fires. Batches close only on +`flush_interval_seconds`, which defaults to 30 seconds. The issue measured +first output at 31 seconds instead of 0. Nothing in the logs names the cause. + +The situation on `main` is worse than the issue describes. PR #241 replaced the +hand-written schema with one reflected from the config structs. `BatchSize` +carries `yaml:"batch_size,omitempty"`, and the reflector reads `omitempty` as +optional. The v1.0 schema listed `batch_size` under `pipeline.required`; the +reflected schema lists only `source`, `handler` and `sink`. So today `validate` +accepts the config too, and no command catches the missing key. + +Three facts, each verified on `main`: + +| Fact | Where | +| --- | --- | +| `run` calls `config.LoadRendered` and never `validate.Validate` | `internal/cli/run/root.go:137` | +| `pipeline.required` is `[source, handler, sink]` | `internal/validate/schemas/config.json` | +| `batch_size` of 0 makes the count flush unreachable | `internal/core/turbine.go:694` | + +The README still says `batch_size` is required and at least 1. The schema no +longer says so, and the engine does not check. + +## The change + +Two parts. The first is the issue's ask. The second is the regression that +made the first insufficient. + +### 1. `run` validates before it loads + +`run` reads the config file, calls `validate.Validate` with the file's text, +and stops on any error-severity diagnostic. Only then does it render and load +the config as it does today. + +Behavior on a failing config: + +- Every error diagnostic is written to stderr in the same text format + `validate` uses, so the two commands print the same lines for the same + fault. +- The command returns `user.config.invalid`, which exits 10. A supervisor + reads that as terminal and does not restart into the same failure. +- Nothing has started: no DuckDB, no source, no sink, no metrics server. + +Warnings do not stop the run. They are written to stderr as warnings. The +one warning class today is an unsupplied template variable with no default. +`validate` demotes the resulting schema error to a warning because an +incomplete shell is not a wrong config. Under `run`, the shell is the +deployment, so an unsupplied variable is a real fault. Failing on it is a +policy change with its own blast radius and is out of scope here. The +warning line makes the empty substitution visible, which is more than `run` +says today. + +`sqlflow dev` and `sqlflow tail` load a config the same way `run` does and +have the same gap. Both get the same call. One helper in `internal/cli` +performs the read, validate, and report, and the three commands call it. The +helper is the only place that decides what stops a run, so the commands +cannot drift. + +`validate.Validate` renders the template once and `LoadRendered` renders it +again. Rendering is cheap and happens once per process. Sharing the rendered +bytes between the two is a refactor of `config.LoadRendered` for no +observable gain, so the double render stays. + +### 2. `batch_size` is required again, and at least 1 + +The struct tag changes from `yaml:"batch_size,omitempty"` to +`yaml:"batch_size" jsonschema:"minimum=1"`. The reflector then lists +`batch_size` under `pipeline.required` and emits `"minimum": 1`, the same +way the Kafka fetch bounds already do. `make schema` regenerates the committed +schema and the `config example` golden. + +Dropping `omitempty` also changes how a `Conf` marshals: a zero `BatchSize` +now serializes as `batch_size: 0`. Nothing in the engine marshals a `Conf` +back to YAML except the golden test, which is regenerated. The tests that +build a `Conf` in Go are unaffected, because they never go through the +schema. + +Requiring the key rather than defaulting it is the issue's option 1. It +matches what the v1.0 schema declared, what the README documents, and what +the Python engine required. Option 2 needs a default, and the only defensible +one is 1, which is the slowest setting and would surprise anyone who lost the +line to a bad merge in the opposite direction. + +The engine itself does not change. A defensive `>=` in the consume loop would +make 0 mean "flush every message", which is a second silent behavior for the +same mistake. The schema is the one place the rule lives. + +### Tests + +Schema: + +- `internal/schema`: `pipeline.required` contains `batch_size`, and + `batch_size` carries `minimum: 1`. This is the assertion that would have + caught #241's regression. It sits beside the golden test rather than inside + it, so a future regeneration cannot update it away. +- `internal/validate`: a config without `batch_size` fails `config.schema` + with a diagnostic at `/pipeline`. A config with `batch_size: 0` fails with a + diagnostic at `/pipeline/batch_size`. + +CLI: + +- `internal/cli/run`: `run` on a config missing `batch_size` returns an error + whose code is `user.config.invalid`, and the message names `batch_size`. + The test asserts that the error returns before the DuckDB open, by running + against a config whose source would fail to connect if reached. +- `internal/cli/run`: `run` on a config with an unsupplied template variable + and everything else valid proceeds past validation. This pins the + warnings-do-not-stop rule. +- `internal/cli`: `dev` and `tail` on the same invalid config return the same + code. One table-driven test over the three commands. +- `internal/cli/examples_test.go` already loads every example config. It + gains a validate pass, so an example that the schema rejects fails the + suite. Every example on `main` sets `batch_size`, so this is a guard, not a + fix. + +### Acceptance + +The issue's own measurement, repeated: + +``` +$ sqlflow run pipeline-without-batch-size.yml +pipeline-without-batch-size.yml:3:1: error: [user.config.invalid] at '/pipeline': missing property 'batch_size' +Error: [user.config.invalid] pipeline-without-batch-size.yml is invalid +$ echo $? +10 +``` + +And `validate` on the same file prints the same diagnostic line. + +### Docs + +README line 309 says an unset `flush_interval_seconds` means only +`batch_size` triggers a batch. `flushIntervalFor` defaults it to 30 seconds, +and the v1.0.0 changelog says so. The line changes to state the default. +Line 308 already says `batch_size` is required and at least 1, and becomes +true again. + +The v1.2.1 changelog entry names the defect: `run` accepted configs that +`validate` rejected, and `batch_size` had become optional in the schema +without anyone deciding it should be. + +## What breaks if this is wrong + +A config in the wild that omits `batch_size` stops starting. On `main` that +config runs with a 30 second flush and no count-based batching, which is +almost never what its author meant. The error names the key and the fix is one +line. That is the trade the issue asks for. + +A config that `validate` rejects for a reason `run` tolerated is the same +story. Strict YAML decoding already rejects unknown keys, so the new rejections +are type, enum, minimum and required violations. Each of those was a config +the engine ran with a value it did not expect. + +## Out of scope + +- Failing `run` on an unsupplied template variable. Noted above as a policy + change with its own blast radius. +- SQL validation under `run`. #142 and #169 cover the SQL checks, and they are + not part of `validate` yet. +- Sharing the rendered bytes between `validate` and `LoadRendered`. diff --git a/docs/superpowers/specs/2026-09-12-schema-registry-mvp-design.md b/docs/superpowers/specs/2026-09-12-schema-registry-mvp-design.md new file mode 100644 index 0000000..3b2e565 --- /dev/null +++ b/docs/superpowers/specs/2026-09-12-schema-registry-mvp-design.md @@ -0,0 +1,581 @@ +# Schema registry MVP: Avro and JSON Schema through Kafka, both directions + +Issue #272. Verified against `main` at ae8fd28 on 2026-09-12. + +## The problem + +Every byte the engine reads is assumed to be JSON, and every byte the Kafka +sink writes is JSON. A customer's topics carry Confluent-framed Avro. Nothing +in the config can say so, and nothing in the engine can decode it. + +## Scope + +In: + +- Kafka source: `json` (today), `json_schema`, `avro`. The last two are framed + by a Confluent-compatible schema registry. +- Kafka sink: the same three, with the sink registering or looking up its + output schema. +- The dev stack gains a schema registry. +- One example config per direction, a framed-record producer, README docs. +- The memory soak runs once per format, raw JSON included. +- Go benchmarks per format, an A/B against `main` with `benchstat`, and a + container throughput run per format. + +Out, each a follow-up issue: + +- Protobuf. Decoding needs `protocompile` plus `dynamicpb`. Encoding needs a + descriptor synthesized from an Arrow schema. Neither belongs in an MVP. +- Message keys. `core.Message` has no key field today. +- `InferredDiskBatch` and `StructuredBatch` with a registry format. The disk + path stages JSON files; a typed path would stage Parquet, which is its own + change. `StructuredBatch` derives its schema from a DuckDB table, and a + registry format has its own schema. Both combinations fail at start with + `user.config.invalid`. +- More than one topic on a registry-backed source. One reader schema per + pipeline, so one subject. +- `validate` fetching the registry schema and binding the SQL against it. + The config surface below is shaped so this drops in. +- AWS Glue Schema Registry. Different header, different client. +- Subject naming strategies other than `-value`. + +## Design + +Three axes, each with one config key. The format decides bytes to Arrow and +back. The handler decides staging. The registry is shared. + +### Config + +```yaml +pipeline: + schema_registry: # new, optional block + url: http://localhost:8081 + auth: # optional + username: '{{ SQLFLOW_SR_USER }}' + password: '{{ SQLFLOW_SR_PASS }}' + # or: bearer_token: '{{ SQLFLOW_SR_TOKEN }}' + ssl: # optional, same shape as kafka.ssl + ca_location: /etc/certs/ca.pem + + source: + type: kafka + kafka: + topics: [orders] + value: # new, optional block + format: avro # json | json_schema | avro; json is the default + + handler: + type: handlers.InferredMemBatch # unchanged + sql: SELECT ... + + sink: + type: kafka + kafka: + topic: orders-enriched + value: # new, optional block + format: avro # json | json_schema | avro; json is the default + subject: orders-enriched-value # default -value + auto_register: true # default true +``` + +Why `value` and not `format`: `sink.format.type: parquet` already exists, +parsed and ignored since the Python engine. Reusing the key would change its +meaning. `value` also names what it is, the record's value as opposed to its +key, which is where keys go when they arrive. + +Why `schema_registry` is on the pipeline: a source and a sink almost always +share one registry, and a shared block is one URL and one credential. A +per-side override is not in the MVP. + +Config rules, checked when the pipeline is built, each failing with +`user.config.invalid`: + +- `format` other than `json` requires `pipeline.schema_registry`. +- A registry-backed source requires exactly one topic and a handler of type + `handlers.InferredMemBatch`. +- `subject` and `auto_register` are sink-only keys. +- `auth` carries either `username` and `password` or `bearer_token`, not both. + +The three struct changes: + +```go +// Pipeline gains +SchemaRegistry *SchemaRegistry `yaml:"schema_registry,omitempty"` + +type SchemaRegistry struct { + URL string `yaml:"url"` + Auth *SchemaRegistryAuth `yaml:"auth,omitempty"` + SSL *KafkaSSL `yaml:"ssl,omitempty"` +} + +type SchemaRegistryAuth struct { + Username string `yaml:"username,omitempty"` + Password string `yaml:"password,omitempty"` + BearerToken string `yaml:"bearer_token,omitempty"` +} + +// KafkaSource and KafkaSink both gain +Value *KafkaValue `yaml:"value,omitempty"` + +type KafkaValue struct { + Format string `yaml:"format,omitempty" jsonschema:"enum=json,enum=json_schema,enum=avro"` + Subject string `yaml:"subject,omitempty"` + AutoRegister *bool `yaml:"auto_register,omitempty"` +} +``` + +`AutoRegister` is a pointer so an absent key means true. `make schema` +regenerates the JSON Schema from these tags. + +### The `serde` package + +A new package, `internal/serde`, owns everything between raw bytes and Arrow +that is not JSON inference: + +```go +// Registry wraps the franz-go client with the caches the client itself does +// not have. Schemas by ID are immutable and cached for the process lifetime. +type Registry struct { ... } +func NewRegistry(conf config.SchemaRegistry) (*Registry, error) +func (r *Registry) Probe(ctx context.Context) error +func (r *Registry) SchemaByID(ctx context.Context, id int) (sr.Schema, error) +func (r *Registry) Latest(ctx context.Context, subject string) (sr.SubjectSchema, error) +func (r *Registry) Register(ctx context.Context, subject string, s sr.Schema) (int, error) +func (r *Registry) Lookup(ctx context.Context, subject string, s sr.Schema) (int, error) + +// Decoder turns one framed record into one row of the reader schema. +type Decoder interface { + Schema() *arrow.Schema + Append(b *array.RecordBuilder, value []byte) error +} +func NewDecoder(ctx context.Context, format string, topic string, reg *Registry) (Decoder, error) + +// Encoder turns a batch into framed records, one per row. +type Encoder interface { + Encode(ctx context.Context, rec arrow.Record) ([][]byte, error) +} +func NewEncoder(format string, subject string, autoRegister bool, reg *Registry) (Encoder, error) +``` + +The franz-go module `github.com/twmb/franz-go/pkg/sr` v1.8.0 supplies the +client and `ConfluentHeader`, whose `DecodeID` and `AppendEncode` are the +header. `github.com/hamba/avro/v2` v2.31.0 supplies Avro. Both are new +dependencies. arrow-go's `arrow/avro` package was considered and not used: it +maps Avro enums to Arrow dictionaries, which a SQL user reads as a string, and +its type mapping panics into a recovered generic error where this engine wants +a coded one. The mapping below is about a hundred lines and is ours. + +### Decoding + +The reader schema is the subject's latest version at pipeline start, where +the subject is `-value`. The decoder converts it to an Arrow schema +once. That schema is the `batch` table for the life of the run, which is what +the handler SQL binds against and what `validate` will fetch later. + +Each record: + +1. `ConfluentHeader.DecodeID` strips the header. A record without the magic + byte is `user.data.malformed`. +2. The writer schema is fetched by ID from the registry, through the cache. + An ID the registry does not know is `user.data.schema_unknown`. A registry + that does not answer is `system.source.unreachable`, and the fetch is + retried three times over about a second before it is reported, because the + alternative is a good record in the DLQ. +3. Avro: hamba's `SchemaCompatibility.Resolve(reader, writer)` produces the + resolved schema, cached per writer ID, and `avro.Unmarshal` decodes the + payload into a `map[string]any` shaped like the reader. This is Avro schema + resolution as the specification defines it: added fields take their + defaults, removed fields are dropped, and a change the reader cannot + resolve is `user.data.malformed` naming both schemas. A payload that does + not decode against its own writer schema is `user.data.malformed`. +4. JSON Schema: the payload after the header is JSON. It is extracted against + the Arrow schema with the `appendJSONValue` machinery `StructuredBatch` + already has. No inference runs. The schema is not validated against the + payload in the MVP; a field the schema declares and the payload lacks is + null. +5. The row is appended to the record builder. Kafka metadata columns are + appended by the handler, as today. + +Avro to Arrow: + +| Avro | Arrow | +| --- | --- | +| `null` | `null` | +| `boolean` | `bool` | +| `int` | `int32` | +| `long` | `int64` | +| `float` | `float32` | +| `double` | `float64` | +| `bytes` | `binary` | +| `string`, `enum`, `uuid` | `utf8` | +| `fixed` | `binary` | +| `record` | `struct` | +| `array` | `list` | +| `map` | `map` | +| `["null", T]` in either order | nullable `T` | +| `date` | `date32` | +| `time-millis`, `time-micros` | `time32[ms]`, `time64[us]` | +| `timestamp-millis`, `timestamp-micros` | `timestamp[ms, UTC]`, `timestamp[us, UTC]` | +| `local-timestamp-*` | `timestamp` with no zone | +| `decimal` | `decimal128(precision, scale)` | +| union with two or more non-null branches, `duration` | `user.sql.type_unsupported` at start | + +`user.sql.type_unsupported` already exists: "A message field has a type the +handler cannot convert." The error names the field path and the Avro type. + +JSON Schema to Arrow, draft 7 and 2020-12 vocabulary that the MVP reads: + +| JSON Schema | Arrow | +| --- | --- | +| `string` | `utf8` | +| `integer` | `int64` | +| `number` | `float64` | +| `boolean` | `bool` | +| `object` with `properties` | `struct` | +| `array` with `items` | `list` | +| `type: [T, "null"]` | nullable `T` | +| `$ref`, `oneOf`, `anyOf`, `object` without `properties`, `array` without `items` | `user.sql.type_unsupported` at start | + +`format: date-time` stays `utf8` in the MVP; a `CAST` in the handler SQL is +the workaround, and it is the same workaround the JSON path needs today. + +### The typed handler + +`handlers.New` gains a functional option, `WithDecoder(serde.Decoder)`. When +the option is set and the config names `handlers.InferredMemBatch`, the +builder returns a new `TypedBatchHandler` instead. From the outside it is the +same kind, `inferred_mem`, so no new handler type reaches the coverage registry +or the config schema. + +`TypedBatchHandler` mirrors `InferredMemBatchHandler` in everything but the +decode. It holds one `array.RecordBuilder` over the decoder's schema plus the +metadata columns from `withMetadataFields`. `Write` calls `decoder.Append`, so +a record that does not decode fails at write time, which is the phase the +error policies key off and the reason `InferredMemBatch` validates JSON at +`Write` rather than `Invoke`. `Invoke` takes the record, binds it to the same +create-mode ingest statement into `batch`, runs the SQL, and reports +`RowsRead`. `Init` drops `batch`, as today. + +### Encoding + +The Kafka sink gains an `Encoder`. `WriteTable` calls `encoder.Encode` in +place of `tableRowsAsJSON`, and the `json` encoder is `tableRowsAsJSON`, so +the default path does not move. + +The Avro and JSON Schema encoders derive an output schema from the batch's +Arrow schema, obtain a schema ID, and frame each row. + +Obtaining the ID, cached by the Arrow schema's fingerprint so a stable SQL +result costs one registry call per run: + +- `auto_register: true`: `CreateSchema` on the subject. The registry returns + the existing ID for a schema it already holds, and 409 for one that breaks + the subject's compatibility rule. 409 is `user.sink.schema_incompatible`. +- `auto_register: false`: `LookupSchema` on the subject. 404 is + `user.sink.schema_unregistered`. This is the mode for a production registry + where pipelines are not allowed to register. + +Both codes are class `user`, so the retry ladder from #269 makes one attempt +and the process exits 10. That is the point: a SQL edit that changes the +output shape is stopped by the registry before any consumer sees it. + +Arrow to Avro, the inverse of the table above with these rules: the record is +named from the subject with characters outside `[A-Za-z0-9_]` replaced by +`_`, in namespace `io.turbolytics.sqlflow`; a nullable field is +`["null", T]` with default `null`; `int8` and `int16` widen to `int`; +`uint8` through `uint32` widen to `long`; `uint64`, `large_utf8`, +`large_binary` and `dictionary` are `user.sink.type_unsupported`, which +already exists for the ClickHouse sink. Cell values come from the `arrowValue` +extractor the ClickHouse sink already has, which returns `time.Time` for +timestamps and dates, the types hamba expects for those logical types. + +Arrow to JSON Schema: an `object` with one property per column, `required` +listing the non-nullable ones, and the type mapping inverted. The row bytes +are the JSON `tableRowsAsJSON` already produces, with the header prepended. + +The sink implements `Prober`. `Probe` calls `Registry.Probe`, which lists +subjects with a short timeout, so a registry that is down fails the start once +with `system.sink.unreachable`, the way a ClickHouse that is down does. + +### Error policy and the error class + +`applyErrorPolicy` applies `IGNORE` and `DLQ` to every write error. Today that +is safe because handlers return only `user.data.malformed` from `Write`. The +decoder introduces the first system-class write error, a registry that stops +answering mid-run. Under `DLQ` policy that would divert good records to the +dead-letter queue and commit their offsets. + +The fix is one guard: `IGNORE` and `DLQ` apply to class `user` only. A +system-class write error stops the pipeline whatever the policy, and the +process exits by the code's class, 11 for `system.source.unreachable`, which a +supervisor reads as retryable. `TestErrorDLQ_SystemClassWriteErrorIsNotDiverted` +pins it. + +### New error codes + +Appended to the registry, each with a summary and an action: + +| Code | When | Action | +| --- | --- | --- | +| `user.data.schema_unknown` | A record's schema ID is not in the registry | Check the producer registers against the same registry the pipeline reads | +| `user.sink.schema_incompatible` | The registry refused the output schema under the subject's compatibility rule | Change the handler SQL to keep the output shape, or change the subject's compatibility level | +| `user.sink.schema_unregistered` | `auto_register` is false and the output schema is not registered | Register the schema, or set `auto_register: true` where the registry allows it | + +`codes.golden` gains three lines. Everything else reuses `user.data.malformed`, +`user.sql.type_unsupported`, `user.sink.type_unsupported`, +`user.config.invalid`, `system.source.unreachable` and +`system.sink.unreachable`. + +### Dev stack + +`dev/kafka-single.yml` gains: + +```yaml + schema-registry: + image: confluentinc/cp-schema-registry:7.3.2 + hostname: schema-registry + container_name: schema-registry + ports: + - "8081:8081" + environment: + SCHEMA_REGISTRY_HOST_NAME: schema-registry + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka1:19092 + SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081 + depends_on: + - kafka1 +``` + +Same image line as the broker, so the two upgrade together. + +### The framed-record producer + +`cmd/publish-framed/main.go` registers a fixture schema under +`-value` and produces Confluent-framed records with franz-go and +hamba. It is the registry twin of `cmd/publish-test-data.py`, and four things +use it: the README walkthrough, the integration tests, the soak, and the +benchmark. The logic lives in `internal/serde/serdetest` so the tests import +it and the command is a thin wrapper. + +Flags: `--format avro|json_schema`, `--brokers`, `--registry`, `--topic`, +and either `--num-messages N` for a fixed count or `--rate R --until +` for a steady stream. The fixture is the soak payload's shape, +`sensor_id` int, `ts` timestamp-millis, `value` double, so every soak and +benchmark config runs the same SQL whatever the format. `ts` is a real +timestamp in the typed formats and a string in raw JSON, which is the +difference the typed path exists to make. + +`kafka-producer-perf-test` cannot produce these records: its payload file is +line-delimited text, and a framed Avro record is binary that can contain a +newline. So the producer runs inside the docker network the way the container +benchmark runs the engine: built as a linux binary in the `golang` image and +run in `debian:bookworm-slim` on `dev_default`, against `kafka1:19092` and +`http://schema-registry:8081`. From the host through Docker Desktop's port +forwarding it could not reach the rate the soak needs. + +`make publish-framed FORMAT=avro TOPIC=orders NUM_MESSAGES=1000` wraps the +fixed-count mode for the walkthrough. + +### Examples + +Two configs under `dev/config/examples/`, every variable with a default so +the example test renders them: + +- `kafka.avro.yml`: Avro in from the registry, aggregate in SQL, Avro out + under a new subject with `auto_register: true`. +- `kafka.json-schema.yml`: JSON Schema in, JSON Schema out. + +The example test builds the sink and the handler. The sink's `Probe` reaches +`localhost:8081`, and the test's `checkBuildError` skips on a connection error +and fails on "not supported" or "requires a". Messages from this change use +neither phrase for a resource that is merely absent. + +### Tests + +Coverage features, in `docs/coverage/features.yml`: + +```yaml + - id: serde.registry + description: Resolves, caches, registers and looks up schemas in a Confluent-compatible registry. + requires: [unit, integration] + - id: serde.avro + description: Decodes Confluent-framed Avro to typed Arrow and encodes Arrow batches back. + requires: [unit, integration] + - id: serde.json_schema + description: Decodes Confluent-framed JSON Schema records to typed Arrow and encodes Arrow batches back. + requires: [unit, integration] +``` + +Test names follow the registry's rule: `TestSerdeAvro_*`, +`TestSerdeJsonSchema_*`, `TestSerdeRegistry_*`, and +`TestIntegrationSerdeAvro_*` for the container pass. Release level is not +required for the MVP. + +Unit, no network. A fake registry over `httptest.Server` implements the five +routes the client uses: schema by ID, latest version, create, lookup, and +subjects. It records calls so a test can assert the cache. + +- Header: a record without the magic byte is `user.data.malformed`; a record + with an unknown ID is `user.data.schema_unknown`; the same ID is fetched + once across a thousand records. +- Avro to Arrow: one test per row of the mapping table, plus the two + unsupported cases with their code and field path. +- Avro decode: a record with every supported type round-trips values; a + writer schema with an added defaulted field and one with a removed field + both decode against the reader; an incompatible type change is + `user.data.malformed` naming both schemas. +- JSON Schema to Arrow: one test per row, plus the unsupported cases. +- Typed handler: `Write` of a bad record fails with the decoder's code and + the batch still ingests the good ones; metadata columns are present and + correct; `RowsRead` matches. +- Encoder: Arrow to Avro schema for every supported type and each + unsupported one; `auto_register: true` calls create once per distinct + schema; `auto_register: false` on an unregistered schema is + `user.sink.schema_unregistered`; a 409 is `user.sink.schema_incompatible`; + a framed record decodes back to the same values. +- Config: each rule in the Config section fails the build with + `user.config.invalid` and a message naming the key. +- Error policy: a system-class write error under `DLQ` stops the pipeline + and writes nothing to the DLQ. +- Registry: the append-only golden gains three lines; every new code exits + 10 and is not retryable. + +Integration, one Redpanda container through +`testcontainers-go/modules/redpanda` v0.44.0, which ships a +Confluent-compatible registry in the same process as the broker. The test +fails rather than skips when it cannot start one, per the coverage rules. + +- `TestIntegrationSerdeAvro_RoundTrip`: register a schema, produce framed + records, run a pipeline with an Avro source and an Avro sink, consume the + output topic, decode it with the schema the sink registered, and compare + values and types. +- `TestIntegrationSerdeAvro_WriterSchemaEvolvesMidRun`: produce under version + 1, register version 2 with an added field, produce under it, and assert the + pipeline keeps running and the added field reads as its default. +- `TestIntegrationSerdeJsonSchema_RoundTrip`: the same shape for JSON Schema. +- `TestIntegrationSerdeRegistry_DownAtStartFailsOnce`: a wrong port fails the + start with `system.source.unreachable` and no retry ladder. + +### Soak, one per format + +The memory gate runs once per supported serialization format, and the PR +carries three verdict blocks: + +| Format | Config | Producer | +| --- | --- | --- | +| raw JSON | `dev/config/soak/inferred.noop.yml`, unchanged | `kafka-producer-perf-test`, unchanged | +| JSON Schema | `dev/config/soak/json_schema.noop.yml` | `publish-framed --format json_schema --rate --until` | +| Avro | `dev/config/soak/avro.noop.yml` | `publish-framed --format avro --rate --until` | + +`make soak SOAK_FORMAT=json|json_schema|avro`, default `json`. +`scripts/soak.sh` picks the config and the producer from the format and +passes `SQLFLOW_SCHEMA_REGISTRY_URL=http://schema-registry:8081` through +`SOAK_ENV` for the framed ones. Everything else is the existing script: the +same topic retention, the same deadline-driven producer loop, the same +sampler and verdict. The rate stays just under what the engine consumes, so +the gate is decided from the same producer position for every format. + +All three keep the noop sink. The gate isolates the decode path on purpose: +what grows under Avro and not under raw JSON grew in the decoder, the +registry cache, or the typed handler, and nowhere else. The encoders are +covered by the benchmarks below, whose `allocs/op` is flat by construction +or the benchmark says so. + +The raw JSON soak is the regression check. The decoder seam touches the +handler builder and the consume loop's write path, and the gate proves the +default path did not pick up an allocation on the way. + +### Benchmarks, Go `testing.B` and an A/B against `main` + +Two levels. The first is `go test -bench`, compared with `benchstat`. The +second is the container throughput run, which is the number the README +quotes. + +Go benchmarks, in `internal/serde/bench_test.go` and beside the existing +handler benchmarks, each with `ReportAllocs` and `SetBytes` on the payload: + +| Benchmark | Measures | +| --- | --- | +| `BenchmarkDecode/json`, `/json_schema`, `/avro` | One `Decoder.Append` per record over a 1,000-record fixture, registry pre-warmed | +| `BenchmarkEncode/json`, `/json_schema`, `/avro` | One `Encoder.Encode` over a 1,000-row batch, schema ID cached | +| `BenchmarkTypedBatch/json_schema`, `/avro` | Write plus Invoke through `TypedBatchHandler`, beside `BenchmarkInferredMemBatch` and `BenchmarkStructuredBatch` on the same fixture | + +The A/B set is every benchmark that runs on both `main` and the branch: +`BenchmarkInferredMemBatch`, `BenchmarkStructuredBatch`, +`BenchmarkConsumeLoopWritePath`, and `BenchmarkDecode/json` plus +`BenchmarkEncode/json` once they exist on both. `scripts/bench-ab.sh +` checks the baseline out into a temporary worktree, runs the +set on both with `-count 10`, and prints `benchstat baseline.txt branch.txt`. +`benchstat` comes from `golang.org/x/perf/cmd/benchstat`, run with `go run`, +so nothing is installed. + +Acceptance for the A/B: no benchmark in the set shows a statistically +significant regression. `benchstat` marks a delta with `p < 0.05`, and a +marked slowdown on the raw JSON path fails the PR. The typed formats have no +`main` counterpart; they are reported as absolute numbers and as a ratio to +`BenchmarkDecode/json` on the branch, so the cost of the typed path is a +number in the PR rather than a guess. + +Container throughput, one run per format, `make benchmark-container +FORMAT=json|json_schema|avro`, with `benchmark.json-schema.mem.yml` and +`benchmark.avro.mem.yml` beside the existing benchmark configs and the +producer chosen by format. Two rounds each, same machine, in the PR as a +table with `main` as the first row, the shape #260 used. + +### Acceptance + +Against the dev stack: + +``` +make start-backing-services +make publish-framed FORMAT=avro TOPIC=orders NUM_MESSAGES=1000 +./bin/sqlflow run dev/config/examples/kafka.avro.yml --max-msgs 1000 +``` + +The pipeline exits 0. The output topic holds framed Avro records, and +`curl localhost:8081/subjects` lists both subjects. The console shows the +aggregate with typed columns, a `timestamp` not a string. Removing +`pipeline.schema_registry` from the config fails the start with +`user.config.invalid` naming the key. Stopping the registry container fails +the start with exit 11. + +### Docs + +README gains a "Schema registry" section under Sources: the config block, the +three formats, the two mapping tables, the sink modes, the error codes, and a +"Not yet" list that is the Out section of this spec. The `kafka` source and +sink examples reference it. + +## What breaks if this is wrong + +If the reader-schema-at-start rule is wrong, a producer that adds a field +mid-run does not surface it until the pipeline restarts. That is documented +and it is the Avro specification's own resolution behavior. A pipeline that +needs the new field restarts. + +If the class guard on the error policy is wrong, a system-class write error +that used to be silently ignored under `IGNORE` now stops the pipeline. No +handler emits one today, so nothing in the wild changes behavior. + +If the type mapping is wrong, a column reaches DuckDB with a type the SQL did +not expect. Every mapped type is in the unit tests, and every unmapped one +fails at start rather than at the first record. + +## Build order + +1. Config structs, rules, schema regeneration, error codes. +2. `serde.Registry` with the fake registry and its tests. +3. Avro to Arrow mapping and the Avro decoder, with `BenchmarkDecode/avro`. +4. `TypedBatchHandler`, the `WithDecoder` wiring through `run`, and + `BenchmarkTypedBatch`. +5. Error policy class guard. +6. JSON Schema to Arrow mapping and decoder, with its benchmark. +7. Encoders and the Kafka sink wiring, Avro then JSON Schema, with + `BenchmarkEncode`. +8. `serdetest` and `cmd/publish-framed`, the dev stack, examples, README. +9. Integration tests on Redpanda. +10. `scripts/bench-ab.sh`, the soak and benchmark configs and the format + switches in `scripts/soak.sh` and `scripts/benchmark-container.sh`. +11. Run the gates: three soaks, the `benchstat` A/B, three container runs. + Their output is the PR body. + +Each step lands green on its own. The pipeline reads Avro after step 4. diff --git a/docs/superpowers/specs/2026-09-12-sink-encode-failure-design.md b/docs/superpowers/specs/2026-09-12-sink-encode-failure-design.md new file mode 100644 index 0000000..3c64f46 --- /dev/null +++ b/docs/superpowers/specs/2026-09-12-sink-encode-failure-design.md @@ -0,0 +1,212 @@ +# A value the sink cannot encode is permanent, not unreachable + +Issue #233. Verified against `main` at a3da41e on 2026-09-12. + +## The problem + +The ClickHouse sink builds its batch client side. When the driver refuses a +value during `batch.Append`, the sink returns the error with no code. The retry +ladder retries anything uncoded, so the same value is re-encoded on every +attempt, and after the ladder is spent the failure is reported as +`system.sink.unreachable`. Nothing was unreachable, the class blames sqlflow +for a payload fault, and the exit code tells a supervisor to restart into the +same failure. + +Two probes against `main` confirm the issue and find a second instance of it. + +The issue's value still fails. `temporalFromString` accepts the driver's own +layouts plus a zone-less form read as UTC. `2026-09-01T12:00:00Z` matches +neither, so it goes to the driver unchanged and fails there: + +``` +temporalFromString(DateTime, "2026-09-01T12:00:00Z") ok=false +``` + +A coded user error is retried too. `retryable` lists three codes that are not +retried and retries everything else. `user.sink.type_unsupported`, which +`arrowValue` returns for a column type the sink cannot convert, is not on the +list: + +``` +inner error: [user.sink.type_unsupported] bad type +attempts=4 code=system.sink.unreachable +err=[system.sink.unreachable] sink still failing after 4 attempts: [user.sink.type_unsupported] bad type +``` + +So the fault is in two layers: + +1. The sink hands an encode failure up without a code. +2. The ladder decides retryability by listing what not to retry, so any + permanent failure it did not anticipate is retried. + +Fixing only the first leaves the second to bite the next code someone adds. +Fixing only the second leaves the uncoded `append row` error retried, because +an uncoded error has to stay retryable: a driver's timeout or reset arrives +without a code, and those are what the ladder exists for. + +## The change + +### 1. `retryable` decides by class + +A failure is retried unless another attempt cannot change the outcome. The +rule, in order: + +| Error | Retried | Why | +| --- | --- | --- | +| Class `user` | No | The config, SQL or data is wrong. It fails identically every time. | +| `system.sink.write_failed` | No | The destination answered and refused. Same result next time. | +| `system.sink.unreachable` | Yes | The destination may come back. | +| Uncoded | Yes | A driver's timeout or reset arrives unclassified. The deadline bounds the cost of guessing wrong. | +| Any other `system` code | Yes | Unchanged from today. | + +The implementation replaces the three-code switch with `errs.ClassOf(err) == +errs.ClassUser` plus the `write_failed` case. `user.config.invalid` and +`user.sink.invalid`, the two user codes on today's list, are covered by the +class rule. The doc comment on `retryable` states the rule as the table +above. + +This is the part that is not ClickHouse specific. Any sink that codes an +encode failure as a user error gets the right retry behavior from the ladder +with no further wiring. + +### 2. A code for a value the sink cannot encode + +New registry entry: + +``` +user.sink.encode_failed +Summary: The sink's client could not encode a result value for the destination + column. It fails the same way on every attempt and is not retried. +Action: Cast or format the column in the handler SQL to match the destination + column's type. The message names the column and the value. +``` + +Class `user` exits 10. A supervisor reads that as terminal, which is correct: +the value is in the topic and a restart re-reads it. + +The existing `user.sink.type_unsupported` stays for what it means today, a +column whose Arrow type the sink cannot convert. That is a property of the +schema and fails for every row. `encode_failed` is a property of one value in +a column whose type is supported. An operator who reads "type unsupported" +for a string that is merely formatted wrong would look at the wrong thing. + +`codes.golden` gains the line. The registry is append-only and the test +enforces that. + +### 3. The ClickHouse sink codes its encode failures + +In `appendTables`, the `batch.Append` error is wrapped with +`user.sink.encode_failed`. The driver's message already names the column and +quotes the value, so the wrap adds only the sink name: + +``` +[user.sink.encode_failed] clickhouse sink: encode row: clickhouse [AppendRow]: +dt_plain parsing time "2026-09-01T12:00:00Z" as "2006-01-02 15:04:05": +cannot parse "T12:00:00Z" as " " +``` + +The `arrowValue` path keeps `user.sink.type_unsupported`. Both are class +`user`, so both stop the ladder after one attempt. + +`PrepareBatch` and `batch.Send` keep `sinkError`, which separates unreachable +from write-failed by inspecting the error. Those two calls are where bytes +cross the network, and their classification is already right. + +The sink still requeues the tables on failure. That is unchanged: whether the +batch can be dropped is the pipeline's error policy's decision, not the +sink's. + +### The Iceberg sink + +Iceberg is the other sink the ladder wraps. `table.AppendTable` converts and +writes in one call, and its error is returned uncoded, so an encode failure +there is retried today too. iceberg-go does not separate the two phases at +the call boundary, and the error types it returns for a conversion fault are +not distinguishable from its I/O faults without string matching. This spec +leaves Iceberg as it is and records the gap. A false "permanent" on a +transient Iceberg fault would drop a batch, which is worse than three wasted +attempts. + +### Tests + +Retry ladder, `internal/sinks/retry_test.go`: + +- A sink failing with `user.sink.encode_failed` is attempted once. The + returned error keeps that code. +- A sink failing with `user.sink.type_unsupported` is attempted once. This is + the second probe above, inverted into an assertion. +- A sink failing with an uncoded error is attempted `MaxAttempts` times and + the result is `system.sink.unreachable`. This pins the rule that uncoded + stays retryable, so a future tidy-up cannot turn a driver timeout into a + terminal failure. +- The existing `write_failed` and `sink.invalid` tests continue to pass + unchanged. + +ClickHouse sink, `internal/sinks/clickhouse_test.go`: + +- `appendTables` against a fake `driver.Batch` whose `Append` returns the + driver's parse error yields `user.sink.encode_failed`, and the driver's + message survives in the chain. `driver.Batch` is an interface, so the fake + needs no server. +- `temporalFromString` on `2026-09-01T12:00:00Z` returns `ok == false`. This + documents that the value reaches the driver, which is the precondition for + the encode path. It is not a claim that the value should fail; see + follow-ups. + +Registry, `internal/errs`: + +- `codes.golden` regenerated. The append-only test passes. +- `ExitCode` of a `user.sink.encode_failed` error is 10. + +End to end, if the ClickHouse conformance harness runs in this change: the +issue's config and message produce `user.sink.encode_failed` on the first +attempt, no retry counter increment, and exit 10. This is the acceptance the +issue describes, and it is the only test here that needs a ClickHouse. + +### Acceptance + +The issue's own repro, after the change: + +``` +Error: [user.sink.encode_failed] clickhouse sink: encode row: clickhouse +[AppendRow]: dt_plain parsing time "2026-09-01T12:00:00Z" as +"2006-01-02 15:04:05": cannot parse "T12:00:00Z" as " " +$ echo $? +10 +``` + +One attempt. `sink_retry_count_total` does not move. + +### Docs + +The v1.2.1 changelog entry names the defect: an encode failure was retried +and then reported as unreachable, and any user-class sink error took the same +path. It names the new code and the retry rule. + +The error code reference, wherever `errs.All()` is rendered, picks up the new +entry from the registry. + +## What breaks if this is wrong + +If the class rule is too broad, a transient fault that some sink coded as +`user` stops being retried. Today no sink codes a transient fault as `user`: +the user codes are config, SQL, type and, now, encode. The retry tests pin +each case, so a sink that starts doing this fails a test rather than a +pipeline. + +If the ClickHouse wrap is wrong, `batch.Append` has a failure mode that a +retry could fix. It does not: `Append` validates and buffers in memory, and +the driver sends nothing until `Send`. Checked in clickhouse-go v2.48.0: both +the native `batch.Append` and `httpBatch.Append` call `block.Append` in memory +and return. Neither touches the connection. + +## Follow-ups, not in this change + +- Accept RFC 3339 in `temporalFromString`. The issue's value is the most + common JSON timestamp form, and with this change it fails fast and clearly + rather than succeeding. Whether the Python engine, which sends the string + to the server, accepts it depends on the server's `date_time_input_format`, + and that parity question is unverified. It is a type-matrix change with its + own spec, not a classification fix. +- Code Iceberg's encode failures when iceberg-go exposes a way to tell them + from I/O failures.