wuchong commented on a change in pull request #15018:
URL: https://github.com/apache/flink/pull/15018#discussion_r588026681
##########
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
##########
@@ -260,12 +261,17 @@ protected TableEnvironmentImpl(
isStreamingMode);
}
+ public static TableEnvironmentImpl create(Configuration configuration) {
+ return create(EnvironmentSettings.fromConfiguration(configuration));
Review comment:
This will lose other configs in `configuration`, we can add a private
`create(Configuration conf, EnvironmentSettings settings)` method as the basic
implementation.
##########
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/TableConfigOptions.java
##########
@@ -34,6 +35,15 @@
public class TableConfigOptions {
private TableConfigOptions() {}
+ @Documentation.TableOption(execMode =
Documentation.ExecMode.BATCH_STREAMING)
+ public static final ConfigOption<PlannerType> TABLE_PLANNER =
+ key("table.planner")
+ .enumType(PlannerType.class)
+ .defaultValue(PlannerType.BLINK)
+ .withDescription(
+ "Use blink planner or old planner. The default
value for the option is blink. "
+ + "The value of the option will also
influence time semantics and so on. Currently it's forbidden to set this option
after initialization of the table environment.");
Review comment:
We don't need to mention the time semantics differences... if we want to
list differences, we can list a full page.
Suggestions:
```
Use either 'blink' planner or 'old' planner. Default is blink planner. For
TableEnvironment, this option is used to construct a TableEnvironment, but this
option can't be changed after that. However there is no such limitation for SQL
Client.
```
##########
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java
##########
@@ -75,7 +75,17 @@
private MathContext decimalContext = MathContext.DECIMAL128;
/** A configuration object to hold all key/value configuration. */
- private Configuration configuration = new Configuration();
+ private final Configuration configuration;
+
+ /** Create TableConfig with default configuration. */
+ public TableConfig() {
+ this.configuration = new Configuration();
+ }
+
+ /** Create TableConfig with specified configuration. */
+ public TableConfig(Configuration configuration) {
Review comment:
This looks like more internal used. Can we replace the usage to the
following ?
```java
val tableConfig = new TableConfig
tableConfig.addConfiguration(conf)
```
##########
File path:
flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/utils/BatchTestBase.scala
##########
@@ -73,6 +77,7 @@ class BatchTestBase extends BatchAbstractTestBase {
@Before
def before(): Unit = {
+ conf.getConfiguration.set(ExecutionOptions.RUNTIME_MODE,
RuntimeExecutionMode.BATCH)
Review comment:
Do we need this? It seems `TestingTableEnvironment.create()` already
cover this.
##########
File path:
flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/api/TableEnvironmentITCase.scala
##########
@@ -80,6 +84,49 @@ class TableEnvironmentITCase(tableEnvName: String,
isStreaming: Boolean) extends
TestTableSourceSinks.createPersonCsvTemporaryTable(tEnv, "MyTable")
}
+ @Test
+ def testSetPlannerType: Unit = {
+ tEnv.getConfig.getConfiguration.set(TableConfigOptions.TABLE_PLANNER,
PlannerType.OLD)
+
+ TestTableSourceSinks.createCsvTemporarySinkTable(
+ tEnv, new TableSchema(Array("first"), Array(STRING)), "MySink1")
+
+ val expected = "Mismatch between configured planner and actual planner. " +
+ "Currently, the 'table.planner' can only be set " +
+ "when instantiating the table environment. Subsequent changes are not
supported. " +
+ "Please instantiate a new TableEnvironment if necessary."
+
+ try {
+ tEnv.executeSql("insert into MySink1 select first from MyTable")
+ } catch {
+ case e: IllegalArgumentException => Assert.assertEquals(expected,
e.getMessage); return
+ case e: Throwable => throw e
+ }
+ fail("Don't trigger validation.")
Review comment:
We can use `thrown.expect(...)` to easily assert exceptions.
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
##########
@@ -443,4 +447,14 @@ abstract class PlannerBase(
private def getClassLoader: ClassLoader = {
Thread.currentThread().getContextClassLoader
}
+
+ protected def isSpecifiedPlanner(): Unit = {
Review comment:
This method sounds return a boolean type and doesn't do any validation.
Besides, I think we can also move `planner.overrideEnvParallelism()` and
`getExecEnv.configure(getTableConfig.getConfiguration, classloader)` into this
method. So that we will not miss to override configuraiton. What about calling
the new method `validateAndOverrideConfiguration()` and add a comment on the
method?
Besides, I suggest to call this method in the implementations of
`Planner#translate` , `Planner#explain`, `Planner#getJsonPlan`,
`Planner#explainJsonPlan`, `Planner#translateJsonPlan`. This should cover all
cases and will be easier to track in the future.
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/BatchPlanner.scala
##########
@@ -110,6 +112,7 @@ class BatchPlanner(
)
translateToRel(modifyOperation)
case _ =>
+ isSpecifiedPlanner()
Review comment:
This is not needed because we will call `optimize` later, and we have
validated in `optimize`.
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/delegation/BatchPlanner.scala
##########
@@ -177,10 +177,11 @@ class BatchPlanner(
super.isSpecifiedPlanner();
if (!config.getConfiguration.get(ExecutionOptions.RUNTIME_MODE)
.equals(RuntimeExecutionMode.BATCH)) {
- throw new IllegalArgumentException("Expect STREAMING mode but get BATCH
mode. " +
- "Please make sure `execution.runtime-mode` is consistent with the " +
- "current TableEnvironment. Otherwise rebuild a new TableEnvironment
that is satisfied " +
- "the requirement.")
+ throw new IllegalArgumentException(
+ "Mismatch between configured planner and actual planner. " +
Review comment:
`Mismatch between configured runtime mode and actual runtime mode`?
##########
File path:
flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
##########
@@ -1246,6 +1248,8 @@ case class BatchTableTestUtil(
conf: TableConfig = new TableConfig)
extends TableTestUtil(test, isStreamingMode = false, catalogManager, conf) {
+ conf.getConfiguration.set(ExecutionOptions.RUNTIME_MODE,
RuntimeExecutionMode.BATCH)
Review comment:
Do we need this? In theory, users don't need to explicitly set the
configuraiton when using existing APIs.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
