Catalogs

Catalogs provide metadata, such as databases, tables, partitions, views, and functions and information needed to access data stored in a database or other external systems.

One of the most crucial aspects of data processing is managing metadata. It may be transient metadata like temporary tables, or UDFs registered against the table environment. Or permanent metadata, like that in a Hive Metastore. Catalogs provide a unified API for managing metadata and making it accessible from the Table API and SQL Queries.

Catalog enables users to reference existing metadata in their data systems, and automatically maps them to Flink’s corresponding metadata. For example, Flink can map JDBC tables to Flink table automatically, and users don’t have to manually re-writing DDLs in Flink. Catalog greatly simplifies steps required to get started with Flink with users’ existing system, and greatly enhanced user experiences.

Catalog Types

GenericInMemoryCatalog

The GenericInMemoryCatalog is an in-memory implementation of a catalog. All objects will be available only for the lifetime of the session.

JdbcCatalog

The JdbcCatalog enables users to connect Flink to relational databases over JDBC protocol. Postgres Catalog and MySQL Catalog are the only two implementations of JDBC Catalog at the moment. See JdbcCatalog documentation for more details on setting up the catalog.

HiveCatalog

The HiveCatalog serves two purposes; as persistent storage for pure Flink metadata, and as an interface for reading and writing existing Hive metadata. Flink’s Hive documentation provides full details on setting up the catalog and interfacing with an existing Hive installation.

The Hive Metastore stores all meta-object names in lower case. This is unlike GenericInMemoryCatalog which is case-sensitive

User-Defined Catalog

Catalogs are pluggable and users can develop custom catalogs by implementing the Catalog interface.

In order to use custom catalogs with Flink SQL, users should implement a corresponding catalog factory by implementing the CatalogFactory interface. The factory is discovered using Java’s Service Provider Interfaces (SPI). Classes that implement this interface can be added to META_INF/services/org.apache.flink.table.factories.Factory in JAR files. The provided factory identifier will be used for matching against the required type property in a SQL CREATE CATALOG DDL statement.

Since Flink v1.16, TableEnvironment introduces a user class loader to have a consistent class loading behavior in table programs, SQL Client and SQL Gateway. The user classloader manages all user jars such as jar added by ADD JAR or CREATE FUNCTION .. USING JAR .. statements. User-defined catalogs should replace Thread.currentThread().getContextClassLoader() with the user class loader to load classes. Otherwise, ClassNotFoundException maybe thrown. The user class loader can be accessed via CatalogFactory.Context#getClassLoader.

Interface in Catalog for supporting time travel

Starting from version 1.18, the Flink framework supports time travel to query historical data of a table. To query the historical data of a table, users should implement getTable(ObjectPath tablePath, long timestamp) method for the catalog that the table belongs to.

  1. public class MyCatalogSupportTimeTravel implements Catalog {
  3.     @Override
  4.     public CatalogBaseTable getTable(ObjectPath tablePath, long timestamp)
  5.             throws TableNotExistException {
  6.         // Build a schema corresponding to the specific time point.
  7.         Schema schema = buildSchema(timestamp);
  8.         // Set parameters to read data at the corresponding time point.
  9.         Map<String, String> options = buildOptions(timestamp);
  10.         // Build CatalogTable
  11.         CatalogTable catalogTable =
  12.                 CatalogTable.of(schema, "", Collections.emptyList(), options, timestamp);
  13.         return catalogTable;
  14.     }
  15. }
  17. public class MyDynamicTableFactory implements DynamicTableSourceFactory {
  18.     @Override
  19.     public DynamicTableSource createDynamicTableSource(Context context) {
  20.         final ReadableConfig configuration =
  21.                 Configuration.fromMap(context.getCatalogTable().getOptions());
  23.         // Get snapshot from CatalogTable
  24.         final Optional<Long> snapshot = context.getCatalogTable().getSnapshot();
  26.         // Build DynamicTableSource using snapshot options.
  27.         final DynamicTableSource dynamicTableSource = buildDynamicSource(configuration, snapshot);
  29.         return dynamicTableSource;
  30.     }
  31. }

Using SQL DDL

Users can use SQL DDL to create tables in catalogs in both Table API and SQL.

Java

  1. TableEnvironment tableEnv = ...;
  3. // Create a HiveCatalog 
  4. Catalog catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>");
  6. // Register the catalog
  7. tableEnv.registerCatalog("myhive", catalog);
  9. // Create a catalog database
  10. tableEnv.executeSql("CREATE DATABASE mydb WITH (...)");
  12. // Create a catalog table
  13. tableEnv.executeSql("CREATE TABLE mytable (name STRING, age INT) WITH (...)");
  15. tableEnv.listTables(); // should return the tables in current catalog and database.

Scala

  1. val tableEnv = ...
  3. // Create a HiveCatalog 
  4. val catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>")
  6. // Register the catalog
  7. tableEnv.registerCatalog("myhive", catalog)
  9. // Create a catalog database
  10. tableEnv.executeSql("CREATE DATABASE mydb WITH (...)")
  12. // Create a catalog table
  13. tableEnv.executeSql("CREATE TABLE mytable (name STRING, age INT) WITH (...)")
  15. tableEnv.listTables() // should return the tables in current catalog and database.

Python

  1. from pyflink.table.catalog import HiveCatalog
  3. # Create a HiveCatalog
  4. catalog = HiveCatalog("myhive", None, "<path_of_hive_conf>")
  6. # Register the catalog
  7. t_env.register_catalog("myhive", catalog)
  9. # Create a catalog database
  10. t_env.execute_sql("CREATE DATABASE mydb WITH (...)")
  12. # Create a catalog table
  13. t_env.execute_sql("CREATE TABLE mytable (name STRING, age INT) WITH (...)")
  15. # should return the tables in current catalog and database.
  16. t_env.list_tables()

SQL Client

  1. // the catalog should have been registered via yaml file
  2. Flink SQL> CREATE DATABASE mydb WITH (...);
  4. Flink SQL> CREATE TABLE mytable (name STRING, age INT) WITH (...);
  6. Flink SQL> SHOW TABLES;
  7. mytable

For detailed information, please check out Flink SQL CREATE DDL.

Using Java, Scala or Python

Users can use Java, Scala or Python to create catalog tables programmatically.

Java

  1. import org.apache.flink.table.api.*;
  2. import org.apache.flink.table.catalog.*;
  3. import org.apache.flink.table.catalog.hive.HiveCatalog;
  5. TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode());
  7. // Create a HiveCatalog 
  8. Catalog catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>");
  10. // Register the catalog
  11. tableEnv.registerCatalog("myhive", catalog);
  13. // Create a catalog database 
  14. catalog.createDatabase("mydb", new CatalogDatabaseImpl(...));
  16. // Create a catalog table
  17. final Schema schema = Schema.newBuilder()
  18.     .column("name", DataTypes.STRING())
  19.     .column("age", DataTypes.INT())
  20.     .build();
  22. tableEnv.createTable("myhive.mydb.mytable", TableDescriptor.forConnector("kafka")
  23.     .schema(schema)
  24.     // …
  25.     .build());
  27. List<String> tables = catalog.listTables("mydb"); // tables should contain "mytable"

Scala

  1. import org.apache.flink.table.api._
  2. import org.apache.flink.table.catalog._
  3. import org.apache.flink.table.catalog.hive.HiveCatalog
  5. val tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode())
  7. // Create a HiveCatalog 
  8. val catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>")
  10. // Register the catalog
  11. tableEnv.registerCatalog("myhive", catalog)
  13. // Create a catalog database 
  14. catalog.createDatabase("mydb", new CatalogDatabaseImpl(...))
  16. // Create a catalog table
  17. val schema = Schema.newBuilder()
  18.     .column("name", DataTypes.STRING())
  19.     .column("age", DataTypes.INT())
  20.     .build()
  22. tableEnv.createTable("myhive.mydb.mytable", TableDescriptor.forConnector("kafka")
  23.     .schema(schema)
  24.     // …
  25.     .build())
  27. val tables = catalog.listTables("mydb") // tables should contain "mytable"

Python

  1. from pyflink.table import *
  2. from pyflink.table.catalog import HiveCatalog, CatalogDatabase, ObjectPath, CatalogBaseTable
  4. settings = EnvironmentSettings.in_batch_mode()
  5. t_env = TableEnvironment.create(settings)
  7. # Create a HiveCatalog
  8. catalog = HiveCatalog("myhive", None, "<path_of_hive_conf>")
  10. # Register the catalog
  11. t_env.register_catalog("myhive", catalog)
  13. # Create a catalog database
  14. database = CatalogDatabase.create_instance({"k1": "v1"}, None)
  15. catalog.create_database("mydb", database)
  17. # Create a catalog table
  18. schema = Schema.new_builder() \
  19.     .column("name", DataTypes.STRING()) \
  20.     .column("age", DataTypes.INT()) \
  21.     .build()
  23. catalog_table = t_env.create_table("myhive.mydb.mytable", TableDescriptor.for_connector("kafka")
  24.     .schema(schema)
  25.     # …
  26.     .build())
  28. # tables should contain "mytable"
  29. tables = catalog.list_tables("mydb")

Catalog API

Note: only catalog program APIs are listed here. Users can achieve many of the same functionalities with SQL DDL. For detailed DDL information, please refer to SQL CREATE DDL.

Database operations

Java/Scala

  1. // create database
  2. catalog.createDatabase("mydb", new CatalogDatabaseImpl(...), false);
  4. // drop database
  5. catalog.dropDatabase("mydb", false);
  7. // alter database
  8. catalog.alterDatabase("mydb", new CatalogDatabaseImpl(...), false);
  10. // get database
  11. catalog.getDatabase("mydb");
  13. // check if a database exist
  14. catalog.databaseExists("mydb");
  16. // list databases in a catalog
  17. catalog.listDatabases();

Python

  1. from pyflink.table.catalog import CatalogDatabase
  3. # create database
  4. catalog_database = CatalogDatabase.create_instance({"k1": "v1"}, None)
  5. catalog.create_database("mydb", catalog_database, False)
  7. # drop database
  8. catalog.drop_database("mydb", False)
  10. # alter database
  11. catalog.alter_database("mydb", catalog_database, False)
  13. # get database
  14. catalog.get_database("mydb")
  16. # check if a database exist
  17. catalog.database_exists("mydb")
  19. # list databases in a catalog
  20. catalog.list_databases()

Table operations

Java/Scala

  1. // create table
  2. catalog.createTable(new ObjectPath("mydb", "mytable"), new CatalogTableImpl(...), false);
  4. // drop table
  5. catalog.dropTable(new ObjectPath("mydb", "mytable"), false);
  7. // alter table
  8. catalog.alterTable(new ObjectPath("mydb", "mytable"), new CatalogTableImpl(...), false);
  10. // rename table
  11. catalog.renameTable(new ObjectPath("mydb", "mytable"), "my_new_table");
  13. // get table
  14. catalog.getTable("mytable");
  16. // check if a table exist or not
  17. catalog.tableExists("mytable");
  19. // list tables in a database
  20. catalog.listTables("mydb");

Python

  1. from pyflink.table import *
  2. from pyflink.table.catalog import CatalogBaseTable, ObjectPath
  3. from pyflink.table.descriptors import Kafka
  5. table_schema = TableSchema.builder() \
  6.     .field("name", DataTypes.STRING()) \
  7.     .field("age", DataTypes.INT()) \
  8.     .build()
  10. table_properties = Kafka() \
  11.     .version("0.11") \
  12.     .start_from_earlist() \
  13.     .to_properties()
  15. catalog_table = CatalogBaseTable.create_table(schema=table_schema, properties=table_properties, comment="my comment")
  17. # create table
  18. catalog.create_table(ObjectPath("mydb", "mytable"), catalog_table, False)
  20. # drop table
  21. catalog.drop_table(ObjectPath("mydb", "mytable"), False)
  23. # alter table
  24. catalog.alter_table(ObjectPath("mydb", "mytable"), catalog_table, False)
  26. # rename table
  27. catalog.rename_table(ObjectPath("mydb", "mytable"), "my_new_table")
  29. # get table
  30. catalog.get_table("mytable")
  32. # check if a table exist or not
  33. catalog.table_exists("mytable")
  35. # list tables in a database
  36. catalog.list_tables("mydb")

View operations

Java/Scala

  1. // create view
  2. catalog.createTable(new ObjectPath("mydb", "myview"), new CatalogViewImpl(...), false);
  4. // drop view
  5. catalog.dropTable(new ObjectPath("mydb", "myview"), false);
  7. // alter view
  8. catalog.alterTable(new ObjectPath("mydb", "mytable"), new CatalogViewImpl(...), false);
  10. // rename view
  11. catalog.renameTable(new ObjectPath("mydb", "myview"), "my_new_view", false);
  13. // get view
  14. catalog.getTable("myview");
  16. // check if a view exist or not
  17. catalog.tableExists("mytable");
  19. // list views in a database
  20. catalog.listViews("mydb");

Python

  1. from pyflink.table import *
  2. from pyflink.table.catalog import CatalogBaseTable, ObjectPath
  4. table_schema = TableSchema.builder() \
  5.     .field("name", DataTypes.STRING()) \
  6.     .field("age", DataTypes.INT()) \
  7.     .build()
  9. catalog_table = CatalogBaseTable.create_view(
  10.     original_query="select * from t1",
  11.     expanded_query="select * from test-catalog.db1.t1",
  12.     schema=table_schema,
  13.     properties={},
  14.     comment="This is a view"
  15. )
  17. catalog.create_table(ObjectPath("mydb", "myview"), catalog_table, False)
  19. # drop view
  20. catalog.drop_table(ObjectPath("mydb", "myview"), False)
  22. # alter view
  23. catalog.alter_table(ObjectPath("mydb", "mytable"), catalog_table, False)
  25. # rename view
  26. catalog.rename_table(ObjectPath("mydb", "myview"), "my_new_view", False)
  28. # get view
  29. catalog.get_table("myview")
  31. # check if a view exist or not
  32. catalog.table_exists("mytable")
  34. # list views in a database
  35. catalog.list_views("mydb")

Partition operations

Java/Scala

  1. // create view
  2. catalog.createPartition(
  3.     new ObjectPath("mydb", "mytable"),
  4.     new CatalogPartitionSpec(...),
  5.     new CatalogPartitionImpl(...),
  6.     false);
  8. // drop partition
  9. catalog.dropPartition(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...), false);
  11. // alter partition
  12. catalog.alterPartition(
  13.     new ObjectPath("mydb", "mytable"),
  14.     new CatalogPartitionSpec(...),
  15.     new CatalogPartitionImpl(...),
  16.     false);
  18. // get partition
  19. catalog.getPartition(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));
  21. // check if a partition exist or not
  22. catalog.partitionExists(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));
  24. // list partitions of a table
  25. catalog.listPartitions(new ObjectPath("mydb", "mytable"));
  27. // list partitions of a table under a give partition spec
  28. catalog.listPartitions(new ObjectPath("mydb", "mytable"), new CatalogPartitionSpec(...));
  30. // list partitions of a table by expression filter
  31. catalog.listPartitionsByFilter(new ObjectPath("mydb", "mytable"), Arrays.asList(epr1, ...));

Python

  1. from pyflink.table.catalog import ObjectPath, CatalogPartitionSpec, CatalogPartition
  3. catalog_partition = CatalogPartition.create_instance({}, "my partition")
  5. catalog_partition_spec = CatalogPartitionSpec({"third": "2010", "second": "bob"})
  6. catalog.create_partition(
  7.     ObjectPath("mydb", "mytable"),
  8.     catalog_partition_spec,
  9.     catalog_partition,
  10.     False)
  12. # drop partition
  13. catalog.drop_partition(ObjectPath("mydb", "mytable"), catalog_partition_spec, False)
  15. # alter partition
  16. catalog.alter_partition(
  17.     ObjectPath("mydb", "mytable"),
  18.     CatalogPartitionSpec(...),
  19.     catalog_partition,
  20.     False)
  22. # get partition
  23. catalog.get_partition(ObjectPath("mydb", "mytable"), catalog_partition_spec)
  25. # check if a partition exist or not
  26. catalog.partition_exists(ObjectPath("mydb", "mytable"), catalog_partition_spec)
  28. # list partitions of a table
  29. catalog.list_partitions(ObjectPath("mydb", "mytable"))
  31. # list partitions of a table under a give partition spec
  32. catalog.list_partitions(ObjectPath("mydb", "mytable"), catalog_partition_spec)

Function operations

Java/Scala

  1. // create function
  2. catalog.createFunction(new ObjectPath("mydb", "myfunc"), new CatalogFunctionImpl(...), false);
  4. // drop function
  5. catalog.dropFunction(new ObjectPath("mydb", "myfunc"), false);
  7. // alter function
  8. catalog.alterFunction(new ObjectPath("mydb", "myfunc"), new CatalogFunctionImpl(...), false);
  10. // get function
  11. catalog.getFunction("myfunc");
  13. // check if a function exist or not
  14. catalog.functionExists("myfunc");
  16. // list functions in a database
  17. catalog.listFunctions("mydb");

Python

  1. from pyflink.table.catalog import ObjectPath, CatalogFunction
  3. catalog_function = CatalogFunction.create_instance(class_name="my.python.udf")
  5. # create function
  6. catalog.create_function(ObjectPath("mydb", "myfunc"), catalog_function, False)
  8. # drop function
  9. catalog.drop_function(ObjectPath("mydb", "myfunc"), False)
  11. # alter function
  12. catalog.alter_function(ObjectPath("mydb", "myfunc"), catalog_function, False)
  14. # get function
  15. catalog.get_function("myfunc")
  17. # check if a function exist or not
  18. catalog.function_exists("myfunc")
  20. # list functions in a database
  21. catalog.list_functions("mydb")

Table API and SQL for Catalog

Registering a Catalog

Users have access to a default in-memory catalog named default_catalog, that is always created by default. This catalog by default has a single database called default_database. Users can also register additional catalogs into an existing Flink session.

Java/Scala

  1. tableEnv.registerCatalog(new CustomCatalog("myCatalog"));

Python

  1. t_env.register_catalog(catalog)

YAML

All catalogs defined using YAML must provide a type property that specifies the type of catalog. The following types are supported out of the box.

CatalogType Value
GenericInMemorygeneric_in_memory
Hivehive
  1. catalogs:
  2.    - name: myCatalog
  3.      type: custom_catalog
  4.      hive-conf-dir: ...

Changing the Current Catalog And Database

Flink will always search for tables, views, and UDF’s in the current catalog and database.

Java/Scala

  1. tableEnv.useCatalog("myCatalog");
  2. tableEnv.useDatabase("myDb");

Python

  1. t_env.use_catalog("myCatalog")
  2. t_env.use_database("myDb")

SQL

  1. Flink SQL> USE CATALOG myCatalog;
  2. Flink SQL> USE myDB;

Metadata from catalogs that are not the current catalog are accessible by providing fully qualified names in the form catalog.database.object.

Java/Scala

  1. tableEnv.from("not_the_current_catalog.not_the_current_db.my_table");

Python

  1. t_env.from_path("not_the_current_catalog.not_the_current_db.my_table")

SQL

  1. Flink SQL> SELECT * FROM not_the_current_catalog.not_the_current_db.my_table;

List Available Catalogs

Java/Scala

  1. tableEnv.listCatalogs();

Python

  1. t_env.list_catalogs()

SQL

  1. Flink SQL> show catalogs;

List Available Databases

Java/Scala

  1. tableEnv.listDatabases();

Python

  1. t_env.list_databases()

SQL

  1. Flink SQL> show databases;

List Available Tables

Java/Scala

  1. tableEnv.listTables();

Python

  1. t_env.list_tables()

SQL

  1. Flink SQL> show tables;

Catalog Modification Listener

Flink supports registering customized listener for catalog modification, such as database and table ddl. Flink will create a CatalogModificationEvent event for ddl and notify CatalogModificationListener. You can implement a listener and do some customized operations when receiving the event, such as report the information to some external meta-data systems.

Implement Catalog Listener

There are two interfaces for the catalog modification listener: CatalogModificationListenerFactory to create the listener and CatalogModificationListener to receive and process the event. You need to implement these interfaces and below is an example.

  1. /** Factory used to create a {@link CatalogModificationListener} instance. */
  2. public class YourCatalogListenerFactory implements CatalogModificationListenerFactory {
  3.     /** The identifier for the customized listener factory, you can named it yourself. */
  4.     private static final String IDENTIFIER = "your_factory";
  6.     @Override
  7.     public String factoryIdentifier() {
  8.         return IDENTIFIER;
  9.     }
  11.     @Override
  12.     public CatalogModificationListener createListener(Context context) {
  13.         return new YourCatalogListener(Create http client from context);
  14.     }
  15. }
  17. /** Customized catalog modification listener. */
  18. public class YourCatalogListener implements CatalogModificationListener {
  19.     private final HttpClient client;
  21.     YourCatalogListener(HttpClient client) {
  22.         this.client = client;
  23.     }
  25.     @Override
  26.     public void onEvent(CatalogModificationEvent event) {
  27.         // Report the database and table information via http client.
  28.     }
  29. }

Register Catalog Listener

After implemented above catalog modification factory and listener, you can register it to the table environment.

  1. Configuration configuration = new Configuration();
  3. // Add the factory identifier, you can set multiple listeners in the configuraiton.
  4. configuration.set(TableConfigOptions.TABLE_CATALOG_MODIFICATION_LISTENERS, Arrays.asList("your_factory"));
  5. TableEnvironment env = TableEnvironment.create(
  6.             EnvironmentSettings.newInstance()
  7.                 .withConfiguration(configuration)
  8.                 .build());
  10. // Create/Alter/Drop database and table.
  11. env.executeSql("CREATE TABLE ...").wait();

For sql-gateway, you can add the option table.catalog-modification.listeners in the flink-conf.yaml and start the gateway, or you can also use SET to specify the listener for ddl, for example, in sql-client or jdbc-driver.

  1. Flink SQL> SET 'table.catalog-modification.listeners' = 'your_factory';
  2. Flink SQL> CREATE TABLE test_table(...);

Catalog Store

Catalog Store is used to store the configuration of catalogs. When using Catalog Store, the configurations of catalogs created in the session will be persisted in the corresponding external system of Catalog Store. Even if the session is reconstructed, previously created catalogs can still be retrieved from Catalog Store.

Configure Catalog Store

Users can configure the Catalog Store in different ways, one is to use the Table API, and another is to use YAML configuration.

Register a catalog store using catalog store instance:

  1. // Initialize a catalog Store instance
  2. CatalogStore catalogStore = new FileCatalogStore("file:///path/to/catalog/store/");
  4. // set up the catalog store
  5. final EnvironmentSettings settings =
  6.         EnvironmentSettings.newInstance().inBatchMode()
  7.         .withCatalogStore(catalogStore)
  8.         .build();

Register a catalog store using configuration:

  1. // Set up configuration
  2. Configuration configuration = new Configuration();
  3. configuration.set("table.catalog-store.kind", "file");
  4. configuration.set("table.catalog-store.file.path", "file:///path/to/catalog/store/");
  5. // set up the configuration.
  6. final EnvironmentSettings settings =
  7.         EnvironmentSettings.newInstance().inBatchMode()
  8.         .withConfiguration(configuration)
  9.         .build();
  11. final TableEnvironment tableEnv = TableEnvironment.create(settings);

In SQL Gateway, it is recommended to configure the settings in a yaml file so that all sessions can automatically use the pre-created Catalog. Usually, you need to configure the kind of Catalog Store and other required parameters for the Catalog Store.

  1. table.catalog-store.kind: file
  2. table.catalog-store.file.path: file:///path/to/catalog/store/

Catalog Store Type

Flink has two built-in Catalog Stores, namely GenericInMemoryCatalogStore and FileCatalogStore, but the Catalog Store model is extendable, so users can also implement their own custom Catalog Store.

GenericInMemoryCatalogStore

GenericInMemoryCatalogStore is an implementation of CatalogStore that saves configuration information in memory. All catalog configurations are only available within the sessions’ lifecycle, and the stored catalog configurations will be automatically cleared after the session is closed.

By default, if no Catalog Store related configuration is specified, the system uses this implementation.

FileCatalogStore

FileCatalogStore can save the Catalog configuration to a file. To use FileCatalogStore, you need to specify the directory where the Catalog configurations needs to be saved. Each Catalog will have its own file named the same as the Catalog Name.

The FileCatalogStore implementation supports both local and remote file systems that are available via the Flink FileSystem abstraction. If the given Catalog Store path does not exist either completely or partly, FileCatalogStore will try to create the missing directories.

If the given Catalog Store path does not exist and FileCatalogStore fails to create a directory, the Catalog Store cannot be initialized, hence an exception will be thrown. In case the FileCatalogstore initialization is not successful, both SQL Client and SQL Gateway will be broken.

Here is an example directory structure representing the storage of Catalog configurations using FileCatalogStore:

  1. - /path/to/save/the/catalog/
  2.   - catalog1.yaml
  3.   - catalog2.yaml
  4.   - catalog3.yaml

Catalog Store Configuration

The following options can be used to adjust the Catalog Store behavior.

KeyDefaultTypeDescription
table.catalog-store.kind
“generic_in_memory”StringThe kind of catalog store to be used. Out of the box, ‘generic_in_memory’ and ‘file’ options are supported.
table.catalog-store.file.path
(none)StringThe configuration option for specifying the path to the file catalog store root directory.

Custom Catalog Store

Catalog Store is extensible, and users can customize Catalog Store by implementing its interface. If SQL CLI or SQL Gateway needs to use Catalog Store, the corresponding CatalogStoreFactory interface also needs to be implemented for this Catalog Store.

  1. public class CustomCatalogStoreFactory implements CatalogStoreFactory {
  3.     public static final String IDENTIFIER = "custom-kind";
  5.     // Used to connect external storage systems
  6.     private CustomClient client;
  8.     @Override
  9.     public CatalogStore createCatalogStore() {
  10.         return new CustomCatalogStore();
  11.     }
  13.     @Override
  14.     public void open(Context context) throws CatalogException {
  15.         // initialize the resources, such as http client
  16.         client = initClient(context);
  17.     }
  19.     @Override
  20.     public void close() throws CatalogException {
  21.         // release the resources
  22.     }
  24.     @Override
  25.     public String factoryIdentifier() {
  26.         // table store kind identifier
  27.         return IDENTIFIER;
  28.     }
  30.     public Set<ConfigOption<?>> requiredOptions() {
  31.         // define the required options
  32.         Set<ConfigOption> options = new HashSet();
  33.         options.add(OPTION_1);
  34.         options.add(OPTION_2);
  36.         return options;
  37.     }
  39.     @Override
  40.     public Set<ConfigOption<?>> optionalOptions() {
  41.         // define the optional options
  42.     }
  43. }
  45. public class CustomCatalogStore extends AbstractCatalogStore {
  47.     private Client client;
  49.     public CustomCatalogStore(Client client) {
  50.         this.client = client;
  51.     }
  53.     @Override
  54.     public void storeCatalog(String catalogName, CatalogDescriptor catalog)
  55.             throws CatalogException {
  56.         // store the catalog
  57.     }
  59.     @Override
  60.     public void removeCatalog(String catalogName, boolean ignoreIfNotExists)
  61.             throws CatalogException {
  62.         // remove the catalog descriptor
  63.     }
  65.     @Override
  66.     public Optional<CatalogDescriptor> getCatalog(String catalogName) {
  67.         // retrieve the catalog configuration and build the catalog descriptor
  68.     }
  70.     @Override
  71.     public Set<String> listCatalogs() {
  72.         // list all catalogs
  73.     }
  75.     @Override
  76.     public boolean contains(String catalogName) {
  77.     }
  78. }