<a id="cmf-environment-catalog"></a>

# Manage Environment Catalogs in Confluent Manager for Apache Flink

Starting with version 2.4.0, Confluent Manager for Apache Flink® (CMF) provides an *environment catalog*
for every environment. The environment catalog is a writable Flink SQL catalog
that stores metadata you create with Data Definition Language (DDL) statements,
such as [user-defined functions](../jobs/sql-statements/user-defined-functions.md#cmf-sql-udf) and
[tables that use custom connectors](../jobs/sql-statements/custom-connectors.md#cmf-sql-custom-connectors).

The built-in Kafka catalogs cannot store this metadata. The catalogs expose Kafka
topics as tables and derive their schemas from Schema Registry, but they do not accept
`CREATE FUNCTION`, and they cannot store tables that are backed by systems
other than Kafka. The environment catalog fills that gap. You register
functions and custom-connector tables in it. Because the catalog is scoped to a
single environment, the objects you create are visible only within that
environment.

#### NOTE
The environment catalog stores metadata only. You deliver the function and
connector code by uploading it to [artifact management](artifacts.md#cmf-artifacts)
and referencing it from your DDL.

## How it works

When the environment catalog is enabled, CMF adds one catalog to every
environment automatically. You do not create or delete it.

- The catalog name is `_env_` followed by the environment name. For example,
  environment `prod` has the catalog `_env_prod`. The `_env_` prefix keeps
  the name from colliding with your Kafka catalogs.
- Each environment catalog includes a `default` database that CMF creates
  automatically and that you cannot delete. CMF creates it the first time a
  statement in the environment opens the catalog, not when the environment is
  created. You can create additional databases with `CREATE DATABASE` and
  drop them with `DROP DATABASE`.
- CMF stores the catalog’s objects (functions and tables) in its own database.
  No Schema Registry or Kafka cluster is required.

The objects you can create in the environment catalog are:

- **Functions:** scalar, table, and aggregate user-defined functions, registered
  with `CREATE FUNCTION`. See [User-Defined Functions in Confluent Manager for Apache Flink](../jobs/sql-statements/user-defined-functions.md#cmf-sql-udf).
- **Tables:** tables backed by custom connectors and formats, created with
  `CREATE TABLE ... WITH ('connector' = '...')`. See
  [Custom Connectors and Formats in Confluent Manager for Apache Flink](../jobs/sql-statements/custom-connectors.md#cmf-sql-custom-connectors).

<a id="cmf-environment-catalog-enable"></a>

## Enable the environment catalog

The environment catalog is an opt-in feature and is disabled by default. Enable
it in your Helm values file:

```yaml
cmf:
  sql:
    environmentCatalog:
      enabled: true
```

While the feature is disabled, the environment catalog is not registered, and
the [catalog endpoint](#cmf-environment-catalog-view) returns HTTP 404.

To register and use functions or custom connectors, also enable
[artifact management](artifacts.md#cmf-artifacts-enable), which stores the JAR files
that your DDL references. See [User-Defined Functions in Confluent Manager for Apache Flink](../jobs/sql-statements/user-defined-functions.md#cmf-sql-udf) and
[Custom Connectors and Formats in Confluent Manager for Apache Flink](../jobs/sql-statements/custom-connectors.md#cmf-sql-custom-connectors).

## Use the environment catalog

To create or use objects in the environment catalog, set the statement’s current
catalog and database to the environment catalog and its `default` database.

With the Confluent CLI, pass the `--catalog` and `--database` flags:

```bash
confluent flink statement create show-functions \
  --url http://cmf:8080 \
  --environment prod \
  --compute-pool pool \
  --catalog _env_prod \
  --database default \
  --sql "SHOW USER FUNCTIONS;"
```

With the REST API, set the current catalog and database in the statement’s
properties:

```json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Statement",
  "metadata": {
    "name": "show-functions"
  },
  "spec": {
    "statement": "SHOW USER FUNCTIONS;",
    "computePoolName": "pool",
    "properties": {
      "sql.current-catalog": "_env_prod",
      "sql.current-database": "default"
    }
  }
}
```

Within the environment catalog you can run function DDL (`CREATE FUNCTION`,
`DROP FUNCTION`), table DDL (`CREATE TABLE`, `ALTER TABLE`, `DROP TABLE`),
database DDL (`CREATE DATABASE`, `DROP DATABASE`), and queries (`SELECT`,
`INSERT INTO`) that reference those objects.

<a id="cmf-environment-catalog-view"></a>

## View the environment catalog

CMF exposes a read-only view of an environment catalog and its databases
through the REST API:

```bash
curl -v -X GET http://cmf:8080/cmf/api/v1/environments/prod/catalog
```

The response contains the catalog name and its databases. CMF creates the
`default` database on first use. Until a statement has run in the
environment, the list of databases can be empty. The response does not list the
functions in the catalog; to list functions, run a `SHOW USER FUNCTIONS;`
statement against the catalog. The endpoint returns HTTP 404 when the
environment catalog feature is disabled.

## Limitations

- `CREATE TABLE AS` (CTAS) and `CREATE OR REPLACE TABLE AS` (RTAS) statements
  are not supported in the environment catalog.
- `UPDATE` and `DELETE` statements are not supported, in line with Flink SQL,
  which does not support these DML statements in streaming mode. Use
  `INSERT INTO` to write rows and `SELECT` to read them.
- `CREATE TEMPORARY FUNCTION` and `CREATE TEMPORARY SYSTEM FUNCTION` are not
  supported. Register functions with `CREATE FUNCTION` instead.
- The environment catalog and its `default` database cannot be renamed or
  deleted.
- User-defined functions must be implemented in Java. See [User-Defined Functions in Confluent Manager for Apache Flink](../jobs/sql-statements/user-defined-functions.md#cmf-sql-udf).

## Related content

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