/** The basic listener for lineage in Flink, all lineage related listener should extend this interface. */
public interface LineageListener {}

/**
 * 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 extends LineageListener {
    /* Event fired after a catalog/database/table is modified. */
    void onEvent(CatalogEvent catalogEvent);

    /* The basic class for catalog related event. */
    public abstract class CatalogEvent {
        /* The catalog of the event. */
        Catalog catalog();
        /* The name of catalog. */
        String catalogName();
    }

    /* The basic class for database related event. */
    public abstract class BaseDatabaseEvent extends CatalogEvent {
        String databaseName();  
    }

    /* Event for creating database. */
    @PublicEvolving
    public class CreateDatabaseEvent extends BaseDatabaseEvent {
        CatalogDatabase database();
        boolean ignoreIfExists();
    }

    /* Event for altering database. */
    @PublicEvolving
    public class AlterDatabaseEvent extends BaseDatabaseEvent {
        CatalogDatabase newDatabase();
        boolean ignoreIfNotExists();
    }

    /* Event for dropping database. */
    @PublicEvolving
    public class DropDatabaseEvent extends BaseDatabaseEvent {
        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 abstract class BaseTableEvent extends CatalogEvent {
        ObjectIdentifier identifier();  
        CatalogBaseTable table();

        /* Get the storage identifier for the table.*/
        StorageIdentifier storage()
    }

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

    /* Event for altering table, provides all changes for old table. */
    @PublicEvolving
    public class AlterTableEvent extends BaseTableEvent {
        List<TableChange> tableChanges();
        boolean ignoreIfExists();
    }

    /* Event for dropping table. */
    @PublicEvolving
    public class DropTableEvent extends BaseTableEvent {
        boolean ignoreIfExists();   
    }
}

/* Factory for catalog listener. */
@PublicEvolving
public interface CatalogModificationListenerFactory {
    public CatalogModificationListener createListener(Configuration configuration, ClassLoader classLoader);
}

/* Storage identifier for different physical table. */
@PublicEvolving
public class StorageIdentifier {
    /* Storage type such as kafka, hive, iceberg or paimon. */
    String type();

    /* Properties for the storage identifier, users can get value from different keys for different storages, such as broker and topic for kafka. */
    Map<String, String> properties();
}

/* 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

/**
 * When job status is changed, Flink will generate job event and notify job status changed listener.
 */
@PublicEvolving
public interface JobStatusChangedListener {
    /* Event fired after job status has been changed. */ 
    void onEvent(JobStatusEvent jobStatusEvent);

    /** Basic job status event. */
    abstract class JobStatusEvent {
        JobID jobId();
        String jobName();
    }

    /** Job created event with job lineage. */
    @PublicEvolving
    class JobCreatedEvent extends JobStatusEvent {
		/* Lineage for the current job. */
        JobLineage lineage();

    	/* Job type, TableOrSQL or DataStream. */
    	String jobType();

    	/* Job execution type, BATCH or STREAMING. */
    	String executionType(); 
 
    	/* Job configuration. */
    	Map<String, String> config();  
    }

    /** Job status changed event for runtime. */
	@PublicEvolving
    class JobExecutionStatusEvent extends JobStatusEvent {
        JobStatus oldStatus();
        JobStatus newStatus();
        @Nullable Throwable exception();
    }

/* Factory for job execution listener. */
@PublicEvolving
public interface JobStatusChangedListenerFactory {
    public JobStatusChangedListener createListener(Configuration configuration, ClassLoader classLoader);
}

/**
 * Job lineage is built according to StreamGraph. Users can get sources, sinks and the relationship between nodes from lineage.
 */
@PublicEvolvig
public interface JobLineage {
    /* Source lineage list. */
    List<SourceLineage> sources();

    /* Sink lineage list. */
    List<SinkLineage> sinks();
}

/** Base connector lineage interface. */
public interface ConnectorLineage {
    StorageIdentifier connector();
}

/** Base source lineage. */
@PublicEvolving
public interface SourceLineage extends ConnectorLineage {}

/** Base sink lineage. */
@PublicEvolving
public interface SinkLineage extends ConnectorLineage {
    List<SourceLineage> sources();
}

/** Basic table lineage and listeners can get catalog and table from it. */
public abstract class TableLineage implements ConnectorLineage {
    /* The catalog of the table lineage. */
    public Catalog catalog();

    /* The table of the table lineage. */
    public CatalogBaseTable table();
}

/** Source lineage for table. */
@PublicEvolving
public class TableSourceLineage extends TableLineage implements SourceLineage {
    public StorageIdentifier connector();

    /* Output columns for the source and the detailed column information such as data type are in the table. */
    public List<String> columns();
}

/** Sink lineage for table. */
@PublicEvolving
public class TableSinkLineage extends TableLineage implements SinkLineage {
    public StorageIdentifier connector(); 

    /* Modify type, INSERT/UPDATE/DELETE. */
    String modifyType();
 
    /* Update mode, APPEND/RETRACT/UPSERT. */
    String updateMode();
    boolean overwrite();

    /* The source lineages for the sink table. */
    public List<SourceLineage> sources();

    /* The output columns for the sink table. */
    public List<String> columns();

    /* The source column lineages for each target column in sink table, this will be supported by Flink in the future. */
    public Map<String, List<TableColumnLineage>> columnLineages();    

    /* Source table and columns for the target column in sink lineage. Multiple source table columns would generate one sink column. */
	@PublicEvolving
    public class TableColumnLineage {
        /* The source table for column lineage. */
        public TableSourceLineage source();

        /* The columns in source lineage for column lineage. */
        public List<String> columns();
    }
}

/**
 * Add set source lineage method in data stream source.
 */
@Public
public class DataStreamSource {
    private SourceLineage lineage;
 
    public DataStreamSource setLineage(SourceLineage lineage);
}
 
/**
 * Add set sink lineage method in data stream sink.
 */
@Public
public class DataStreamSink {
    private SinkLineage lineage;
 
    public DataStreamSink setLineage(SinkLineage lineage);
}

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

# Config for job status changed listeners.
lineage.job-status-changed.listeners: {job status changed listener factory1},{job status changed listener factory2}

/** User defined vector source lineage. */
public class VectorSourceLineage implements SourceLineage {
    /* The capacity of source lineage. */
    int capacity();

    /* The value type in the vector. */
    String valueType();
}

/** User defined vector sink lineage. */
public class VectorSinkLineage implements SinkLineage {
    List<SourceLineage> sources();
    int capacity();
    String valueType();
}
 
/* User can use vector source/sink lineages in datastream job. */
Map<String, String> kafkaSourceConf = ...;
Map<String, String> kafkaSinkConf = ...;
StreamExecutionEnvironment env = ...;

VectorSourceLineage sourceLineage = new VectorSourceLineage(new StorageIdentifier("kafka", kafkaSourceConf), 10, "int");
env.fromSource(...).setLineage(sourceLineage)
	.map(...).keyBy(..).
	.sinkTo(...).setLineage(new VectorSinkLineages(new StorageIdentifier("kafka", kafkaSinkConf), 20, "int", Arrays.asList(sourceLineage)));
env.execute();

-- 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(CatalogEvent event) {
        listeners.forEach(listener -> listener.onEvent(event));
    }

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

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

/* Listeners and related operations in AbstractCatalog. */
public abstract class AbstractCatalog implements Catalog {
    private final List<CatalogModificationListener> listeners;

    /* Create the catalog with listeners. */
    public AbstractCatalog(String name, String defaultDatabase, List<CatalogModificationListener> listeners); 

    /**
     * Notify the listeners with given database event, after the customized implementation of AbstractCatalog create/alter/drop a database,
     * it can create the specific event and call the notify method.
     */
    protected void notify(BaseDatabaseEvent event) {
        for (CatalogModificationListener listener : listeners) {
            listener.onEvent(event);
        }
    }
}

/* Create default catalog context with listeners. */
public DefaultCatalogContext {
    public DefaultCatalogContext(
            String name,
            Map<String, String> options,
            ReadableConfig configuration,
            ClassLoader classLoader,
            List<CatalogModificationListener> listeners);

    /* Get catalog event listeners from the context. */
    public List<CatalogModificationListener> listeners() {
        return listeners;
    }
}

/* Set pipeline to client before it submit job graph.  */
public class AbstractSessionClusterExecutor {
    public CompletableFuture<JobClient> execute(
            @Nonnull final Pipeline pipeline,
            @Nonnull final Configuration configuration,
            @Nonnull final ClassLoader userCodeClassloader)
            throws Exception {
        ....

        final ClusterClientProvider<ClusterID> clusterClientProvider =
                    clusterDescriptor.retrieve(clusterID);
        ClusterClient<ClusterID> clusterClient = clusterClientProvider.getClusterClient();

        // Set pipeline to the cluster client before submit job graph
        clusterClient.setPipeline(pipeline); 

        return clusterClient
                    .submitJob(jobGraph)
                    ....  
    }
}

/* Create job submission event and notify listeners before it submit job graph. */
public class RestClusterClient {
    private final List<JobStatusChangedListener> listeners;
    private Pipeline pipeline;

    @Override
    public CompletableFuture<JobID> submitJob(@Nonnull JobGraph jobGraph) {
        // Create event and notify listeners before the job graph is submitted.
        JobSubmissionEvent event = createEventFromPipeline(pipeline);
        if (event != null) {
            listeners.forEach(listener -> listener.onEvent(event));
        }
        
        ....;
    }
}

/* Set pipeline to client before it submit job graph.  */
public class DefaultExecutionGraph {
    private final List<JobStatusChangedListener> executionListeners;

    private boolean transitionState(JobStatus current, JobStatus newState, Throwable error) {
        ....;
        notifyJobStatusHooks(newState, error);
        // notify job execution listeners
        notifyJobStatusChangedListeners(current, newState, error);
        ....;
    }

    private void notifyJobStatusChangedListeners(JobStatus current, JobStatus newState, Throwable error) {
        JobStatusEvent event = create job status event;
        executionListeners.forEach(listener -> listener.onEvent(event));
    }
}