CREATE SINK INTO

Use the CREATE SINK INTO command to create a sink into RisingWave’s table.

syntax

CREATE SINK [ IF NOT EXISTS ] sink_name INTO table_name [ ( col_name [ , ... ] ) ]
[FROM sink_from | AS select_query]
[ WITH ( sink_option = value [, ...] ) ];

Parameters

Parameter or clause	Description
sink_name	The name of the sink. If a schema name is given (for example, CREATE SINK <schema>.<sink> ...), then the sink is created in the specified schema. Otherwise it is created in the current schema.
col_name	The corresponding table columns in the sink result. For those columns not listed, it will be inserted as the default value defined in the table.
WITH clause	You can use options such as type = 'append-only' and force_append_only = 'true' to explicitly control the append-only behavior.

A table without a primary key can only accept the append-only sink.

You can use ALTER TABLE ... ADD COLUMN to add columns to a table that has sinks.

Enable sink decoupling

Added in v2.5.0. It is currently in technical preview stage.

By default, sink decoupling is disabled for sinks created with CREATE SINK INTO. To enable it, set sink_decouple = true before creating the sink.

SET sink_decouple = true;

CREATE TABLE t (
  v1 INT PRIMARY KEY,
  v2 INT
);

CREATE TABLE t2 (
  v1 INT PRIMARY KEY,
  v2 INT
);

CREATE SINK s1 INTO t FROM t2;

-- If this downstream MV is slow, it will not block the sink s1,
-- since the upstream and downstream are decoupled.
CREATE MATERIALIZED VIEW m1 AS
SELECT v1, SUM(v2)
FROM t
GROUP BY v1;

Examples

You can union data from two different Kafka topics.

CREATE TABLE orders (
    id int primary key,
    price int,
    item_id int,
    customer_id int
);

CREATE source orders_s0 (
    id int primary key,
    price int,
    item_id int,
    customer_id int
) WITH (
    connector = 'kafka',
    topic = 'topic_0',
    ...
) FORMAT PLAIN ENCODE JSON;

CREATE source orders_s1 (
    id int primary key,
    price int,
    item_id int,
    customer_id int
) WITH (
    connector = 'kafka',
    topic = 'topic_1',
    ...
) FORMAT PLAIN ENCODE JSON;

CREATE SINK orders_sink0 INTO orders FROM orders_s0;
CREATE SINK orders_sink1 INTO orders FROM orders_s1;

If you don’t want one of the topics, you can drop it.

DROP SINK orders_sink0;

Troubleshooting

1. CREATE SINK ... INTO aligns rows to be inserted based on column order, not column names. If you see data correctness issues, verify that the column order in the SELECT list matches the target table.
2. The sink target table may have a different primary key than the upstream table. If primary keys overlap downstream, rows can be overwritten. This can make your downstream table have fewer rows than your upstream table.

See also

CREATE SINK

Create a sink into an external target

DROP SINK

Remove a sink

SHOW CREATE SINK

Show the SQL statement used to create a sink