CREATE TABLE

Rows can be added using the INSERT command. When creating a table, you can specify connector settings and data format.

​

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

The append-only table does not support delete or update operations, nor does it support non-append-only upstream connectors. There are two main reasons for using append-only tables:

• 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

In a CREATE TABLE statement, you can specify ENGINE = iceberg to store table data in the Apache Iceberg format on external object storage. This provides an alternative to RisingWave’s default internal row-based storage, making it easier to persist data for long-term use and cross-system access. For more information on the syntax and usage, see Iceberg table engine.

​

Watermarks

RisingWave supports generating watermarks when creating an append-only streaming table. Watermarks are like markers or signals that track the progress of event time, allowing you to process events within their corresponding time windows. For more information on the syntax on how to create a watermark, see Watermarks.

​

TTL of append-only table

Data in an append-only table cannot be deleted with DELETE statements. Therefore, RisingWave provides TTL (Time-To-Live) feature to automatically clean up expired data in the table. However, please note that this cleanup only applies to the data within the table itself—it does not affect downstream materialized views or other streaming jobs. If you need to clean up data in downstream materialized views used in queries, you should use a Temporal Filter in the downstream streaming job.

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

​

PK conflict behavior

Records with insert operations could introduce duplicate records with the same primary key in the table. In that case, an alternative action specified by the ON CONFLICT conflict_action clause will be adopted. The record can come from an INSERT statement, external connectors of the table, or sinks into the table.

The conflict_action could be one of the following. A version column can be specified together for DO UPDATE FULL and DO UPDATE IF NOT NULL. When a version column is specified, the insert operation will take effect only when the newly inserted row’s version column is greater than or equal to the existing row’s version column, and is not NULL.

• 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

A version column is a non-primary-key column that can be used with OVERWRITE and DO UPDATE IF NOT NULL to tune the conflict handling behaviors.

Common use cases include:

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

The statement below creates a table that has three columns.

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

The statement below creates a table that includes nested tables.

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>);

The statement below creates a table with a Kafka broker as the source.

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;