Skip to content

Commit cf339b4

Browse files
committed
docs(streaming): one --topic vocabulary for kafka and iggy, and how continuous behaves
--topic serves every log engine: an entry carries the engine's hierarchy as a path (kafka 'orders', iggy 'events/orders'), so no family-specific flag exists. The help and the skill say so, and the skill gains the streaming facts an agent needs: tail-by-batch on --every, a committed position per topic, keyed upsert on partition+offset, a loud failure at a retention gap, and start=latest as a continuous-only seed. Describes dlthubworker #289 and #290; merge after they deploy.
1 parent 8492fa4 commit cf339b4

2 files changed

Lines changed: 28 additions & 12 deletions

File tree

‎skills/hotdata/SKILL.md‎

Lines changed: 17 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -321,7 +321,7 @@ hotdata ingest sources add --family sql --config @source.json --display-name "pr
321321
# "credentials": {"username": …, "password": …}}
322322
# --config also accepts a bare config object, @- (stdin), or inline JSON.
323323
# --credentials takes the secret half separately. Keep secrets out of argv.
324-
# Families: sql, filesystem (buckets), kafka, iceberg, delta, ducklake, rest.
324+
# Families: sql, filesystem (buckets), kafka, iggy, iceberg, delta, ducklake, rest.
325325
# The fields each half takes: hotdata ingest sources fields <family>.
326326
# Two config fields have a flag of their own, which BUILDS that JSON:
327327
hotdata ingest sources add --family filesystem --bucket-url s3://events-prod
@@ -346,7 +346,8 @@ hotdata ingest create --datasource-id ds_01J --type one-time \
346346
# selector.json is family-specific (what subset to read) — its fields, and the
347347
# write modes this family accepts: hotdata ingest sources fields <family>.
348348
# destination.json is {"database_id", "schema", "table", "write_mode"} —
349-
# write_mode: replace | upsert (upsert needs a continuous bucket ingest).
349+
# write_mode: replace | upsert (upsert needs a continuous ingest of a family
350+
# that stamps a row key: filesystem, kafka, iggy).
350351
# Selector and destination are both IMMUTABLE after creation.
351352
# CREATE STARTS NOTHING, for every type. It returns no run id, and
352353
# `ingest logs <id>` is EMPTY until the scheduler claims the ingest — normal,
@@ -369,13 +370,21 @@ hotdata ingest create --source "prod postgres" --table orders --schema public \
369370
# --schema <name> source schema (sql)
370371
# --format csv|jsonl|parquet, --glob "**/*.parquet" (bucket sources)
371372
# --record-shape otel_traces|mqtt_observations (bucket sources)
372-
# --topic <name> Kafka topic, REPEATABLE — topics live on the ingest,
373-
# not the datasource, which is the cluster (kafka)
373+
# --topic <name> log topic, REPEATABLE, one flag for every log engine
374+
# (kafka, iggy). Topics live on the ingest, not the
375+
# datasource, which is the cluster/server. An entry
376+
# carries the engine's hierarchy as a path: kafka has
377+
# one level (--topic orders), iggy two
378+
# (--topic events/orders). The table is named from the
379+
# whole path (events_orders).
374380
# --table-path <path> ONE Delta table under the datasource root, e.g.
375381
# warehouse/orders — name it with --dest-table (delta)
376382
# --all everything the datasource exposes (buckets need --format)
377383
# --limit N stop after N source rows
378-
# --stream shorthand for --type continuous (still needs --every)
384+
# --stream shorthand for --type continuous (still needs --every).
385+
# Offered only where `hotdata ingest sources fields
386+
# <family>` says `continuous: yes` — today filesystem,
387+
# derived, kafka, iggy; the server refuses it elsewhere.
379388
# Destination flags instead of --destination:
380389
# --database-id (required) --dest-schema (default public) --write-mode (default replace)
381390
# --dest-table <name> for sources that land ONE table: buckets, delta,
@@ -414,6 +423,9 @@ hotdata ingest run <run-id> # exits 0 succeeded / 1 failed|cancelled /
414423
```
415424

416425
Agent tips:
426+
- **Streaming (kafka, iggy) is tail-by-batch, not a live subscription.** A continuous ingest runs on its `--every` interval; each run resumes from the committed position (per topic: partition → last offset read), loads only what is new, and commits the new position after a clean load. Rows carry the partition and offset as a key (`_kafka_partition`/`_kafka_offset` on kafka, `_source_partition`/`_source_offset` on iggy), so a re-read upserts rather than duplicates. Latency is one interval plus job start-up. A run that finds nothing new succeeds as `no new messages`.
427+
- **Retention gaps fail loudly.** If the source has expired messages past the committed offset, the run **fails** naming the partition and the count, and the position is not advanced — the ingest stops progressing rather than skipping data. Recreating the ingest is how to accept the loss; lengthening retention or shortening `--every` is how to avoid it.
428+
- **`start: latest` is a continuous-only selector field (kafka).** On the first run it seeds the position at the head without reading; the bounded one-time/scheduled reader refuses it.
417429
- **There is no `trigger-import` / run-now verb, by design.** Nothing you can call starts a run — the scheduler dispatches every one. A one-time ingest is created *due*, so it is claimed on the next tick; scheduled/continuous ones are claimed on their schedule, and each run recovers from the last committed state. To make the next scheduled run happen now: `hotdata ingest schedule <ingest-id> --next now`. To load again from scratch: create another one-time ingest.
418430
- **An empty run list right after `ingest create` is normal.** The scheduler has not claimed the ingest yet. Poll `ingest logs <ingest-id>` until a run appears rather than treating the gap as a failed create — re-creating the ingest here is how you end up with two loads into one table.
419431
- **`pause` means both halves** — stop the current run *and* stop future dispatch. `resume` is its inverse and is deliberately not a trigger.

‎src/commands/ingest.rs‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -163,13 +163,17 @@ pub enum IngestCommands {
163163
#[arg(long = "table-path", conflicts_with_all = ["tables", "topics", "format", "glob"])]
164164
table_path: Option<String>,
165165

166-
/// Kafka topic to read, repeatable (Kafka sources).
166+
/// Topic to read, repeatable (Kafka sources, and Iggy sources as
167+
/// `stream/topic`).
167168
///
168-
/// The datasource is the CLUSTER — one credential opens it and each
169-
/// ingest picks its own topics — which is why topics are named here
170-
/// and not on the datasource. Each topic lands in a destination table
171-
/// of its own name; put them under a common one with
172-
/// --dest-table-prefix.
169+
/// One flag for every log engine. The datasource is the CLUSTER or
170+
/// SERVER — one credential opens it and each ingest picks its own
171+
/// topics — which is why topics are named here and not on the
172+
/// datasource. An entry carries the engine's own hierarchy as a
173+
/// path: Kafka has one level (`--topic orders`), Iggy two
174+
/// (`--topic events/orders`). Each topic lands in a destination
175+
/// table named from the whole path (`events_orders`); put them under
176+
/// a common prefix with --dest-table-prefix.
173177
#[arg(long = "topic", conflicts_with_all = ["tables", "format", "glob"])]
174178
topics: Vec<String>,
175179

@@ -835,7 +839,7 @@ fn build_selector(plan: &CreatePlan) -> Result<(serde_json::Value, Option<String
835839
if selector.is_empty() {
836840
return Err(
837841
"nothing to read — pass --table <name> (SQL, Iceberg, DuckLake), --table-path \
838-
<path> (Delta), --topic <name> (Kafka), --format with an optional --glob \
842+
<path> (Delta), --topic <name> (Kafka; Iggy as stream/topic), --format with an optional --glob \
839843
(buckets), --sql, --raw-sql, --all, or the whole --selector as JSON. A REST \
840844
source is --selector only: its resources carry endpoints, not just names \
841845
('hotdata ingest sources fields rest')"

0 commit comments

Comments
 (0)