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

Create a view

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

{ 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:

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:

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:

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:

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:

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:

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:

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:

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.

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.

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.

When a statement creates, changes, or reads from a view that calls a user-defined function, 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.

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.