Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
As discussed above, currently flink HybridSource is released, but it only be used in DataStream. We Need to add sql and table support for many table & sql end users.
so we propose this flip.
Basic Idea
Add a new built-in hybrid connector. First, In the HybridTableSourceFactory, use 'source-identifiers'(final name could be changed) option to concat ordered some child sources.
Next, we deal with indexed concrete child source option and pass to child source table factory to create child table source instances.
When child table source instances are ready, we use child table source ScanRuntimeProvider to get the actual child Source(FLIP-27 new Source API)
Finally, we bind child sources to HybridSource.
core options:
source-identifiers：Use comma delimiter identifier string to indicate concatenated child sources. It's in order. The boundedness of hybrid source is last child source's boundedness. (required)
$identifier.option: The concrete child source options. Because we may use same connectors as connected sources, so we can't use connector name as prefix, so use user-specific identifier as child source option prefix. it will be taken-off the identifier to get real option and pass to concrete child sources.(required)
note: we use [A-Za-z0-9_]+ as child source identifier pattern.
switched-start-position-enabled: Currently, the DataStream API expose two start ways about next source. Use this option to enable the Switched-Start-Position(default is false). if it's false, the hybrid will use Fixed-Start-Position. The default is Fixed-Start-Position.
1.ddl (boundedness, just use 2 filesystem sources to explain it)
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'='filesystem'
'realtime.path' = '/tmp/a.csv',
'realtime.format' = 'csv'
);
table api (boundedness, just use 2 filesystem sources to explain it)
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", "filesystem")
.option("realtime.path", "/tmp/b.csv")
.option("realtime.format", "csv")
.build();
tEnv.createTable("hybrid_source", tableDescriptor);
Table table = tEnv.from("hybrid_source").select($("f0"), $("f1"), $("f2"));
2.ddl (unbounded data, use source.monitor-interval specify the second source is unbounded)
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'='filesystem'
'realtime.path' = '/tmp/a.csv',
'realtime.format' = 'csv'
'realtime.source.monitor-interval' = '5s'
);
table api (unbounded data, use source.monitor-interval specify the second source is unbounded)
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", "filesystem")
.option("realtime.path", "/tmp/b.csv")
.option("realtime.format", "csv")
.option("realtime.source.monitor-interval", "5s")
.build();
tEnv.createTable("hybrid_source", tableDescriptor);
Table table = tEnv.from("hybrid_source").select($("f0"), $("f1"), $("f2"));
3.ddl(normal scenario, with kafka)
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'
);
table api(normal scenario, with kafka)
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"));
4.ddl(with more child sources)
create table hybrid_source(
f0 varchar,
f1 varchar,
f2 bigint
) with(
'connector'='hybrid',
'sources'='historical01,historical02,realtime',
'historical01.connector'='filesystem'
'historical01.path' = '/tmp/a.csv',
'historical01.format' = 'csv',
'historical02.connector'='filesystem'
'historical02.path' = '/tmp/a.csv',
'historical02.format' = 'csv',
'realtime.connector'='kafka'
'realtime.properties.bootstrap.servers' = 'localhost:9092',
'realtime.properties.group.id' = 'testGroup',
'realtime.scan.startup.mode' = 'earliest-offset',
'realtime.format' = 'csv'
);
table api
similar with other cases above
Start position conversion:
Currently, the base SplitEnumerator and built-in source not expose the end position, we can't use it pass to the next streaming source. We may need to update the SplitEnumerator interface to add getEndTimestamp() support.
Then we use switched-start-position-enabled option to choose switched start position or fixed start position. how it works? u can see the draft implementation below.
Prototype implementation
HybridTableSource（Switched start position & Fixed start position）
public class HybridTableSource implements ScanTableSource {
public HybridTableSource(
String tableName,
@Nonnull List<Source<RowData, ?, ?>> childSources,
Configuration configuration,
ResolvedSchema tableSchema) {
this.tableName = tableName;
this.tableSchema = tableSchema;
this.childSources = childSources;
this.configuration = configuration;
}
@Override
public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
Preconditions.checkArgument(childSources.size() > 0);
HybridSource<RowData> hybridSource;
if (configuration.getBoolean(
HybridConnectorOptions.OPTIONAL_SWITCHED_START_POSITION_ENABLED)) {
HybridSource.HybridSourceBuilder<RowData, SplitEnumerator> builder =
HybridSource.builder(childSources.get(0));
for (int i = 1; i < childSources.size(); i++) {
final int sourceIndex = i;
Boundedness boundedness = childSources.get(sourceIndex).getBoundedness();
builder.addSource(
switchContext -> {
SplitEnumerator previousEnumerator =
switchContext.getPreviousEnumerator();
// how to pass to kafka or other connector ? We may add a method in new
// source api like startTimestamp();
long switchedTimestamp = previousEnumerator.getEndTimestamp();
childSources.get(sourceIndex).setStartTimetamp(switchedTimestamp);
return childSources.get(sourceIndex);
},
boundedness);
}
hybridSource = builder.build();
} else {
HybridSource.HybridSourceBuilder<RowData, HybridSourceSplitEnumerator> builder =
HybridSource.builder(childSources.get(0));
for (int i = 1; i < childSources.size(); i++) {
builder.addSource(childSources.get(i));
}
hybridSource = builder.build();
}
return SourceProvider.of(hybridSource);
}
}
HybridTableSource bind accepted child sources with given order to final HybridSource.
HybridTableSourceFactory
The core logic is that we read hybrid source schema and identifier-prefix child source options and takeoff it's prefix to generate new child source schema.
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
final FactoryUtil.TableFactoryHelper helper =
FactoryUtil.createTableFactoryHelper(this, context);
final Configuration tableOptions = (Configuration) helper.getOptions();
String tableName = context.getObjectIdentifier().toString();
ResolvedCatalogTable hybridCatalogTable = context.getCatalogTable();
ResolvedSchema tableSchema = hybridCatalogTable.getResolvedSchema();
// validate params
Set<ConfigOption<?>> optionalOptions = new HashSet<>();
for (String optionKey : tableOptions.toMap().keySet()) {
if (optionKey.matches(HybridConnectorOptions.SOURCE_OPTION_REGEX)) {
ConfigOption<String> childOption = key(optionKey).stringType().noDefaultValue();
optionalOptions.add(childOption);
}
}
optionalOptions.addAll(optionalOptions());
FactoryUtil.validateFactoryOptions(requiredOptions(), optionalOptions, tableOptions);
Set<String> consumedOptionKeys = new HashSet<>();
consumedOptionKeys.add(CONNECTOR.key());
consumedOptionKeys.add(HybridConnectorOptions.SOURCES.key());
optionalOptions.stream().map(ConfigOption::key).forEach(consumedOptionKeys::add);
FactoryUtil.validateUnconsumedKeys(
factoryIdentifier(), tableOptions.keySet(), consumedOptionKeys);
// process source option
String sourceStr = tableOptions.get(HybridConnectorOptions.SOURCES);
List<String> sourceList = Arrays.asList(sourceStr.split(","));
if (sourceList.size() < 2) {
throw new TableException(
String.format(
"Hybrid source '%s' option must specify at least 2 sources.",
HybridConnectorOptions.SOURCES.key()));
}
// parse schema-field-mappings
ObjectMapper mapper = JacksonMapperFactory.createObjectMapper();
String fieldMappingStr =
tableOptions.get(HybridConnectorOptions.OPTIONAL_SCHEMA_FIELD_MAPPINGS);
List<Map<String, String>> fieldMappingsList = null;
if (!StringUtils.isEmpty(fieldMappingStr)) {
try {
fieldMappingsList =
mapper.readValue(
fieldMappingStr, new TypeReference<List<Map<String, String>>>() {});
} catch (Exception e) {
throw new TableException(
String.format(
"Failed to parse hybrid source '%s' option.",
HybridConnectorOptions.OPTIONAL_SCHEMA_FIELD_MAPPINGS.key()),
e);
}
if (fieldMappingsList.size() != sourceList.size()) {
throw new IllegalArgumentException(
String.format(
"Hybrid source '%s' option split nums must equals child source nums.",
HybridConnectorOptions.OPTIONAL_SCHEMA_FIELD_MAPPINGS.key()));
}
}
LOG.info("Hybrid source is consisted of {}.", sourceList);
// generate concrete child sources & concat sources to final hybrid source
List<Source<RowData, ?, ?>> childSources = new ArrayList<>();
ClassLoader cl = HybridTableSourceFactory.class.getClassLoader();
for (int i = 0; i < sourceList.size(); i++) {
ResolvedCatalogTable childCatalogTable;
ResolvedSchema childResolvedSchema = null;
// override schema
if (fieldMappingsList != null) {
Map<String, String> fieldMappings = fieldMappingsList.get(i);
childResolvedSchema =
createChildSchemaUsingFieldMappings(tableSchema, fieldMappings);
}
if (childResolvedSchema != null) {
childCatalogTable =
new ResolvedCatalogTable(
hybridCatalogTable
.getOrigin()
.copy(
extractChildSourceOptions(
hybridCatalogTable.getOptions(), i)),
childResolvedSchema);
} else {
childCatalogTable =
hybridCatalogTable.copy(
extractChildSourceOptions(hybridCatalogTable.getOptions(), i));
}
DynamicTableSource tableSource =
FactoryUtil.createDynamicTableSource(
null,
context.getObjectIdentifier(),
childCatalogTable,
Collections.emptyMap(),
context.getConfiguration(),
cl,
true);
if (tableSource instanceof ScanTableSource) {
ScanTableSource.ScanRuntimeProvider provider =
((ScanTableSource) tableSource)
.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE);
if (provider instanceof SourceProvider) {
Source<RowData, ?, ?> source = ((SourceProvider) provider).createSource();
childSources.add(source);
} else {
throw new UnsupportedOperationException(
provider.getClass().getCanonicalName() + " is unsupported now.");
}
} else {
// hybrid source only support ScanTableSource
throw new TableException(
String.format(
"%s is not a ScanTableSource, please check it.",
tableSource.getClass().getName()));
}
}
LOG.info("Generate hybrid child sources with: {}.", childSources);
// post check
Preconditions.checkArgument(sourceList.size() == childSources.size());
return new HybridTableSource(tableName, childSources, tableOptions, tableSchema);
}
protected Map<String, String> extractChildSourceOptions(
Map<String, String> originalOptions, int index) {
if (originalOptions == null || originalOptions.isEmpty()) {
return originalOptions;
}
String optionPrefix = index + HybridConnectorOptions.SOURCE_DELIMITER;
Map<String, String> sourceOptions =
originalOptions.entrySet().stream()
.filter(entry -> entry.getKey().startsWith(optionPrefix))
.collect(
Collectors.toMap(
entry ->
StringUtils.removeStart(
entry.getKey(), optionPrefix),
Map.Entry::getValue));
String[] sources = originalOptions.get(HybridConnectorOptions.SOURCES.key()).split(",");
sourceOptions.put(FactoryUtil.CONNECTOR.key(), sources[index]);
return sourceOptions;
}
TableSourceFactory is using for Java SPI to search hybrid source implementation.
HybridConnectorOptions
public class HybridConnectorOptions {
public static final String SOURCE_DELIMITER = ".";
public static final ConfigOption<String> SOURCES =
ConfigOptions.key("sources")
.stringType()
.noDefaultValue()
.withDescription(
"Use comma delimiter and identifier indicate child sources that need to be concatenated. e.g. sources='historical,realtime'");
public static final ConfigOption<Boolean> OPTIONAL_SWITCHED_START_POSITION_ENABLED =
ConfigOptions.key("switched-start-position-enabled")
.booleanType()
.noDefaultValue()
.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 ts and pass to this unbounded streaming source. ");
public static final ConfigOption<String> OPTIONAL_SCHEMA_FIELD_MAPPINGS =
ConfigOptions.key("schema-field-mappings")
.stringType()
.noDefaultValue()
.withDescription(
"Use json kv to match the different field names with ddl field. e.g. '[{\"f0\":\"A\"},{}]' it means the "
+ "first child source column A is match to ddl column f0, the second source no matching.");
private HybridConnectorOptions() {}
}
Options for creating HybridTableSource.
Future Work
add a metadata column to indicate the data source type, let user know the data is from bounded source or unbounded source.
for example, sources='historical,realtime'. historical.connector='filesystem', realtime.connector='kafka'. This metadata will be 'historical' when read from filesystem and 'realtime' when read from kafka.
Then user will use this metadata column deal some custom sql logics.
Compatibility, Deprecation, and Migration Plan
It's a new support without migration currently.
Test Plan
Add unit test case for HybridTableSourceFactory and HybridTableSource.
Add integration test cases for hybrid source sql ddl.
Rejected Alternatives
to be added.