{% highlight sql %}
CREATE CATALOG mypg WITH(
'type' = 'jdbc',
'default-database' = '...',
'username' = '...',
'password' = '...',
'base-url' = '...'
);
USE CATALOG mypg;
{% endhighlight %}
{% highlight java %}
EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
TableEnvironment tableEnv = TableEnvironment.create(settings);
String name = "mypg";
String defaultDatabase = "mydb";
String username = "...";
String password = "...";
String baseUrl = "..."
JdbcCatalog catalog = new JdbcCatalog(name, defaultDatabase, username, password, baseUrl);
tableEnv.registerCatalog("mypg", catalog);
// set the JdbcCatalog as the current catalog of the session
tableEnv.useCatalog("mypg");
{% endhighlight %}
{% highlight scala %}
val settings = EnvironmentSettings.newInstance().inStreamingMode().build()
val tableEnv = TableEnvironment.create(settings)
val name = "mypg"
val defaultDatabase = "mydb"
val username = "..."
val password = "..."
val baseUrl = "..."
val catalog = new JdbcCatalog(name, defaultDatabase, username, password, baseUrl)
tableEnv.registerCatalog("mypg", catalog)
// set the JdbcCatalog as the current catalog of the session
tableEnv.useCatalog("mypg")
{% endhighlight %}
{% highlight python %}
from pyflink.table.catalog import JdbcCatalog
environment_settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
t_env = StreamTableEnvironment.create(environment_settings=environment_settings)
name = "mypg"
default_database = "mydb"
username = "..."
password = "..."
base_url = "..."
catalog = JdbcCatalog(name, default_database, username, password, base_url)
t_env.register_catalog("mypg", catalog)
# set the JdbcCatalog as the current catalog of the session
t_env.use_catalog("mypg")
{% endhighlight %}
{% highlight yaml %}
execution:
planner: blink
...
current-catalog: mypg # set the JdbcCatalog as the current catalog of the session
current-database: mydb
catalogs:
- name: mypg
type: jdbc
default-database: mydb
username: ...
password: ...
base-url: ...
{% endhighlight %}
#### PostgresSQL Metaspace Mapping
PostgresSQL has an additional namespace as `schema` besides database. A Postgres instance can have multiple databases, each database can have multiple schemas with a default one named "public", each schema can have multiple tables.
In Flink, when querying tables registered by Postgres catalog, users can use either `schema_name.table_name` or just `table_name`. The `schema_name` is optional and defaults to "public".
Therefor the metaspace mapping between Flink Catalog and Postgres is as following:
| Flink Catalog Metaspace Structure | Postgres Metaspace Structure |
| :------------------------------------| :-----------------------------------|
| catalog name (defined in Flink only) | N/A |
| database name | database name |
| table name | [schema_name.]table_name |
The full path of Postgres table in Flink should be ``"