create table hybrid_source(
 f0 varchar,
 f1 varchar,
 f2 bigint
) with(
 'connector'='hybrid',
 'sources'='historical,realtime',
 'historical.connector'='filesystem'
 'historical.path' = '/tmp/a.csv',
 'historical.format' = 'csv',
 'realtime.connector'='kafka',
 'realtime.topic' = 'test',
 'realtime.properties.bootstrap.servers' = 'localhost:9092',
 'realtime.properties.group.id' = 'test',
 'realtime.scan.startup.mode' = 'earliest-offset',
 'realtime.format' = 'csv' 
);

TableDescriptor tableDescriptor =
                TableDescriptor.forConnector("hybrid")
                        .schema(
                                Schema.newBuilder()
                                        .column("f0", DataTypes.STRING())
                                        .column("f1", DataTypes.STRING())
                                        .column("f2", DataTypes.BIGINT())
                                        .build())
                        .option("source-identifiers", "historical,realtime")
                        .option("historical.connector", "filesystem")
                        .option("historical.path", "/tmp/a.csv")
                        .option("historical.format", "csv")
                        .option("realtime.connector", "kafka")
                        .option("realtime.topic", "test")
                        .option("realtime.properties.bootstrap.servers", "localhost:9092")
						.option("realtime.group.id", "test")
						.option("realtime.scan.startup.mode", "earliest-offset")
                        .build();
        tEnv.createTable("hybrid_source", tableDescriptor);

        Table table = tEnv.from("hybrid_source").select($("f0"), $("f1"), $("f2"));

/**
 * A decorative interface of {@link SplitEnumerator} which allows to get end timestamp.
 *
 * <p>The split enumerator must implement this interface if it needs to support switched start
 * position in hybrid source scenario and other needed situations.
 */
public interface SupportsGetEndTimestamp {

    /** Get the end timestamp for current source or split enumerator. */
    long getEndTimestamp();
}

/**
 * Unlike predefined start offset in table source, this interface allow to apply a dynamic switched
 * timestamp to a {@link ScanTableSource}, then table source can use this switched timestamp to
 * re-initialize the start offsets. However, it depends on concrete source implementation. A common
 * scenario is that HybridSource next source use this ability to switch from previous one.
 *
 * <p>Note: if the table source can use in HybridSource when enable switched-start-position then it
 * must be implemented, otherwise will cause exception.
 */
public interface SupportsSwitchedStartTimestamp {      
	 
     /** Apply given switched start timestamp to source. */
     void applySwitchedStartTimestamp(long startTimestamp);
}

public class HybridTableSource implements ScanTableSource {

    private final String tableName;
    private final ResolvedSchema tableSchema;
    private final List<ScanTableSource> childTableSources;
    private final Configuration configuration;

    public HybridTableSource(
            String tableName,
            @Nonnull List<ScanTableSource> childTableSources,
            Configuration configuration,
            ResolvedSchema tableSchema) {
        this.tableName = tableName;
        this.tableSchema = tableSchema;
        Preconditions.checkArgument(
                childTableSources.size() >= 2, "child table sources must at least 2 sources");
        this.childTableSources = childTableSources;
        this.configuration = configuration;
    }

    @Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
        HybridSource.HybridSourceBuilder<RowData, SplitEnumerator> builder;
        ScanTableSource firstTableSource = childTableSources.get(0);
        ScanTableSource.ScanRuntimeProvider firstProvider =
                firstTableSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE);
        Source<RowData, ?, ?> firstSource = checkProviderAndGetSource(firstProvider);
        builder = HybridSource.builder(firstSource);

        if (configuration.getBoolean(OPTIONAL_SWITCHED_START_POSITION_ENABLED)) {
            // switched start position
            for (int i = 1; i < childTableSources.size(); i++) {
                ScanTableSource nextTableSource = childTableSources.get(i);
                Boundedness boundedness =
                        nextTableSource
                                        .getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE)
                                        .isBounded()
                                ? Boundedness.BOUNDED
                                : Boundedness.CONTINUOUS_UNBOUNDED;
                builder.addSource(
                        switchContext -> {
                            SplitEnumerator previousEnumerator =
                                    switchContext.getPreviousEnumerator();
                            // how to get and apply timestamp depends on specific enumerator
                            long switchedTimestamp =
                                    checkAndCastSplitEnumerator(previousEnumerator)
                                            .getEndTimestamp();
                            checkAndCastTableSource(nextTableSource)
                                    .applySwitchedStartTimestamp(switchedTimestamp);
                            ScanTableSource.ScanRuntimeProvider nextProvider =
                                    nextTableSource.getScanRuntimeProvider(
                                            ScanRuntimeProviderContext.INSTANCE);
                            return checkProviderAndGetSource(nextProvider);
                        },
                        boundedness);
            }
        } else {
            // fixed start position
            for (int i = 1; i < childTableSources.size(); i++) {
                ScanTableSource.ScanRuntimeProvider nextProvider =
                        childTableSources
                                .get(i)
                                .getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE);
                builder.addSource(checkProviderAndGetSource(nextProvider));
            }
        }

        return SourceProvider.of(builder.build());
    }
}

public class HybridConnectorOptions {

    public static final String SOURCE_IDENTIFIER_REGEX = "[A-Za-z0-9_]+";
    public static final String SOURCE_IDENTIFIER_DELIMITER = ".";

    public static final ConfigOption<String> SOURCE_IDENTIFIERS =
            ConfigOptions.key("source-identifiers")
                    .stringType()
                    .noDefaultValue()
                    .withDescription(
                            "Use comma delimiter and identifier indicate child sources that need to be concatenated. "
                                    + "e.g. source-identifiers='historical,realtime'");

    public static final ConfigOption<Boolean> OPTIONAL_SWITCHED_START_POSITION_ENABLED =
            ConfigOptions.key("switched-start-position-enabled")
                    .booleanType()
                    .defaultValue(false)
                    .withDescription(
                            "Whether to enable switched start position, default is false for using fixed start position. "
                                    + "If it is true, then hybrid source will call the previous source SplitEnumerator#getEndTimestamp "
                                    + "to get end timestamp and pass to next unbounded streaming source. ");
}