database inputs

The postgres and clickhouse inputs run a query on a timer. Each row becomes one message, with the column names as its fields.

// config.json — follow a table as rows are added to it
{ "type": "postgres", "connection": "local-postgres",
  "table": "readings",
  "interval_secs": 5,
  "mode": { "type": "incremental", "field": "id" },
  "max_batch": 100 }

// ...or poll a query as reference data
{ "type": "clickhouse", "connection": "local-clickhouse",
  "query": "SELECT sensor, max(value) AS peak FROM sensor_readings GROUP BY sensor",
  "interval_secs": 30,
  "mode": { "type": "snapshot" } }

The server converts each row to JSON: postgres with row_to_json, ClickHouse with JSONEachRow. A timestamp is ISO 8601, a numeric keeps its digits, and a jsonb column gives the value that it holds:

{"id": 42, "sensor": "press-3", "value": 21.5, "recorded_at": "2026-01-01T12:00:00.123456+00:00"}

a table and a query are the same thing

Give exactly one of table and query. The input puts it in a subquery. Then it adds the columns projection, the cursor condition, the order and the page limit around it.

• An incremental read of a query needs no placeholder and no ORDER BY. The query must only return the field that the input follows.
• The query must be one SELECT with no semicolon at the end. Anything that the server accepts in a subquery is valid, including WITH.
• columns is a list of the columns to select. To leave out a large column, do not put it in the list. To remove a field later in the pipeline, use a map.

the two modes

snapshot

snapshot reads the whole relation on each read, in one query. It sends every row. Use it for reference data, for example a table of recipes or thresholds that a remember transform keeps. Use it also for an aggregate that the server calculates.

A snapshot has no page limit. Use it only for a relation that fits in memory.

With an envelope, all rows of one read have the same polled_at. Use it to tell the rows of one snapshot from the rows of the previous one.

incremental

incremental follows a column that grows, for example an id or an updated_at. It reads only the rows above the highest value that it sent. This value is the watermark. The input reads in pages, in the order of the column.

• field: the column to follow. The input never reads rows where this column is null.
• start_from: newest (the default) or oldest. With newest, the first read finds the highest value and reads only rows above it. With oldest, the input reads the whole table first, page by page, and then follows it.
• lag_secs: for a timestamp column only. The input leaves rows that are newer than now() minus this many seconds for a later read. This gives a late transaction time to commit. The server refuses it on a numeric column.

Rules for the watermark:

• It is in memory. After a restart, the input starts again from start_from. Thus an incremental input is at least once across a restart. If the server checkpoints, the watermark is part of each checkpoint, and a restart continues exactly after it.
• It moves when the input sends the rows. It does not wait for the outputs. The run loop acknowledges a batch whether or not its outputs succeed (see acknowledging an input). Thus ack: on_delivery fails at build time. Under checkpointing, the watermark waits for the checkpoint, not for the acknowledgement.
• The input handles ties at a page boundary. Many rows can have the same timestamp. The input cuts a full page before its last distinct value. It reads those rows again, complete, on the next page. If all rows on a page have the same value, the input cannot cut the page. It sends the page as it is and logs a warning. Increase page_size, or follow a field with fewer ties.
• The input does not see rows that commit late. A long transaction can commit a row with a value below the watermark. Use lag_secs for this case. For every change at commit time, you need change data capture. This input does not do that.
• The input does not see deletes. It sees an update only if the update increases the cursor field. Use an updated_at column for this.

Index the field that you follow. Each read is WHERE field > $1 ORDER BY field LIMIT n. On an indexed column, this is one index lookup. On a column without an index, it is a scan of the whole table on each read.

paging, batching and the interval

• page_size: the largest number of rows in one query. The default is 1000. A full page starts the next query at once. A short page ends the read.
• interval_secs: the wait after the end of one read and before the next read. Thus a slow read never overlaps the next one. The first read occurs when the pipeline starts.
• max_batch: the largest number of rows in one batch. The default is 1. The input does not wait for a batch to fill.

performance

Set max_batch. With the default of 1, a read of 1000 rows makes 1000 passes through the run loop. With max_batch: 1000, it makes one pass.

failures

If the database is down, the read fails. The input reports the error one time and retries with backoff. The watermark does not change, so the input skips no rows. A table that does not exist yet is the same case. The input starts to read when the table exists.

sampling

To see some rows before you create the pipeline, post the input to POST /api/inputs/sample:

curl -X POST localhost:6767/api/inputs/sample \
     -H 'content-type: application/json' \
     -d '{"input": {"type": "postgres", "connection": "local-postgres",
          "table": "readings", "interval_secs": 5,
          "mode": {"type": "incremental", "field": "id"}}}'

If the input starts from the newest rows, the sample reads from the oldest rows. The notes in the reply say so. The sample does not change the pipeline. The web UI uses the same endpoint.

what is on the wire

The watermark goes to the server as text. The server casts it to the type of the column.

• Postgres returns (field)::text with each row. The next query uses ($1::text)::<type>. The input reads the type from a prepared statement.
• ClickHouse returns toString(field). The next query uses CAST({cursor:String} AS <type>). The input reads the type with DESCRIBE.

Text keeps the exact value of every type. The database outputs use the same method in the other direction.

The ClickHouse input sends two settings with each request:

• output_format_json_quote_64bit_integers=0, so an Int64 arrives as a number. By default, ClickHouse sends it as a string.
• date_time_output_format=iso, so a DateTime arrives as 2026-01-01T12:00:00Z. The postgres output refuses the default format.

postgres

Reads a postgres table, view or query at an interval and sends each row as a message.

The input runs a query each interval_secs. In snapshot mode, it reads the full relation. In incremental mode, it reads only the rows after the last read, by a column that increases. Each row becomes one JSON object with the column names as fields. Postgres makes the object with row_to_json. A timestamp is ISO 8601, a numeric keeps its digits, and a jsonb column is a nested value:

json {"id": 42, "sensor": "press-3", "value": 21.5, "recorded_at": "2026-01-01T12:00:00.123456+00:00"}

An incremental read is at-least-once across a restart. The input keeps the watermark in memory, and after a restart it starts again from start_from. An incremental read does not see a deleted row. Put an index on the column that it follows, or each read scans the full table.

field	type		description
connection	postgres connection	required	The name of the postgres connection to read through. Declare the connection in the connections file.
interval_secs	integer	required	The time between two reads, in seconds. The time starts at the end of a read. Thus, two reads never overlap. The first read occurs when the pipeline starts.
mode	snapshot | incremental	required	Whether each read returns all rows (snapshot) or only the rows after the last read (incremental).
columns	list of string	optional	The columns to read, in order. Leave it empty to read all columns. For an incremental input, the list must include field.
max_batch	integer	optional	The maximum number of rows in one batch. The default is 1. The input puts rows that it already read into a batch. It does not wait for more rows to fill a batch.
page_size	integer	optional	The maximum number of rows that one query returns. The default is 1000. When a page is full, the input reads the next page immediately. The interval starts after a page that is not full. A snapshot ignores this field and reads all rows in one query.
query	string	optional	A SELECT to read in place of a table. The input puts the query in a subquery. It adds the cursor condition, the ORDER BY and the page limit outside the subquery. Thus, the query needs no placeholder and no ORDER BY. Write one statement with no semicolon at the end. You can use all SQL that the server accepts in a subquery, for example WITH.
table	string	optional	The table or view to read, as name or schema.name. Give exactly one of table and query.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

mode — type: "snapshot"

This component takes no configuration.

mode — type: "incremental"

field	type		description
field	string	required	The column that the input follows. The watermark is the highest value that the input sent. Each read asks for the rows above the watermark. The input does not read rows where the column is null.
lag_secs	integer	optional	For a timestamp column: the time to stay behind the current time, in seconds. A later read gets the rows that are less than this time before now(). This gives late transactions time to commit. Do not use it with a numeric column. The server refuses the query.
start_from	oldest | newest	optional	Where the first read starts. With newest, the input reads only the rows added after the pipeline started. With oldest, it reads all rows first and then follows the table. The default is newest.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
connection	The name of the connection that the input read from.
polled_at	The start time of the read that returned the row, as RFC 3339. All rows of one read have the same value. Use it to tell one snapshot from the next. The time comes from the clock of kayak, not from the database server.

clickhouse

Reads a ClickHouse table, view or query at an interval and sends each row as a message. It uses the HTTP interface of ClickHouse.

ClickHouse sends the rows as JSONEachRow. A DateTime is ISO 8601, an Int64 is a number, and a Decimal keeps its digits. The modes, the watermark and the limits of an incremental read are the same as on the postgres input.

field	type		description
connection	clickhouse connection	required	The name of the clickhouse connection to read through. Declare the connection in the connections file.
interval_secs	integer	required	The time between two reads, in seconds. The time starts at the end of a read. Thus, two reads never overlap. The first read occurs when the pipeline starts.
mode	snapshot | incremental	required	Whether each read returns all rows (snapshot) or only the rows after the last read (incremental).
columns	list of string	optional	The columns to read, in order. Leave it empty to read all columns. For an incremental input, the list must include field.
max_batch	integer	optional	The maximum number of rows in one batch. The default is 1. The input puts rows that it already read into a batch. It does not wait for more rows to fill a batch.
page_size	integer	optional	The maximum number of rows that one query returns. The default is 1000. When a page is full, the input reads the next page immediately. The interval starts after a page that is not full. A snapshot ignores this field and reads all rows in one query.
query	string	optional	A SELECT to read in place of a table. The input puts the query in a subquery. It adds the cursor condition, the ORDER BY and the page limit outside the subquery. Thus, the query needs no placeholder and no ORDER BY. Write one statement with no semicolon at the end. You can use all SQL that the server accepts in a subquery, for example WITH.
table	string	optional	The table or view to read, as name or schema.name. Give exactly one of table and query.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

mode — type: "snapshot"

This component takes no configuration.

mode — type: "incremental"

field	type		description
field	string	required	The column that the input follows. The watermark is the highest value that the input sent. Each read asks for the rows above the watermark. The input does not read rows where the column is null.
lag_secs	integer	optional	For a timestamp column: the time to stay behind the current time, in seconds. A later read gets the rows that are less than this time before now(). This gives late transactions time to commit. Do not use it with a numeric column. The server refuses the query.
start_from	oldest | newest	optional	Where the first read starts. With newest, the input reads only the rows added after the pipeline started. With oldest, it reads all rows first and then follows the table. The default is newest.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
connection	The name of the connection that the input read from.
polled_at	The start time of the read that returned the row, as RFC 3339. All rows of one read have the same value. Use it to tell one snapshot from the next. The time comes from the clock of kayak, not from the database server.