/**
 * Different events will be fired when a catalog/database/table is modified. The customized listener can receive these events and then do some specific operations according to the event type.
 */
@PublicEvolving
public interface CatalogEventListener {
    /* Event fired after a catalog/database/table is modified. */
    void onEvent(CatalogEvent catalogEvent);

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

    /* Event for catalog registration. */
    @PublicEvolving
    interface RegisterCatalogEvent extends CatalogEvent { }
 
    /* Event for catalog unregistration. */ 
    @PublicEvolving
    interface UnregisterCatalogEvent extends CatalogEvent {
        boolean ignoreIfNotExists();
    }

    /* Event for database creation. */
    @PublicEvolving
    interface CreateDatabaseEvent extends CatalogEvent {
        CatalogDatabase database();
        String databaseName(); 
        boolean ignoreIfExists();
    }

    /* Event for dropping database. */
    @PublicEvolving
    interface DropDatabaseEvent extends CatalogEvent  {
        String databaseName(); 
        boolean ignoreIfExists();
    }

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

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

    /* Event for altering table, provides all information in old table and new table. */
    @PublicEvolving
    interface AlterTableEvent extends BaseTableEvent {
        List<TableChange> tableChanges();
        boolean ignoreIfExists();
    }

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

@PublicEvolving
public interface JobListener {
    /* Event is fired before a job is submitted. */
    void onEvent(JobSubmissionEvent submissionEvent);

    /* Event for job submission. */
    @PublicEvolving
    public interface JobSubmissionEvent {
        JobID jobId();
        String jobName();
        JobLogicalPlan plan();
    }

    /* Use JobSubmittedEvent instead. */
    @Deprecated
    void onJobSubmitted(@Nullable JobClient jobClient, @Nullable Throwable throwable);

    /* Use JobExecutedEvent instead. */
    @Deprecated
    void onJobExecuted(
            @Nullable JobExecutionResult jobExecutionResult, @Nullable Throwable throwable);

    @PublicEvolving
    public interface JobPreSubmitEvent extends JobSubmissionEvent { }

    @PublicEvolving
    public interface JobSubmittedEvent extends JobSubmissionEvent {
        @Nullable JobClient jobClient();
        @Nullable Throwable throwable();
    }

    @PublicEvolving
    public interface JobExecutedEvent extends JobSubmissionEvent {
        @Nullable JobExecutionResult jobExecutionResult();
        @Nullable Throwable throwable();
    }
}

/**
 * When job status is changed in job manager, it will generate job event and notify job execution listener.
 */
@PublicEvolving
public interface JobExecutionListener {
    /* Event fired after job status has been changed. */ 
    void onJobStatusChanged(JobStatusEvent jobStatusEvent);

    /* Event fired when a checkpoint is started/completed/aborted. */
    void onCheckpoint(CheckpointEvent checkpointEvent);

    /* Job status event with plan. */
    @PublicEvolving
    public interface JobStatusEvent {
        JobLogicalPlan plan();
        JobStatus oldStatus();
        JobStatus newStatus();
    }

    /* Event for job checkpoint. */
    @PublicEvolving
    public interface CheckpointEvent {
        /* Snapshot type, checkpoint or savepoint. */
        String snapshotType(); 
        long checkpoint();
        @Nullable String externalSavepointLocation();
        boolean isPeriodic;
        long timestamp();
        Map<String, String> config(); 
        
    }

    /* Checkpoint started/completed/aborted event. */
    @PublicEvolving
    public interface CheckpointStartedEvent extends CheckpointEvent {}
    @PublicEvolving 
    public interface CheckpointCompletedEvent extends CheckpointEvent {}
    @PublicEvolving   
    public interface CheckpointStartedEvent extends CheckpointEvent {} 
}

/**
 * Job logical plan is built according to JobGraph. Users can get sources, sinks and the relationship between nodes from plan.
 */
@PublicEvolvig
public interface JobLogicalPlan {
    JobID jobId();
    String jobName();

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

    /* Job execution mode, PIPELINED/PIPELINED_FORCED/BATCH/BATCH_FORCED. */
    ExecutionMode executionMode();

    /* Job type, BATCH or STREAMING. */
    String jobType();

    /* Source vertex list. */
    List<JobPlanVertex> sources();

    /* Sink vertex list. */
    List<JobPlanVertex> sinks();

    /* Get all vertex list. */
    List<JobPlanVertex> getVerticesSortedTopologicallyFromSources();

    /* Get specific vertex by id. */
    JobPlanVertex vertex(String id); 

    /* Job configuration. */
    Configuration jobConfiguration();

    /* Vertex in job logical plan based on JobVertex. */
    @PublicEvolving
    public interface JobPlanVertex {
        String id();
        String name();
        String operatorName();
        String operatorDescription();
        int parallelism(); 
        String invokableClassName();
        boolean supportsConcurrentExecutionAttempts();
        List<JobPlanEdge> inputs();
    }

    /* Edge between vertexes in the logical plan. */
    @PublicEvolving
    public interface JobPlanEdge {
        JobPlanVertex source();
        JobPlanVertex target();
        String distribution();
        String shipStrategyName();
        boolean isBroadcast();
        boolean isForward();
    }
}

/* Table scan source and sink base interface, datastream source/sink vertex can be added based on the requirements in the future. */
public interface JobPlanTableVertex extends JobPlanVertex {
    /* `catalog`.`database`.`table` for scan source. */
    ObjectIdentifier table();

    /* For Scan source, the type is Values or Table; for sink, the type is CollectSink or ModifySink. */
    String type();

    /* Table options. */
    Map<String, String> config();

    /* For scan source, columns which are consumed by source; for sink, columns which are produced by sink. */
    RowType columns();
 
    /* Table scan source vertex. */
	@PublicEvolving
	public interface JobPlanTableSourceVertex extends JobPlanTableVertex {}
 
    /* Table sink vertex. */
	@PublicEvolving
	public interface JobPlanTableSinkVertex extends JobPlanTableVertex {
        /* Modify type, INSERT/UPDATE/DELETE. */
        String modifyType();

        /* Update mode, APPEND/RETRACT/UPSERT. */
        String updateMode();
        boolean overwrite();
        Map<String, String> staticPartitions();
    } 
}

# Config catalog event listeners.
table.catalog.listeners: {job catalog listener class1},{job catalog listener class2}

# Existing config job submission listeners.
execution.job-listeners: {job submission listener class1},{job submission listener class2}

# Config job execution listeners.
jobmanager.execution.listeners: {job execution listener class1},{job execution listener class2}