TimescaleDB Sink Node

The timescaledbsink node writes records into a TimescaleDB hypertable over JDBC. It coerces a configured time column to a TIMESTAMPTZ, optionally promotes the target table to a hypertable on startup, and writes rows in transactional batches using either INSERT or UPSERT.

TimescaleDB is a PostgreSQL extension, so the connection is an ordinary PostgreSQL JDBC URL — a jdbc:postgresql: prefix is enforced at validation time.

Key Features

• Flexible time input: the time column accepts a numeric epoch (in a configurable unit) or an ISO-8601 string, and is converted to a UTC TIMESTAMPTZ
• Automatic hypertable creation: on startup the node can run create_hypertable(..., if_not_exists => TRUE) so a plain table becomes a hypertable without manual DDL
• INSERT and UPSERT modes: append rows, or insert-or-update on a conflict target you choose
• Transactional batching: rows in a batch commit atomically — a failure rolls the whole batch back
• Schema qualification: target a table in a specific schema, with identifiers safely quoted
• Fail-fast on missing time: a record without the time column is rejected as a single-record error rather than corrupting the batch

Configuration

Field	Type	Required	Default	Description
jdbcUrl	String	Yes	—	PostgreSQL JDBC URL, e.g. jdbc:postgresql://host:5432/metrics. Must start with jdbc:postgresql:
credentialId	String	No	—	ID of a username/password credential in jobContext.otherProperties. Omit to connect without credentials
tableName	String	Yes	—	Target table. It must already exist — the node does not create it
schemaName	String	No	—	Schema qualifying the table (e.g. public). Omit to use the database default
timeColumn	String	Yes	—	Record field / column used as the hypertable time dimension. Coerced to TIMESTAMPTZ
timeUnit	String	No	MILLISECONDS	Unit for numeric time values: SECONDS, MILLISECONDS, MICROSECONDS, NANOSECONDS. Ignored for ISO-8601 strings
createHypertable	boolean	No	true	Run create_hypertable(..., if_not_exists => TRUE) on startup. Set false if the hypertable exists or the role lacks DDL rights
writeMode	String	No	INSERT	INSERT or UPSERT
upsertKeyColumns	String	No	—	Comma-separated conflict-target columns. Required when writeMode is UPSERT
batchSize	int	No	1000	Rows per batch / transaction

Time Column Handling

Every record must carry the field named by timeColumn; a record missing it (or holding null) is rejected as a single-record error and the rest of the batch proceeds.

The value is converted to a UTC OffsetDateTime based on its type:

Value type	Interpretation
Number	Epoch offset in timeUnit — e.g. 1753280531000 with MILLISECONDS
String	ISO-8601 timestamp. Parsed with an offset if present (2026-07-23T14:02:11Z), otherwise as an instant. timeUnit is ignored
Anything else	Rejected — the record errors with a type message

Hypertable Creation

With createHypertable left at true, the node executes the following once at startup, before any writes:

SELECT create_hypertable('"schema"."table"', 'time_column', if_not_exists => TRUE)

This is idempotent — an existing hypertable is left alone. Two cases call for setting it to false:

• The database role lacks DDL privileges, so the call would fail even though the hypertable already exists.
• You created the hypertable yourself with non-default settings (chunk interval, partitioning column, compression policy) and don't want an unconfigured call in the startup path.

The table itself is never created. Create it, with the column types you want, before running the pipeline.

Write Modes

INSERT

Each record becomes a new row. A unique or primary key violation fails the batch and rolls it back.

config:
  jdbcUrl: "jdbc:postgresql://localhost:5432/metrics"
  credentialId: "pg-cred"
  tableName: "device_metrics"
  timeColumn: "ts"
  writeMode: "INSERT"

UPSERT

On conflict with the columns listed in upsertKeyColumns, the existing row is updated. upsertKeyColumns is required in this mode and must match a unique constraint or index on the table.

On a hypertable, any unique index must include the partitioning (time) column, so the conflict target is typically the time column plus an identity column:

config:
  jdbcUrl: "jdbc:postgresql://localhost:5432/metrics"
  credentialId: "pg-cred"
  tableName: "device_metrics"
  timeColumn: "ts"
  writeMode: "UPSERT"
  upsertKeyColumns: "device_id,ts"

DAG Example

jobContext:
  otherProperties:
    pg-cred:
      username: zephflow
      password: supersecret
  metricTags: {}
  dlqConfig:

dag:
  - id: "source"
    commandName: "stdin"
    config:
      encodingType: "JSON_OBJECT_LINE"
    outputs:
      - "sink"

  - id: "sink"
    commandName: "timescaledbsink"
    config:
      jdbcUrl: "jdbc:postgresql://timescale:5432/metrics"
      credentialId: "pg-cred"
      schemaName: "public"
      tableName: "device_metrics"
      timeColumn: "ts"
      timeUnit: "MILLISECONDS"
      createHypertable: true
      writeMode: "INSERT"
      batchSize: 1000

Matching table definition:

CREATE TABLE device_metrics (
  ts          TIMESTAMPTZ      NOT NULL,
  device_id   TEXT             NOT NULL,
  temperature DOUBLE PRECISION,
  humidity    DOUBLE PRECISION
);

The node promotes this to a hypertable on ts at startup.

Related Nodes

• jdbcsink: Write to any JDBC database without hypertable handling
• influxdbsink: Alternative time-series destination using the InfluxDB v2 write API
• jdbcsource: Read records from a JDBC database