> ## Documentation Index
> Fetch the complete documentation index at: https://docs.risingwave.com/llms.txt
> Use this file to discover all available pages before exploring further.

# CREATE SINK INTO

> Use the `CREATE SINK INTO` command to create a sink into RisingWave's table.

## syntax

```sql theme={null}
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.                                                                |

<Note>
  A table without a primary key can only accept the append-only sink.
</Note>

<Note>
  You can use [`ALTER TABLE ... ADD COLUMN`](/sql/commands/sql-alter-table#add-column) to add columns to a table that has sinks.
</Note>

## Enable sink decoupling

<Note>
  Added in v2.5.0. It is currently in **[technical preview](/changelog/product-lifecycle#product-release-lifecycle)** stage.
</Note>

By default, [sink decoupling](/delivery/overview#sink-decoupling) is disabled for sinks created with `CREATE SINK INTO`. To enable it, set `sink_decouple = true` before creating the sink.

```sql theme={null}
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.

```sql theme={null}
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.

```sql theme={null}
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

<CardGroup>
  <Card title="CREATE SINK" icon="plus" icontype="solid" href="/sql/commands/sql-create-sink">
    Create a sink into an external target
  </Card>

  <Card title="DROP SINK" icon="trash" icontype="solid" href="/sql/commands/sql-drop-sink">
    Remove a sink
  </Card>

  <Card title="SHOW CREATE SINK" icon="eye" icontype="solid" href="/sql/commands/sql-show-create-sink">
    Show the SQL statement used to create a sink
  </Card>
</CardGroup>