/** Dynamic option provider for source and sink tables. */
@PublicEvolving
public interface TableDynamicOptionProvider {
    /**
     * Provide dynamic options for source and sink tables.
     *
     * @param context The context for job to get table dynamic options.
     * @return The source dynamic options.
     */
    TableDynamicOpitons getDynamicOptions(TableOptionsContext context);
    
    @PublicEvolving
    static interface TableOptionsContext {
        /**
         * Configuration for the job which provide job customized options, for example, 
         * different strategy to generate table dynamic options.
         **/
        ReadableConfig jobConfiguration();
        
        /** 
         * Get catalog context for given catalog which is used to identify physical catalog.
         * There will be some information about options in `CatalogContext` such as the location
         * where the option values for tables are stored.
         **/
        CatalogContext catalogContext(String catalog);
        
        /** All source tables for the job. */
        Set<ObjectIdentifier> sources();
        
        /**
         * Get current options for the source table which is used to determine the dynamic 
         * options for the table, for example, jdbc table names in the options will be sued to
         * generate username and password.
         **/
        Map<String, String> sourceOptions(ObjectIdentifier table);
        
        /** All sink table for the job. */
        Set<ObjectIdentifier> sinks();
        
        /**
         * Get current options for the sink table which is used to determine the dynamic 
         * options for the table, for example, jdbc table names in the options will be sued to
         * generate username and password. 
         **/
        Map<String, String> sinkOptions(ObjectIdentifier table);
    }
    
    /** Table dynamic option result from the provider. */
    @PublicEvolving
    static interface TableDynamicOpitons {
        /** Get source options for given table identifier. */
        Map<String, String> sourceOptions(ObjectIdentifier table);
        
        /** Get sink options for given table identifier. */
        Map<String, String> sinkOptions(ObjectIdentifier table);
    }
}
 
/** Factory to create table dynamic option provider. */
@PublicEvolving
public interface TableDynamicOptionProviderFactory extends Factory {
    TableDynamicOptionProvider createProvider(Context context);
 
    /** Context for dynamic option provider factory. */
    static interface Context {
        ReadableConfig getConfiguration();
 
        ClassLoader getClassLoader();
    }
}

/** Add register provider method in table environment. */
public interface TableEnvironment {
    void registerTableOptionProvider(TableDynamicOptionProvider provider);
}

sql-gateway.table-dynamic-option-provider-factory: {provider factory class identifier}

/** Load the provider factory and create provider for DefaultContext. */
public class DefaultContext {
    private final TableDynamicOptionProvider tableOptionProvider;
    
    ...;
    
    public static DefaultContext load(
            Configuration dynamicConfig,
            List<URL> dependencies,
            boolean discoverExecutionConfig,
            boolean discoverPythonJar) {
        ...;
        
        TableDynamicOptionProviderFactory providerFactory = FactoryUtil.discoverFactory(
            DefaultContext.class.getClassLoader(),
            TableDynamicOptionProviderFactory.class,
            dynamicConfig.get(SOURCE_DYNAMIC_OPTIONS_FACTORY));
        TableDynamicOptionProvider optionsProvider = 
            providerFactory.create(new Context() {
                ...;
            });
        
        ...;
        
        return new DefaultContext(..., optionsProvider);
    }
}

public class TableEnvironmentImpl implements TableEnvironmentInternal {
    @Nullable private TableDynamicOptionProvider tableDynamicOptionProvider;
     
    /** Register source dynamic option provider. */
    @Override
    public void registerTableDynamicOptionProvider(
            TableDynamicOptionProvider TableDynamicOptionProvider) {
        this.TableDynamicOptionProvider = TableDynamicOptionProvider;
    }
     
    /** Get options for source/sink tables and add them to specified operation. */
    private void provideTableDynamicOptions(List<Operation> operations) {
        Map<ObjectIdentifier, List<SourceQueryOperation>> identifierSourceOperations = traversesourceOperations(operations);
        TableDynamicOpitons tableDynamicOptions = 
          tableDynamicOptionProvider.getDynamicOptions(
              new TableOptionsContext() { ... });
        // Add options in QueryOperation and ModifyOperation.
        ...;
    }
}

public class SourceQueryOperation implements QueryOperation {
    private final Map<String, String> dynamicOptions
    ....;
    
    /** Add options to dynamicOptions if it is not existed. */
    public void addDynamicOptions(Map<String, String> options) {
        options.forEach((k, v) -> dynamicOptions.putIfAbsent(k, v));
    }
}

public class SinkModifyOperation implements ModifyOperation {
    private final Map<String, String> dynamicOptions;
    
    public void addDynamicOptions(Map<String, String> options) {
        options.forEach((k, v) -> dynamicOptions.putIfAbsent(k, v));
    }
}

SET 'sql-gateway.table-dynamic-option-provider-factory'='paimon-provider'; // Set the paimon provider for Sql-Gateway

SELECT * FROM tableA JOIN tableB ON ... JOIN tableC ON ...;

SET 'sql-gateway.table-dynamic-option-provider-factory'='jdbc-security-provider'; // Set the jdbc security provider for Sql-Gateway

CREATE TABLE my_jdbc_table (
    val1 INT,
    val2 INT)
  WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://localhost:3306/test',
    'table-name' = 'mysql_table'
    //'username' = '****'    We don't need to provide username and password for job here
    //'password' = '****'
  );

INSERT INTO my_jdbc_table SELECT ... FROM ... GROUP BY ...;