emitter.dot

// walshadow — CH emitter component view
// Parallel decode+insert pipeline: reorder coordinator (commit order,
// side-effect-free transaction plan then execute, DDL/TRUNCATE barrier)
// → decode pool ×M → InsertBatcher (per-table
// TableEncoder, budget/deadline seal) → inserter pool ×N (one complete
// INSERT per sealed batch) → ack collector (contiguous-done watermark).
// Zoomed-in view of cluster_ch + cluster_ddl from internals.dot.
//
// regeneration spec:
// sources of truth: plans/emitter.md · src/emit/pipeline/{reorder,planner,plan_spool,decode,batcher,inserter,ack,tail}.rs · src/emit/{ch_emitter,ch_ddl}.rs · src/catalog/type_bridge.rs
// subsumes: plans/emitter.md § "Stage walk" + "Transaction planner" + "Barrier fence" + "Ack-LSN tracking" + "DdlApplicator"
// quality bar:
//   - decode ×M and inserter ×N read as pools (stacked node or ×M/×N label), not single tasks
//   - barrier fence visually orders DDL strictly after earlier data durable (placed → FlushAll → durable)
//   - ack side-channel (Register/Placed/Acked → watermark) distinct from row path; dotted #b380b0
//   - rows path solid #BF8C5F, DDL path dashed #BF8C5F — same colour, different style
// shared style: palette.md
digraph emitter {
  rankdir=TB;
  compound=true;
  graph [fontname="Helvetica", labelloc="t", label="walshadow emitter — one ordered pipeline, ClickHouse or metrics only", fontsize=14, splines=spline, nodesep=0.4, ranksep=0.5, bgcolor="#272623", fontcolor="#ECE1D7"];
  node  [fontname="Helvetica", fontsize=10, shape=box, style="rounded,filled", color="#6E6963", fontcolor="#ECE1D7"];
  edge  [fontname="Helvetica", fontsize=9, arrowsize=0.8, color="#c1a78e", fontcolor="#ECE1D7"];

  // ═════ reorder coordinator (upstream, commit-order boundary) ═════
  subgraph cluster_src {
    label="reorder coordinator — single-threaded commit order  (emit/pipeline/reorder.rs, inner sink of QueueingRecordSink)";
    style="rounded,filled"; color="#4c4641"; fillcolor="#34302c"; fontcolor="#ECE1D7";

    reorder [label="transaction reorder\nwalk rows and changes in WAL order\nempty commits and aborts still advance progress\nschema and truncate changes wait for earlier rows", fillcolor="#4D4128"];
    plan    [label="transaction planner (planner.rs + plan_spool.rs)\nplan first, execute after: detoast + route +\nraw decode, side-effect-free; frozen route view\n(whole-xact config granularity)\nplan ≤1 MiB resident else .plan spool\n[len|crc32c|body]*, seal frame; error = abandon,\nnothing emitted", fillcolor="#4D4128"];
    cat     [label="ShadowCatalog::subscribe\nSchemaEvent\nAdded / Changed / Dropped\n(unbounded mpsc, rides xact buffer\n as ordered_events)", fillcolor="#4D4D28", shape=parallelogram];
    reorder -> plan [label="drain walk\n(plan → seal → execute)"];
  }

  // ═════ decode pool ═════
  subgraph cluster_decode {
    label="decode pool ×M  (emit/pipeline/decode.rs)";
    style="rounded,filled"; color="#4c4641"; fillcolor="#34302c"; fontcolor="#ECE1D7";

    decode [label="decode worker ×M\nplanned envelopes: descriptor + route attached,\nvalues resolved at planning\noracle PgPending resolve\nchunk 1024 rows / 4 MiB", fillcolor="#4D4128"];
  }

  // ═════ batcher hub ═════
  subgraph cluster_batch {
    label="InsertBatcher — single hub, per-table accumulation  (emit/pipeline/batcher.rs)";
    style="rounded,filled"; color="#4c4641"; fillcolor="#34302c"; fontcolor="#ECE1D7";

    enc  [label="TableEncoder per dest table\nColumnBuf slabs: Fixed / String /\nNullableFixed / NullableString\ncolumn-major + synthetic\n_lsn / _xid / _commit_ts / _is_deleted", fillcolor="#5D4628", shape=note];
    trip [label="flush trigger\nrow_budget 65536 · byte_budget 1 MiB\nper-table deadline (flush_timeout,\n 0 → 100 ms floor) · FlushAll", fillcolor="#5D4628", shape=diamond];
    seal [label="seal InsertBatch\nowned slabs + per_seq row counts\nFlushAll also bumps schema_epoch\n(rebuild plans post-DDL)", fillcolor="#5D4628"];

    enc -> trip -> seal;
  }

  // ═════ inserter pool ═════
  subgraph cluster_insert {
    label="inserter pool ×N  (emit/pipeline/inserter.rs)";
    style="rounded,filled"; color="#4c4641"; fillcolor="#34302c"; fontcolor="#ECE1D7";

    ins [label="inserter ×N — any idle takes any batch\nTypeAst cache per (table, epoch)\nBlockBuilder over owned slabs\nsend_query → send_data → EndOfStream\nsend_with_retry: reconnect + backoff,\ninsert_timeout 30 s; exhaustion = Fatal", fillcolor="#5D4628"];
  }

  // ═════ DDL path: barrier + applicator (own CH connection) ═════
  subgraph cluster_ddl {
    label="ordered control barriers — DDL / config / TOAST lifecycle";
    style="rounded,filled"; color="#4c4641"; fillcolor="#34302c"; fontcolor="#ECE1D7";

    fence  [label="barrier_fence\n1. wait all seqs placed\n2. batcher FlushAll (+ reply)\n3. wait all seqs durable", fillcolor="#5D4628"];
    bridge [label="type_bridge::map\nRelAttr → ResolvedColumn\n(reject type change,\n pk strips Nullable)", fillcolor="#5D4628"];
    apply  [label="apply ordered control\nCatalog → DdlApplicator\nConfig → resolver republish\nowner TRUNCATE → dest + toast mirror wipe\nToastBarrier → rewrite O−B\ntoast Dropped → durable retire enqueue", fillcolor="#5D4628"];

    fence -> apply;
    bridge -> apply [style=dashed, label="resolve types"];
  }

  // ═════ CH endpoint ═════
  subgraph cluster_ch {
    label="ClickHouse — main insert pool + DDL + TOAST store connections";
    style="rounded,filled"; color="#4c4641"; fillcolor="#34302c"; fontcolor="#ECE1D7";

    chrows [label="insert connections ×N\nclickhouse-c-rs AsyncClient\nReplacingMergeTree(_lsn, _is_deleted)\nNative rows", fillcolor="#4D4128", shape=cylinder];
    chddl  [label="DDL connection\nclickhouse-c-rs AsyncClient\nALTER / CREATE /\n DROP / TRUNCATE", fillcolor="#4D4128", shape=cylinder];
    chtoast [label="TOAST store connection\npg_toast_<relid> mirrors\nput / as-of fetch / truncate / O−B", fillcolor="#4D4128", shape=cylinder];
  }

  toastput [label="ToastResolver\nput births + tombstones before publish\nas-of fetch for pre-window values", fillcolor="#4D4D28"];
  ledger [label="toast_retires.toml\nsaved mirror retirements", fillcolor="#4D3850", shape=note];

  nulltail [label="metrics-only tail\nno ClickHouse connection\nacknowledge rows immediately", fillcolor="#4D3A28", shape=note];

  // ═════ ack collector (durability watermark) ═════
  ack [label="ack collector\nwait for every earlier commit\npublish latest durable position\nfeed manifest + source feedback", fillcolor="#4D3A28", shape=note];

  // ═════ row path ═════
  plan -> decode [color="#BF8C5F", penwidth=2, lhead=cluster_decode, label="execute_plan:\nDecodeJob\n(mpmc, bound 4M)"];
  decode  -> enc    [color="#BF8C5F", penwidth=2, lhead=cluster_batch, label="BatcherMsg::Rows\n(FIFO mpsc 256 —\nshared with FlushAll)"];
  seal    -> ins    [color="#BF8C5F", penwidth=2, lhead=cluster_insert, label="InsertBatch\n(mpmc)"];
  ins     -> chrows [color="#BF8C5F", penwidth=2, label="one complete INSERT\nper sealed batch"];
  decode  -> nulltail [color="#BF8C5F", style=dashed, constraint=false, label="metrics only"];
  reorder -> toastput [color="#BF8C5F", style=dashed, label="TOAST changes"];
  decode  -> toastput [color="#CBA85E", style=dashed, label="fetch"];
  toastput -> chtoast [color="#BF8C5F", penwidth=2];

  // ═════ DDL path ═════
  cat     -> reorder [color="#CBA85E", style=dashed, label="SchemaEvent\ndrains at commit"];
  reorder -> fence   [color="#BF8C5F", style=dashed, lhead=cluster_ddl, label="barrier xact:\nper event / TRUNCATE"];
  apply   -> chddl   [color="#BF8C5F", style=dashed, penwidth=2, label="DDL SQL\n(own connection)"];
  apply   -> chtoast [color="#BF8C5F", style=dashed, label="truncate / O−B"];
  apply   -> ledger  [color="#6E6963", style=dashed, label="toast DROP enqueue"];
  ledger  -> chtoast [color="#BF8C5F", style=dashed, label="retire after saved\nrestart point passes"];
  fence   -> trip    [style=dashed, color="#B58B86", constraint=false, label="FlushAll"];

  // ═════ ack side-channel ═════
  reorder -> ack [style=dotted, color="#b380b0", constraint=false, label="Register(seq, commit_lsn)"];
  decode  -> ack [style=dotted, color="#b380b0", constraint=false, label="Placed(seq, rows)"];
  ins     -> ack [style=dotted, color="#b380b0", label="Acked(per_seq)\nonly after EndOfStream"];
  nulltail -> ack [style=dotted, color="#b380b0", constraint=false, label="done"];
  ack     -> fence [style=dotted, color="#b380b0", constraint=false, label="wait_placed_through /\nwait_through"];

  // ═════ Legend ═════
  legend [shape=plaintext, label=<
    <TABLE BORDER="0" CELLBORDER="1" CELLSPACING="0" CELLPADDING="4">
      <TR><TD COLSPAN="2" BGCOLOR="#34302c"><B>node fill — role</B></TD></TR>
      <TR><TD BGCOLOR="#4D4128" WIDTH="28"></TD><TD>reorder / decode pool / ClickHouse</TD></TR>
      <TR><TD BGCOLOR="#4D4D28"></TD><TD>ShadowCatalog schema-event tx</TD></TR>
      <TR><TD BGCOLOR="#5D4628"></TD><TD>batcher / inserters / DdlApplicator</TD></TR>
      <TR><TD BGCOLOR="#4D3A28"></TD><TD>ack collector / null tail / manifest side-channel</TD></TR>
      <TR><TD COLSPAN="2" BGCOLOR="#34302c"><B>edge style</B></TD></TR>
      <TR><TD><FONT COLOR="#BF8C5F"><B>━━</B></FONT></TD><TD>row path (Native rows, solid)</TD></TR>
      <TR><TD><FONT COLOR="#BF8C5F"><B>┄┄</B></FONT></TD><TD>DDL path (own connection, dashed)</TD></TR>
      <TR><TD><FONT COLOR="#CBA85E"><B>┄┄</B></FONT></TD><TD>SchemaEvent off ShadowCatalog</TD></TR>
      <TR><TD><FONT COLOR="#b380b0"><B>···</B></FONT></TD><TD>ack events / barrier waits</TD></TR>
      <TR><TD COLSPAN="2" BGCOLOR="#34302c"><B>synthetic columns (every dest table)</B></TD></TR>
      <TR><TD><FONT FACE="monospace">_lsn</FONT></TD><TD>UInt64 — commit_lsn; ReplacingMergeTree dedup key</TD></TR>
      <TR><TD><FONT FACE="monospace">_xid</FONT></TD><TD>UInt32 — source xid; recovers xact boundary</TD></TR>
      <TR><TD><FONT FACE="monospace">_commit_ts</FONT></TD><TD>DateTime64(6, 'UTC'); shifted from PG epoch</TD></TR>
      <TR><TD><FONT FACE="monospace">_is_deleted</FONT></TD><TD>Bool — 1 on delete; ReplacingMergeTree is_deleted arg unless soft_delete</TD></TR>
      <TR><TD COLSPAN="2" BGCOLOR="#34302c"><B>watermark rule</B></TD></TR>
      <TR><TD COLSPAN="2">commits may finish out of order; progress advances only after every earlier commit is durable. Failed batch stops process so restart can replay it.</TD></TR>
    </TABLE>
  >];
  chrows -> legend [style=invis];
  chddl  -> legend [style=invis];
}