storage: Add Debezium JSON envelope support for Kafka sources

[bosconi]

Summary

• Adds support for JSON-encoded Debezium envelope messages on Kafka sources, previously limited to Avro only
• Introduces ENVELOPE DEBEZIUM (MODE = 'TICDC') syntax for TiCDC dialect support
• New UpsertStyle::DebeziumJson variant handles envelope extraction at render time, producing [key: jsonb, data: jsonb, ...metadata] output schema

Motivation

Enables ingestion from CDC systems like TiCDC that emit Debezium-format JSON rather than Avro.

Syntax

-- Generic Debezium JSON
CREATE TABLE t FROM SOURCE s (REFERENCE "topic")
  KEY FORMAT JSON VALUE FORMAT JSON
  ENVELOPE DEBEZIUM;

-- TiCDC dialect
CREATE TABLE t FROM SOURCE s (REFERENCE "topic")
  KEY FORMAT JSON VALUE FORMAT JSON
  ENVELOPE DEBEZIUM (MODE = 'TICDC');

Changes

• Parser (sql-parser): SourceEnvelope::Debezium is now a struct variant with optional DebeziumMode
• Planning (sql): JSON encoding routes to UpsertStyle::DebeziumJson; error message updated to "AVRO or JSON"
• Storage types (storage-types): New DebeziumJsonMode enum and UpsertStyle::DebeziumJson variant with descriptor transform
• Render (storage): extract_debezium_json() parses Jsonb datum to extract op/after fields, supports payload-wrapped format
• Catalog (catalog): Exhaustive match updated for new variant

Test plan

• Unit tests: mz-sql-parser, mz-storage-types, mz-sql, mz-storage, mz-catalog all pass
• Parser datadriven tests rewritten for struct variant
• SLT subscribe_outputs.slt passes (SUBSCRIBE DEBEZIUM unaffected)
• Testdrive kafka-debezium-json-sources.td: CRUD, snapshot, payload-wrapped, INCLUDE OFFSET, error cases
• Testdrive kafka-debezium-json-ticdc-sources.td: CRUD, delete, reinsertion, MODE syntax, invalid mode
• Existing kafka-upsert-debezium-sources.td updated and passes (error message change)

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 noreply@anthropic.com

[github-actions]

Thanks for opening this PR! Here are a few tips to help make the review process smooth for everyone.

PR title guidelines

• Use imperative mood: "Fix X" not "Fixed X" or "Fixes X"
• Be specific: "Fix panic in catalog sync when controller restarts" not "Fix bug" or "Update catalog code"
• Prefix with area if helpful: compute: , storage: , adapter: , sql:

Pre-merge checklist

• The PR title is descriptive and will make sense in the git log.
• This PR has adequate test coverage / QA involvement has been duly considered. (trigger-ci for additional test/nightly runs)
• If this PR includes major user-facing behavior changes, I have pinged the relevant PM to schedule a changelog post.
• This PR has an associated up-to-date design doc, is a design doc (template), or is sufficiently small to not require a design.
• If this PR evolves an existing $T ⇔ Proto$T mapping (possibly in a backwards-incompatible way), then it is tagged with a T-proto label.
• If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label (example).

[bosconi]

@claude what sayest thou?

[claude]

Claude finished @bosconi's task in 4m 33s —— View job

---

Code Review

• Read CLAUDE.md for repo guidelines
• Analyze changed files
• Review parser changes
• Review planning/DDL changes
• Review storage-types and render changes
• Review tests
• Post findings

---

Overall this is a solid, well-structured PR. The layered approach (parser → planner → storage-types → render) is clean, and the test coverage is good. A few issues worth addressing before merge:

---

Issues

1. _mode is unused — TiCDC and Generic behave identically

src/storage/src/render/sources.rs:655

fn extract_debezium_json(
    row: &Row,
    _mode: &DebeziumJsonMode,   // <-- intentionally unused
    include_envelope: bool,
) -> Result<Option<(Row, Option<Row>)>, DecodeError> {

The underscore prefix is an explicit statement that TiCDC mode is not differentiated at extraction time. If that's intentional (TiCDC uses standard Debezium format with extra fields passable via INCLUDE DEBEZIUM METADATA), that's fine, but it deserves a comment explaining why no special handling is needed rather than just silently ignoring it. If TiCDC eventually needs different behavior (e.g., different op semantics, null-key tombstone patterns), removing the underscore silently is a footgun. Consider either removing the mode parameter entirely (and the DebeziumJsonMode storage enum) or adding a // TiCDC uses standard Debezium op/after semantics; mode is reserved for future dialect-specific handling comment.

Fix this →

---

2. INCLUDE KEY AS alias silently ignored for JSON Debezium

src/sql/src/plan/statement/ddl.rs:1390

The planner allows INCLUDE KEY AS mykey when using JSON Debezium:

(ast::SourceEnvelope::Debezium { .. }, _, Some(DataEncoding::Json)) => {
    // JSON Debezium: key is included as a separate column, INCLUDE KEY is fine
}

But UnplannedSourceEnvelope::desc() for DebeziumJson completely ignores key_envelope and always emits a hardcoded "key": Jsonb column (src/storage-types/src/sources/envelope.rs:256). Any user alias specified via INCLUDE KEY AS mykey is silently discarded — the column always ends up as "key". This is a silent mismatch that will confuse users.

Either reject INCLUDE KEY for JSON Debezium (matching the Avro behavior conceptually, since the key is already materialized), or actually honour the alias from key_envelope.

Fix this →

---

3. No feature flag for new user-facing syntax

New storage envelope modes in this repo are typically gated behind a scx.require_feature_flag(...) to allow staged rollout and testing in cloud before GA. The JSON Debezium path has no flag:

Some(DataEncoding::Json) => {
    // JSON path: skip typecheck_debezium, use DebeziumJson style
    let storage_mode = match mode { ... };
    ...
    UnplannedSourceEnvelope::Upsert { style: UpsertStyle::DebeziumJson { ... } }
}

Compare with e.g. ENVELOPE MATERIALIZE which calls scx.require_feature_flag(&vars::ENABLE_ENVELOPE_MATERIALIZE)?;. Is the intent to GA this immediately? If so, the PR checklist item "major user-facing behavior changes" needs a PM ping.

---

Nits / Minor

src/storage/src/render/sources.rs:671 — local variable named envelope shadows the Debezium concept of "envelope":

let envelope = match json_value.get("payload") {
    Some(payload) if payload.is_object() => payload,
    _ => &json_value,
};

This is the resolved message payload, not the envelope. Naming it message or payload would reduce confusion when reading error messages below that reference envelope.

src/sql-parser/src/ast/defs/ddl.rs:645 — DebeziumMode::None in the AST implies absence. The storage-types version is named DebeziumJsonMode::Generic which is more descriptive. For future consistency consider DebeziumMode::Generic in the AST layer too (minor, not blocking).

src/storage/src/render/sources.rs:691 — tracing::warn! on every truncate event (op='t') could be noisy at scale. Consider using tracing::debug! or a rate-limited warn.

src/storage-types/src/sources/envelope.rs:257 — key: Jsonb.nullable(false) — is this guaranteed? If Kafka key bytes are empty, the JSON decoder emits Datum::JsonNull. That violates the non-null constraint at runtime. This may be fine in practice (empty keys are unusual for CDC), but worth a comment or a check.

---

Tests

The testdrive coverage is solid: CRUD, snapshot, payload-wrapped, INCLUDE OFFSET, INCLUDE DEBEZIUM METADATA with/without alias, TiCDC metadata fields (commit_ts, ts_ms), delete + re-insert, and error cases. One gap: no test for INCLUDE KEY with JSON Debezium — given issue #2 above, this path is currently broken/misleading and a test would catch it.

---

Pre-merge checklist item

The T-proto checklist item: UpsertStyle and SourceEnvelope are Serialize/Deserialize types used for catalog persistence. Based on the code, they appear to use serde (not protobuf) for encoding — so a T-proto label is likely not required, but worth confirming with the storage team.

[bosconi]

Experiment concluded.