CREATE CATALOG paimon_catalog1 WITH (
    ... // other options
    'metadata.store' = 'jdbc',
    'metadata.jdbc.driver' = 'com.mysql.jdbc.Driver',
    'metadata.jdbc.url' = 'XXXXX',
    'metadata.jdbc.database' = 'paimon_cata1',    // The default systemMetadata databaseDatabase name is `paimon`
    'metadata.jdbc.username' = 'XXX',
    'metadata.jdbc.password' = 'XXX'
);

/** Factor to create metadata store. */
public interface MetadataStoreFactory {
    /* The identifier for metadata store factor. */
	String identifier();

    /* Create metadata store from specific context. */
	MetadataStore create(MetadataStoreContext context);

    /** The context used to create metadata store in the factory. */
	public interface MetadataStoreContext {
	    Options options();
    }
}

/* Metadata store will manage the options, table lineage and data lineage information for the catalog. */
public interface MetadataStore {
	/**
	 * Store the source table and job lineage.
	 *
	 * @param entity the table lineage entity
	 */
	void storeSourceTableLineage(TableLineageEntity entity);

	/**
	 * Delete the source table lineage for given job.
	 *
	 * @param job the job for table lineage
	 */
	void deleteSourceTableLineage(String job);

   /**
	 * Get source table and job lineages.
	 *
	 * @param predicate the predicate for the table lineages
	 * @return the iterator for source table and job lineages
	 */
	Iterator<TableLineageEntity> sourceTableLineages(@Nullable Predicate predicate);

	/**
	 * Store the sink table and job lineage.
	 *
	 * @param entity the table lineage entity
	 */
	void storeSinkTableLineage(TableLineageEntity entity);

	/**
	 * Get sink table and job lineages.
	 *
	 * @param predicate the predicate for the table lineages
	 * @return the iterator for source table and job lineages
	 */
	Iterator<TableLineageEntity> sinkTableLineages(@Nullable Predicate predicate sinkTableLineages(@Nullable Predicate predicate);

	/**
	 * Delete the sink table lineage for given job.
	 *
	 * @param job the job for table lineage
	 */
	void deleteSinkTableLineage(String job);

	   /**
	 * Store the source table and job lineage.
	 *
	 * @param entity the data lineage entity
	 */
	void storeSourceDataLineage(DataLineageEntity entity);

	/**
	 * Get source data and job lineages.
	 *
	 * @param predicate the predicate for the table lineages
	 * @return the iterator for source table and job lineages
	 */
	Iterator<DataLineageEntity> sourceDataLineages(@Nullable Predicate predicate);

	/**
	 * Store the source table and job lineage.
	 *
	 * @param entity the data lineage entity
	 */
	void storeSinkDataLineage(DataLineageEntity entity);

	/**
	 * Get sink data and job lineages.
	 *
	 * @param predicate the predicate for the table lineages
	 * @return the iterator for source table and job lineages
	 */
	Iterator<DataLineageEntity> sinkDataLineages(@Nullable Predicate predicate);
}

/**
 * Table lineage entity with database, table and job for table source and sink lineage.
 */
public interface TableLineageEntity {
	public String getDatabase();

	public String getTable();

	public String getJob();

    public String getCreateTime();
}

/**
 * Data lineage entity with table lineage, barrier id and snapshot id for table source and sink lineage.
 */
public interface DataLineageEntity extends TableLineageEntity {
	public long getBarrierId();

	public long getSnapshotId();
}

public class JdbcMetadataStoreOptions {
	/**
	 * The jdbc driver class for metadata store.
	 */
	public static ConfigOption<String> METADATA_JDBC_DRIVER =
			key("metadata.jdbc.driver")
					.stringType()
					.noDefaultValue()
					.withDescription("The jdbc driver class for metadata store.");
	/**
	 * The jdbc url for metadata store.
	 */
	public static ConfigOption<String> METADATA_JDBC_URL =
			key("metadata.jdbc.url")
					.stringType()
					.noDefaultValue()
					.withDescription("The jdbc url for metadata store.");

	/**
	 * The jdbc url for metadata store.
	 */
	public static ConfigOption<String> METADATA_JDBC_USERNAME =
			key("metadata.jdbc.username")
					.stringType()
					.noDefaultValue()
					.withDescription("The jdbc username for metadata store.");

	/**
	 * The jdbc url for metadata store.
	 */
	public static ConfigOption<String> METADATA_JDBC_PASSWORD =
			key("metadata.jdbc.password")
					.stringType()
					.noDefaultValue()
					.withDescription("The jdbc password for metadata store.");

	/**
	 * The jdbc url for metadata store.
	 */
	public static ConfigOption<String> METADATA_JDBC_DATABASE =
			key("metadata.jdbc.database")
					.stringType()
					.defaultValue("paimon")
					.withDescription("The jdbc database for metadata store and default database is `paimon_system``paimon`");
}

/* Jdbc metadata store factory. */
public class JdbcMetadataStoreFactory implements MetadataStoreFactory {

	public static final String IDENTIFIER = "jdbc";

	@Override
	public String identifier() {
		return IDENTIFIER;
	}

	@Override
	public MetadataStore create(MetadataStoreContext context) {
        // Create DataSource from the context
		return new JdbcMetadataStore(`The DataSource of database`);
	}
}

/* Jdbc metadata store. */
public class JdbcMetadataStore implements MetadataStore {
    private DataSource dataSource;
}