[
https://issues.apache.org/jira/browse/FLINK-17887?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu updated FLINK-17887:
----------------------------
Description:
There is some problem with current ScanForamtFactory and SinkFormatFactory
interfaces:
1) {{ScanFormat#createScanFormat}} only accepts {{ScanTableSource.Context}},
which means it can’t work in lookup source.
2) The naming of {{ScanFormat}} also indicates it is only used in scan source.
But a lookup source should be able to work with format too.
3) It’s confusing that {{ScanFormatFactory#createScanFormat}} and
{{ScanFormat#createScanFormat}} (create itself?)
The proposed changes:
1. Have a common interface DynamicTableSource.Context, and make Context of
ScanTableSource and LookupTableSource extend it, and rename them to
LookupContext and ScanContext
2. Change parameter of ScanFormat.createScanFormat from ScanTableSource.Context
to DynamicTableSource.Context
3. Rename ScanFormat.createScanFormat to DecodingFormat#createRuntimeDecoder()
4. Rename SinkFormat.createSinkFormat to EncodingFormat#createRuntimeEncoder()
5. Rename ScanFormatFactory to DecodingFormatFactory
6. Rename SinkFormatFactory to EncodingFormatFactory
was:
There is some problem with current ScanForamtFactory and SinkFormatFactory
interfaces:
1) {{ScanFormat#createScanFormat}} only accepts {{ScanTableSource.Context}},
which means it can’t work in lookup source.
2) The naming of {{ScanFormat}} also indicates it is only used in scan source.
But a lookup source should be able to work with format too.
3) It’s confusing that {{ScanFormatFactory#createScanFormat}} and
{{ScanFormat#createScanFormat}} (create itself?)
The proposed new interface strucutre:
{code:java}
interface DeserializationFormatFactory {
DeserializationSchemaProvider createDeserializationSchemaProvider(
DynamicTableFactory.Context context, ReadableConfig formatOptions);
}
interface SerializationFormatFactory {
SerializationSchemaProvider createSerializationSchemaProvider(
DynamicTableFactory.Context context, ReadableConfig formatOptions);
}
interface DeserializationSchemaProvider {
DeserializationSchema createDeserializationSchema(
DynamicTableSource.Context context, DataType producedDataType)
}
interface SerializationSchemaProvider{
SerializationSchema createSerializationSchema(
DynamicTableSink.Context context, DataType consumedDataType);
}
{code}
The top-level user-facing interfaces are still {{DeserializationFormatFactory}}
and {{SerializationFormatFactory}}. This is also good for future evolution,
e.g. introducing new {{EncoderFormatFactory}} and {{DecoderFormatFactory}} if
we have a new runtime {{Encoder}} {{Decoder}} in the future. This also solves
our concerns: no concepts of SourceFormat and SinkFormat .
> Improve interface of ScanFormatFactory and SinkFormatFactory
> ------------------------------------------------------------
>
> Key: FLINK-17887
> URL: https://issues.apache.org/jira/browse/FLINK-17887
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / API
> Reporter: Jark Wu
> Assignee: Jark Wu
> Priority: Blocker
> Labels: pull-request-available
> Fix For: 1.11.0
>
>
> There is some problem with current ScanForamtFactory and SinkFormatFactory
> interfaces:
> 1) {{ScanFormat#createScanFormat}} only accepts {{ScanTableSource.Context}},
> which means it can’t work in lookup source.
> 2) The naming of {{ScanFormat}} also indicates it is only used in scan
> source. But a lookup source should be able to work with format too.
> 3) It’s confusing that {{ScanFormatFactory#createScanFormat}} and
> {{ScanFormat#createScanFormat}} (create itself?)
> The proposed changes:
> 1. Have a common interface DynamicTableSource.Context, and make Context of
> ScanTableSource and LookupTableSource extend it, and rename them to
> LookupContext and ScanContext
> 2. Change parameter of ScanFormat.createScanFormat from
> ScanTableSource.Context to DynamicTableSource.Context
> 3. Rename ScanFormat.createScanFormat to DecodingFormat#createRuntimeDecoder()
> 4. Rename SinkFormat.createSinkFormat to EncodingFormat#createRuntimeEncoder()
> 5. Rename ScanFormatFactory to DecodingFormatFactory
> 6. Rename SinkFormatFactory to EncodingFormatFactory
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
