Add protocols tu support iceberg catalogs

[andrea-gioia]

We can use the same properties used by flik SQL...

Flink SQL: CREATE CATALOG Examples

Introduction

In Apache Flink, the CREATE CATALOG command is used to define a new catalog that provides access to tables, databases, and functions stored in an external system (such as Apache Iceberg, Hive, JDBC databases, and Kafka Schema Registry).

---

1. Iceberg Catalog

Apache Iceberg supports multiple catalog backends, such as Hadoop, Hive, and REST.

1.1 Iceberg with Hadoop Catalog (HDFS/S3/Unitstore)

CREATE CATALOG iceberg_hadoop WITH (
  'type' = 'iceberg',
  'catalog-type' = 'hadoop',
  'warehouse' = 's3://my-bucket/warehouse/'
);

➡ Queries:

USE CATALOG iceberg_hadoop;
SELECT * FROM my_db.my_table;

1.2 Iceberg with Hive Metastore

CREATE CATALOG iceberg_hive WITH (
  'type' = 'iceberg',
  'catalog-type' = 'hive',
  'uri' = 'thrift://hive-metastore:9083'
);

➡ Queries:

USE CATALOG iceberg_hive;
SELECT * FROM my_db.my_table;

1.3 Iceberg with REST Metastore

CREATE CATALOG iceberg_rest WITH (
  'type' = 'iceberg',
  'catalog-type' = 'rest',
  'uri' = 'http://iceberg-rest-server:8181'
);

➡ Queries:

USE CATALOG iceberg_rest;
SELECT * FROM my_db.my_table;

---

2. Hive Catalog

Connects Flink to Hive Metastore.

CREATE CATALOG my_hive WITH (
  'type' = 'hive',
  'hive.metastore.uris' = 'thrift://hive-metastore:9083'
);

➡ Queries:

USE CATALOG my_hive;
SELECT * FROM my_hive_db.my_table;

---

3. JDBC Catalog (MySQL, PostgreSQL, etc.)

JDBC Catalog connects Flink to relational databases.

3.1 JDBC (MySQL Example)

CREATE CATALOG my_mysql WITH (
  'type' = 'jdbc',
  'default-database' = 'my_db',
  'username' = 'root',
  'password' = 'password',
  'base-url' = 'jdbc:mysql://localhost:3306/'
);

➡ Queries:

USE CATALOG my_mysql;
SELECT * FROM my_table;

3.2 JDBC (PostgreSQL Example)

CREATE CATALOG my_postgres WITH (
  'type' = 'jdbc',
  'default-database' = 'my_db',
  'username' = 'postgres',
  'password' = 'mypassword',
  'base-url' = 'jdbc:postgresql://localhost:5432/'
);

➡ Queries:

USE CATALOG my_postgres;
SELECT * FROM my_table;

---

4. Generic In-Memory Catalog

This is a default built-in catalog for Flink.

CREATE CATALOG my_memory_catalog WITH (
  'type' = 'generic_in_memory'
);

➡ Queries:

USE CATALOG my_memory_catalog;
CREATE DATABASE my_db;
CREATE TABLE my_db.my_table (id INT, name STRING);
INSERT INTO my_db.my_table VALUES (1, 'Alice');
SELECT * FROM my_db.my_table;

---

5. Kafka Schema Registry Catalog

This catalog integrates Apache Kafka Schema Registry (Avro/Protobuf/JSON).

CREATE CATALOG kafka_registry WITH (
  'type' = 'kafka',
  'kafka.bootstrap.servers' = 'kafka-broker:9092',
  'kafka.schema.registry.url' = 'http://schema-registry:8081'
);

➡ Queries:

USE CATALOG kafka_registry;
SELECT * FROM kafka_topic_table;

---

6. Dropping a Catalog

If you need to remove a catalog, you can use the DROP CATALOG command:

DROP CATALOG my_hive;
DROP CATALOG my_mysql;

---

7. Summary

Catalog Type	Storage Backend / System	Example Configuration
Iceberg (Hadoop)	HDFS, S3, Unitstore	'catalog-type' = 'hadoop'
Iceberg (Hive)	Hive Metastore	'catalog-type' = 'hive'
Iceberg (REST)	Iceberg REST	'catalog-type' = 'rest'
Hive	Apache Hive	'type' = 'hive'
JDBC	MySQL, PostgreSQL	'type' = 'jdbc'
Generic In-Memory	Local memory (for testing)	'type' = 'generic_in_memory'
Kafka Schema Registry	Kafka topics with Avro/Protobuf/JSON schemas	'type' = 'kafka'

---