Supported catalogs

The Gravitino Flink connector exposes Hive, Iceberg, Paimon, and JDBC catalogs to Flink. Each is backed by the corresponding Flink connector and behaves the way that connector behaves. The difference is that catalogs come from Gravitino rather than from CREATE CATALOG statements in Flink.

Setup

Every catalog type needs the two connector jars from Usage in Flink's lib directory. Some types need more jars next to them.

Catalog	Additional jars
Hive	None
Iceberg	Iceberg Flink runtime, version-matched (see Iceberg)
Paimon	Paimon Flink connector for your Flink minor version
JDBC	Flink JDBC connector for your Flink minor version, plus the database's JDBC driver

A catalog is selected by its Gravitino name with USE CATALOG.

Catalog properties prefixed with flink.bypass. pass to the underlying Flink connector with the prefix removed, so flink.bypass.clients sets clients. The Hive catalog accepts only the flink.bypass. keys listed in its table below.

Across all types, function operations, partition operations, querying UDFs, LOAD, UNLOAD, CREATE TABLE LIKE, TRUNCATE TABLE, UPDATE, DELETE, and CALL are not supported.

Hive

Requires Hive metastore 2.x and HDFS 2.x or 3.x. Most other DDL and DML work.

Flink distinguishes Hive tables from generic tables, following Flink's own Hive catalog. A table created with 'connector'='hive' is a Hive table with a normal Hive schema. Any other connector, or none, makes a generic table: Gravitino stores an empty Hive schema and serializes the schema and partition keys into flink.* properties, detected on load through is_generic, flink.connector, and flink.connector.type. Set 'connector'='hive' explicitly for a raw Hive table, because generic is the default. Flink 1.18 deprecated the managed table APIs, so do not rely on managed tables.

Create a Hive table and a generic table

USE CATALOG {hive_catalog};
CREATE DATABASE IF NOT EXISTS {database};
USE {database};

CREATE TABLE IF NOT EXISTS employees (id INT, name STRING, dt INT)
PARTITIONED BY (dt) WITH ('connector'='hive');

CREATE TABLE IF NOT EXISTS jdbc_table (id INT, name STRING) WITH (
  'connector'='jdbc',
  'url'='jdbc:postgresql://{postgres_host}:5432/{database}',
  'table-name'='jdbc_table',
  'username'='{user}',
  'password'='{password}'
);

Views support CREATE VIEW, DROP VIEW, ALTER VIEW (rename and replace the definition), list, and load. The connector stores the SQL with the flink dialect and records flink.schema.num-columns, the dialect marker the Hive catalog requires. Loading tries the flink dialect, then hive. A view created by another engine, such as Spark, appears in SHOW VIEWS but cannot be loaded.

Gravitino catalog property	Flink Hive connector setting	Description
metastore.uris	hive.metastore.uris	Hive metastore URI
flink.bypass.default-database	default-database	Hive default database
flink.bypass.hive-conf-dir	hive-conf-dir	Hive conf directory
flink.bypass.hive-version	hive-version	Hive version
flink.bypass.hadoop-conf-dir	hadoop-conf-dir	Hadoop conf directory

Catalog properties starting with hadoop., dfs., fs., or hive. also pass through and override values from the conf directories.

Iceberg

CREATE DATABASE, CREATE TABLE, DROP TABLE, ALTER TABLE, INSERT INTO, INSERT OVERWRITE, and SELECT work. Metadata tables such as {table}$snapshots are not supported.

The Iceberg version on the Flink side differs from the server's 1.11.0 for older Flink versions. Take the runtime jar for your Flink version, and do not mix Iceberg versions on one classpath.

Flink	Scala	Iceberg	Iceberg Flink runtime jar
1.18	2.12	1.9.2	iceberg-flink-runtime-1.18-1.9.2.jar
1.19	2.12	1.10.2	iceberg-flink-runtime-1.19-1.10.2.jar
1.20	2.12	1.11.0	iceberg-flink-runtime-1.20-1.11.0.jar

Create, write and query an Iceberg table

USE CATALOG {iceberg_catalog};
CREATE DATABASE IF NOT EXISTS {database};
USE {database};

CREATE TABLE sample (id BIGINT COMMENT 'unique id', data STRING NOT NULL)
PARTITIONED BY (data) WITH ('format-version'='2');

INSERT INTO sample VALUES (1, 'A'), (2, 'B');
SELECT * FROM sample WHERE data = 'B';

Views support CREATE VIEW, DROP VIEW, ALTER VIEW (rename and replace the definition), list, and load, for views the Iceberg backend manages. The connector stores the SQL with the flink dialect and loads with flink, then hive. Other dialects on the same view, such as spark, are kept.

Gravitino catalog property	Flink Iceberg connector setting	Description
catalog-backend	catalog-type	Catalog backend: hive and rest are tested; jdbc is not
uri	uri	Catalog backend URI
warehouse	warehouse	Catalog backend warehouse
io-impl	io-impl	The FileIO implementation
oss-endpoint	oss.endpoint	Aliyun OSS endpoint
oss-access-key-id	client.access-key-id	OSS access key
oss-secret-access-key	client.access-key-secret	OSS secret key

An OSS warehouse also needs aliyun-sdk-oss-3.10.2.jar, hamcrest-core-1.1.jar, and jdom2-2.0.6.jar on the Flink classpath, all from the Aliyun OSS SDK.

Paimon

Append-only tables and primary key tables work. Distribution is hash only, set with the bucket-key and bucket table properties; with bucket-key alone, the bucket count is chosen automatically. Paimon 1.2.0 is the tested version, with the jar paimon-flink-{flink_minor}-1.2.0.jar for Flink 1.18, 1.19, or 1.20.

The connector passes the catalog backend and the S3 and OSS credentials through to Paimon, so the filesystem, hive, and jdbc backends all work, as does a filesystem warehouse on HDFS, S3, or OSS. Object storage also needs Paimon's file IO jar for that store (paimon-s3 or paimon-oss) on the Flink classpath. The connector's tests cover the three backends on a local or HDFS warehouse, not object storage.

Create a bucketed primary key table

USE CATALOG {paimon_catalog};

CREATE TABLE paimon_bucketed_table (
    id BIGINT,
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH ('bucket-key' = 'id', 'bucket' = '4');

Views support CREATE VIEW, DROP VIEW, ALTER VIEW (rename and replace the definition), list, and load, where the Paimon backend implements views. The connector stores the SQL twice, with the flink dialect and with Paimon's query dialect, and loads with flink, then hive, then query.

Gravitino catalog property	Flink Paimon connector setting	Description
catalog-backend	metastore	filesystem, hive, or jdbc
uri	uri	Metastore URI or JDBC URL, for the hive and jdbc backends
warehouse	warehouse	Warehouse path, such as hdfs://{namenode}/{path}, s3://{bucket}/{path}, or oss://{bucket}/{path}
s3-endpoint	s3.endpoint	S3 endpoint
s3-access-key-id	s3.access-key	S3 access key
s3-secret-access-key	s3.secret-key	S3 secret key
oss-endpoint	fs.oss.endpoint	OSS endpoint
oss-access-key-id	fs.oss.accessKeyId	OSS access key
oss-secret-access-key	fs.oss.accessKeySecret	OSS secret key

JDBC

MySQL and PostgreSQL catalogs work with the open source connector jar. Oracle and SQL Server work through the Datastrato extension jar described in Usage.

Flink	Flink JDBC connector
1.18	3.2.0-1.18
1.19	3.3.0-1.19
1.20	3.3.0-1.20

Set the catalog property flink.bypass.default-database to the database Flink opens by default.

Use a JDBC catalog

USE CATALOG {jdbc_catalog};
CREATE DATABASE {database};
USE {database};

CREATE TABLE jdbc_table_a (aa BIGINT NOT NULL PRIMARY KEY NOT ENFORCED, bb BIGINT);
INSERT INTO jdbc_table_a VALUES (1, 2);
INSERT INTO jdbc_table_a VALUES (1, 4);
SELECT * FROM jdbc_table_a;

With a primary key, the second insert updates the row in place, so the query returns 1, 4.

Gravitino catalog property	Flink JDBC connector setting	Description
jdbc-url	base-url	JDBC URL of the database
jdbc-user	username	Database user
jdbc-password	password	Password for that user
flink.bypass.default-database	default-database	Default database to connect to

Related

• Flink connector
• Flink authentication
• Connecting a catalog