/**
 * 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. */
    void onEvent(CatalogModificationEvent event);
}

/* Basic interface for catalog modification. */
@PublicEvolving
public interface class CatalogModificationEvent {

/**
 * Job deployment listener, client will create specific event and notify this listener when it submits a job.
 */
@PublicEvolving
public interface JobDeploymentListener {
    /* Event fired after a catalog has been registered. */
    void onRegisterCatalog(RegisterCatalogEvent catalogEvent);

    /* Event fired after a catalog has been unregistered. */ 
    void onUnregisterCatalog(UnregisterCatalogEvent catalogEvent);

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

    /* EventContext firedfor after a database has been droppedthe event. */
    CatalogContext void onDropDatabase(DropDatabaseEvent databaseEventcontext();
}

/* Context for catalog /*modification Eventand firedjob after a table has been createdlineage events. */
@PublicEvolving
public interface CatalogContext {
  void onCreateTable(CreateTableEvent tableEvent);

    /* EventThe firedname after a table has been changedof catalog. */
    voidString onAlterTablegetCatalogName(AlterTableEvent tableEvent);

    /* EventClass firedof afterthe a table has been dropped. */
    void onDropTable(DropTableEvent tableEvent); catalog. */
    Class<? extends Catalog> getClass();

    /* EventIdentifier firedfor beforethe acatalog jobfrom iscatalog submittedfactory, tosuch do some validationsas jdbc/iceberg/paimon. */
    voidOptional<String> onJobPreSubmissiongetFactoryIdentifier(JobSubmissionEvent submitEvent); 

    /* EventConfig firedfor after a job is submitted. catalog. */
    Configuration void onJobSubmission(JobSubmissionEvent submitEventgetConfiguration(); 
}
 
/* The basic  /* Eventclass for catalogdatabase registration,related provides catalog name, default database, database list and properties in the Catalogevent. */
public interface DatabaseModificationEvent extends CatalogModificationEvent {
    CatalogDatabase database();
}

/* Event for creating database. */
    @PublicEvolving
public interface CreateDatabaseEvent extends interface RegisterCatalogEventDatabaseModificationEvent {
        String catalogboolean ignoreIfExists();
        Catalog catalog();
    }
 
    /* Event for catalogaltering unregistrationdatabase. */ 
    @PublicEvolving
public interface AlterDatabaseEvent extends interface UnregisterCatalogEventDatabaseModificationEvent {
        String catalogCatalogDatabase newDatabase();
        boolean ignoreIfNotExists();
    }

    /* Event for dropping database creation, provides catalog name, database name, comment and properties of the database. */
    @PublicEvolving
    interface CreateDatabaseEvent {
        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();
        boolean ignoreIfExistsCatalogBaseTable table();
    }

    /* Event for droppingcreating databasetable. */
    @PublicEvolving
public interface CreateTableEvent extends interface DropDatabaseEventCatalogModificationEvent {
        String catalogboolean ignoreIfExists();
}

/* Event for altering table, provides all changes for String name(); 
        boolean ignoreIfExistsold table. */
@PublicEvolving
public interface AlterTableEvent extends CatalogModificationEvent {
    List<TableChange> tableChanges();
    boolean ignoreIfExists();
}

    /* TableEvent informationfor event,dropping provides column list, primary keys, partition keys, watermarks and properties in the table. The table can be source or sinktable. */
@PublicEvolving
public interface DropTableEvent extends CatalogModificationEvent {
    boolean ignoreIfExists();
}

/* Factory for catalog modification listener. */
    @PublicEvolving
public    interface TableEventCatalogModificationListenerFactory {
     CatalogModificationListener    ObjectIdentifier identifier(createListener(Context context);  

        CatalogBaseTable@PublicEvolving table();
    }

public interface   /* Event for table creation. */Context {
    @PublicEvolving
    interface CreateTableEvent extends TableEvent {Configuration getConfiguration();
        booleanClassLoader ignoreIfExistsgetUserClassLoader();
    }
	/*
    /* Event	* Get an Executor pool for alteringthe table,listener providesto allrun informationasync inoperations oldthat tablecan andpotentially newbe table. */IO-heavy.
    @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 {
        JobID jobId();
        String jobName();
        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);
    }
}