/**
 * Different events will be fired when a catalog/database/table is modified. The customized listener can get and
 * report specific information from the event according to the event type.
 */
@PublicEvolving
public interface CatalogModificationListener {
    /** The event will be fired when the database/table is modified

/**
 * Job creation listener, client will create specific event and notify this listener.
 */
@PublicEvolving
public interface JobCreationListener {
	/* Start the listener. */
	void start(Map<String, String> config);

    /* Event fired after a catalog has been registered. */
    void onRegisterCatalog(CatalogEvent catalogEvent);

    void onUnregisterCatalog(UnregisterCatalogEvent catalogEvent);

    /* Event fired after a database has been created. */
    void onCreateDatabase(DatabaseEvent databaseEvent);

    /* Event fired after a database has been dropped. */
    void onDropDatabaseonEvent(DatabaseEventCatalogModificationEvent databaseEventevent);
}

    /* EventBasic firedinterface afterfor a table has been createdcatalog modification. */
@PublicEvolving
public interface class  void onCreateTable(CreateTableEvent tableEvent);

CatalogModificationEvent {
    /* EventContext firedfor after a table has been changedthe event. */
    CatalogContext void onAlterTable(AlterTableEvent tableEventcontext();
}

    /* EventContext firedfor aftercatalog amodification tableand hasjob beenlineage droppedevents. */
@PublicEvolving
public interface CatalogContext {
 void onDropTable(DropTableEvent tableEvent); 

    /* EventThe firedname beforeof a job is submitted to do some validations. */catalog. */
    voidString onJobPreSubmissiongetCatalogName(JobSubmissionEvent submitEvent); 

    /* EventClass firedof after a job is submittedthe catalog. */
    Class<? extends voidCatalog> onJobSubmissiongetClass(JobSubmissionEvent submitEvent); 
 
    /* EventIdentifier for the catalog registration, providesfrom catalog namefactory, defaultsuch database, database list and properties in the Catalogas jdbc/iceberg/paimon. */
    @PublicEvolvingOptional<String> getFactoryIdentifier();

    interface/* CatalogEventConfig {
for catalog. */
       String catalog();
        Catalog catalog();
    }
 
    Configuration getConfiguration();
}

/* The basic class for database related event. */
public interface DatabaseModificationEvent extends CatalogModificationEvent {
    CatalogDatabase database();
}

/* Event for catalogcreating unregistrationdatabase. */ 
    @PublicEvolving
public interface CreateDatabaseEvent extends interface UnregisterCatalogEventDatabaseModificationEvent {
    boolean ignoreIfExists();
}

/* Event for altering String catalog();
    database. */
@PublicEvolving
public interface AlterDatabaseEvent extends DatabaseModificationEvent {
    booleanCatalogDatabase ignoreIfNotExistsnewDatabase();
    boolean ignoreIfNotExists();
}

    /* Event for dropping database creation, provides catalog name, database name, comment and properties of the database. */
    @PublicEvolving
    interface DatabaseEvent {
        String catalog();
        String name();
        CatalogDatabase database();
        . */
@PublicEvolving
public interface DropDatabaseEvent extends DatabaseModificationEvent {
    boolean ignoreIfExists();
}

/**
 * Base table event, provides column list, primary keys, partition keys, watermarks and properties in
 * CatalogBaseTable. The table can be source or sink.
 */
public interface TableModificationEvent extends CatalogModificationEvent {
    ObjectIdentifier identifier();
    CatalogBaseTable table();
}

/* Event for creating table. */
@PublicEvolving
public interface CreateTableEvent extends CatalogModificationEvent {
    boolean ignoreIfExists();
    }

/* Event for altering table, provides /*all Eventchanges for droppingold databasetable. */
@PublicEvolving
public interface AlterTableEvent extends CatalogModificationEvent @PublicEvolving{
    interface DropDatabaseEvent {List<TableChange> tableChanges();
        String catalogboolean ignoreIfExists();
}

/* Event for dropping table. */
@PublicEvolving
public interface DropTableEvent String name(); 
    extends CatalogModificationEvent {
    boolean ignoreIfExists();
    }

/* Factory for catalog /*modification Table information event, provides column list, primary keys, partition keys, watermarks and properties in the table. The table can be source or sink. */
    @PublicEvolving
    interface TableEvent {
        ObjectIdentifier identifier();  
        CatalogBaseTable table();
    }

    /* Event for table creation. */
    @PublicEvolving
    interface CreateTableEvent extends TableEvent {
        boolean ignoreIfExists();
    }

    /* Event for altering table, provides all information in old table and new table. */
    @PublicEvolving
    interface AlterTableEvent extends TableEvent {
        CreateTableEvent newTable();
        boolean ignoreIfExists();
    }

    /* Event for dropping table. */
    @PublicEvolving
    interface DropTableEvent {
        ObjectIdentifier identifier();
        boolean ignoreIfExists();   
    }

    /* Event for before and after job is submitted, provides source tables and sink tables. */
    @PublicEvolving
    interface JobSubmissionEvent {
        List<TableEvent> sources();
        List<TableEvent> sinks();
    }
}

/* Kafka storage identifier options. */
"properties.bootstrap.servers" for Kafka bootstrap servers
"topic" for Kafka Topic
"properties.group.id" for Kafka group id
"topic-pattern" for Kafka topic pattern

# Config for catalog modification listeners.
table.catalog-modification.listeners: {table catalog listener factory1},{table catalog listener factory2}

-- Create a table my_kafka_table1 for kafka topic 'kafka_topic'
CREATE TABLE my_kafka_table1 (
  val1 STRING,
  val2 STRING,
  val3 STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'kafka_topic',
  'properties.bootstrap.servers' = 'kafka_bootstrap_server',
  'properties.group.id' = 'kafka_group',
  'format' = 'json'
);

-- Create a Paimon catalog and table for warehouse 'paimon_path1'
CREATE CATALOG paimon_catalog WITH (
    'type'='paimon',
    'warehouse'='paimon_path1'
);
USE CATALOG paimon_catalog;
CREATE TABLE my_paimon_table (
    val1 STRING,
    val2 STRING,
    val3 STRING
) WITH (...);

-- Insert data to Paimon table from Kafka
INSERT INTO my_paimon_table SELECT ... FROM default_catalog.default_database.my_kafka_table1 WHERE ...;

-- Create another table my_kafka_table2 for kafka topic 'kafka_topic' which is same as my_kafka_table1 above
CREATE TABLE my_kafka_table2 (
  val1 STRING,
  val2 STRING,
  val3 STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'kafka_topic',
  'properties.bootstrap.servers' = 'kafka_bootstrap_server',
  'properties.group.id' = 'kafka_group',
  'format' = 'json'
);

-- Create a Paimon catalog with the same name 'paimon_catalog' for different warehouse 'paimon_path2'
CREATE CATALOG paimon_catalog WITH (
    'type'='paimon',
    'warehouse'='paimon_path2'
);
USE CATALOG paimon_catalog;
CREATE TABLE my_paimon_table (
    val1 STRING,
    val2 STRING,
    val3 STRING
) WITH (...);

-- Insert data to Paimon table from Kafka
INSERT INTO my_paimon_table SELECT ... FROM default_catalog.default_database.my_kafka_table2 WHERE ...;

-- Create two catalogs for warehouse 'paimon_path1' and 'paimon_path2', there are two different tables 'my_paimon_table'
CREATE CATALOG paimon_catalog1 WITH (
    'type'='paimon',
    'warehouse'='paimon_path1'
);
CREATE CATALOG paimon_catalog2 WITH (
    'type'='paimon',
    'warehouse'='paimon_path2'
);

-- Create mysql table
CREATE TABLE mysql_table (
    val1 STRING,
    val2 STRING,
    val3 STRING
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://HOST:PORT/my_database',
    'table-name' = 'my_mysql_table',
    'username' = '***',
    'password' = '***'
);

-- Insert data to mysql table from two paimon tales
INSERT INTO mysql_table 
    SELECT ... FROM paimon_catalog1.default.my_paimon_table
        JOIN paimon_catalog2.default.my_paimon_table
    ON ... WHERE ...;

/* Listeners and related operations in the catalog manager. */
public final class CatalogManager {
    private final List<CatalogModificationListener> listeners;

    /* Create catalog manager with listener list. */
    private CatalogManager(
            String defaultCatalogName,
            Catalog defaultCatalog,
            DataTypeFactory typeFactory,
            ManagedTableListener managedTableListener,
            List<CatalogModificationListener> listeners);

    /* Notify the listeners with given catalog event. */
    private void notify(CatalogModificationEvent event) {
        listeners.forEach(listener -> listener.onEvent(event));
    }

    /* Notify listener for tables. */
    public void createTable/dropTable/alterTable(...) {
        ....;
        notify(Create Different Table Modification Event With Context);
    }

    /* Add database ddls and notify listener for databases. */
    public void createDatabase/dropDatabase/alterDatabase(...) {
        ....;
        notify(Create Different Database Modification Event With Context); 
    }

    /* Add listeners in Builder for catalog manager. */
    public static final class Builder {
        Builder listeners(List<CatalogModificationListener> listeners);
    }
}