You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by go...@apache.org on 2021/01/09 03:24:51 UTC
[flink] 02/04: [FLINK-20857][table-planner-blink] Introduce
BatchPhysicalHashWindowAggregate & BatchPhysicalLocalHashWindowAggregate,
and make BatchExecHashWindowAggregate only extended from ExecNode
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a8e591ad0461e67554ffc1bc9261055f9405ecbc
Author: godfreyhe <go...@163.com>
AuthorDate: Thu Jan 7 14:48:38 2021 +0800
[FLINK-20857][table-planner-blink] Introduce BatchPhysicalHashWindowAggregate & BatchPhysicalLocalHashWindowAggregate, and make BatchExecHashWindowAggregate only extended from ExecNode
This closes #14574
---
.../exec/batch/BatchExecHashWindowAggregate.java | 149 +++++++++++++++++++++
.../agg/batch/HashWindowCodeGenerator.scala | 2 +-
.../agg/batch/SortWindowCodeGenerator.scala | 2 +-
.../codegen/agg/batch/WindowCodeGenerator.scala | 33 ++---
.../metadata/AggCallSelectivityEstimator.scala | 4 +-
.../plan/metadata/FlinkRelMdColumnInterval.scala | 2 +-
.../planner/plan/metadata/FlinkRelMdSize.scala | 2 +-
.../BatchExecPythonGroupWindowAggregate.scala | 6 +-
.../batch/BatchExecSortWindowAggregateBase.scala | 6 +-
...cala => BatchPhysicalHashWindowAggregate.scala} | 42 +++---
... => BatchPhysicalHashWindowAggregateBase.scala} | 67 +--------
...=> BatchPhysicalLocalHashWindowAggregate.scala} | 39 ++++--
.../planner/plan/rules/FlinkBatchRuleSets.scala | 2 +-
.../batch/BatchPhysicalSortLimitRule.scala | 8 +-
...cala => BatchPhysicalWindowAggregateRule.scala} | 24 ++--
.../batch/RemoveRedundantLocalHashAggRule.scala | 2 +-
.../batch/RemoveRedundantLocalSortAggRule.scala | 2 +-
.../table/planner/plan/utils/FlinkRelMdUtil.scala | 6 +-
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 18 +--
19 files changed, 270 insertions(+), 146 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecHashWindowAggregate.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecHashWindowAggregate.java
new file mode 100644
index 0000000..d8b147d
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecHashWindowAggregate.java
@@ -0,0 +1,149 @@
+/*
+ * 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.plan.nodes.exec.batch;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.calcite.FlinkRelBuilder.PlannerNamedWindowProperty;
+import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.planner.codegen.agg.batch.HashWindowCodeGenerator;
+import org.apache.flink.table.planner.codegen.agg.batch.WindowCodeGenerator;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.logical.LogicalWindow;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
+import org.apache.flink.table.planner.plan.utils.AggregateInfoList;
+import org.apache.flink.table.planner.plan.utils.AggregateUtil;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+import org.apache.flink.table.runtime.generated.GeneratedOperator;
+import org.apache.flink.table.runtime.operators.CodeGenOperatorFactory;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.calcite.rel.core.AggregateCall;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+/** Batch {@link ExecNode} for hash-based window aggregate operator. */
+public class BatchExecHashWindowAggregate extends ExecNodeBase<RowData>
+ implements BatchExecNode<RowData> {
+
+ private final int[] grouping;
+ private final int[] auxGrouping;
+ private final AggregateCall[] aggCalls;
+ private final LogicalWindow window;
+ private final int inputTimeFieldIndex;
+ private final boolean inputTimeIsDate;
+ private final PlannerNamedWindowProperty[] namedWindowProperties;
+ private final RowType aggInputRowType;
+ private final boolean enableAssignPane;
+ private final boolean isMerge;
+ private final boolean isFinal;
+
+ public BatchExecHashWindowAggregate(
+ int[] grouping,
+ int[] auxGrouping,
+ AggregateCall[] aggCalls,
+ LogicalWindow window,
+ int inputTimeFieldIndex,
+ boolean inputTimeIsDate,
+ PlannerNamedWindowProperty[] namedWindowProperties,
+ RowType aggInputRowType,
+ boolean enableAssignPane,
+ boolean isMerge,
+ boolean isFinal,
+ ExecEdge inputEdge,
+ RowType outputType,
+ String description) {
+ super(Collections.singletonList(inputEdge), outputType, description);
+ this.grouping = grouping;
+ this.auxGrouping = auxGrouping;
+ this.aggCalls = aggCalls;
+ this.window = window;
+ this.inputTimeFieldIndex = inputTimeFieldIndex;
+ this.inputTimeIsDate = inputTimeIsDate;
+ this.namedWindowProperties = namedWindowProperties;
+ this.aggInputRowType = aggInputRowType;
+ this.enableAssignPane = enableAssignPane;
+ this.isMerge = isMerge;
+ this.isFinal = isFinal;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner) {
+ final ExecNode<RowData> inputNode = (ExecNode<RowData>) getInputNodes().get(0);
+ final Transformation<RowData> inputTransform = inputNode.translateToPlan(planner);
+
+ final AggregateInfoList aggInfos =
+ AggregateUtil.transformToBatchAggregateInfoList(
+ aggInputRowType,
+ JavaScalaConversionUtil.toScala(Arrays.asList(aggCalls)),
+ null,
+ null);
+ final TableConfig tableConfig = planner.getTableConfig();
+ final RowType inputRowType = (RowType) inputNode.getOutputType();
+ final HashWindowCodeGenerator hashWindowCodeGenerator =
+ new HashWindowCodeGenerator(
+ new CodeGeneratorContext(tableConfig),
+ planner.getRelBuilder(),
+ window,
+ inputTimeFieldIndex,
+ inputTimeIsDate,
+ JavaScalaConversionUtil.toScala(Arrays.asList(namedWindowProperties)),
+ aggInfos,
+ inputRowType,
+ grouping,
+ auxGrouping,
+ enableAssignPane,
+ isMerge,
+ isFinal);
+ final int groupBufferLimitSize =
+ tableConfig
+ .getConfiguration()
+ .getInteger(ExecutionConfigOptions.TABLE_EXEC_WINDOW_AGG_BUFFER_SIZE_LIMIT);
+ final Tuple2<Long, Long> windowSizeAndSlideSize = WindowCodeGenerator.getWindowDef(window);
+ final GeneratedOperator<OneInputStreamOperator<RowData, RowData>> generatedOperator =
+ hashWindowCodeGenerator.gen(
+ inputRowType,
+ (RowType) getOutputType(),
+ groupBufferLimitSize,
+ 0, // windowStart
+ windowSizeAndSlideSize.f0,
+ windowSizeAndSlideSize.f1);
+
+ final long managedMemory =
+ ExecNodeUtil.getMemorySize(
+ tableConfig, ExecutionConfigOptions.TABLE_EXEC_RESOURCE_HASH_AGG_MEMORY);
+ return ExecNodeUtil.createOneInputTransformation(
+ inputTransform,
+ getDesc(),
+ new CodeGenOperatorFactory<>(generatedOperator),
+ InternalTypeInfo.of(getOutputType()),
+ inputTransform.getParallelism(),
+ managedMemory);
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashWindowCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashWindowCodeGenerator.scala
index a1d0a65..76de099 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashWindowCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashWindowCodeGenerator.scala
@@ -72,7 +72,7 @@ class HashWindowCodeGenerator(
inputTimeIsDate: Boolean,
namedProperties: Seq[PlannerNamedWindowProperty],
aggInfoList: AggregateInfoList,
- inputRowType: RelDataType,
+ inputRowType: RowType,
grouping: Array[Int],
auxGrouping: Array[Int],
enableAssignPane: Boolean = true,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/SortWindowCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/SortWindowCodeGenerator.scala
index 3677b41..bce1f56 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/SortWindowCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/SortWindowCodeGenerator.scala
@@ -62,7 +62,7 @@ class SortWindowCodeGenerator(
inputTimeIsDate: Boolean,
namedProperties: Seq[PlannerNamedWindowProperty],
aggInfoList: AggregateInfoList,
- inputRowType: RelDataType,
+ inputRowType: RowType,
inputType: RowType,
outputType: RowType,
buffLimitSize: Int,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala
index b28c813..f30568c 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala
@@ -18,11 +18,7 @@
package org.apache.flink.table.planner.codegen.agg.batch
-import org.apache.calcite.avatica.util.DateTimeUtils
-import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.core.AggregateCall
-import org.apache.calcite.tools.RelBuilder
-import org.apache.commons.math3.util.ArithmeticUtils
+import org.apache.flink.api.java.tuple.{Tuple2 => JTuple2}
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.data.binary.BinaryRowData
import org.apache.flink.table.data.utils.JoinedRowData
@@ -52,6 +48,11 @@ import org.apache.flink.table.types.logical.LogicalTypeRoot.INTERVAL_DAY_TIME
import org.apache.flink.table.types.logical._
import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasRoot
+import org.apache.calcite.avatica.util.DateTimeUtils
+import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.tools.RelBuilder
+import org.apache.commons.math3.util.ArithmeticUtils
+
import scala.collection.JavaConversions._
abstract class WindowCodeGenerator(
@@ -61,14 +62,15 @@ abstract class WindowCodeGenerator(
inputTimeIsDate: Boolean,
namedProperties: Seq[PlannerNamedWindowProperty],
aggInfoList: AggregateInfoList,
- inputRowType: RelDataType,
+ inputRowType: RowType,
grouping: Array[Int],
auxGrouping: Array[Int],
enableAssignPane: Boolean = true,
val isMerge: Boolean,
val isFinal: Boolean) {
- protected lazy val builder: RelBuilder = relBuilder.values(inputRowType)
+ protected lazy val builder: RelBuilder = relBuilder.values(
+ FlinkTypeFactory.INSTANCE.buildRelNodeRowType(inputRowType))
protected lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
@@ -79,14 +81,12 @@ abstract class WindowCodeGenerator(
AggCodeGenHelper.getAggBufferNames(auxGrouping, aggInfos)
protected lazy val aggBufferTypes: Array[Array[LogicalType]] = AggCodeGenHelper.getAggBufferTypes(
- inputType,
+ inputRowType,
auxGrouping,
aggInfos)
- protected lazy val groupKeyRowType: RowType = AggCodeGenHelper.projectRowType(inputType, grouping)
-
- private lazy val inputType: RowType =
- FlinkTypeFactory.toLogicalType(inputRowType).asInstanceOf[RowType]
+ protected lazy val groupKeyRowType: RowType =
+ AggCodeGenHelper.projectRowType(inputRowType, grouping)
protected lazy val timestampInternalType: LogicalType =
if (inputTimeIsDate) new IntType() else new BigIntType()
@@ -116,7 +116,7 @@ abstract class WindowCodeGenerator(
(groupKeyTypes :+ timestampInternalType) ++ aggBuffTypes,
((groupKeyNames :+ "assignedTs$") ++ aggBuffNames).toArray)
} else {
- FlinkTypeFactory.toLogicalRowType(inputRowType)
+ inputRowType
}
}
@@ -680,7 +680,8 @@ abstract class WindowCodeGenerator(
remainder)),
literal(index * slideSize))
exprCodegen.generateExpression(new CallExpressionResolver(relBuilder).resolve(expr).accept(
- new ExpressionConverter(relBuilder.values(inputRowType))))
+ new ExpressionConverter(
+ relBuilder.values(FlinkTypeFactory.INSTANCE.buildRelNodeRowType(inputRowType)))))
}
def getGrouping: Array[Int] = grouping
@@ -726,7 +727,7 @@ abstract class WindowCodeGenerator(
object WindowCodeGenerator {
- def getWindowDef(window: LogicalWindow): (Long, Long) = {
+ def getWindowDef(window: LogicalWindow): JTuple2[JLong, JLong] = {
val (windowSize, slideSize): (Long, Long) = window match {
case TumblingGroupWindow(_, _, size) if isTimeIntervalLiteral(size) =>
(asLong(size), asLong(size))
@@ -736,7 +737,7 @@ object WindowCodeGenerator {
// count tumbling/sliding window and session window not supported now
throw new UnsupportedOperationException(s"Window $window is not supported right now.")
}
- (windowSize, slideSize)
+ new JTuple2[JLong, JLong](windowSize, slideSize)
}
def asLong(expr: Expression): Long = extractValue(expr, classOf[JLong]).get()
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimator.scala
index acc90d5..26eba34 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimator.scala
@@ -19,7 +19,7 @@
package org.apache.flink.table.planner.plan.metadata
import org.apache.flink.table.planner.JDouble
-import org.apache.flink.table.planner.plan.nodes.physical.batch.{BatchExecLocalHashWindowAggregate, BatchExecLocalSortWindowAggregate, BatchPhysicalGroupAggregateBase, BatchPhysicalWindowAggregateBase}
+import org.apache.flink.table.planner.plan.nodes.physical.batch.{BatchExecLocalSortWindowAggregate, BatchPhysicalGroupAggregateBase, BatchPhysicalLocalHashWindowAggregate, BatchPhysicalWindowAggregateBase}
import org.apache.flink.table.planner.plan.stats._
import org.apache.flink.table.planner.plan.utils.AggregateUtil
@@ -62,7 +62,7 @@ class AggCallSelectivityEstimator(agg: RelNode, mq: FlinkRelMetadataQuery)
(rel.getGroupSet.toArray ++ auxGroupSet, otherAggCalls)
case rel: BatchPhysicalGroupAggregateBase =>
(rel.grouping ++ rel.auxGrouping, rel.getAggCallList)
- case rel: BatchExecLocalHashWindowAggregate =>
+ case rel: BatchPhysicalLocalHashWindowAggregate =>
val fullGrouping = rel.grouping ++ Array(rel.inputTimeFieldIndex) ++ rel.auxGrouping
(fullGrouping, rel.getAggCallList)
case rel: BatchExecLocalSortWindowAggregate =>
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnInterval.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnInterval.scala
index 4b80a55..b1eec16 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnInterval.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnInterval.scala
@@ -545,7 +545,7 @@ class FlinkRelMdColumnInterval private extends MetadataHandler[ColumnInterval] {
case agg: BatchExecLocalSortWindowAggregate =>
// grouping + assignTs + auxGrouping
agg.grouping ++ Array(agg.inputTimeFieldIndex) ++ agg.auxGrouping
- case agg: BatchExecLocalHashWindowAggregate =>
+ case agg: BatchPhysicalLocalHashWindowAggregate =>
// grouping + assignTs + auxGrouping
agg.grouping ++ Array(agg.inputTimeFieldIndex) ++ agg.auxGrouping
case agg: BatchPhysicalWindowAggregateBase => agg.grouping ++ agg.auxGrouping
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala
index e408203..199e5d4 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala
@@ -213,7 +213,7 @@ class FlinkRelMdSize private extends MetadataHandler[BuiltInMetadata.Size] {
val mapInputToOutput: Map[Int, Int] = windowAgg match {
case agg: WindowAggregate =>
AggregateUtil.checkAndGetFullGroupSet(agg).zipWithIndex.toMap
- case agg: BatchExecLocalHashWindowAggregate =>
+ case agg: BatchPhysicalLocalHashWindowAggregate =>
// local win-agg output type: grouping + assignTs + auxGrouping + aggCalls
agg.grouping.zipWithIndex.toMap ++
agg.auxGrouping.zipWithIndex.map {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecPythonGroupWindowAggregate.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecPythonGroupWindowAggregate.scala
index ba84b24..a5e5d59 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecPythonGroupWindowAggregate.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecPythonGroupWindowAggregate.scala
@@ -126,7 +126,7 @@ class BatchExecPythonGroupWindowAggregate(
val outputType = FlinkTypeFactory.toLogicalRowType(getRowType)
val inputType = FlinkTypeFactory.toLogicalRowType(inputRowType)
- val (windowSize: Long, slideSize: Long) = WindowCodeGenerator.getWindowDef(window)
+ val windowSizeAndSlideSize = WindowCodeGenerator.getWindowDef(window)
val groupBufferLimitSize = planner.getTableConfig.getConfiguration.getInteger(
ExecutionConfigOptions.TABLE_EXEC_WINDOW_AGG_BUFFER_SIZE_LIMIT)
@@ -137,8 +137,8 @@ class BatchExecPythonGroupWindowAggregate(
outputType,
inputTimeFieldIndex,
groupBufferLimitSize,
- windowSize,
- slideSize,
+ windowSizeAndSlideSize.f0,
+ windowSizeAndSlideSize.f1,
getConfig(planner.getExecEnv, planner.getTableConfig))
if (isPythonWorkerUsingManagedMemory(planner.getTableConfig.getConfiguration)) {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecSortWindowAggregateBase.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecSortWindowAggregateBase.scala
index 5ad4827..2395de8 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecSortWindowAggregateBase.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecSortWindowAggregateBase.scala
@@ -103,13 +103,13 @@ abstract class BatchExecSortWindowAggregateBase(
val groupBufferLimitSize = planner.getTableConfig.getConfiguration.getInteger(
ExecutionConfigOptions.TABLE_EXEC_WINDOW_AGG_BUFFER_SIZE_LIMIT)
- val (windowSize: Long, slideSize: Long) = WindowCodeGenerator.getWindowDef(window)
+ val windowSizeAndSlideSize = WindowCodeGenerator.getWindowDef(window)
val generator = new SortWindowCodeGenerator(
ctx, planner.getRelBuilder, window, inputTimeFieldIndex,
inputTimeIsDate, namedWindowProperties,
- aggInfos, inputRowType, inputType, outputType,
- groupBufferLimitSize, 0L, windowSize, slideSize,
+ aggInfos, inputType, inputType, outputType,
+ groupBufferLimitSize, 0L, windowSizeAndSlideSize.f0, windowSizeAndSlideSize.f1,
grouping, auxGrouping, enableAssignPane, isMerge, isFinal)
val generatedOperator = if (grouping.isEmpty) {
generator.genWithoutKeys()
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecHashWindowAggregate.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashWindowAggregate.scala
similarity index 71%
rename from flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecHashWindowAggregate.scala
rename to flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashWindowAggregate.scala
index f3ce9fe..6ec67fa 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecHashWindowAggregate.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashWindowAggregate.scala
@@ -20,8 +20,10 @@ package org.apache.flink.table.planner.plan.nodes.physical.batch
import org.apache.flink.table.functions.UserDefinedFunction
import org.apache.flink.table.planner.calcite.FlinkRelBuilder.PlannerNamedWindowProperty
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.logical.LogicalWindow
-import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge
+import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
+import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecHashWindowAggregate
import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
import org.apache.calcite.rel.RelNode
@@ -30,9 +32,10 @@ import org.apache.calcite.rel.core.AggregateCall
import java.util
-import scala.collection.JavaConversions._
-
-class BatchExecHashWindowAggregate(
+/**
+ * Batch physical RelNode for (global) hash-based window aggregate.
+ */
+class BatchPhysicalHashWindowAggregate(
cluster: RelOptCluster,
traitSet: RelTraitSet,
inputRel: RelNode,
@@ -47,25 +50,22 @@ class BatchExecHashWindowAggregate(
namedWindowProperties: Seq[PlannerNamedWindowProperty],
enableAssignPane: Boolean = false,
isMerge: Boolean)
- extends BatchExecHashWindowAggregateBase(
+ extends BatchPhysicalHashWindowAggregateBase(
cluster,
traitSet,
inputRel,
outputRowType,
- aggInputRowType,
grouping,
auxGrouping,
aggCallToAggFunction,
window,
- inputTimeFieldIndex,
- inputTimeIsDate,
namedWindowProperties,
enableAssignPane,
isMerge,
isFinal = true) {
override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
- new BatchExecHashWindowAggregate(
+ new BatchPhysicalHashWindowAggregate(
cluster,
traitSet,
inputs.get(0),
@@ -82,10 +82,22 @@ class BatchExecHashWindowAggregate(
isMerge)
}
- //~ ExecNode methods -----------------------------------------------------------
-
- override def getInputEdges: util.List[ExecEdge] = List(
- ExecEdge.builder()
- .damBehavior(ExecEdge.DamBehavior.END_INPUT)
- .build())
+ override def translateToExecNode(): ExecNode[_] = {
+ new BatchExecHashWindowAggregate(
+ grouping,
+ auxGrouping,
+ getAggCallList.toArray,
+ window,
+ inputTimeFieldIndex,
+ inputTimeIsDate,
+ namedWindowProperties.toArray,
+ FlinkTypeFactory.toLogicalRowType(aggInputRowType),
+ enableAssignPane,
+ isMerge,
+ true, // isFinal is always true
+ ExecEdge.builder().damBehavior(ExecEdge.DamBehavior.END_INPUT).build(),
+ FlinkTypeFactory.toLogicalRowType(getRowType),
+ getRelDetailedDescription
+ )
+ }
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecHashWindowAggregateBase.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashWindowAggregateBase.scala
similarity index 54%
rename from flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecHashWindowAggregateBase.scala
rename to flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashWindowAggregateBase.scala
index 92bfeca..06f6fea 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecHashWindowAggregateBase.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashWindowAggregateBase.scala
@@ -18,44 +18,31 @@
package org.apache.flink.table.planner.plan.nodes.physical.batch
-import org.apache.flink.api.dag.Transformation
-import org.apache.flink.configuration.MemorySize
-import org.apache.flink.table.api.config.ExecutionConfigOptions
-import org.apache.flink.table.data.RowData
import org.apache.flink.table.functions.UserDefinedFunction
import org.apache.flink.table.planner.calcite.FlinkRelBuilder.PlannerNamedWindowProperty
-import org.apache.flink.table.planner.calcite.FlinkTypeFactory
-import org.apache.flink.table.planner.codegen.CodeGeneratorContext
-import org.apache.flink.table.planner.codegen.agg.batch.{HashWindowCodeGenerator, WindowCodeGenerator}
-import org.apache.flink.table.planner.delegation.BatchPlanner
import org.apache.flink.table.planner.plan.cost.{FlinkCost, FlinkCostFactory}
import org.apache.flink.table.planner.plan.logical.LogicalWindow
-import org.apache.flink.table.planner.plan.nodes.exec.LegacyBatchExecNode
-import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil
-import org.apache.flink.table.planner.plan.utils.AggregateUtil.transformToBatchAggregateInfoList
import org.apache.flink.table.planner.plan.utils.FlinkRelMdUtil
-import org.apache.flink.table.runtime.operators.CodeGenOperatorFactory
-import org.apache.flink.table.runtime.typeutils.InternalTypeInfo
+import org.apache.flink.table.runtime.util.collections.binary.BytesMap
+
import org.apache.calcite.plan.{RelOptCluster, RelOptCost, RelOptPlanner, RelTraitSet}
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.AggregateCall
import org.apache.calcite.rel.metadata.RelMetadataQuery
-import org.apache.calcite.tools.RelBuilder
-import org.apache.flink.table.runtime.util.collections.binary.BytesMap
-abstract class BatchExecHashWindowAggregateBase(
+/**
+ * Batch physical RelNode for hash-based window aggregate.
+ */
+abstract class BatchPhysicalHashWindowAggregateBase(
cluster: RelOptCluster,
traitSet: RelTraitSet,
inputRel: RelNode,
outputRowType: RelDataType,
- aggInputRowType: RelDataType,
grouping: Array[Int],
auxGrouping: Array[Int],
aggCallToAggFunction: Seq[(AggregateCall, UserDefinedFunction)],
window: LogicalWindow,
- inputTimeFieldIndex: Int,
- inputTimeIsDate: Boolean,
namedWindowProperties: Seq[PlannerNamedWindowProperty],
enableAssignPane: Boolean = false,
isMerge: Boolean,
@@ -72,8 +59,7 @@ abstract class BatchExecHashWindowAggregateBase(
namedWindowProperties,
enableAssignPane,
isMerge,
- isFinal)
- with LegacyBatchExecNode[RowData] {
+ isFinal) {
override def computeSelfCost(planner: RelOptPlanner, mq: RelMetadataQuery): RelOptCost = {
val numOfGroupKey = grouping.length
@@ -96,43 +82,4 @@ abstract class BatchExecHashWindowAggregateBase(
val costFactory = planner.getCostFactory.asInstanceOf[FlinkCostFactory]
costFactory.makeCost(rowCnt, hashCpuCost + aggFunctionCpuCost, 0, 0, memCost)
}
-
- //~ ExecNode methods -----------------------------------------------------------
-
- override protected def translateToPlanInternal(
- planner: BatchPlanner): Transformation[RowData] = {
- val config = planner.getTableConfig
- val input = getInputNodes.get(0).translateToPlan(planner)
- .asInstanceOf[Transformation[RowData]]
- val ctx = CodeGeneratorContext(config)
- val outputType = FlinkTypeFactory.toLogicalRowType(getRowType)
- val inputRowType = getInput.getRowType
- val inputType = FlinkTypeFactory.toLogicalRowType(inputRowType)
-
- val aggInfos = transformToBatchAggregateInfoList(
- FlinkTypeFactory.toLogicalRowType(aggInputRowType), aggCallToAggFunction.map(_._1))
-
- val groupBufferLimitSize = config.getConfiguration.getInteger(
- ExecutionConfigOptions.TABLE_EXEC_WINDOW_AGG_BUFFER_SIZE_LIMIT)
-
- val (windowSize: Long, slideSize: Long) = WindowCodeGenerator.getWindowDef(window)
-
- val generatedOperator = new HashWindowCodeGenerator(
- ctx, planner.getRelBuilder, window, inputTimeFieldIndex,
- inputTimeIsDate, namedWindowProperties,
- aggInfos, inputRowType, grouping, auxGrouping, enableAssignPane, isMerge, isFinal).gen(
- inputType, outputType, groupBufferLimitSize, 0,
- windowSize, slideSize)
- val operator = new CodeGenOperatorFactory[RowData](generatedOperator)
-
- val managedMemory = MemorySize.parse(config.getConfiguration.getString(
- ExecutionConfigOptions.TABLE_EXEC_RESOURCE_HASH_AGG_MEMORY)).getBytes
- ExecNodeUtil.createOneInputTransformation(
- input,
- getRelDetailedDescription,
- operator,
- InternalTypeInfo.of(outputType),
- input.getParallelism,
- managedMemory)
- }
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecLocalHashWindowAggregate.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalLocalHashWindowAggregate.scala
similarity index 70%
rename from flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecLocalHashWindowAggregate.scala
rename to flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalLocalHashWindowAggregate.scala
index 2e19199..937f941 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecLocalHashWindowAggregate.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalLocalHashWindowAggregate.scala
@@ -20,8 +20,10 @@ package org.apache.flink.table.planner.plan.nodes.physical.batch
import org.apache.flink.table.functions.UserDefinedFunction
import org.apache.flink.table.planner.calcite.FlinkRelBuilder.PlannerNamedWindowProperty
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.logical.LogicalWindow
-import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge
+import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecHashWindowAggregate
+import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
import org.apache.calcite.rel.RelNode
@@ -30,9 +32,10 @@ import org.apache.calcite.rel.core.AggregateCall
import java.util
-import scala.collection.JavaConversions._
-
-class BatchExecLocalHashWindowAggregate(
+/**
+ * Batch physical RelNode for local hash-based window aggregate.
+ */
+class BatchPhysicalLocalHashWindowAggregate(
cluster: RelOptCluster,
traitSet: RelTraitSet,
inputRel: RelNode,
@@ -46,25 +49,22 @@ class BatchExecLocalHashWindowAggregate(
inputTimeIsDate: Boolean,
namedWindowProperties: Seq[PlannerNamedWindowProperty],
enableAssignPane: Boolean = false)
- extends BatchExecHashWindowAggregateBase(
+ extends BatchPhysicalHashWindowAggregateBase(
cluster,
traitSet,
inputRel,
outputRowType,
- inputRowType,
grouping,
auxGrouping,
aggCallToAggFunction,
window,
- inputTimeFieldIndex,
- inputTimeIsDate,
namedWindowProperties,
enableAssignPane,
isMerge = false,
isFinal = false) {
override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
- new BatchExecLocalHashWindowAggregate(
+ new BatchPhysicalLocalHashWindowAggregate(
cluster,
traitSet,
inputs.get(0),
@@ -80,7 +80,22 @@ class BatchExecLocalHashWindowAggregate(
enableAssignPane)
}
- //~ ExecNode methods -----------------------------------------------------------
-
- override def getInputEdges: util.List[ExecEdge] = List(ExecEdge.DEFAULT)
+ override def translateToExecNode(): ExecNode[_] = {
+ new BatchExecHashWindowAggregate(
+ grouping,
+ auxGrouping,
+ getAggCallList.toArray,
+ window,
+ inputTimeFieldIndex,
+ inputTimeIsDate,
+ namedWindowProperties.toArray,
+ FlinkTypeFactory.toLogicalRowType(inputRowType),
+ enableAssignPane,
+ false, // isMerge is always false
+ false, // isFinal is always false
+ ExecEdge.DEFAULT,
+ FlinkTypeFactory.toLogicalRowType(getRowType),
+ getRelDetailedDescription
+ )
+ }
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
index 1f5809e..bd25c16 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
@@ -423,7 +423,7 @@ object FlinkBatchRuleSets {
// over agg
BatchExecOverAggregateRule.INSTANCE,
// window agg
- BatchExecWindowAggregateRule.INSTANCE,
+ BatchPhysicalWindowAggregateRule.INSTANCE,
BatchExecPythonWindowAggregateRule.INSTANCE,
// join
BatchExecHashJoinRule.INSTANCE,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortLimitRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortLimitRule.scala
index ef7c0ba..9296afd 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortLimitRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortLimitRule.scala
@@ -61,17 +61,17 @@ class BatchPhysicalSortLimitRule
override def convert(rel: RelNode): RelNode = {
val sort = rel.asInstanceOf[FlinkLogicalSort]
- // create local BatchExecSortLimit
+ // create local BatchPhysicalSortLimit
val localRequiredTrait = sort.getInput.getTraitSet.replace(FlinkConventions.BATCH_PHYSICAL)
val localInput = RelOptRule.convert(sort.getInput, localRequiredTrait)
- // if fetch is null, there is no need to create local BatchExecSortLimit
+ // if fetch is null, there is no need to create local BatchPhysicalSortLimit
val inputOfExchange = if (sort.fetch != null) {
val limit = SortUtil.getLimitEnd(sort.offset, sort.fetch)
val rexBuilder = sort.getCluster.getRexBuilder
val intType = rexBuilder.getTypeFactory.createSqlType(SqlTypeName.INTEGER)
val providedLocalTraitSet = localRequiredTrait.replace(sort.getCollation)
- // for local BatchExecSortLimit, offset is always 0, and fetch is `limit`
+ // for local BatchPhysicalSortLimit, offset is always 0, and fetch is `limit`
new BatchPhysicalSortLimit(
rel.getCluster,
providedLocalTraitSet,
@@ -90,7 +90,7 @@ class BatchPhysicalSortLimitRule
.replace(FlinkRelDistribution.SINGLETON)
val newInput = RelOptRule.convert(inputOfExchange, requiredTrait)
- // create global BatchExecSortLimit
+ // create global BatchPhysicalSortLimit
val providedGlobalTraitSet = requiredTrait.replace(sort.getCollation)
new BatchPhysicalSortLimit(
rel.getCluster,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecWindowAggregateRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalWindowAggregateRule.scala
similarity index 96%
rename from flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecWindowAggregateRule.scala
rename to flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalWindowAggregateRule.scala
index bbd96ac..ff79e51 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecWindowAggregateRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalWindowAggregateRule.scala
@@ -27,7 +27,7 @@ import org.apache.flink.table.planner.plan.`trait`.FlinkRelDistribution
import org.apache.flink.table.planner.plan.logical.{LogicalWindow, SlidingGroupWindow, TumblingGroupWindow}
import org.apache.flink.table.planner.plan.nodes.FlinkConventions
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalWindowAggregate
-import org.apache.flink.table.planner.plan.nodes.physical.batch.{BatchExecHashWindowAggregate, BatchExecLocalHashWindowAggregate, BatchExecLocalSortWindowAggregate, BatchExecSortWindowAggregate}
+import org.apache.flink.table.planner.plan.nodes.physical.batch.{BatchExecLocalSortWindowAggregate, BatchExecSortWindowAggregate, BatchPhysicalHashWindowAggregate, BatchPhysicalLocalHashWindowAggregate}
import org.apache.flink.table.planner.plan.utils.AggregateUtil
import org.apache.flink.table.planner.plan.utils.AggregateUtil.hasTimeIntervalType
import org.apache.flink.table.planner.plan.utils.PythonUtil.isPythonAggregate
@@ -48,15 +48,15 @@ import scala.collection.JavaConversions._
/**
* Rule to convert a [[FlinkLogicalWindowAggregate]] into a
* {{{
- * BatchExecHash(or Sort)WindowAggregate (global)
+ * BatchPhysicalHash(or Sort)WindowAggregate (global)
* +- BatchPhysicalExchange (hash by group keys if group keys is not empty, else singleton)
- * +- BatchExecLocalHash(or Sort)WindowAggregate (local)
+ * +- BatchPhysicalLocalHash(or Sort)WindowAggregate (local)
* +- input of window agg
* }}}
* when all aggregate functions are mergeable
* and [[OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY]] is TWO_PHASE, or
* {{{
- * BatchExecHash(or Sort)WindowAggregate
+ * BatchPhysicalHash(or Sort)WindowAggregate
* +- BatchPhysicalExchange (hash by group keys if group keys is not empty, else singleton)
* +- input of window agg
* }}}
@@ -67,12 +67,12 @@ import scala.collection.JavaConversions._
* this rule will try to create two possibilities above, and chooses the best one based on cost.
* if all aggregate function buffer are fix length, the rule will choose hash window agg.
*/
-class BatchExecWindowAggregateRule
+class BatchPhysicalWindowAggregateRule
extends RelOptRule(
operand(classOf[FlinkLogicalWindowAggregate],
operand(classOf[RelNode], any)),
FlinkRelFactories.LOGICAL_BUILDER_WITHOUT_AGG_INPUT_PRUNE,
- "BatchExecWindowAggregateRule")
+ "BatchPhysicalWindowAggregateRule")
with BatchPhysicalAggRuleBase {
override def matches(call: RelOptRuleCall): Boolean = {
@@ -163,7 +163,7 @@ class BatchExecWindowAggregateRule
input.getRowType, call.builder(), window.timeAttribute)
val inputTimeFieldType = agg.getInput.getRowType.getFieldList.get(inputTimeFieldIndex).getType
val inputTimeIsDate = inputTimeFieldType.getSqlTypeName == SqlTypeName.DATE
- // local-agg output order: groupset | assignTs | aucGroupSet | aggCalls
+ // local-agg output order: groupSet | assignTs | auxGroupSet | aggCalls
val newInputTimeFieldIndexFromLocal = groupSet.length
val config = input.getCluster.getPlanner.getContext.unwrap(classOf[FlinkContext]).getTableConfig
@@ -180,7 +180,7 @@ class BatchExecWindowAggregateRule
val newLocalInput = RelOptRule.convert(input, localRequiredTraitSet)
val localProvidedTraitSet = localRequiredTraitSet
- new BatchExecLocalHashWindowAggregate(
+ new BatchPhysicalLocalHashWindowAggregate(
agg.getCluster,
localProvidedTraitSet,
newLocalInput,
@@ -234,7 +234,7 @@ class BatchExecWindowAggregateRule
// hash
val newGlobalAggInput = RelOptRule.convert(localAgg, globalRequiredTraitSet)
- new BatchExecHashWindowAggregate(
+ new BatchPhysicalHashWindowAggregate(
agg.getCluster,
aggProvidedTraitSet,
newGlobalAggInput,
@@ -293,7 +293,7 @@ class BatchExecWindowAggregateRule
// case 2: Sliding window without pane optimization
val newInput = RelOptRule.convert(input, requiredTraitSet)
- new BatchExecHashWindowAggregate(
+ new BatchPhysicalHashWindowAggregate(
agg.getCluster,
aggProvidedTraitSet,
newInput,
@@ -430,6 +430,6 @@ class BatchExecWindowAggregateRule
}
}
-object BatchExecWindowAggregateRule {
- val INSTANCE: RelOptRule = new BatchExecWindowAggregateRule
+object BatchPhysicalWindowAggregateRule {
+ val INSTANCE: RelOptRule = new BatchPhysicalWindowAggregateRule
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRule.scala
index 4484f2e..0a711f4 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRule.scala
@@ -49,7 +49,7 @@ class RemoveRedundantLocalHashAggRule extends RelOptRule(
localAgg.grouping,
localAgg.auxGrouping,
// Use the localAgg agg calls because the global agg call filters was removed,
- // see BatchExecHashAggRule for details.
+ // see BatchPhysicalHashAggRule for details.
localAgg.getAggCallToAggFunction,
isMerge = false)
call.transformTo(newGlobalAgg)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalSortAggRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalSortAggRule.scala
index 32920f9..833cc08 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalSortAggRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalSortAggRule.scala
@@ -48,7 +48,7 @@ abstract class RemoveRedundantLocalSortAggRule(
localAgg.grouping,
localAgg.auxGrouping,
// Use the localAgg agg calls because the global agg call filters was removed,
- // see BatchExecSortAggRule for details.
+ // see BatchPhysicalSortAggRule for details.
localAgg.getAggCallToAggFunction,
isMerge = false)
call.transformTo(newGlobalAgg)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelMdUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelMdUtil.scala
index 2afbf0c..81b436e 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelMdUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelMdUtil.scala
@@ -23,7 +23,7 @@ import org.apache.flink.table.planner.JDouble
import org.apache.flink.table.planner.calcite.FlinkRelBuilder.PlannerNamedWindowProperty
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.nodes.calcite.{Expand, Rank, WindowAggregate}
-import org.apache.flink.table.planner.plan.nodes.physical.batch.{BatchExecLocalHashWindowAggregate, BatchExecLocalSortWindowAggregate, BatchPhysicalGroupAggregateBase, BatchPhysicalWindowAggregateBase}
+import org.apache.flink.table.planner.plan.nodes.physical.batch.{BatchExecLocalSortWindowAggregate, BatchPhysicalGroupAggregateBase, BatchPhysicalLocalHashWindowAggregate, BatchPhysicalWindowAggregateBase}
import org.apache.flink.table.runtime.operators.rank.{ConstantRankRange, RankRange}
import org.apache.flink.table.runtime.operators.sort.BinaryIndexedSortable
import org.apache.flink.table.runtime.typeutils.BinaryRowDataSerializer.LENGTH_SIZE_IN_BYTES
@@ -334,7 +334,7 @@ object FlinkRelMdUtil {
// grouping + assignTs + auxGrouping
(agg.getAggCallList,
agg.grouping ++ Array(agg.inputTimeFieldIndex) ++ agg.auxGrouping)
- case agg: BatchExecLocalHashWindowAggregate =>
+ case agg: BatchPhysicalLocalHashWindowAggregate =>
// grouping + assignTs + auxGrouping
(agg.getAggCallList,
agg.grouping ++ Array(agg.inputTimeFieldIndex) ++ agg.auxGrouping)
@@ -388,7 +388,7 @@ object FlinkRelMdUtil {
}
/**
- * Split groupKeys on Aggregate/ BatchExecGroupAggregateBase/ BatchExecWindowAggregateBase
+ * Split groupKeys on Aggregate/ BatchPhysicalGroupAggregateBase/ BatchPhysicalWindowAggregateBase
* into keys on aggregate's groupKey and aggregate's aggregateCalls.
*
* @param agg the aggregate
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
index 5a5de0c..5015479 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
@@ -1244,7 +1244,7 @@ class FlinkRelMdHandlerTestBase {
Array("count$0")).toList // agg calls
val localWindowAggRowType = typeFactory.createStructType(
localWindowAggTypes, localWindowAggNames)
- val batchLocalWindowAgg = new BatchExecLocalHashWindowAggregate(
+ val batchLocalWindowAgg = new BatchPhysicalLocalHashWindowAggregate(
batchCalc.getCluster,
batchPhysicalTraits,
batchCalc,
@@ -1260,7 +1260,7 @@ class FlinkRelMdHandlerTestBase {
enableAssignPane = false)
val batchExchange2 = new BatchPhysicalExchange(
cluster, batchPhysicalTraits.replace(hash01), batchLocalWindowAgg, hash01)
- val batchWindowAggWithLocal = new BatchExecHashWindowAggregate(
+ val batchWindowAggWithLocal = new BatchPhysicalHashWindowAggregate(
cluster,
batchPhysicalTraits,
batchExchange2,
@@ -1277,7 +1277,7 @@ class FlinkRelMdHandlerTestBase {
isMerge = true
)
- val batchWindowAggWithoutLocal = new BatchExecHashWindowAggregate(
+ val batchWindowAggWithoutLocal = new BatchPhysicalHashWindowAggregate(
batchExchange1.getCluster,
batchPhysicalTraits,
batchExchange1,
@@ -1383,7 +1383,7 @@ class FlinkRelMdHandlerTestBase {
Array("count$0")).toList // agg calls
val localWindowAggRowType = typeFactory.createStructType(
localWindowAggTypes, localWindowAggNames)
- val batchLocalWindowAgg = new BatchExecLocalHashWindowAggregate(
+ val batchLocalWindowAgg = new BatchPhysicalLocalHashWindowAggregate(
batchCalc.getCluster,
batchPhysicalTraits,
batchCalc,
@@ -1399,7 +1399,7 @@ class FlinkRelMdHandlerTestBase {
enableAssignPane = false)
val batchExchange2 = new BatchPhysicalExchange(
cluster, batchPhysicalTraits.replace(hash1), batchLocalWindowAgg, hash1)
- val batchWindowAggWithLocal = new BatchExecHashWindowAggregate(
+ val batchWindowAggWithLocal = new BatchPhysicalHashWindowAggregate(
cluster,
batchPhysicalTraits,
batchExchange2,
@@ -1416,7 +1416,7 @@ class FlinkRelMdHandlerTestBase {
isMerge = true
)
- val batchWindowAggWithoutLocal = new BatchExecHashWindowAggregate(
+ val batchWindowAggWithoutLocal = new BatchPhysicalHashWindowAggregate(
batchExchange1.getCluster,
batchPhysicalTraits,
batchExchange1,
@@ -1527,7 +1527,7 @@ class FlinkRelMdHandlerTestBase {
Array("count$0")).toList // agg calls
val localWindowAggRowType = typeFactory.createStructType(
localWindowAggTypes, localWindowAggNames)
- val batchLocalWindowAggWithAuxGroup = new BatchExecLocalHashWindowAggregate(
+ val batchLocalWindowAggWithAuxGroup = new BatchPhysicalLocalHashWindowAggregate(
batchCalc.getCluster,
batchPhysicalTraits,
batchCalc,
@@ -1543,7 +1543,7 @@ class FlinkRelMdHandlerTestBase {
enableAssignPane = false)
val batchExchange2 = new BatchPhysicalExchange(
cluster, batchPhysicalTraits.replace(hash0), batchLocalWindowAggWithAuxGroup, hash0)
- val batchWindowAggWithLocalWithAuxGroup = new BatchExecHashWindowAggregate(
+ val batchWindowAggWithLocalWithAuxGroup = new BatchPhysicalHashWindowAggregate(
cluster,
batchPhysicalTraits,
batchExchange2,
@@ -1560,7 +1560,7 @@ class FlinkRelMdHandlerTestBase {
isMerge = true
)
- val batchWindowAggWithoutLocalWithAuxGroup = new BatchExecHashWindowAggregate(
+ val batchWindowAggWithoutLocalWithAuxGroup = new BatchPhysicalHashWindowAggregate(
batchExchange1.getCluster,
batchPhysicalTraits,
batchExchange1,