ferenc-csaky commented on code in PR #161:
URL:
https://github.com/apache/flink-connector-jdbc/pull/161#discussion_r2549526092
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/RowDataResultExtractor.java:
##########
Review Comment:
I'd rather put this class into the `extractor` package, next to its
interface and `RowExtractor`. Based on the current grouping IMO it makes more
sense.
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/source/JdbcDynamicTableSource.java:
##########
@@ -314,4 +328,30 @@ private Serializable[][] replicatePushdownParamsForN(int
n) {
}
return allPushdownParams;
}
+
+ private static class JdbcDataStreamScanProvider implements
DataStreamScanProvider {
+
+ private final JdbcSource<RowData> source;
+
+ public JdbcDataStreamScanProvider(JdbcSource<RowData> source) {
+ this.source = Preconditions.checkNotNull(source);
+ }
+
+ @Override
+ public DataStream<RowData> produceDataStream(
+ ProviderContext providerContext, StreamExecutionEnvironment
execEnv) {
+ DataStreamSource<RowData> sourceStream =
+ execEnv.fromSource(
+ source,
+ WatermarkStrategy.noWatermarks(),
+ JdbcDynamicTableSource.class.getSimpleName());
Review Comment:
I think `JdbcSource.class` makes more sense here, since we produce a
`DataStream` here.
I'd also include the relevant Flink table name here so it will be trivial
which table this DS represents. We can add a `String tableIdentifier` class
field into `JdbcDynamicTableSource` and it should be set to
`context.getObjectIdentifier().asSummaryString()` in the
`JdbcDynamicTableFactory`. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
