create table hybrid_source(
 f0 varchar,
 f1 varchar,
 f2 bigint
) with(
 'connector'='hybrid',
 'source-identifiers'='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"));

create table hybrid_source(
 f0 varchar,
 f1 varchar,
 f2 bigint
) with(
 'connector'='hybrid',
 'source-identifiers'='historical,realtime',
 'switched-start-position-enabled'='true'
 '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' = 'testGroup',
  --'realtime.scan.startup.mode' = 'earliest-offset',
 'realtime.format' = 'csv' 
);

/**
 * 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.
 */
@PublicEvolving
public interface SupportsGetEndTimestamp {

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

/**
 * Unlike predefined start offset in source, this interface allow to apply a dynamic switched
 * timestamp to a source, then 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's next source use this ability to switch from previous one.
 */
@PublicEvolving
public interface SupportsSwitchedStartTimestamp {

    /** Apply given switched start timestamp to source. */
    void applySwitchedStartTimestamp(long startTimestamp);
}

/**
 * Provider that extends for {@link DataStreamScanProvider} to let DataStreamScanProvider support
 * expose inner source.
 *
 * <p>Usually a {@link SourceProvider} expose {@link Source}, {@link SourceFunctionProvider} expose
 * {@link SourceFunction} and {@link InputFormatProvider} expose {@link InputFormat}, but {@link
 * DataStreamScanProvider} just produce datastream, we can not get the inner source(note this source
 * could be SourceFunction or new Source or InputFormat). But sometimes we need it, a common
 * scenario is HybridSource, we need to extract inner source from provider.
 *
 * @param <T> source type
 */

@PublicEvolving
public interface DataStreamScanSourceAbilityProvider<T> extends DataStreamScanProvider {

    /**
     * Expose DataStreamScanProvider inner source.
     *
     * @return inner source
     */
    T createSource();
}

public class HybridTableSource implements ScanTableSource, SupportsReadingMetadata, SupportsProjectionPushDown {      
	
    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 DynamicTableSource copy() {
        return new HybridTableSource(tableName, childTableSources, configuration, tableSchema);
    }

    @Override
    public String asSummaryString() {
        return "HybridTableSource";
    }

    @Override
    public ChangelogMode getChangelogMode() {
        Set<RowKind> kinds = new HashSet<>();
        for (ScanTableSource childSource : childTableSources) {
            kinds.addAll(childSource.getChangelogMode().getContainedKinds());
        }
        ChangelogMode.Builder builder = ChangelogMode.newBuilder();
        for (RowKind kind : kinds) {
            builder.addContainedKind(kind);
        }
        return builder.build();
    }

    @Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
        HybridSource.HybridSourceBuilder<RowData, SplitEnumerator> builder;
        ScanTableSource firstTableSource = childTableSources.get(0);
        ScanTableSource.ScanRuntimeProvider firstProvider =
                validateAndGetProvider(firstTableSource);
        Source<RowData, ?, ?> firstSource = validateAndGetSource(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);
                ScanTableSource.ScanRuntimeProvider nextProvider =
                        validateAndGetProvider(nextTableSource);
                Source<RowData, ?, ?> nextSource = validateAndGetSource(nextProvider);
                Boundedness boundedness = nextSource.getBoundedness();
                final Source<RowData, ?, ?> localCopySource;
                try {
                    localCopySource = InstantiationUtil.clone(nextSource);
                } catch (ClassNotFoundException | IOException e) {
                    throw new IllegalStateException("Unable to clone the hybrid child source.", e);
                }
                // builder#addSource below is a serialized-lambda. if we use nextSource, the
                // lambda captured variables will be HybridTableSource and nextSource
                // while nextSource can be serialized, but HybridTableSource can not be
                // serialized, it will cause serialized exception. So here we do a deepCopy to
                // address it.
                builder.addSource(
                        switchContext -> {
                            SplitEnumerator previousEnumerator =
                                    switchContext.getPreviousEnumerator();
                            // how to get and apply timestamp depends on specific enumerator
                            long switchedTimestamp =
                                    validateAndCastSplitEnumerator(previousEnumerator)
                                            .getEndTimestamp();
                            validateAndCastSource(localCopySource)
                                    .applySwitchedStartTimestamp(switchedTimestamp);
                            return localCopySource;
                        },
                        boundedness);
            }
        } else {
            // fixed start position
            for (int i = 1; i < childTableSources.size(); i++) {
                ScanTableSource.ScanRuntimeProvider provider =
                        validateAndGetProvider(childTableSources.get(i));
                builder.addSource(validateAndGetSource(provider));
            }
        }

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

    @Override
    public Map<String, DataType> listReadableMetadata() {
        final Map<String, DataType> metadataMap = new HashMap<>();
        tableSchema.getColumns().stream()
                .filter(column -> column instanceof Column.MetadataColumn)
                .forEach(
                        column ->
                                metadataMap.put(
                                        ((Column.MetadataColumn) column)
                                                .getMetadataKey()
                                                .orElse(column.getName()),
                                        column.getDataType()));
        return metadataMap;
    }

    @Override
    public void applyReadableMetadata(List<String> metadataKeys, DataType producedDataType) {
        // we pass the ddl metadata fields to each child sources, if child source has this
        // metadata then return its value else metadata value is null.
        for (ScanTableSource childSource : childTableSources) {
            Preconditions.checkState(
                    childSource instanceof SupportsReadingMetadata,
                    "The table source %s must implement "
                            + "SupportsReadingMetadata interface to be used in hybrid source.",
                    childSource.getClass().getName());
            ((SupportsReadingMetadata) childSource)
                    .applyReadableMetadata(metadataKeys, producedDataType);
        }
    }

    @Override
    public void applyProjection(int[][] projectedFields, DataType producedDataType) {
        for (ScanTableSource childSource : childTableSources) {
            Preconditions.checkState(
                    childSource instanceof SupportsProjectionPushDown,
                    "The table source %s must implement "
                            + "SupportsProjectionPushDown interface to be used in hybrid source.",
                    childSource.getClass().getName());
            ((SupportsProjectionPushDown) childSource)
                    .applyProjection(projectedFields, producedDataType);
        }
    }

    @Override
    public boolean supportsNestedProjection() {
        return false;
    }

    // ------------------------------------------------------------------------

    private static ScanTableSource.ScanRuntimeProvider validateAndGetProvider(
            ScanTableSource tableSource) {
        ScanTableSource.ScanRuntimeProvider provider =
                tableSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE);
        Preconditions.checkState(
                provider instanceof SourceProvider
                        || provider instanceof DataStreamScanSourceAbilityProvider,
                "Provider %s is not a SourceProvider or DataStreamScanSourceAbilityProvider.",
                provider.getClass().getName());
        return provider;
    }

    private static Source<RowData, ?, ?> validateAndGetSource(
            ScanTableSource.ScanRuntimeProvider provider) {
        if (provider instanceof DataStreamScanSourceAbilityProvider) {
            Object source = ((DataStreamScanSourceAbilityProvider<?>) provider).createSource();
            Preconditions.checkState(
                    source instanceof Source,
                    "Child source %s is not a new Source.",
                    source.getClass().getName());
            return (Source<RowData, ?, ?>) source;
        } else {
            return ((SourceProvider) provider).createSource();
        }
    }

    private static SupportsGetEndTimestamp validateAndCastSplitEnumerator(
            SplitEnumerator splitEnumerator) {
        Preconditions.checkState(
                splitEnumerator instanceof SupportsGetEndTimestamp,
                "The split enumerator %s must implement "
                        + "SupportsGetEndTimestamp interface to be used in hybrid source when %s "
                        + "is true.",
                splitEnumerator.getClass().getName(),
                OPTIONAL_SWITCHED_START_POSITION_ENABLED.key());
        return (SupportsGetEndTimestamp) splitEnumerator;
    }

    private static SupportsSwitchedStartTimestamp validateAndCastSource(Source source) {
        Preconditions.checkState(
                source instanceof SupportsSwitchedStartTimestamp,
                "The dynamic table source %s must implement "
                        + "SupportsSwitchedStartTimestamp interface"
                        + "to be used in hybrid source when %s is true.",
                source.getClass().getName(),
                OPTIONAL_SWITCHED_START_POSITION_ENABLED.key());
        return (SupportsSwitchedStartTimestamp) source;
    }
 }

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. ");
}