feat: add schema DSL support for Kafka table engine

[armitageee]

Summary

Add first-class schema DSL / migration support for ClickHouse integration engines, starting with ENGINE = Kafka (and ideally Distributed + the common Kafka → MV → storage pattern).

Today these objects can only be managed via chkit generate --empty + hand-written SQL, which breaks the “schema as TypeScript” workflow for a very common production use case.

Motivation

We manage multiple ClickHouse instances with chkit (schema TS → generate → SQL migrations → CI migrate).

Ingestion from Kafka is a core path for us, e.g.:

Kafka topic
  → ENGINE = Kafka          (queue)
  → MATERIALIZED VIEW
  → ENGINE = Distributed    (optional)
  → MergeTree / *MergeTree  (storage)

MergeTree tables and ordinary MVs fit chkit well. Kafka (and Distributed) do not, so we end up with a split workflow:

• schema objects → TypeScript + generated migrations
• Kafka pipeline → manual empty migrations

That makes review, drift, and pull inconsistent across the same database.

Current behavior (chkit 0.1.2-beta.5)

From docs + source:

1. ClickHouse compatibility focuses on MergeTree-family engines for vanilla ClickHouse.
2. table() always models MergeTree-like structure (primaryKey / orderBy required in the type model).
3. SQL emission always appends PRIMARY KEY (...) and ORDER BY (...), which is invalid / meaningless for ENGINE = Kafka.
4. settings rendering does not appear to quote string values, so Kafka settings like kafka_broker_list = 'host:9092' cannot be expressed safely via the DSL.
5. There is no model for CREATE TABLE ... AS ... ENGINE = Distributed(...).
6. No docs/tests/issues currently mention Kafka support (that we could find).

chkit generate --empty works as an escape hatch, but loses schema-as-code benefits (diff from TS, pull round-trip, drift against desired state).

Proposed support

Must-have (Kafka)

• DSL for Kafka engine tables without requiring PRIMARY KEY / ORDER BY.
• Correct DDL generation, e.g.:

CREATE TABLE IF NOT EXISTS db.events_queue
(
  ...
)
ENGINE = Kafka
SETTINGS
  kafka_broker_list = '...',
  kafka_topic_list = '...',
  kafka_group_name = '...',
  kafka_format = 'JSONEachRow',
  kafka_num_consumers = 1,
  kafka_auto_offset_reset = 'earliest',
  input_format_skip_unknown_fields = 1;

• Proper quoting/escaping for string settings.
• chkit pull can introspect Kafka tables into the DSL (or at least not silently mis-model them as MergeTree).
• chkit drift / check understand Kafka settings as part of desired state.

Nice-to-have

• Distributed table support (ENGINE = Distributed(cluster, db, local_table, sharding_key) and/or CREATE TABLE ... AS local ENGINE = Distributed(...)).
• Documented pattern: Kafka queue + MV TO storage table (MergeTree / Distributed).
• Guidance on what is in/out of scope for drift (e.g. consumer offsets are runtime state, not schema).

Example DSL sketch (illustrative)

import { schema, table, materializedView } from '@chkit/core'

const events_queue = table({
  database: 'analytics',
  name: 'events_queue',
  engine: 'Kafka',
  columns: [
    { name: 'id', type: 'String' },
    { name: 'evtTime', type: 'DateTime64(3)' },
    // ...
  ],
  // no primaryKey / orderBy for Kafka
  settings: {
    kafka_broker_list: 'kafka01:9092,kafka02:9092',
    kafka_topic_list: 'my_topic',
    kafka_group_name: 'chkit_analytics_my_topic',
    kafka_format: 'JSONEachRow',
    kafka_num_consumers: 1,
    kafka_auto_offset_reset: 'earliest',
    input_format_skip_unknown_fields: 1,
  },
})

const events_local = table({
  database: 'analytics',
  name: 'events_local',
  engine: 'MergeTree()',
  columns: [/* ... */],
  primaryKey: ['evtTime', 'id'],
  orderBy: ['evtTime', 'id'],
  partitionBy: 'toYYYYMM(evtTime)',
})

const events_mv = materializedView({
  database: 'analytics',
  name: 'events_mv',
  to: { database: 'analytics', name: 'events_local' },
  as: 'SELECT * FROM analytics.events_queue',
})

export default schema(events_queue, events_local, events_mv)

Alternatives considered

• Keep using generate --empty for Kafka DDL (works, but fragments GitOps).
• Store Kafka DDL only in external scripts (loses chkit journal / CI unify).
• Abuse engine: 'Kafka' today with dummy orderBy — produces incorrect SQL and bad drift.

Environment

• chkit: 0.1.2-beta.4 / 0.1.2-beta.5
• ClickHouse: 25.3
• Target: self-hosted single-node / small multi-instance GitOps setup (not only ObsessionDB)

Ask

Is Kafka (and more broadly integration engines) on the roadmap?
If yes, we’d be happy to refine the DSL shape or contribute tests based on a real Kafka → MV → MergeTree pipeline.

Thanks!