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.~~
I got it. Please ignore it.
--
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: commits-unsubscr...@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
