You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2022/03/22 19:13:53 UTC
[flink] 05/08: [FLINK-26777][table-planner] Remove `PlannerConfig` from `PlannerBase`
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b7fb46469a05e8b171b2034e765365456630d92c
Author: Marios Trivyzas <ma...@gmail.com>
AuthorDate: Mon Mar 21 14:52:28 2022 +0200
[FLINK-26777][table-planner] Remove `PlannerConfig` from `PlannerBase`
Remove `PlannerConfig` delegation class, since `TableConfig` is now the
one that holds the complete view of `rootConfiguration` (environment
config) + `configuration` (the application specific configuration).
This closes #19197.
(cherry picked from commit f1bedbe4110c288f561f02e1909b67b694d5278b)
---
.../table/planner/delegation/PlannerConfig.java | 56 ----------------------
.../planner/plan/nodes/exec/ExecNodeBase.java | 4 +-
.../planner/plan/nodes/exec/ExecNodeConfig.java | 47 +++++++-----------
.../exec/processor/DeadlockBreakupProcessor.java | 2 +-
.../MultipleInputNodeCreationProcessor.java | 2 +-
.../table/planner/delegation/PlannerBase.scala | 13 +----
.../table/planner/delegation/StreamPlanner.scala | 2 +-
.../BatchCommonSubGraphBasedOptimizer.scala | 2 +-
.../StreamCommonSubGraphBasedOptimizer.scala | 10 ++--
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 6 +--
.../batch/EnforceLocalHashAggRuleTest.scala | 2 +-
.../batch/EnforceLocalSortAggRuleTest.scala | 2 +-
.../sql/agg/AggregateJoinTransposeITCase.scala | 2 +-
.../flink/table/planner/utils/TableTestBase.scala | 11 ++---
14 files changed, 38 insertions(+), 123 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/PlannerConfig.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/PlannerConfig.java
deleted file mode 100644
index 3563d76..0000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/PlannerConfig.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.table.planner.delegation;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ReadableConfig;
-
-import java.util.Optional;
-
-/**
- * Configuration view that combines the API specific table configuration and the executor
- * configuration. The table configuration has precedence.
- */
-@Internal
-public final class PlannerConfig implements ReadableConfig {
-
- private final ReadableConfig tableConfig;
-
- private final ReadableConfig executorConfig;
-
- PlannerConfig(ReadableConfig tableConfig, ReadableConfig executorConfig) {
- this.tableConfig = tableConfig;
- this.executorConfig = executorConfig;
- }
-
- @Override
- public <T> T get(ConfigOption<T> option) {
- return tableConfig.getOptional(option).orElseGet(() -> executorConfig.get(option));
- }
-
- @Override
- public <T> Optional<T> getOptional(ConfigOption<T> option) {
- final Optional<T> tableValue = tableConfig.getOptional(option);
- if (tableValue.isPresent()) {
- return tableValue;
- }
- return executorConfig.getOptional(option);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeBase.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeBase.java
index 337230c..dbb8810 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeBase.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeBase.java
@@ -129,9 +129,7 @@ public abstract class ExecNodeBase<T> implements ExecNode<T> {
translateToPlanInternal(
(PlannerBase) planner,
new ExecNodeConfig(
- ((PlannerBase) planner).getConfiguration(),
- ((PlannerBase) planner).getTableConfig(),
- new Configuration()));
+ ((PlannerBase) planner).getTableConfig(), new Configuration()));
if (this instanceof SingleTransformationTranslator) {
if (inputsContainSingleton()) {
transformation.setParallelism(1);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeConfig.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeConfig.java
index 183ce14..b19ec9f 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeConfig.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/ExecNodeConfig.java
@@ -20,57 +20,44 @@ package org.apache.flink.table.planner.plan.nodes.exec;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
-import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
import org.apache.flink.table.planner.delegation.PlannerBase;
-import org.apache.flink.table.planner.delegation.PlannerConfig;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.CommonPythonUtil;
import java.time.ZoneId;
import java.util.Optional;
/**
- * Configuration view which is used combine the {@link PlannerConfig} with the {@link
+ * Configuration view which is used combine the {@link PlannerBase#getTableConfig()} with the {@link
* ExecNodeBase#getNodeConfig()} configuration. The persisted configuration of the {@link ExecNode}
- * which is deserialized from the JSON plan has precedence over the {@link PlannerConfig}.
+ * which is deserialized from the JSON plan has precedence over the {@link
+ * PlannerBase#getTableConfig()}.
*/
@Internal
public final class ExecNodeConfig implements ReadableConfig {
- private final ReadableConfig plannerConfig;
-
// See https://issues.apache.org/jira/browse/FLINK-26190
// Used only for the deprecated getMaxIdleStateRetentionTime to also satisfy tests which
// manipulate maxIdleStateRetentionTime, like OverAggregateHarnessTest.
- private final TableConfig originalTableConfig;
- // See https://issues.apache.org/jira/browse/FLINK-26190
private final TableConfig tableConfig;
private final ReadableConfig nodeConfig;
- ExecNodeConfig(
- ReadableConfig plannerConfig, TableConfig tableConfig, ReadableConfig nodeConfig) {
- this.plannerConfig = plannerConfig;
+ ExecNodeConfig(TableConfig tableConfig, ReadableConfig nodeConfig) {
this.nodeConfig = nodeConfig;
- this.originalTableConfig = tableConfig;
- this.tableConfig = TableConfig.getDefault();
- this.tableConfig.setNullCheck(tableConfig.getNullCheck());
- this.tableConfig.setDecimalContext(tableConfig.getDecimalContext());
- this.tableConfig.addConfiguration(tableConfig.getConfiguration());
- this.tableConfig.addConfiguration((Configuration) nodeConfig);
+ this.tableConfig = tableConfig;
}
/**
- * Return the merged {@link TableConfig} from {@link PlannerBase#getTableConfig()} and {@link
- * ExecNodeBase#getNodeConfig()}.
+ * Return the {@link PlannerBase#getTableConfig()}.
*
- * @return the {@link TableConfig}.
- * @deprecated This method is used only for {@link CodeGeneratorContext} and related methods,
- * which end up passing the {@link TableConfig} to the {@link CodeGeneratorContext}. It
- * should be removed once {@link CodeGeneratorContext#nullCheck()} is removed, since for all
- * other usages it's possible to use the {@link ReadableConfig}.
+ * @return the {@link PlannerBase#getTableConfig()}.
+ * @deprecated This method is used only for {@link
+ * CommonPythonUtil#getMergedConfig(StreamExecutionEnvironment, TableConfig)}. It should be
+ * removed when this method is refactored to accept a {@link ReadableConfig} instead.
*/
// See https://issues.apache.org/jira/browse/FLINK-26190
@Deprecated
@@ -80,7 +67,7 @@ public final class ExecNodeConfig implements ReadableConfig {
@Override
public <T> T get(ConfigOption<T> option) {
- return nodeConfig.getOptional(option).orElseGet(() -> plannerConfig.get(option));
+ return nodeConfig.getOptional(option).orElseGet(() -> tableConfig.get(option));
}
@Override
@@ -89,7 +76,7 @@ public final class ExecNodeConfig implements ReadableConfig {
if (tableValue.isPresent()) {
return tableValue;
}
- return plannerConfig.getOptional(option);
+ return tableConfig.getOptional(option);
}
/** @return The duration until state which was not updated will be retained. */
@@ -99,13 +86,13 @@ public final class ExecNodeConfig implements ReadableConfig {
// See https://issues.apache.org/jira/browse/FLINK-26190
/**
- * Using {@link #originalTableConfig} to satisify tests like {@code OverAggregateHarnessTest},
- * which use {@code HarnessTestBase#TestTableConfig} to individually manipulate the
+ * Using {@link #tableConfig} to satisify tests like {@code OverAggregateHarnessTest}, which use
+ * {@code HarnessTestBase#TestTableConfig} to individually manipulate the
* maxIdleStateRetentionTime. See {@link TableConfig#getMaxIdleStateRetentionTime()}.
*/
@Deprecated
public long getMaxIdleStateRetentionTime() {
- return originalTableConfig.getMaxIdleStateRetentionTime();
+ return tableConfig.getMaxIdleStateRetentionTime();
}
// See https://issues.apache.org/jira/browse/FLINK-26190
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/DeadlockBreakupProcessor.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/DeadlockBreakupProcessor.java
index 9959262..6c1e769 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/DeadlockBreakupProcessor.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/DeadlockBreakupProcessor.java
@@ -44,7 +44,7 @@ public class DeadlockBreakupProcessor implements ExecNodeGraphProcessor {
execGraph.getRootNodes(),
InputProperty.DamBehavior.END_INPUT,
StreamExchangeMode.BATCH,
- context.getPlanner().getConfiguration());
+ context.getPlanner().getTableConfig());
resolver.detectAndResolve();
return execGraph;
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/MultipleInputNodeCreationProcessor.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/MultipleInputNodeCreationProcessor.java
index 8cc3cc4..fc245d4 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/MultipleInputNodeCreationProcessor.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/MultipleInputNodeCreationProcessor.java
@@ -79,7 +79,7 @@ public class MultipleInputNodeCreationProcessor implements ExecNodeGraphProcesso
execGraph.getRootNodes(),
InputProperty.DamBehavior.BLOCKING,
StreamExchangeMode.PIPELINED,
- context.getPlanner().getConfiguration());
+ context.getPlanner().getTableConfig());
resolver.detectAndResolve();
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
index 05d7baf..a3a5c24 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
@@ -103,10 +103,6 @@ abstract class PlannerBase(
private var parser: Parser = _
private var currentDialect: SqlDialect = getTableConfig.getSqlDialect
- private val plannerConfiguration: ReadableConfig = new PlannerConfig(
- tableConfig.getConfiguration,
- executor.getConfiguration)
-
@VisibleForTesting
private[flink] val plannerContext: PlannerContext =
new PlannerContext(
@@ -148,14 +144,7 @@ abstract class PlannerBase(
def getFlinkContext: FlinkContext = plannerContext.getFlinkContext
/**
- * Gives access to both API specific table configuration and executor configuration.
- *
- * This configuration should be the main source of truth in the planner module.
- */
- def getConfiguration: ReadableConfig = plannerConfiguration
-
- /**
- * @deprecated Do not use this method anymore. Use [[getConfiguration]] to access options.
+ * @deprecated Do not use this method anymore. Use [[getTableConfig]] to access options.
* Create transformations without it. A [[StreamExecutionEnvironment]] is a mixture
* of executor and stream graph generator/builder. In the long term, we would like
* to avoid the need for it in the planner module.
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/StreamPlanner.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/StreamPlanner.scala
index c3839bc..1c4a512 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/StreamPlanner.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/StreamPlanner.scala
@@ -211,7 +211,7 @@ class StreamPlanner(
override def beforeTranslation(): Unit = {
super.beforeTranslation()
- val runtimeMode = getConfiguration.get(ExecutionOptions.RUNTIME_MODE)
+ val runtimeMode = getTableConfig.get(ExecutionOptions.RUNTIME_MODE)
if (runtimeMode != RuntimeExecutionMode.STREAMING) {
throw new IllegalArgumentException(
"Mismatch between configured runtime mode and actual runtime mode. " +
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala
index 62d2d9a..ba08b50 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala
@@ -41,7 +41,7 @@ class BatchCommonSubGraphBasedOptimizer(planner: BatchPlanner)
override protected def doOptimize(roots: Seq[RelNode]): Seq[RelNodeBlock] = {
// build RelNodeBlock plan
- val rootBlocks = RelNodeBlockPlanBuilder.buildRelNodeBlockPlan(roots, planner.getConfiguration)
+ val rootBlocks = RelNodeBlockPlanBuilder.buildRelNodeBlockPlan(roots, planner.getTableConfig)
// optimize recursively RelNodeBlock
rootBlocks.foreach(optimizeBlock)
rootBlocks
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala
index 6205d93..a449858 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala
@@ -49,17 +49,17 @@ class StreamCommonSubGraphBasedOptimizer(planner: StreamPlanner)
extends CommonSubGraphBasedOptimizer {
override protected def doOptimize(roots: Seq[RelNode]): Seq[RelNodeBlock] = {
- val config = planner.getConfiguration
+ val tableConfig = planner.getTableConfig
// build RelNodeBlock plan
- val sinkBlocks = RelNodeBlockPlanBuilder.buildRelNodeBlockPlan(roots, config)
+ val sinkBlocks = RelNodeBlockPlanBuilder.buildRelNodeBlockPlan(roots, tableConfig)
// infer trait properties for sink block
sinkBlocks.foreach { sinkBlock =>
// don't require update before by default
sinkBlock.setUpdateBeforeRequired(false)
- val miniBatchInterval: MiniBatchInterval = if (config.get(
+ val miniBatchInterval: MiniBatchInterval = if (tableConfig.get(
ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED)) {
- val miniBatchLatency = config.get(
+ val miniBatchLatency = tableConfig.get(
ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY).toMillis
Preconditions.checkArgument(miniBatchLatency > 0,
"MiniBatch Latency must be greater than 0 ms.", null)
@@ -156,7 +156,7 @@ class StreamCommonSubGraphBasedOptimizer(planner: StreamPlanner)
val config = planner.getTableConfig
val calciteConfig = TableConfigUtils.getCalciteConfig(config)
val programs = calciteConfig.getStreamProgram
- .getOrElse(FlinkStreamProgram.buildProgram(config.getConfiguration))
+ .getOrElse(FlinkStreamProgram.buildProgram(config))
Preconditions.checkNotNull(programs)
val context = relNode.getCluster.getPlanner.getContext.unwrap(classOf[FlinkContext])
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
index ef9db28..b1a1265 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
@@ -978,7 +978,7 @@ class FlinkRelMdHandlerTestBase {
cluster, streamPhysicalTraits, streamTs, program, program.getOutputRowType)
val streamExchange = new StreamPhysicalExchange(
cluster, streamPhysicalTraits.replace(hash01), streamCalc, hash01)
- val emitStrategy = WindowEmitStrategy(tableConfig.getConfiguration, tumblingGroupWindow)
+ val emitStrategy = WindowEmitStrategy(tableConfig, tumblingGroupWindow)
val streamWindowAgg = new StreamPhysicalGroupWindowTableAggregate(
cluster,
streamPhysicalTraits,
@@ -1620,7 +1620,7 @@ class FlinkRelMdHandlerTestBase {
cluster, streamPhysicalTraits, streamTs, program, program.getOutputRowType)
val streamExchange = new StreamPhysicalExchange(
cluster, streamPhysicalTraits.replace(hash01), streamCalc, hash01)
- val emitStrategy = WindowEmitStrategy(tableConfig.getConfiguration, tumblingGroupWindow)
+ val emitStrategy = WindowEmitStrategy(tableConfig, tumblingGroupWindow)
val streamWindowAgg = new StreamPhysicalGroupWindowAggregate(
cluster,
streamPhysicalTraits,
@@ -1759,7 +1759,7 @@ class FlinkRelMdHandlerTestBase {
cluster, streamPhysicalTraits, streamTs, program, program.getOutputRowType)
val streamExchange = new StreamPhysicalExchange(
cluster, streamPhysicalTraits.replace(hash1), streamCalc, hash1)
- val emitStrategy = WindowEmitStrategy(tableConfig.getConfiguration, tumblingGroupWindow)
+ val emitStrategy = WindowEmitStrategy(tableConfig, tumblingGroupWindow)
val streamWindowAgg = new StreamPhysicalGroupWindowAggregate(
cluster,
streamPhysicalTraits,
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalHashAggRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalHashAggRuleTest.scala
index 9d93773..bfbae11 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalHashAggRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalHashAggRuleTest.scala
@@ -38,7 +38,7 @@ class EnforceLocalHashAggRuleTest extends EnforceLocalAggRuleTestBase {
@Before
override def setup(): Unit = {
super.setup()
- val program = FlinkBatchProgram.buildProgram(util.tableEnv.getConfig.getConfiguration)
+ val program = FlinkBatchProgram.buildProgram(util.tableEnv.getConfig)
// remove the original BatchExecHashAggRule and add BatchExecHashAggRuleForOnePhase
// to let the physical phase generate one phase aggregate
program.getFlinkRuleSetProgram(FlinkBatchProgram.PHYSICAL)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalSortAggRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalSortAggRuleTest.scala
index 29520c5..03a72a1 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalSortAggRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalSortAggRuleTest.scala
@@ -42,7 +42,7 @@ class EnforceLocalSortAggRuleTest extends EnforceLocalAggRuleTestBase {
super.setup()
util.addFunction("weightedAvg", new WeightedAvg)
- val program = FlinkBatchProgram.buildProgram(util.tableEnv.getConfig.getConfiguration)
+ val program = FlinkBatchProgram.buildProgram(util.tableEnv.getConfig)
// remove the original BatchExecSortAggRule and add BatchExecSortAggRuleForOnePhase
// to let the physical phase generate one phase aggregate
program.getFlinkRuleSetProgram(FlinkBatchProgram.PHYSICAL)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateJoinTransposeITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateJoinTransposeITCase.scala
index c125235..1668880 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateJoinTransposeITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateJoinTransposeITCase.scala
@@ -43,7 +43,7 @@ class AggregateJoinTransposeITCase extends BatchTestBase {
@Before
override def before(): Unit = {
super.before()
- val programs = FlinkBatchProgram.buildProgram(tEnv.getConfig.getConfiguration)
+ val programs = FlinkBatchProgram.buildProgram(tEnv.getConfig)
// remove FlinkAggregateJoinTransposeRule from logical program (volcano planner)
programs.getFlinkRuleSetProgram(FlinkBatchProgram.LOGICAL)
.getOrElse(throw new TableException(s"${FlinkBatchProgram.LOGICAL} does not exist"))
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
index cb56447..83818a5 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
@@ -32,7 +32,6 @@ import org.apache.flink.streaming.api.environment.{LocalStreamEnvironment, Strea
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment => ScalaStreamExecEnv}
import org.apache.flink.streaming.api.{TimeCharacteristic, environment}
import org.apache.flink.table.api._
-import org.apache.flink.table.api.bridge.internal.AbstractStreamTableEnvironmentImpl
import org.apache.flink.table.api.bridge.java.{StreamTableEnvironment => JavaStreamTableEnv}
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment => ScalaStreamTableEnv}
import org.apache.flink.table.api.config.ExecutionConfigOptions
@@ -1261,7 +1260,7 @@ case class StreamTableTestUtil(
}
def buildStreamProgram(firstProgramNameToRemove: String): Unit = {
- val program = FlinkStreamProgram.buildProgram(tableEnv.getConfig.getConfiguration)
+ val program = FlinkStreamProgram.buildProgram(tableEnv.getConfig)
var startRemove = false
program.getProgramNames.foreach {
name =>
@@ -1285,8 +1284,7 @@ case class StreamTableTestUtil(
def getStreamProgram(): FlinkChainedProgram[StreamOptimizeContext] = {
val tableConfig = tableEnv.getConfig
val calciteConfig = TableConfigUtils.getCalciteConfig(tableConfig)
- calciteConfig.getStreamProgram.getOrElse(FlinkStreamProgram.buildProgram(
- tableConfig.getConfiguration))
+ calciteConfig.getStreamProgram.getOrElse(FlinkStreamProgram.buildProgram(tableConfig))
}
def enableMiniBatch(): Unit = {
@@ -1346,7 +1344,7 @@ case class BatchTableTestUtil(
extends TableTestUtil(test, isStreamingMode = false, catalogManager, conf) {
def buildBatchProgram(firstProgramNameToRemove: String): Unit = {
- val program = FlinkBatchProgram.buildProgram(tableEnv.getConfig.getConfiguration)
+ val program = FlinkBatchProgram.buildProgram(tableEnv.getConfig)
var startRemove = false
program.getProgramNames.foreach {
name =>
@@ -1370,8 +1368,7 @@ case class BatchTableTestUtil(
def getBatchProgram(): FlinkChainedProgram[BatchOptimizeContext] = {
val tableConfig = tableEnv.getConfig
val calciteConfig = TableConfigUtils.getCalciteConfig(tableConfig)
- calciteConfig.getBatchProgram.getOrElse(FlinkBatchProgram.buildProgram(
- tableConfig.getConfiguration))
+ calciteConfig.getBatchProgram.getOrElse(FlinkBatchProgram.buildProgram(tableConfig))
}
def createCollectTableSink(