JDBC catalog

Connect RisingWave to a JDBC-compliant catalog for Apache Iceberg.

A JDBC catalog stores Iceberg table metadata in a relational database like PostgreSQL. You can use a JDBC catalog with RisingWave when creating an Iceberg SOURCE, SINK, or CONNECTION (for internal Iceberg tables).

For a complete list of all catalog-related parameters, see the main Catalog configuration page.

Example

Internal tables (recommended)

CREATE CONNECTION jdbc_conn WITH (
    type = 'iceberg',
    warehouse.path = 's3://my-bucket/warehouse/',
    s3.region = 'us-west-2',
    s3.access.key = '...',
    s3.secret.key = '...',
    enable_config_load = false,

    catalog.type = 'jdbc',
    catalog.uri = 'jdbc:postgresql://postgres:5432/iceberg_db',
    catalog.jdbc.user = 'user',
    catalog.jdbc.password = 'password'
);

SET iceberg_engine_connection = 'public.jdbc_conn';

-- `commit_checkpoint_interval` controls Iceberg commit frequency. Default: about every 60 seconds; set to 1 for faster commits and visibility.
CREATE TABLE my_internal_iceberg_table (
  id INT PRIMARY KEY,
  name VARCHAR
)
WITH (commit_checkpoint_interval = 1)
ENGINE = iceberg;

Write (CREATE SINK)

CREATE SINK my_jdbc_sink FROM my_mv WITH (
    connector = 'iceberg',
    type = 'upsert',
    primary_key = 'id',
    catalog.type = 'jdbc',
    catalog.uri = 'jdbc:postgresql://postgres:5432/iceberg_db',
    catalog.jdbc.user = 'user',
    catalog.jdbc.password = 'password',
    warehouse.path = 's3://my-bucket/warehouse/',
    s3.region = 'us-west-2',
    s3.access.key = '...',
    s3.secret.key = '...',
    database.name = 'my_db',
    create_table_if_not_exists = 'true',
    table.name = 'my_table'
);

Read (CREATE SOURCE)

CREATE SOURCE my_jdbc_source WITH (
    connector = 'iceberg',
    catalog.type = 'jdbc',
    catalog.uri = 'jdbc:postgresql://postgres:5432/iceberg_db',
    catalog.jdbc.user = 'user',
    catalog.jdbc.password = 'password',
    warehouse.path = 's3://my-bucket/warehouse/',
    s3.region = 'us-west-2',
    s3.access.key = '...',
    s3.secret.key = '...',
    database.name = 'my_db',
    table.name = 'my_table'
);