Hisoka-X commented on code in PR #4918:
URL: https://github.com/apache/seatunnel/pull/4918#discussion_r1338375371
##########
seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkAbstractPluginExecuteProcessor.java:
##########
@@ -76,14 +80,41 @@ protected Optional<DataStream<Row>> fromSourceTable(Config
pluginConfig) {
if (pluginConfig.hasPath(SOURCE_TABLE_NAME)) {
StreamTableEnvironment tableEnvironment =
flinkRuntimeEnvironment.getStreamTableEnvironment();
- Table table =
tableEnvironment.from(pluginConfig.getString(SOURCE_TABLE_NAME));
- return
Optional.ofNullable(TableUtil.tableToDataStream(tableEnvironment, table, true));
+ String tableName = pluginConfig.getString(SOURCE_TABLE_NAME);
+ Table table = tableEnvironment.from(tableName);
+ return Optional.ofNullable(
+ TableUtil.tableToDataStream(
+ tableEnvironment, table,
isAppendMap.getOrDefault(tableName, true)));
}
return Optional.empty();
}
protected void registerResultTable(Config pluginConfig, DataStream<Row>
dataStream) {
- flinkRuntimeEnvironment.registerResultTable(pluginConfig, dataStream);
+ if (pluginConfig.hasPath(RESULT_TABLE_NAME.key())) {
+ String resultTable =
pluginConfig.getString(RESULT_TABLE_NAME.key());
+ if (pluginConfig.hasPath(SOURCE_TABLE_NAME)) {
+ String sourceTable = pluginConfig.getString(SOURCE_TABLE_NAME);
+ flinkRuntimeEnvironment.registerResultTable(
+ pluginConfig,
+ dataStream,
+ resultTable,
+ isAppendMap.getOrDefault(sourceTable, true));
+ registerAppendStream(pluginConfig);
+ return;
+ }
+ flinkRuntimeEnvironment.registerResultTable(
+ pluginConfig,
+ dataStream,
+ resultTable,
+ isAppendMap.getOrDefault(resultTable, true));
Review Comment:
Hi @Carl-Zhou-CN , could you explain for me why we should register table as
source table name when both have `RESULT_TABLE_NAME` and `SOURCE_TABLE_NAME`?
Thanks.
--
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]
