CREATE TABLE

​

Syntax

CREATE TABLE [ IF NOT EXISTS ] table_name (
    col_name data_type [ NOT NULL ] [ PRIMARY KEY ] [ DEFAULT default_expr ] [ AS generation_expression ],
    ...
    [ PRIMARY KEY (col_name, ... ) ]
    [ watermark_clause ]
)
[ APPEND ONLY ]
[ ON CONFLICT conflict_action ]
[INCLUDE { header | key | offset | partition | timestamp } [AS <column_name>]]
[ WITH (
    connector='connector_name',
    connector_parameter='value', ...)]
[FORMAT data_format ENCODE data_encode [ (
    message='message',
    schema.location='location', ...) ]
]
[ ENGINE = iceberg ];

​

Notes

• RisingWave supports NOT NULL constraints in table schemas. For batch inserts and updates, RisingWave throws an error if any row contains NULL in a NOT NULL column; for streaming ingestion, rows with NULL in NOT NULL columns are silently ignored.
• For tables with primary key constraints, if you insert a new data record with an existing key, the new record will overwrite the existing record.
• A generated column that is defined with non-deterministic functions cannot be specified as part of the primary key. For example, if A1 is defined as current_timestamp(), then it cannot be part of the primary key.
• Names and unquoted identifiers are case-insensitive. Therefore, you must double-quote any of these fields for them to be case-sensitive. See also Identifiers.
• The syntax for creating a table with connector settings and the supported connectors are the same as for creating a source. See CREATE SOURCE for a full list of supported connectors and data formats.
• To know when a data record is loaded to RisingWave, you can define a column that is generated based on the processing time (<column_name> timestamptz AS proctime()) when creating the table or source. See also proctime().
• For a table with schema from external connector, use * to represent all columns from the external connector first, so that you can define a generated column on table with an external connector. Alternatively, you can partially combine the schema with generated columns or apply the schema directly to define the table structure. See the examples below:
  Copy

  Ask AI

  -- Use * to represent all columns
  CREATE TABLE from_kafka (
    *,
    gen_i32_field INT AS int32_field + 2,
    PRIMARY KEY (some_key)
  )
  INCLUDE KEY AS some_key
  [INCLUDE { header | offset | partition | timestamp } [AS <column_name>]]
  WITH (
    connector = 'kafka',
    topic = 'test-rw-sink-upsert-avro',
    properties.bootstrap.server = 'message_queue:29092'
  )
  FORMAT upsert ENCODE AVRO (
    schema.registry = 'http://message_queue:8081'
  );
  
  -- Specify partial schema
  CREATE TABLE t2 (bar INT, gen_col INT AS bar + 1)
  WITH (
    connector = 'kafka',
    topic = 'test-rw-sink-upsert-avro',
    properties.bootstrap.server = 'message_queue:29092'
  )
  FORMAT upsert ENCODE AVRO (
    schema.registry = 'http://message_queue:8081'
  );
  

​

Parameters

​

Append only

• Certain features are exclusively available for append-only tables, such as watermark and TTL (Time-To-Live).
• Streaming jobs created downstream from append-only tables can leverage their append-only property to optimize performance.

​

Use Iceberg table engine

​

Watermarks

​

TTL of append-only table

CREATE TABLE t (
    ...
) APPEND ONLY WITH (retention_seconds = 86400);

​

PK conflict behavior

• IGNORE: Ignore the newly inserted record.
• OVERWRITE [WITH VERSION COLUMN(col_name)]: Full update: replace the existing row in the table with the new row.
• DO UPDATE IF NOT NULL [WITH VERSION COLUMN(col_name)]: Partial update: only replace those fields with non-NULL values in the inserted row. NULL values in the inserted row means unchanged in this case.

​

Version column

1. Use an event time column as the version column to ensure temporal ordering and ignore late-arriving data.
2. Use an integer as a logical version number. You can use an Upsert-then-Delete pattern with a logical version column to implement “zero-downtime incremental version upgrade” for data in the table: you insert a new version of data with a higher version number while keeping old data, and then after finishing the upgrade, you clean up old data with a DELETE statement. This can be used to implement in-place streaming job modification, or re-sync external batch data into RisingWave (reverse ETL).

​

Example

CREATE TABLE taxi_trips(
    id VARCHAR,
    distance DOUBLE PRECISION,
    city VARCHAR
);

CREATE TABLE IF NOT EXISTS taxi_trips(
    id VARCHAR,
    distance DOUBLE PRECISION,
    duration DOUBLE PRECISION,
    fare STRUCT<
      initial_charge DOUBLE PRECISION,
      subsequent_charge DOUBLE PRECISION,
      surcharge DOUBLE PRECISION,
      tolls DOUBLE PRECISION>);

CREATE TABLE IF NOT EXISTS table_abc (
   column1 varchar,
   column2 integer,
)
WITH (
   connector='kafka',
   topic='demo_topic',
   properties.bootstrap.server='172.10.1.1:9090,172.10.1.2:9090',
   scan.startup.mode='latest',
   scan.startup.timestamp_millis='140000000',
) FORMAT PLAIN ENCODE JSON;