What are Flink SQL catalogs and why are they important for production

-- Fully qualified reference
SELECT * FROM hive_catalog.nbi_db.accounts;

-- Switch to a catalog and database
USE CATALOG hive_catalog;
USE nbi_db;
SELECT * FROM accounts;  -- resolves to hive_catalog.nbi_db.accounts

package com.bank.flink;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class InMemoryCatalogDemo {
    public static void main(String[] args) {
        TableEnvironment tEnv = TableEnvironment.create(
            EnvironmentSettings.inStreamingMode());

        // This table exists only in memory
        tEnv.executeSql(
            "CREATE TABLE transactions (" +
            "  txn_id STRING, amount DECIMAL(15,2)" +
            ") WITH ('connector'='kafka', 'topic'='nbi-txns', 'format'='json'," +
            "  'properties.bootstrap.servers'='kafka.bank.internal:9092')"
        );

        // When this process exits, the table definition is gone.
        // Next time, you must re-run the CREATE TABLE.
    }
}

-- Register a HiveCatalog
CREATE CATALOG hive_catalog WITH (
    'type' = 'hive',
    'hive-conf-dir' = '/opt/hive-conf'
);

USE CATALOG hive_catalog;
CREATE DATABASE IF NOT EXISTS nbi_db;
USE nbi_db;

-- This table definition is now stored in the Hive Metastore.
-- It survives cluster restarts and is visible to all jobs.
CREATE TABLE transactions (
    txn_id     STRING,
    account_id STRING,
    amount     DECIMAL(15, 2),
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic'     = 'nbi-transactions',
    'format'    = 'json',
    'properties.bootstrap.servers' = 'kafka.bank.internal:9092'
);

package com.bank.flink;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class HiveCatalogDemo {
    public static void main(String[] args) {
        TableEnvironment tEnv = TableEnvironment.create(
            EnvironmentSettings.inStreamingMode());

        // Register HiveCatalog programmatically
        tEnv.executeSql(
            "CREATE CATALOG hive_catalog WITH (" +
            "  'type' = 'hive'," +
            "  'hive-conf-dir' = '/opt/hive-conf'" +
            ")"
        );

        tEnv.executeSql("USE CATALOG hive_catalog");
        // Now tables persist in the Hive Metastore
    }
}

CREATE CATALOG pg_catalog WITH (
    'type'             = 'jdbc',
    'base-url'         = 'jdbc:postgresql://db.bank.internal:5432/',
    'default-database' = 'nbi',
    'username'         = '${JDBC_USER}',
    'password'         = '${JDBC_PASS}'
);

USE CATALOG pg_catalog;

-- The PostgreSQL accounts table is automatically available
SELECT account_id, balance FROM nbi.accounts WHERE status = 'ACTIVE';
-- No CREATE TABLE needed: the catalog discovers it from the database

// Production pattern: register catalog at session start
TableEnvironment tEnv = TableEnvironment.create(
    EnvironmentSettings.inStreamingMode());
tEnv.executeSql("CREATE CATALOG hive_catalog WITH ('type'='hive', ...)");
tEnv.executeSql("USE CATALOG hive_catalog");
tEnv.executeSql("USE nbi_db");
// All tables are now available from the persistent catalog

-- Reusable view stored in the catalog
CREATE VIEW active_accounts AS
SELECT * FROM accounts WHERE status = 'ACTIVE';

-- Reusable function stored in the catalog
CREATE FUNCTION currency_convert AS 'com.bank.udf.CurrencyConvertFunction';

-- Any team can use these without redeclaring
SELECT account_id, currency_convert(balance, 'USD', 'INR') FROM active_accounts;

-- Cross-catalog query: join Kafka stream with PostgreSQL lookup
SELECT t.txn_id, t.amount, a.holder_name
FROM hive_catalog.nbi_db.transactions t
JOIN pg_catalog.nbi.accounts a
    ON t.account_id = a.account_id;