/**
 * 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 before a job is submitted to do some validationsof catalog. */
    voidString onJobPreSubmissiongetCatalogName(JobSubmissionEvent submitEvent); 

    /* EventClass firedof afterthe a job is submitted. */
    void onJobSubmission(JobSubmissionEvent submitEvent); 
 catalog. */
    Class<? extends Catalog> getClass();

    /* EventIdentifier for the catalog registration,from provides catalog namefactory, defaultsuch database, database list and properties in the Catalogas jdbc/iceberg/paimon. */
    @PublicEvolving
  Optional<String> getFactoryIdentifier();

  interface CatalogEvent {
/* Config for catalog. */
    Configuration String cataloggetConfiguration();
}

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

/* Event for creating database. */ 
    @PublicEvolving
public interface CreateDatabaseEvent extends interface UnregisterCatalogEventDatabaseModificationEvent {
        String catalog();
        boolean ignoreIfNotExistsignoreIfExists();
    }

    /* Event for database creation, provides catalog name, database name, comment and properties of the altering database. */
    @PublicEvolving
public interface AlterDatabaseEvent extends interface DatabaseEventDatabaseModificationEvent {
        String catalogCatalogDatabase newDatabase();
        String nameboolean ignoreIfNotExists();
}

/*  Event  for  dropping  CatalogDatabase database();
     database. */
@PublicEvolving
public interface DropDatabaseEvent extends DatabaseModificationEvent {
    boolean ignoreIfExists();
    }

    /**
 Event* forBase droppingtable database. */
    @PublicEvolving
    interface DropDatabaseEvent {
        String catalog();
        String name(); 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 creating /* Table information event, provides column list, primary keys, partition keys, watermarks and properties in the table. The table can be source or sinktable. */
@PublicEvolving
public interface CreateTableEvent extends CatalogModificationEvent {
    boolean ignoreIfExists();
}

/* Event for altering table, provides all changes for old table. */
    @PublicEvolving
public interface AlterTableEvent extends interfaceCatalogModificationEvent TableEvent {
        ObjectIdentifierList<TableChange> identifiertableChanges();  
        CatalogBaseTableboolean tableignoreIfExists();
    }

    /* Event for dropping table creation. */
    @PublicEvolving
   public interface CreateTableEventDropTableEvent extends TableEventCatalogModificationEvent {
        boolean ignoreIfExists();
    }}

    /* EventFactory for alteringcatalog table,modification provides all information in old table and new table. */listener. */
@PublicEvolving
public interface CatalogModificationListenerFactory {
    CatalogModificationListener createListener(Context context);

    @PublicEvolving 
    public interface AlterTableEvent extendsContext TableEvent {
        CreateTableEventConfiguration newTablegetConfiguration();
        booleanClassLoader ignoreIfExistsgetUserClassLoader();
    }
	/*
    /* Event	* Get an Executor pool for the 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)

/**
 * When job status is changed in job manager, it will generate job event to notify job execution listener.
 */
@PublicEvolving
public interface JobExecutionListener extends AutoCloseable {
    /* Start the listener with job config. */
    void start(Map<String, String> config) throws Exception;
  
    /* Event fired after job has been created. */ 
    void onCreated(JobCreatedEvent createdEvent);

    /* Event fired after job has beed finished. */
    void onFinished(JobFinishedEvent finishedEvent);

    /* Event fired after job has beed canceled. */
    void onCanceled(JobCanceledEvent canceledEvent);

    /* Event fired after job has beed failed. */
    void onFailed(JobFailedEvent failedEvent);

    /* Vertex in job plan, provides id, name, parallelism, input edges and output column names. */
    @PublicEvolving
    interface JobVertexInfo {
        String id();
        String name();
        int parallelism(); 
        List<JobEdgeInfo> inputs();
        List<String> outputColumns();
    }
 
    /* Edge in job plan, provides source/target vertex, input columns, distribution, isBroadcast and isForward. */
    @PublicEvolving 
    interface JobEdgeInfo {
        JobVertexInfo source();
        JobVertexInfo target();
        /* Input column names of the edge. */
        List<String> inputColumns();
        String distribution();
        boolean isBroadcast();
        boolean isForward();
    }

    /* Job source vertex, provides source table name, input table columns and source type. */ 
    @PublicEvolving 
    interface JobSourceVertex extends JobVertexInfo {
        /* `Catalog.Database.Table` format name. */
        String sourceName();

        /* Scan or lookup. */
        String type();

        /* Columns from source table, detailed information such as column type is provided in {@Code JobCreationListener#onCreateTable}. */
        List<String> columns();

        /* Source config provides options in source such as source type. */
        Map<String, String> config();
    }

    /* */ 
    @PublicEvolving 
    interface JobSinkVertex extends JobVertexInfo {
        /* `Catalog.Database.Table` format name. */ 
        String sinkName();

        /* Columns from sink table, detailed information such as column type is provided in {@Code JobCreationListener#onCreateTable}. */ 
        List<String> columns();

        /* Sink config provides options in source such as source type. */ 
        Map<String, String> config();
    }

    /* Event for job status is changed. */
    interface JobEvent {
        JobID jobId();
        String jobName();

        /* Scheduler type such as Default/Adaptive/AdaptiveBatch. */
        String scheduler();

        /* Job execution mode. */
        ExecutionMode executionMode();

        /* BATCH or STREAMING. */
        String jobType();
        
        /* Timestamp for current job status. */
        long timestamp();
    }

    /* Event for job is created. */
    @PublicEvolving
    interface JobCreatedEvent extends JobEvent {
        /* Source list. */
        List<JobSourceVertex> sources();
listeners.forEach(listener -> listener.onEvent(event));
    }

    /* SinkNotify list. listener for tables. */
    public void createTable/dropTable/alterTable(...) {
       List<JobSinkVertex> sinks() ....;
    }

    /* Event for job is finished. */notify(Create Different Table Modification Event With Context);
    @PublicEvolving}

    interface/* JobFinishedEventAdd extendsdatabase JobEventddls {and }

notify listener for databases. */*
  Event for jobpublic is canceled. */void createDatabase/dropDatabase/alterDatabase(...) {
    @PublicEvolving
    interface JobCanceledEvent extends JobEvent { }

....;
         /*notify(Create Different Database Modification Event for job is failed. */With Context); 
    }

    @PublicEvolving
/* Add listeners in interfaceBuilder JobFailedEventfor extendscatalog JobEventmanager. {*/
    public static final class Throwable exception();Builder {
    }
}