godfreyhe commented on a change in pull request #14605:
URL: https://github.com/apache/flink/pull/14605#discussion_r556399309
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalOverAggregateBase.scala
##########
@@ -18,88 +18,50 @@
package org.apache.flink.table.planner.plan.nodes.physical.batch
-import org.apache.flink.table.data.RowData
-import org.apache.flink.table.functions.UserDefinedFunction
+import org.apache.flink.table.api.TableException
import org.apache.flink.table.planner.CalcitePair
-import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.`trait`.{FlinkRelDistribution,
FlinkRelDistributionTraitDef}
import org.apache.flink.table.planner.plan.cost.{FlinkCost, FlinkCostFactory}
-import org.apache.flink.table.planner.plan.nodes.exec.{LegacyBatchExecNode,
ExecEdge}
-import
org.apache.flink.table.planner.plan.nodes.physical.batch.OverWindowMode.OverWindowMode
import
org.apache.flink.table.planner.plan.rules.physical.batch.BatchExecJoinRuleBase
+import
org.apache.flink.table.planner.plan.utils.OverAggregateUtil.splitOutOffsetOrInsensitiveGroup
import org.apache.flink.table.planner.plan.utils.{FlinkRelOptUtil,
OverAggregateUtil, RelExplainUtil}
-import com.google.common.collect.ImmutableList
import org.apache.calcite.plan._
import org.apache.calcite.rel.RelDistribution.Type._
import org.apache.calcite.rel._
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.Window.Group
import org.apache.calcite.rel.core.{AggregateCall, Window}
import org.apache.calcite.rel.metadata.RelMetadataQuery
-import org.apache.calcite.rex.RexLiteral
-import org.apache.calcite.sql.fun.SqlLeadLagAggFunction
-import org.apache.calcite.tools.RelBuilder
import org.apache.calcite.util.ImmutableIntList
import java.util
import scala.collection.JavaConversions._
-import scala.collection.mutable.ArrayBuffer
/**
- * Batch physical RelNode for sort-based over [[Window]] aggregate.
- */
-abstract class BatchExecOverAggregateBase(
+ * Base batch physical RelNode for sort-based over [[Window]] aggregate.
+ */
+abstract class BatchPhysicalOverAggregateBase(
cluster: RelOptCluster,
- relBuilder: RelBuilder,
traitSet: RelTraitSet,
inputRel: RelNode,
outputRowType: RelDataType,
inputRowType: RelDataType,
- grouping: Array[Int],
- orderKeyIndices: Array[Int],
- orders: Array[Boolean],
- nullIsLasts: Array[Boolean],
- windowGroupToAggCallToAggFunction: Seq[
- (Window.Group, Seq[(AggregateCall, UserDefinedFunction)])],
+ windowGroups: Seq[Window.Group],
logicWindow: Window)
extends SingleRel(cluster, traitSet, inputRel)
- with BatchPhysicalRel
- with LegacyBatchExecNode[RowData] {
-
- protected lazy val modeToGroupToAggCallToAggFunction:
- Seq[(OverWindowMode, Window.Group, Seq[(AggregateCall,
UserDefinedFunction)])] =
- splitOutOffsetOrInsensitiveGroup()
+ with BatchPhysicalRel {
- protected val constants: ImmutableList[RexLiteral] = logicWindow.constants
- protected val inputTypeWithConstants: RelDataType = {
- val constantTypes = constants.map(c =>
FlinkTypeFactory.toLogicalType(c.getType))
- val inputTypeNamesWithConstants =
- inputType.getFieldNames ++ constants.indices.map(i => "TMP" + i)
- val inputTypesWithConstants = inputType.getChildren ++ constantTypes
- cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
- .buildRelNodeRowType(inputTypeNamesWithConstants,
inputTypesWithConstants)
+ val partitionKeyIndices: Array[Int] = windowGroups.head.keys.toArray
+ windowGroups.tail.foreach { g =>
+ if (!util.Arrays.equals(partitionKeyIndices, g.keys.toArray)) {
+ throw new TableException("" +
+ "BatchPhysicalOverAggregateBase requires all groups should have same
partition key.")
+ }
}
- lazy val aggregateCalls: Seq[AggregateCall] =
- windowGroupToAggCallToAggFunction.flatMap(_._2).map(_._1)
-
- protected lazy val inputType =
FlinkTypeFactory.toLogicalRowType(inputRowType)
-
- protected def isUnboundedWindow(group: Window.Group) =
- group.lowerBound.isUnbounded && group.upperBound.isUnbounded
-
- protected def isUnboundedPrecedingWindow(group: Window.Group) =
- group.lowerBound.isUnbounded && !group.upperBound.isUnbounded
-
- protected def isUnboundedFollowingWindow(group: Window.Group) =
- !group.lowerBound.isUnbounded && group.upperBound.isUnbounded
-
- protected def isSlidingWindow(group: Window.Group) =
- !group.lowerBound.isUnbounded && !group.upperBound.isUnbounded
-
- def getGrouping: Array[Int] = grouping
+ protected lazy val newWindowGroups: Seq[Group] =
splitOutOffsetOrInsensitiveGroup(windowGroups)
Review comment:
how about `offsetAndInsensitiveSensitiveGroups`
----------------------------------------------------------------
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]
