add ClickHouse DLQ support

design-docs/2026-08-dlq.md

@@ -0,0 +1,118 @@
+# DLQ for `ClickhouseProducer` — retrospective design doc
+
+Status: retrospective — written after an initial implementation, to be revised and
+used to align the code with the agreed design.
+
+## Problem
+
+`ClickhouseProducer.produce_arrow` inserts an Arrow batch into ClickHouse via
+`clickhouse_connect`, retried up to 3 times (`_insert_with_retry`) to absorb
+transient failures (network blips, brief unavailability). If all retries fail,
+the exception previously propagated straight to the caller — a single
+malformed or schema-incompatible row anywhere in the batch would fail the
+*entire* batch, and by extension typically crash or stall the consuming node,
+since a `Producer` failure has no other defined outcome.
+
+In a Kafka → ClickHouse pipeline, batches are read from a topic and can contain
+many independent events. One event with a bad value (wrong type, oversized
+string, invalid encoding, a `NULL` in a non-nullable column, etc.) shouldn't be
+able to:
+
+1. Block delivery of every other, valid row in the same batch, or
+2. Take down the node, which would stall consumption of all *subsequent*
+   batches too, since Kafka offsets can't advance past an unprocessed batch.
+
+We need a way to isolate and quarantine the specific row(s) causing an insert
+to fail, while still delivering everything else, and without losing the
+rejected data — it should be inspectable and potentially replayed later.
+
+## Non-goals
+
+* Retrying DLQ'd rows automatically. Once a row lands in the DLQ it's a human/
+  offline concern.
+* Deduplicating or rate-limiting DLQ writes.
+* A DLQ for `KafkaProducer`/`KafkaConsumer` — this doc covers only
+  `ClickhouseProducer`, though the mechanism is intentionally generic (see
+  below) so it could extend to other producer types later.
+
+## Solution
+
+### Isolating bad rows: recursive bisection
+
+`ClickhouseProducer.produce_arrow` still attempts one `_insert_with_retry` for
+the whole incoming batch. If that fails (all 3 attempts exhausted) and a
+`dlq_producer` is configured, it falls back to `_insert_with_dlq_fallback`,
+which bisects the batch recursively:
+
+* Batch of 1 row that fails → that row is the culprit. Wrap it as a DLQ
+  record (see schema below) and send it to `dlq_producer`.
+* Batch of N>1 rows that fails → split it into chunks of
+  `max(1, N // split_factor)` rows and recurse on each chunk independently.
+
+This narrows down to exactly the failing rows without needing to inspect the
+ClickHouse error message, at the cost of extra insert round-trips proportional
+to how deep the recursion goes. `split_factor` (default 10, configurable via
+`ClickHouseOutputSettings.dlq_split_factor`) trades off round-trip count
+against how large a "wasted" chunk is when one bad row poisons a chunk insert:
+a higher split factor means fewer rows re-sent per recursion level but more
+levels of recursion for a large batch with a single bad row.
+
+If no `dlq_producer` is configured, a failed insert still raises after retries
+are exhausted — this preserves today's fail-loud behavior for nodes that
+haven't opted into a DLQ, rather than silently dropping data.
+
+### DLQ record schema
+
+Each rejected row is wrapped into a single-row Arrow table with a fixed
+schema before being handed to `dlq_producer.produce_arrow`:
+
+| column        | type   | contents                                                        |
+|---------------|--------|------------------------------------------------------------------|
+| `producer`    | string | JSON: `{"ch_url", "ch_database", "ch_table"}` — where it came from |
+| `data`        | string | JSON-encoded original row                                        |
+| `err_message` | string | `str(exception)` from the failed insert                          |
+| `time`        | datetime (UTC) | when the row was quarantined                              |
+
+This schema is independent of the source table's schema, so a single DLQ
+sink (topic or table) can receive rejected rows from multiple source tables/
+producers — `producer` is what disambiguates provenance, and `data` carries
+the arbitrary original row as JSON rather than requiring a matching column
+layout.
+
+### `dlq_producer` is just a `Producer`
+
+The DLQ target is typed as `Producer` (`tkati_core.producer.Producer`), the
+same interface used for regular node outputs — not a dedicated DLQ type. In
+practice this means:
+
+* A `ClickhouseProducer` can be its own or another `ClickhouseProducer`'s
+  `dlq_producer` (writing quarantined rows to a `..._dlq` table), exercised by
+  `test_ch_producer_as_dlq_for_another_ch_producer`.
+* A `KafkaProducer` could equally be used to push rejected rows to a DLQ
+  topic instead of a table, with no changes to `ClickhouseProducer`.
+
+`produce_arrow` calls `self._dlq_producer.flush()` once after the recursive
+fallback completes (not once per DLQ'd row), so buffering producers (e.g. a
+future Kafka DLQ) don't need a flush per row.
+
+### Wiring
+
+* `ClickHouseOutputSettings.dlq_split_factor: int = 10` — the only new setting
+  surfaced to config; the DLQ *target* itself is wired by the caller passing a
+  `dlq_producer` to `build_producer`/`ClickhouseProducer.__init__`, not by
+  settings. There's currently no config-driven way to say "this node's output
+  gets a DLQ pointed at table X" — that's assembled in application code.
+* `build_producer` no longer takes a `split_factor` kwarg directly; it derives
+  it from `settings.dlq_split_factor` when settings is a
+  `ClickHouseOutputSettings`.
+
+## Open questions for the revised design
+
+* Should DLQ wiring become config-driven (e.g. an `[output.dlq]` settings
+  block resolved by `build_producer`) rather than assembled by hand in
+  application code?
+* Is unbounded recursive bisection acceptable, or should there be a cap
+  (e.g. give up and DLQ the whole remaining chunk after N levels) to bound
+  worst-case round-trips against ClickHouse when many rows in a batch are bad?
+* `KafkaProducer` has no DLQ fallback path today — is that in scope for this
+  effort or a later one?

packages/tkati-core/CHANGELOG.md

@@ -1,3 +1,9 @@
+# WIP 0.8.4
+
+* DLQ sinks now receive rejected rows using the DLQ schema
+  `producer`/`data`/`err_message`/`time`; `producer` contains the source
+  ClickHouse URL, database, and table as JSON
+
 # 0.8.3
 
 * `[input.config]` and `[output.config]` pass arbitrary librdkafka properties

packages/tkati-core/tests/test_ch_producer.py

@@ -1,3 +1,4 @@
+import json
 from unittest.mock import MagicMock, patch
 
 import clickhouse_connect.driver.exceptions as ch_exc
@@ -72,6 +73,30 @@ def test_a_data_error_is_not_retried() -> None:
     sleep.assert_not_called()
 
 
+def _set_ch_client_metadata(ch_client: MagicMock) -> None:
+    ch_client.uri = "http://clickhouse:8123"
+    ch_client.database = "default"
+
+
+def _assert_dlq_row(
+    sent: pa.Table,
+    *,
+    ch_table: str = "traffic_event",
+    err_message: str,
+    data: dict,
+) -> None:
+    assert len(sent) == 1
+    assert sent.column_names == ["producer", "data", "err_message", "time"]
+    row = sent.to_pylist()[0]
+    producer_info = json.loads(row["producer"])
+    assert producer_info["ch_table"] == ch_table
+    assert producer_info["ch_database"] == "default"
+    assert producer_info["ch_url"] == "http://clickhouse:8123"
+    assert json.loads(row["data"]) == data
+    assert row["err_message"] == err_message
+    assert row["time"] is not None
+
+
 def test_insert_retry_on_non_data_error() -> None:
     """CH unreachable twice then back: insert_arrow called 3x."""
     ch_client = MagicMock()
@@ -131,6 +156,7 @@ def insert_side_effect(table, arrow_table):
 def test_dlq_single_bad_row() -> None:
     """A single row CH always rejects → written to DLQ once, without retries."""
     ch_client = MagicMock()
+    _set_ch_client_metadata(ch_client)
     dlq_producer = MagicMock()
 
     bad_row = _make_arrow_table(1)
@@ -149,7 +175,7 @@ def test_dlq_single_bad_row() -> None:
     sleep.assert_not_called()
     dlq_producer.produce_arrow.assert_called_once()
     sent = dlq_producer.produce_arrow.call_args[0][0]
-    assert len(sent) == 1
+    _assert_dlq_row(sent, err_message=str(_ch_data_error()), data={"uid": "uid-0", "traffic_in": 100})
 
 
 def test_data_error_without_dlq_raises() -> None:
@@ -167,6 +193,7 @@ def test_recursive_descent() -> None:
     """4-row batch rejected; with split_factor=2, recursion finds and DLQs exactly
     the one bad row, and the three good rows still reach ClickHouse."""
     ch_client = MagicMock()
+    _set_ch_client_metadata(ch_client)
     dlq_producer = MagicMock()
 
     def insert_side_effect(table, arrow_table):
@@ -188,8 +215,11 @@ def insert_side_effect(table, arrow_table):
 
     dlq_producer.produce_arrow.assert_called_once()
     sent = dlq_producer.produce_arrow.call_args[0][0]
-    assert len(sent) == 1
-    assert sent.column("uid")[0].as_py() == "uid-2"
+    _assert_dlq_row(
+        sent,
+        err_message=str(_ch_data_error()),
+        data={"uid": "uid-2", "traffic_in": 102},
+    )
 
     inserted = set()
     for call in ch_client.insert_arrow.call_args_list:
@@ -203,6 +233,7 @@ def test_data_error_splits_without_sleeping() -> None:
     """Regression test for audit F3: isolating a bad row used to cost 2s of sleep
     at every level of the descent. It must now cost round-trips only."""
     ch_client = MagicMock()
+    _set_ch_client_metadata(ch_client)
     dlq_producer = MagicMock()
 
     def insert_side_effect(table, arrow_table):
@@ -220,7 +251,11 @@ def insert_side_effect(table, arrow_table):
 
     sleep.assert_not_called()
     dlq_producer.produce_arrow.assert_called_once()
-    assert dlq_producer.produce_arrow.call_args[0][0].column("uid")[0].as_py() == "uid-7"
+    _assert_dlq_row(
+        dlq_producer.produce_arrow.call_args[0][0],
+        err_message=str(_ch_data_error()),
+        data={"uid": "uid-7", "traffic_in": 107},
+    )
     dlq_producer.flush.assert_called_once()
 
 
@@ -271,6 +306,7 @@ def test_non_data_error_mid_descent_aborts_the_whole_descent() -> None:
     the DLQ stay sent, and are sent again once the rewound batch is re-read —
     the DLQ is at-least-once, like the output."""
     ch_client = MagicMock()
+    _set_ch_client_metadata(ch_client)
     dlq_producer = MagicMock()
 
     def insert_side_effect(table, arrow_table):
@@ -292,7 +328,11 @@ def insert_side_effect(table, arrow_table):
     # uid-0 was isolated and filed before CH went away; uid-2 and uid-3 were
     # never reached as singles.
     assert dlq_producer.produce_arrow.call_count == 1
-    assert dlq_producer.produce_arrow.call_args[0][0].column("uid")[0].as_py() == "uid-0"
+    _assert_dlq_row(
+        dlq_producer.produce_arrow.call_args[0][0],
+        err_message=str(_ch_data_error()),
+        data={"uid": "uid-0", "traffic_in": 100},
+    )
     dlq_producer.flush.assert_not_called()
 
     # uid-1 still gets its 3 attempts at that node: a blip mid-descent may pass,
@@ -322,6 +362,7 @@ def test_ch_producer_success_no_dlq_call() -> None:
 def test_ch_producer_failure_with_dlq() -> None:
     """produce_arrow hits a bad row: recursive fallback runs and DLQ is flushed."""
     ch_client = MagicMock()
+    _set_ch_client_metadata(ch_client)
     dlq_producer = MagicMock()
     arrow_table = _make_arrow_table(1)
     ch_client.insert_arrow.side_effect = _ch_data_error()
@@ -347,21 +388,26 @@ def test_ch_producer_failure_without_dlq_raises() -> None:
 
 
 def test_ch_producer_as_dlq_for_another_ch_producer() -> None:
-    """A ClickhouseProducer can itself be used as the dlq_producer for another ClickhouseProducer."""
+    """A ClickhouseProducer used as DLQ receives a ClickHouse DLQ row."""
     primary_ch_client = MagicMock()
     dlq_ch_client = MagicMock()
     arrow_table = _make_arrow_table(1)
 
+    _set_ch_client_metadata(primary_ch_client)
     primary_ch_client.insert_arrow.side_effect = _ch_data_error()
 
     dlq_producer = ClickhouseProducer(ch_client=dlq_ch_client, table="traffic_event_dlq")
     producer = ClickhouseProducer(
-        ch_client=primary_ch_client, table="traffic_event", dlq_producer=dlq_producer
+        ch_client=primary_ch_client,
+        table="traffic_event",
+        dlq_producer=dlq_producer,
     )
 
     producer.produce_arrow(arrow_table)
 
     dlq_ch_client.insert_arrow.assert_called_once()
+    sent = dlq_ch_client.insert_arrow.call_args.kwargs["arrow_table"]
+    _assert_dlq_row(sent, err_message=str(_ch_data_error()), data={"uid": "uid-0", "traffic_in": 100})
 
 
 def test_ch_producer_flush_is_noop() -> None:
@@ -418,6 +464,7 @@ def test_ch_producer_does_not_pass_stats_to_its_dlq() -> None:
     """The DLQ fallback runs inside the primary's deliver block. Handing the
     DLQ the same stats would count that time twice."""
     ch_client = MagicMock()
+    _set_ch_client_metadata(ch_client)
     ch_client.insert_arrow.side_effect = _ch_data_error()
     dlq_producer = MagicMock()
     stats = LoopStats(phases=PRODUCER_PHASES)

packages/tkati-core/tkati_core/clickhouse/producer.py

@@ -1,6 +1,9 @@
+from datetime import UTC, datetime
+
 import clickhouse_connect as ch
 import clickhouse_connect.driver as ch_driver
 import clickhouse_connect.driver.exceptions as ch_exc
+import orjson
 import pyarrow as pa
 from loguru import logger
 from tenacity import (
@@ -114,6 +117,30 @@ def _table_slices(table: pa.Table, chunk_size: int) -> list[pa.Table]:
     return [table.slice(i, chunk_size) for i in range(0, len(table), chunk_size)]
 
 
+def _make_dlq_table(
+    ch_client: ch_driver.Client,
+    ch_table: str,
+    row_table: pa.Table,
+    err: Exception,
+) -> pa.Table:
+    return pa.Table.from_pylist(
+        [
+            {
+                "producer": orjson.dumps(
+                    {
+                        "ch_url": ch_client.uri,
+                        "ch_database": ch_client.database,
+                        "ch_table": ch_table,
+                    }
+                ).decode(),
+                "data": orjson.dumps(row_table.to_pylist()[0]).decode(),
+                "err_message": str(err),
+                "time": datetime.now(UTC),
+            }
+        ]
+    )
+
+
 def _insert_with_dlq_fallback(
     table: pa.Table,
     ch_client: ch_driver.Client,
@@ -139,7 +166,14 @@ def _insert_with_dlq_fallback(
             logger.error(
                 f"Single row rejected by ClickHouse ({_describe(err)}), sending to DLQ"
             )
-            dlq_producer.produce_arrow(table)
+            dlq_producer.produce_arrow(
+                _make_dlq_table(
+                    ch_client,
+                    ch_table,
+                    table,
+                    err,
+                )
+            )
             return
         chunk_size = max(1, len(table) // split_factor)
         logger.warning(

packages/tkati-node-el/tests/test_node_el.py

@@ -138,7 +138,8 @@ def test_ch_producer_isolates_a_bad_row_against_real_clickhouse(
 
     mock_dlq_producer.produce_arrow.assert_called_once()
     sent = mock_dlq_producer.produce_arrow.call_args[0][0]
-    assert sent.column("uid").to_pylist() == ["not-a-uuid"]
+    assert sent.column_names == ["producer", "data", "err_message", "time"]
+    assert orjson.loads(sent.column("data")[0].as_py())["uid"] == "not-a-uuid"
 
 
 def test_every_batch_is_sent_unchanged() -> None: