Wei Zhong created FLINK-14591:
---------------------------------
Summary: Execute PlannerBase#mergeParameters every time of calling PlannerBase#translate method
Key: FLINK-14591
URL:
https://issues.apache.org/jira/browse/FLINK-14591 Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Reporter: Wei Zhong
In current implementation of blink planner, the method "PlannerBase#mergeParameter" will be called by "PlannerBase#translate" method to merge the configuration inside TableConfig into global job parameters:
{code:scala}
override def translate(
modifyOperations: util.List[ModifyOperation]): util.List[Transformation[_]] = {
if (modifyOperations.isEmpty) {
return List.empty[Transformation[_]]
}
mergeParameters()
val relNodes = modifyOperations.map(translateToRel)
val optimizedRelNodes = optimize(relNodes)
val execNodes = translateToExecNodePlan(optimizedRelNodes)
translateToPlan(execNodes)
}
{code}
This translate method is called in every important moment, e.g. execute, toDataStream, insertInto, etc.
But as shown above, there is a chance that the method return directly and not call the "mergeParameters".
In fact if we set some configurations between the "Table#insertInto" method and "TableEnvironment#execute" method, these configurations will not be merged into global job parameters because the "mergeParameters" method is not called:
{code:scala}
val env = StreamExecutionEnvironment.getExecutionEnvironment
val tEnv = StreamTableEnvironment.create(env, EnvironmentSettings.newInstance.useBlinkPlanner.build)
...
...
val result = ...
val sink = TestSinkUtil.configureSink(result, new TestingAppendTableSink)
tEnv.registerTableSink("MySink", sink)
tEnv.getConfig.getConfiguration.setString("jobparam1", "value1")
result.insertInto("MySink")
// the "jobparam2" configuration will loss
tEnv.getConfig.getConfiguration.setString("jobparam2", "value2")
tEnv.execute("test")
val jobConfig = env.getConfig.getGlobalJobParameters.toMap
assertTrue(jobConfig.get("jobparam1")=="value1")
// this assertion will fail:
assertTrue(jobConfig.get("jobparam2")=="value2"){code}
This may bring some confusion to the user. It will be great if we can fix this problem.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
Locked
