<a id="cmf-sql-views"></a>

# Views in Confluent Manager for Apache Flink

Starting with version 2.5.0, Confluent Manager for Apache Flink® (CMF) lets you create Flink SQL views
in the [environment catalog](../../configure/environment-catalog.md#cmf-environment-catalog). A view is a named
query. It stores the query, not data. When a statement reads from a view,
Flink SQL replaces the view with its query and runs that query as part of the
statement.

Use a view to give one name to a query that several statements share, such as a
filter or a join over Kafka tables. A view can read tables from any catalog in its
environment, including the Kafka catalogs, but you can create views only in the
environment catalog. Statements in other environments can’t list or read the
view.

## Prerequisites

- The environment catalog is enabled. For more information, see
  [Enable the environment catalog](../../configure/environment-catalog.md#cmf-environment-catalog-enable).
- You have permission to create statements in the environment. For a view that
  calls a user-defined function, you also need the `Use` permission on the
  artifact that contains the function. For more information, see
  [Access control](#cmf-sql-views-access-control).

## Create a view

Create a view in the environment catalog with `CREATE VIEW` or
`CREATE OR REPLACE VIEW`:

```text
{ CREATE VIEW [IF NOT EXISTS] | CREATE OR REPLACE VIEW }
  [catalog_name.][db_name.]view_name
  [( column_name [, column_name ]* )]
  [COMMENT 'view_comment']
  AS query_expression;
```

- `IF NOT EXISTS`: If a view or a table already has the name, the statement
  completes without an error and leaves that view or table unchanged.
- `OR REPLACE`: If a view already has the name, the statement replaces that
  view’s definition, including its query, column names, and comment. If no view
  or table has the name, it creates the view. CMF rejects a statement that
  uses both `OR REPLACE` and `IF NOT EXISTS`.
- `column_name`: Names for the view’s columns. Without this list, the view
  uses the column names of the query.
- `COMMENT`: A description of the view
- `AS`: The query that defines the view

Tables and views in a database share one set of names, so a view can’t take the
name of a table. `CREATE VIEW` without `IF NOT EXISTS`,
`CREATE OR REPLACE VIEW`, and `ALTER VIEW ... RENAME TO` fail with an
`already exists` error when a table has the name. `CREATE VIEW` without
`IF NOT EXISTS` fails the same way when a view has the name.

The examples on this page set the statement’s current catalog to the
environment catalog, `_env_<environment>`, and its current database to
`default`. Flink SQL resolves the names in the view’s query that aren’t
fully qualified against the statement’s current catalog and database, at the
time you create the view. The view stores the fully qualified names, so later
statements read the same tables, whatever their own current catalog.

The following command creates the view `large_orders` in the `default`
database of the environment catalog for the `prod` environment. The view reads
from the `orders` table of the examples catalog, which exists when CMF is
installed with `cmf.sql.examples-catalog.enabled=true`:

```bash
confluent flink statement create create-large-orders \
  --url http://cmf:8080 \
  --environment prod \
  --compute-pool pool \
  --catalog _env_prod \
  --database default \
  --sql "CREATE VIEW large_orders AS SELECT order_id, customer_id, price FROM examples.marketplace.orders WHERE price > 50;"
```

To read from a Kafka table, name it by catalog, database, and table. Quote names
that contain hyphens with backticks:

```sql
CREATE VIEW large_kafka_orders AS
  SELECT order_id, customer_id, price
  FROM `kafka-cat`.`kafka-db`.`orders`
  WHERE price > 50;
```

## List and inspect views

`SHOW VIEWS` lists the views in the current database. `SHOW TABLES` lists
both the tables and the views.

`SHOW CREATE VIEW` returns a `CREATE VIEW` statement for the view, and
`DESCRIBE` returns its columns and their types:

```sql
SHOW CREATE VIEW large_orders;

DESCRIBE large_orders;
```

The query in the `SHOW CREATE VIEW` output is the form that Flink SQL
stored when you created the view, not the text you submitted. Every table name
in it is fully qualified, and `SELECT *` is replaced by the list of columns.
Any user who can create statements in the environment can run
`SHOW CREATE VIEW`, so don’t put passwords or other secrets in a view’s query.

## Query a view

Read from a view the same way you read from a table:

```sql
SELECT customer_id, price FROM large_orders;

INSERT INTO `kafka-cat`.`kafka-db`.`big_spenders`
  SELECT customer_id, price FROM large_orders;
```

In a statement whose current catalog is not the environment catalog, use the
view’s fully qualified name. `DEFAULT` is a reserved word in Flink SQL, so
quote the `default` database name with backticks:

```sql
SELECT customer_id, price FROM _env_prod.`default`.large_orders;
```

## Change a view

Replace a view’s query with `ALTER VIEW ... AS` or `CREATE OR REPLACE VIEW`,
and rename a view with `ALTER VIEW ... RENAME TO`:

```sql
ALTER VIEW large_orders AS
  SELECT order_id, customer_id, price
  FROM examples.marketplace.orders
  WHERE price > 100;

ALTER VIEW large_orders RENAME TO very_large_orders;
```

A view can’t read from itself, directly or through other views. CMF rejects an
`ALTER VIEW ... AS` or `CREATE OR REPLACE VIEW` statement that would make it
do so, and keeps the view’s current query. For example,
if view `b` reads from view `a`, then `ALTER VIEW a AS SELECT * FROM b`
fails with this error:

```text
View '_env_prod.default.a' would read from itself (_env_prod.default.a -> _env_prod.default.b -> _env_prod.default.a). A view cannot refer to itself, directly or through other views.
```

Changing a view doesn’t change statements that are already running. They keep
the query that the view had when they started. A stopped statement that you
resume also keeps the old query, because resuming doesn’t compile the statement
again. To run a statement with the view’s new query, create the statement again.

## Drop a view

Remove a view from the catalog with `DROP VIEW`:

```sql
DROP VIEW [IF EXISTS] [catalog_name.][db_name.]view_name;
```

`DROP VIEW` drops only views, and `DROP TABLE` drops only tables.

CMF doesn’t check whether other views read from the view or table that you
drop. Those views stay in the catalog. A statement that you create later and
that reads from one of them fails when you create it, with an error that names
the missing table or view. Statements that are already running keep running,
and so do stopped statements that you resume, because they don’t read the
catalog again. Dropping a table in a Kafka catalog is different, because it also
deletes the table’s topic and the topic’s data. Statements that read the table,
directly or through a view, lose that data. For more information, see
[Table Operations in Confluent Manager for Apache Flink](table-operations.md#flink-sql-table-operations).

You can’t delete an environment while its environment catalog holds views or
tables. For more information, see [Configure Environments in Confluent Manager for Apache Flink](../../configure/environments.md#cmf-environments).

<a id="cmf-sql-views-access-control"></a>

## Access control

You create, change, drop, and read views with Flink SQL statements, so you
need permission to create statements in the environment. For the roles that
grant it, see [Configure Access Control for Confluent Manager for Apache Flink](../../configure/access-control.md#cmf-access-control).

When a statement creates, changes, or reads from a view that calls a
[user-defined function](user-defined-functions.md#cmf-sql-udf), CMF checks that you have the
`Use` permission on the artifact that contains the function, the same as when
you call the function directly. Without it, CMF rejects the request to create
the statement with HTTP 403, and a view that the statement would create or
change stays as it was. For more information, see [Access control](../../configure/artifacts.md#cmf-artifacts-rbac).

## Limitations

- You can create views only in the environment catalog. `CREATE VIEW`,
  `ALTER VIEW`, and `DROP VIEW` in a Kafka catalog fail with the error
  `View DDL is not supported on Kafka catalog '<catalog>'.` To query Kafka
  tables through a view, create the view in the environment catalog.
- A view that uses `SELECT *` keeps the columns that the source table had when
  you created or last changed the view. Columns that you add to the table later
  don’t appear in the view, and new statements that read from the view fail if
  you drop a column that the view uses. After you change the table’s columns,
  run `ALTER VIEW ... AS` to update the view.
- A view keeps the column types that it had when you created or last changed it.
  If you drop the source table and create it again with a different column type,
  statements that read from the view cast the values to the old type. For
  example, if the view has an `INT` column and the re-created table has a
  `DOUBLE` column, the view returns `2` for `2.61` without an error. If
  the cast fails, such as for a `STRING` value that isn’t a number, the
  statement’s job fails and keeps restarting. After you re-create a source
  table, update each view that reads it with `ALTER VIEW ... AS` or
  `CREATE OR REPLACE VIEW`, and then create the statements that read those
  views again.
- `ALTER VIEW ... RENAME TO` can’t move a view to another database or catalog.
  If the new name includes another database or catalog, the statement completes
  without an error, but the view stays in its database and takes only the last
  part of the new name. For example, `ALTER VIEW db1.v RENAME TO db2.w` leaves
  the view at `db1.w`. To move a view, create it in the target database with
  the same query and drop the original.

## Related content

- [Manage Environment Catalogs in Confluent Manager for Apache Flink](../../configure/environment-catalog.md#cmf-environment-catalog)
- [User-Defined Functions in Confluent Manager for Apache Flink](user-defined-functions.md#cmf-sql-udf)
- [Custom Connectors and Formats in Confluent Manager for Apache Flink](custom-connectors.md#cmf-sql-custom-connectors)
- [Manage Kafka Catalogs and Databases for Confluent Manager for Apache Flink](../../configure/catalog.md#flink-sql-catalog)
- [Features & Support for Statements in Confluent Manager for Apache Flink](features-support.md#flink-sql-features-support)