

<a id="flink-sql-select"></a>

# SELECT Statement in Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® enables querying the content of your tables by using familiar
SELECT syntax.

## Syntax


```sql
SELECT [DISTINCT] select_list FROM table_expression [ WHERE boolean_expression ] [ LIMIT row_limit ]
```

## Description

The SELECT statement in Flink does what the SQL standard says it must do.
You needn’t look further than standard SQL itself to understand the behavior.

For example, UNION without ALL means that duplicate rows must be removed.

Flink maintains the relation, called a [dynamic table](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables),
specified by the SQL query. Its behavior is always the same as if you ran the
SQL query again, over the current snapshot of the data, each time a new row
arrives for any table in the relation.

This formalism is what enables you to reason about exactly what Flink does
by understanding what any SQL system, such as MySQL, Snowflake, or Oracle,
would do.

Another way to understand what Flink SQL does is to consider the following
statement:

```sql
SELECT * FROM clicks ORDER BY clickTime LIMIT 10;
```

This statement doesn’t only look at 10 rows, sort them, and terminate. It
maintains this relation, and as new orders arrive, the relation changes, always
showing the top 10 most recent orders. This is exactly as if you re-ran the
query each time a new row was written to the `clicks` table. You get the
same result.

## Select list

The `select_list` specification `*` means the query resolves all
columns. But in production, avoid using `*` because it
makes queries less robust to catalog changes. Instead, use a `select_list`
to specify a subset of available columns or make calculations using
the columns. For example, if an `orders` table has columns named
`order_id`, `price`, and `tax` you could write the following query:

```sql
SELECT order_id, price + tax FROM orders
```

## Table expression

The `table_expression` can be any source of data, including a table, view,
or `VALUES` clause, the joined results of multiple existing tables, or a
subquery.

Assuming that an `orders` table is available in the catalog, the
following statement reads all rows from `orders`.

```sql
SELECT * FROM orders;
```

## VALUES clause

Queries can consume from inline data by using the `VALUES` clause. Each tuple
corresponds to one row. You can provide an alias to assign a name to each column.

```sql
 SELECT order_id, price
   FROM (VALUES (1, 2.0), (2, 3.1))
   AS t (order_id, price);
```

Your output should resemble:

```none
order_id price
1        2.0
2        3.1
```

## WHERE clause

Filter rows by using the `WHERE` clause.

```sql
 SELECT price + tax
   FROM orders
   WHERE id = 10;
```

## Functions

You can invoke built-in scalar functions on the columns of a single row.

```sql
SELECT PRETTY_PRINT(order_id) FROM orders;
```

## DISTINCT

If `SELECT DISTINCT` is specified, all duplicate rows are removed from
the result set, which means that one row is kept from each group of duplicates.

For streaming queries, the required state for computing the query result might
grow infinitely. State size depends on the number of distinct rows.


**Table types.** `SELECT DISTINCT` accepts an
[append-only or updating table](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables-updating-table)
as input, and it always produces an updating table, the same as a
[group aggregation](group-aggregation.md#flink-sql-group-aggregation).

```sql
SELECT DISTINCT id FROM orders;
```

## Usage

In the Flink SQL shell or in a Cloud Console workspace, run the
following commands to see examples of the 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. View all rows in the `clicks` table by using a SELECT statement.
   ```sql
   SELECT * FROM clicks;
   ```

   Your output should resemble:
   ```none
   ip_address url                             click_ts_raw
   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
   ```
4. View only unique rows in the `clicks` table by using a SELECT DISTINCT
   statement.
   ```sql
   SELECT DISTINCT * FROM clicks;
   ```

   Your output should resemble:
   ```none
   ip_address url                             click_ts_raw
   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.12  https://apache.org/index.html   1692819375
   ```
5. View only records that have the ip_address of `10.0.0.1` by using a
   SELECT WHERE statement.
   ```sql
   SELECT * FROM clicks WHERE ip_address='10.0.0.1';
   ```

   Your output should resemble:
   ```none
   ip_address url                         click_ts_raw
   10.0.0.1   https://acme.com/index.html 1692812175
   10.0.0.1   https://acme.com/index.html 1692812175
   ```

<a id="flink-sql-select-examples"></a>

## Examples

The following examples show frequently encountered scenarios with SELECT.

### Most minimal statement

The shortest valid SELECT statement returns a single literal value and runs
as a bounded query.

Syntax
: ```sql
  SELECT 1;
  ```

Properties
: - Statement is bounded

### Verify the local time zone configuration

Syntax
: ```sql
  SELECT NOW();
  ```

Properties
: - Statement is bounded
  - NOW() returns a TIMESTAMP_LTZ(3), so if the client is configured correctly,
    it should show a timestamp in your local time zone.

### Combine multiple tables into one

Syntax
: ```sql
  CREATE TABLE t_union_1 (i INT);
  CREATE TABLE t_union_2 (i INT);
  TABLE t_union_1 UNION ALL TABLE t_union_2;
  <br/>
  -- alternate syntax
  SELECT * FROM t_union_1
  UNION ALL
  SELECT * FROM t_union_2;
  ```

### Get insights into the current watermark

Syntax
: ```mysql
  CREATE TABLE t_watermarked_insight (s STRING) DISTRIBUTED INTO 1 BUCKETS;
  <br/>
  INSERT INTO t_watermarked_insight VALUES ('Bob'), ('Alice'), ('Charly');
  <br/>
  SELECT $rowtime, CURRENT_WATERMARK($rowtime) FROM t_watermarked_insight;
  ```
  <br/>
  The output resembles:
  <br/>
  ```none
  $rowtime                EXPR$1
  2024-04-29 11:59:01.080 NULL
  2024-04-29 11:59:01.093 2024-04-04 15:27:37.433
  2024-04-29 11:59:01.094 2024-04-04 15:27:37.433
  ```

Properties
: - The CURRENT_WATERMARK function returns the watermark that arrived at the
    operator evaluating the SELECT statement.
  - The returned watermark is the minimum of all inputs, across all
    tables/topics and their partitions.
  - If a common watermark was not received from all inputs, the function
    returns NULL.
  - The CURRENT_WATERMARK function takes a
    [time attribute](../../concepts/timely-stream-processing.md#flink-sql-time-attributes), which is a column
    that has WATERMARK FOR defined.

A watermark is always emitted after the row has been processed, so the first
row always has a NULL watermark.

The default watermark strategy uses a fixed out-of-orderness tolerance of
180 ms with no warmup period. The watermark advances immediately as data
arrives, trailing the maximum observed event time by 180 ms.

Sources emit watermarks every 200 ms, but within the first 200 ms they emit
per row for powering examples such as this one.

### Flatten fields into columns

Syntax
: ```sql
  CREATE TABLE t_flattening (i INT, r1 ROW<i INT, s STRING>, r2 ROW<other INT>);
  <br/>
  SELECT r1.*, r2.* FROM t_flattening;
  ```

Properties
: You can apply the `*` operator on nested data, which enables flattening
  fields into columns of the table.

## Related content

- [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).