rkhachatryan commented on code in PR #27070:
URL: https://github.com/apache/flink/pull/27070#discussion_r2435803966
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalSink.scala:
##########
@@ -102,13 +103,17 @@ class StreamPhysicalSink(
.reuseOrCreate(cluster.getMetadataQuery)
.getUpsertKeys(inputRel)
+ val config = unwrapTableConfig(this)
new StreamExecSink(
- unwrapTableConfig(this),
+ config,
tableSinkSpec,
inputChangelogMode,
InputProperty.DEFAULT,
FlinkTypeFactory.toLogicalRowType(getRowType),
upsertMaterialize,
+ // persist upsertMaterialize strategy separately in the compiled plan to
make it immutable;
+ // later on, it can't be obtained from the node config because it is
merged with the new environment
+
config.getOptional(TABLE_EXEC_SINK_UPSERT_MATERIALIZE_STRATEGY).orElse(null),
Review Comment:
It can't be `null` here.
But in `StreamExecSink`, it still can be null for the jobs submitted against
older runtime.
I think it would be more clear to write null here as well if the option is
not defined.
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]
