<a id="flink-sql-insert-into-from-select-statement"></a>

# INSERT INTO FROM SELECT Statement in Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® enables inserting SELECT query results directly into a Flink SQL
table.

## Syntax

```sql
[EXECUTE] INSERT { INTO | OVERWRITE } [catalog_name.][database_name.]table_name
  [PARTITION (partition_column_name1=value1 [, partition_column_name2=value2, ...])]
  [(column_name1 [, column_name2, ...])]
  select_statement
```

**OVERWRITE**
: `INSERT OVERWRITE` overwrites all existing data in the table or
  partition. New records are appended.

**PARTITION**
: The `PARTITION` clause contains static partition columns for the
  insertion.

**COLUMN LIST**
: For a table `T(a INT, b INT, c INT)`, Flink supports
  <br/>
  ```sql
  INSERT INTO T(c, b) SELECT x, y FROM S.
  ```
  <br/>
  The `x` result is written to column `c`, and the `y` result is written
  to column `b`. If column `a` is nullable, `a` is set to NULL.

## Description

Insert query results into a table.

Use the INSERT INTO FROM SELECT statement to insert rows into a table from
another table or query.

For example, if you have a table `T` with columns `a`, `b`, and `c`,
and another table `S` with columns `x` and `y`, the following query
writes the values of `x` and `y` from `S` into `c` and `b` of `T`,
respectively.

```sql
INSERT INTO T (c, b) SELECT x, y FROM S
```

If column `a` of `T` is nullable, Flink sets it to NULL.

<!-- .. admonition:: See **INSERT INTO FROM SELECT** in action -->
<!-- The following resources show the INSERT INTO FROM SELECT statement in -->
<!-- various applications. -->
<!-- - TODO -->
<!-- - TODO -->

## Examples

### Insert rows into a simple table

In the Flink SQL shell or in a Cloud Console workspace, run the
following commands to see an example of the INSERT INTO FROM SELECT
statement.

1. Create a table for web page click events.
   ```sql
   -- Create a table for web page click events.
   CREATE TABLE clicks (
     ip_address VARCHAR,
     url VARCHAR,
     click_ts_raw BIGINT
   );
   ```
2. Populate the table with mock clickstream data.
   ```sql
   -- Populate the table with mock clickstream data.
   INSERT INTO clicks
   VALUES( '10.0.0.1',  'https://acme.com/index.html',     1692812175),
         ( '10.0.0.12', 'https://apache.org/index.html',   1692826575),
         ( '10.0.0.13', 'https://confluent.io/index.html', 1692826575),
         ( '10.0.0.1',  'https://acme.com/index.html',     1692812175),
         ( '10.0.0.12', 'https://apache.org/index.html',   1692819375),
         ( '10.0.0.13', 'https://confluent.io/index.html', 1692826575);
   ```

   Press Enter to return to the SQL shell. Because INSERT INTO VALUES is a
   point-in-time statement, it exits after it completes inserting records.
3. Create another table for filtered web page click events.
   ```sql
   CREATE TABLE filtered_clicks (
     ip_address VARCHAR,
     url VARCHAR,
     click_ts_raw BIGINT
   );
   ```
4. Run the following statement to insert filtered rows into the
   `filtered_clicks` table. Only clicks that have an IP address
   of `10.0.0.1` are inserted.
   ```sql
   INSERT INTO filtered_clicks(
     ip_address,
     url,
     click_ts_raw
   )
   SELECT * FROM clicks WHERE ip_address = '10.0.0.1';
   ```
5. View the rows in the `filtered_clicks` table.
   ```sql
   SELECT * FROM filtered_clicks;
   ```

   Your output should resemble:
   ```none
   ip_address url                         click_timestamp
   10.0.0.1   https://acme.com/index.html 2023-08-23 10:36:15
   10.0.0.1   https://acme.com/index.html 2023-08-23 10:36:15
   ```

### Fill a table without specifying all columns

```sql
CREATE TABLE t_insert_gaps (c1 STRING, c2 STRING, c3 STRING, c4 STRING);

INSERT INTO t_insert_gaps (c3) SELECT 'Bob';

INSERT INTO t_insert_gaps (c3, c2) SELECT 'Bob', 'Alice';

SELECT * FROM t_insert_gaps;
```

Properties
: - A column list is defined between the table name and the SELECT in the
    INSERT INTO statement, so the SELECT statement uses a reduced schema.
  - Columns `c1`, `c2`, and `c4` are filled with NULLs.
  - If one of the columns is declared NOT NULL, an error occurs.

### Repartition a source into a sink

If you want to repartition from a source table into a sink table, and you have
already created the sink table with the desired number of partitions, the
following query repartitions the source table into the sink table.

```sql
INSERT INTO sink SELECT * FROM source;
```

If you have not yet created the sink table with the desired number of partitions, the
following query creates the sink with the desired number of partitions.

```sql
CREATE TABLE sink_with_12_partitions
DISTRIBUTED INTO 12 BUCKETS
  AS
  SELECT * FROM source;
```

## Related content

- [Convert the Serialization Format of a Topic](../../how-to-guides/convert-serialization-format.md#flink-sql-convert-format)
- [INSERT VALUES](insert-values.md#flink-sql-insert-values-statement)
- [Flink SQL Queries](overview.md#flink-sql-queries)
- [Flink SQL Functions](../functions/overview.md#flink-sql-functions-overview)
- [Statements](../statements/overview.md#flink-sql-statements-overview)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).