From 7c57b8c8e52380f6b08dcc152a32d0e018e39cb0 Mon Sep 17 00:00:00 2001 From: Jincheng Sun Date: Thu, 1 Dec 2016 17:04:44 +0800 Subject: [PATCH] [FLINK-4693][tableApi] Add session group-windows for batch tables --- .../table/expressions/fieldExpression.scala | 2 +- .../dataset/DataSetWindowAggregate.scala | 272 ++++++++++++++++++ .../api/table/plan/rules/FlinkRuleSets.scala | 1 + .../dataSet/DataSetWindowAggregateRule.scala | 74 +++++ .../runtime/aggregate/AggregateUtil.scala | 241 +++++++++++++++- ...nWindowAggregateCombineGroupFunction.scala | 128 +++++++++ ...onWindowAggregateReduceGroupFunction.scala | 159 ++++++++++ .../DataSetWindowAggregateMapFunction.scala | 75 +++++ .../org/apache/flink/api/table/table.scala | 6 - .../batch/table/AggregationsITCase.scala | 49 ++++ .../scala/batch/table/GroupWindowTest.scala | 70 +++++ .../scala/stream/table/GroupWindowTest.scala | 11 - 12 files changed, 1061 insertions(+), 27 deletions(-) create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetWindowAggregate.scala create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetWindowAggregateRule.scala create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetSessionWindowAggregateCombineGroupFunction.scala create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetSessionWindowAggregateReduceGroupFunction.scala create mode 100644 flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetWindowAggregateMapFunction.scala create mode 100644 flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/GroupWindowTest.scala diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/fieldExpression.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/fieldExpression.scala index c7817bf1bac34e..7eae284773f451 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/fieldExpression.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/fieldExpression.scala @@ -36,7 +36,7 @@ abstract class Attribute extends LeafExpression with NamedExpression { case class UnresolvedFieldReference(name: String) extends Attribute { - override def toString = "\"" + name + override def toString = s"'$name" override private[flink] def withName(newName: String): Attribute = UnresolvedFieldReference(newName) diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetWindowAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetWindowAggregate.scala new file mode 100644 index 00000000000000..9573ebaa2a6157 --- /dev/null +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetWindowAggregate.scala @@ -0,0 +1,272 @@ +/* + * 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.api.table.plan.nodes.dataset + +import org.apache.calcite.plan.{RelOptCluster, RelOptCost, RelOptPlanner, RelTraitSet} +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.rel.{RelNode, RelWriter, SingleRel} +import org.apache.flink.api.common.functions.RichGroupReduceFunction +import org.apache.flink.api.common.operators.Order +import org.apache.flink.api.common.typeinfo.TypeInformation +import org.apache.flink.api.java.DataSet +import org.apache.flink.api.java.operators.MapOperator +import org.apache.flink.api.table.FlinkRelBuilder.NamedWindowProperty +import org.apache.flink.api.table.plan.logical.{EventTimeSessionGroupWindow,LogicalWindow} +import org.apache.flink.api.table.plan.nodes.FlinkAggregate +import org.apache.flink.api.table.runtime.aggregate.AggregateUtil +import org.apache.flink.api.table.runtime.aggregate.AggregateUtil.CalcitePair +import org.apache.flink.api.table.typeutils.{RowTypeInfo, TypeConverter} +import org.apache.flink.api.table.{BatchTableEnvironment, FlinkTypeFactory, Row} + +import scala.collection.JavaConverters._ + +/** + * Flink RelNode which matches along with a LogicalWindowAggregate. + */ +class DataSetWindowAggregate( + window: LogicalWindow, + namedProperties: Seq[NamedWindowProperty], + cluster: RelOptCluster, + traitSet: RelTraitSet, + inputNode: RelNode, + namedAggregates: Seq[CalcitePair[AggregateCall, String]], + rowRelDataType: RelDataType, + inputType: RelDataType, + grouping: Array[Int]) + extends SingleRel(cluster, traitSet, inputNode) + with FlinkAggregate + with DataSetRel { + + override def deriveRowType() = rowRelDataType + + override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = { + new DataSetWindowAggregate( + window, + namedProperties, + cluster, + traitSet, + inputs.get(0), + namedAggregates, + getRowType, + inputType, + grouping) + } + + override def toString: String = { + s"Aggregate(${ + if (!grouping.isEmpty) { + s"groupBy: (${groupingToString(inputType, grouping)}), " + } else { + "" + } + }window: ($window), " + + s"select: (${ + aggregationToString( + inputType, + grouping, + getRowType, + namedAggregates, + namedProperties) + }))" + } + + override def explainTerms(pw: RelWriter): RelWriter = { + super.explainTerms(pw) + .itemIf("groupBy", groupingToString(inputType, grouping), !grouping.isEmpty) + .item("window", window) + .item( + "select", aggregationToString( + inputType, + grouping, + getRowType, + namedAggregates, + namedProperties)) + } + + override def computeSelfCost(planner: RelOptPlanner, metadata: RelMetadataQuery): RelOptCost = { + val child = this.getInput + val rowCnt = metadata.getRowCount(child) + val rowSize = this.estimateRowSize(child.getRowType) + val aggCnt = this.namedAggregates.size + planner.getCostFactory.makeCost(rowCnt, rowCnt * aggCnt, rowCnt * rowSize) + } + + override def translateToPlan( + tableEnv: BatchTableEnvironment, + expectedType: Option[TypeInformation[Any]]): DataSet[Any] = { + + val config = tableEnv.getConfig + val groupingKeys = grouping.indices.toArray + + // get the output types + val fieldTypes: Array[TypeInformation[_]] = + getRowType.getFieldList.asScala + .map(field => FlinkTypeFactory.toTypeInfo(field.getType)) + .toArray + + val rowTypeInfo = new RowTypeInfo(fieldTypes) + + val inputDS = getInput.asInstanceOf[DataSetRel].translateToPlan( + tableEnv, + // tell the input operator that this operator currently only supports Rows as input + Some(TypeConverter.DEFAULT_ROW_TYPE)) + + val aggString = aggregationToString( + inputType, + grouping, + getRowType, + namedAggregates, + namedProperties) + + val aggOpName = + if (grouping.length > 0) { + s"groupBy: (${groupingToString(inputType, grouping)}), " + + s"window: ($window), select: ($aggString)" + } else { + s"window: ($window), select: ($aggString)" + } + + //create mapFunction for initializing the aggregations + val mapFunction = AggregateUtil.createDataSetWindowPrepareMapFunction( + window, + namedAggregates, + grouping, + inputType) + + // create groupReduceFunction for calculating the aggregations + val groupReduceFunction = + AggregateUtil.createDataSetWindowAggregateReduceGroupFunction( + window, + namedProperties, + namedAggregates, + inputType, + rowRelDataType, + grouping) + + + val prepareOpName = s"prepare select: ($aggString)" + + val mappedInput = + inputDS + .map(mapFunction) + .name(prepareOpName) + + // check whether all aggregates support partial aggregate + val result: DataSet[Any] = { + window match { + case EventTimeSessionGroupWindow(_, _, _) => + createEventTimeSessionGroupWindowDataSet( + aggOpName, + groupReduceFunction, + mappedInput) + case _ => + throw new UnsupportedOperationException( + s" [ ${window.getClass.getCanonicalName.split("\\.").last} ] is currently not " + + s"supported on batch tables. ") + } + + } + // if the expected type is not a Row, inject a mapper to convert it to the expected type + expectedType match { + case Some(typeInfo) if typeInfo.getTypeClass != classOf[Row] => + val mapName = s"convert: (${getRowType.getFieldNames.asScala.toList.mkString(", ")})" + result.map( + getConversionMapper( + config = config, + nullableInput = false, + inputType = rowTypeInfo.asInstanceOf[TypeInformation[Any]], + expectedType = expectedType.get, + conversionOperatorName = "DataSetWindowAggregateConversion", + fieldNames = getRowType.getFieldNames.asScala + )) + .name(mapName) + case _ => result + } + } + private[this] def createEventTimeSessionGroupWindowDataSet( + aggOpName: String, + groupReduceFunction: RichGroupReduceFunction[Row, Row], + mappedInput: MapOperator[Any, Row]): DataSet[Any] = { + val groupingKeys = grouping.indices.toArray + val rowTypeInfo = resultRowTypeInfo + + // gets the start and end position of the window in the intermediate result for + // combine and reduce. + val (windowStartPos, windowEndPos) = + AggregateUtil.computeWindowStartEndPropertyIntermediatePos( + namedAggregates, + inputType, + grouping) + + // the position of the rowtime field in the intermediate result for map output + val rowTimeFilePos = windowStartPos + // grouping window + if (groupingKeys.length > 0) { + // do incremental aggregation + if (AggregateUtil.doAllSupportPartialAggregation( + namedAggregates.map(_.getKey), + inputType, + grouping.length)) { + val combineGroupFunction = + AggregateUtil.createDataSetSessionWindowAggregateCombineFunction( + window, + namedAggregates, + inputType, + grouping) + + mappedInput.groupBy(groupingKeys: _*) + .sortGroup(rowTimeFilePos, Order.ASCENDING) + .combineGroup(combineGroupFunction) + .groupBy(groupingKeys: _*) + .sortGroup(windowStartPos, Order.ASCENDING) + .sortGroup(windowEndPos, Order.ASCENDING) + .reduceGroup(groupReduceFunction) + .returns(rowTypeInfo) + .name(aggOpName) + .asInstanceOf[DataSet[Any]] + } + // do incremental aggregation + else { + mappedInput.groupBy(groupingKeys: _*) + .sortGroup(rowTimeFilePos, Order.ASCENDING) + .reduceGroup(groupReduceFunction) + .returns(rowTypeInfo) + .name(aggOpName) + .asInstanceOf[DataSet[Any]] + } + } + // non-grouping window + else { + throw new UnsupportedOperationException( + "Session non-grouping window on event-time are currently not supported.") + } + } + + private[this] def resultRowTypeInfo: RowTypeInfo = { + // get the output types + val fieldTypes: Array[TypeInformation[_]] = + getRowType.getFieldList.asScala + .map(field => FlinkTypeFactory.toTypeInfo(field.getType)) + .toArray + new RowTypeInfo(fieldTypes) + } + +} diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala index 26c025eb1382a5..b07a9eeb5e2ea1 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala @@ -107,6 +107,7 @@ object FlinkRuleSets { DataSetMinusRule.INSTANCE, DataSetSortRule.INSTANCE, DataSetValuesRule.INSTANCE, + DataSetWindowAggregateRule.INSTANCE, BatchTableSourceScanRule.INSTANCE ) diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetWindowAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetWindowAggregateRule.scala new file mode 100644 index 00000000000000..bcd72c4e8f436c --- /dev/null +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetWindowAggregateRule.scala @@ -0,0 +1,74 @@ +/* + * 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.api.table.plan.rules.dataSet + +import org.apache.calcite.plan.{Convention, RelOptRule, RelOptRuleCall, RelTraitSet} +import org.apache.calcite.rel.RelNode +import org.apache.calcite.rel.convert.ConverterRule +import org.apache.flink.api.table.TableException +import org.apache.flink.api.table.plan.logical.rel.LogicalWindowAggregate +import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention,DataSetWindowAggregate} + +import scala.collection.JavaConversions._ + +class DataSetWindowAggregateRule + extends ConverterRule( + classOf[LogicalWindowAggregate], + Convention.NONE, + DataSetConvention.INSTANCE, + "DataSetWindowAggregateRule") { + + override def matches(call: RelOptRuleCall): Boolean = { + val agg: LogicalWindowAggregate = call.rel(0).asInstanceOf[LogicalWindowAggregate] + + // check if it have distinct aggregates + val distinctAggs = agg.getAggCallList.exists(_.isDistinct) + if (distinctAggs) { + throw TableException("DISTINCT aggregates are currently not supported.") + } + + // check if it have grouping sets + val groupSets = agg.getGroupSets.size() != 1 || agg.getGroupSets.get(0) != agg.getGroupSet + if (groupSets || agg.indicator) { + throw TableException("GROUPING SETS are currently not supported.") + } + + !distinctAggs && !groupSets && !agg.indicator + } + + override def convert(rel: RelNode): RelNode = { + val agg: LogicalWindowAggregate = rel.asInstanceOf[LogicalWindowAggregate] + val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE) + val convInput: RelNode = RelOptRule.convert(agg.getInput, DataSetConvention.INSTANCE) + + new DataSetWindowAggregate( + agg.getWindow, + agg.getNamedProperties, + rel.getCluster, + traitSet, + convInput, + agg.getNamedAggCalls, + rel.getRowType, + agg.getInput.getRowType, + agg.getGroupSet.toArray) + } +} + +object DataSetWindowAggregateRule { + val INSTANCE: RelOptRule = new DataSetWindowAggregateRule +} diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala index 4428963bb6a5e5..ac04a23d72352d 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateUtil.scala @@ -19,23 +19,26 @@ package org.apache.flink.api.table.runtime.aggregate import java.util -import org.apache.calcite.rel.`type`._ +import org.apache.calcite.rel.`type`.{RelDataType, _} import org.apache.calcite.rel.core.AggregateCall import org.apache.calcite.sql.{SqlAggFunction, SqlKind} import org.apache.calcite.sql.`type`.SqlTypeName._ import org.apache.calcite.sql.`type`.{SqlTypeFactoryImpl, SqlTypeName} import org.apache.calcite.sql.fun._ -import org.apache.flink.api.common.functions.{MapFunction, RichGroupReduceFunction} -import org.apache.flink.api.common.typeinfo.TypeInformation +import org.apache.flink.api.common.functions.{MapFunction, RichGroupCombineFunction, RichGroupReduceFunction} +import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation} import org.apache.flink.api.java.tuple.Tuple import org.apache.flink.api.table.FlinkRelBuilder.NamedWindowProperty -import org.apache.flink.api.table.expressions.{WindowEnd, WindowStart} -import org.apache.flink.api.table.plan.logical._ -import org.apache.flink.api.table.typeutils.RowTypeInfo +import org.apache.flink.api.table.expressions.{Expression, _} +import org.apache.flink.api.table.plan.logical.{EventTimeGroupWindow, _} +import org.apache.flink.api.table.typeutils.{RowTypeInfo, TimeIntervalTypeInfo} import org.apache.flink.api.table.typeutils.TypeCheckUtils._ import org.apache.flink.api.table.{FlinkTypeFactory, Row, TableException} import org.apache.flink.streaming.api.functions.windowing.{AllWindowFunction, WindowFunction} import org.apache.flink.streaming.api.windowing.windows.{Window => DataStreamWindow} +import java.sql.Timestamp + + import scala.collection.JavaConversions._ import scala.collection.mutable.ArrayBuffer @@ -84,6 +87,63 @@ object AggregateUtil { mapFunction } + /** + * Create a [[org.apache.flink.api.common.functions.MapFunction]] that prepares for aggregates. + * The function returns intermediate aggregate values of all aggregate function which are + * organized by the following format: + * + * {{{ + * avg(x) aggOffsetInRow = 2 count(z) aggOffsetInRow = 5 + * | | + * v v + * +---------+---------+--------+--------+--------+--------+--------+ + * |groupKey1|groupKey2| sum1 | count1 | sum2 | count2 |row-time| + * +---------+---------+--------+--------+--------+--------+--------+ + * ^ ^ + * | | + * sum(y) aggOffsetInRow = 4 row-time + * + * }}} + * + * @param window According to the window type to create the corresponding mapFunction + * + */ + private[flink] def createDataSetWindowPrepareMapFunction( + window: LogicalWindow, + namedAggregates: Seq[CalcitePair[AggregateCall, String]], + groupings: Array[Int], + inputType: RelDataType): MapFunction[Any, Row] = { + + val (aggFieldIndexes, aggregates) = transformToAggregateFunctions( + namedAggregates.map(_.getKey), + inputType, + groupings.length) + + window match { + case EventTimeSessionGroupWindow(_, timeField, gap) => + val mapReturnType: RowTypeInfo = + createAggregateBufferDataType( + groupings, + aggregates, + inputType, + Option(Array(BasicTypeInfo.LONG_TYPE_INFO))) + + val rowTimeFieldPos = getTimeFieldPos(timeField,inputType) + + new DataSetWindowAggregateMapFunction[Row, Row]( + aggregates, + aggFieldIndexes, + groupings, + rowTimeFieldPos, + mapReturnType.asInstanceOf[RowTypeInfo]).asInstanceOf[MapFunction[Any, Row]] + + case _ => + throw new UnsupportedOperationException( + s" [ ${window.getClass.getCanonicalName.split("\\.").last} ] is currently not " + + s"supported on batch tables. ") + } + } + /** * Create a [[org.apache.flink.api.common.functions.GroupReduceFunction]] to compute aggregates. * If all aggregates support partial aggregation, the @@ -134,6 +194,116 @@ object AggregateUtil { groupReduceFunction } + /** + * Create a [[org.apache.flink.api.common.functions.GroupCombineFunction]] that pre-aggregation + * for aggregates. + * The function returns intermediate aggregate values of all aggregate function which are + * organized by the following format: + * + * {{{ + * avg(x) aggOffsetInRow = 2 count(z) aggOffsetInRow = 5 + * | | windowEnd(max(row-time) + * | | | + * v v v + * +---------+---------+--------+--------+--------+--------+-----------+---------+ + * |groupKey1|groupKey2| sum1 | count1 | sum2 | count2 |windowStart|windowEnd| + * +---------+---------+--------+--------+--------+--------+-----------+---------+ + * ^ ^ + * | | + * sum(y) aggOffsetInRow = 4 windowStart(min(row-time)) + * + * }}} + * + */ + private[flink] def createDataSetSessionWindowAggregateCombineFunction( + window: LogicalWindow, + namedAggregates: Seq[CalcitePair[AggregateCall, String]], + inputType: RelDataType, + groupings: Array[Int]): RichGroupCombineFunction[Row,Row] = { + + val aggregates = transformToAggregateFunctions( + namedAggregates.map(_.getKey), + inputType, + groupings.length)._2 + //The two arity used to store window start and properties + val intermediateRowArity = groupings.length + + aggregates.map(_.intermediateDataType.length).sum + 2 + + window match { + case EventTimeSessionGroupWindow(_, _, gap) => + val combineReturnType: RowTypeInfo = + createAggregateBufferDataType( + groupings, + aggregates, + inputType, + Option(Array(BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO))) + + new DataSetSessionWindowAggregateCombineGroupFunction( + aggregates, + groupings, + intermediateRowArity, + asLong(gap), + combineReturnType) + case _ => + throw new UnsupportedOperationException( + s" [ ${window.getClass.getCanonicalName.split("\\.").last} ] is currently not " + + s"supported on batch tables. ") + } + } + + /** + * @param window According to the window type to create the corresponding groupReduceFunction + */ + private[flink] def createDataSetWindowAggregateReduceGroupFunction( + window: LogicalWindow, + properties: Seq[NamedWindowProperty], + namedAggregates: Seq[CalcitePair[AggregateCall, String]], + inputType: RelDataType, + outputType: RelDataType, + groupings: Array[Int]): RichGroupReduceFunction[Row, Row] = { + + val aggregates = transformToAggregateFunctions( + namedAggregates.map(_.getKey), + inputType, + groupings.length)._2 + + val (groupingOffsetMapping, aggOffsetMapping) = + getGroupingOffsetAndAggOffsetMapping( + namedAggregates, + inputType, + outputType, + groupings) + + //The arity used to store window start and properties + val WINDOW_START_END_ARITY = 2 + val intermediateRowArity = groupings.length + + aggregates.map(_.intermediateDataType.length).sum + WINDOW_START_END_ARITY + + val (startPos, endPos) = + if (isTimeWindow(window)) { + computeWindowStartEndPropertyPos(properties) + } else { + (None, None) + } + + window match { + case EventTimeSessionGroupWindow(_, _, gap) => + new DataSetSessionWindowAggregateReduceGroupFunction( + aggregates, + groupingOffsetMapping, + aggOffsetMapping, + intermediateRowArity, + outputType.getFieldCount, + startPos, + endPos, + asLong(gap)) + case _ => + throw new UnsupportedOperationException( + s" [ ${window.getClass.getCanonicalName.split("\\.").last} ] is currently not " + + s"supported on batch tables. ") + } + } + /** * Create a [[org.apache.flink.api.common.functions.ReduceFunction]] for incremental window * aggregation. @@ -305,6 +475,27 @@ object AggregateUtil { } } + private[flink] def getTimestamp(timeField: Any): Long = { + timeField match { + case b: Byte => b.toLong + case t: Character => t.toLong + case s: Short => s.toLong + case i: Int => i.toLong + case l: Long => l + case f: Float => f.toLong + case d: Double => d.toLong + case s: String => s.toLong + case t: Timestamp => t.getTime + case other@_ => + throw new RuntimeException(s"Window time field doesn't support $other type currently") + } + } + + private[flink] def asLong(expr: Expression): Long = expr match { + case Literal(value: Long, TimeIntervalTypeInfo.INTERVAL_MILLIS) => value + case _ => throw new IllegalArgumentException() + } + /** * Return true if all aggregates can be partially computed. False otherwise. */ @@ -318,6 +509,21 @@ object AggregateUtil { groupKeysCount)._2.forall(_.supportPartial) } + private def getTimeFieldPos( + timeField: Expression, + inputType: RelDataType): Int = { + timeField match { + case ResolvedFieldReference(name, resultType) => + val relDataType = inputType.getFieldList.filter(r => name.equals(r.getName)) + if (relDataType.length == 1) { + relDataType.head.getIndex + } else { + throw new IllegalArgumentException() + } + case _ => throw new IllegalArgumentException() + } + } + /** * @return groupingOffsetMapping (mapping relation between field index of intermediate * aggregate Row and output Row.) @@ -357,6 +563,19 @@ object AggregateUtil { } } + private[flink] def computeWindowStartEndPropertyIntermediatePos( + namedAggregates: Seq[CalcitePair[AggregateCall, String]], + inputType: RelDataType, + groupings: Array[Int]): (Int, Int) = { + val aggregates = transformToAggregateFunctions( + namedAggregates.map(_.getKey), + inputType, + groupings.length)._2 + val windowStartPos = groupings.length + + aggregates.map(_.intermediateDataType.length).sum + (windowStartPos, windowStartPos + 1) + } + private def computeWindowStartEndPropertyPos( properties: Seq[NamedWindowProperty]): (Option[Int], Option[Int]) = { @@ -514,7 +733,8 @@ object AggregateUtil { private def createAggregateBufferDataType( groupings: Array[Int], aggregates: Array[Aggregate[_]], - inputType: RelDataType): RowTypeInfo = { + inputType: RelDataType, + windowKeyTypes: Option[Array[TypeInformation[_]]] = None): RowTypeInfo = { // get the field data types of group keys. val groupingTypes: Seq[TypeInformation[_]] = groupings @@ -527,8 +747,11 @@ object AggregateUtil { // get all field data types of all intermediate aggregates val aggTypes: Seq[TypeInformation[_]] = aggregates.flatMap(_.intermediateDataType) - // concat group key types and aggregation types - val allFieldTypes = groupingTypes ++: aggTypes + // concat group key types, aggregation types, and window key types + val allFieldTypes:Seq[TypeInformation[_]] = windowKeyTypes match { + case None => groupingTypes ++: aggTypes + case _ => groupingTypes ++: aggTypes ++: windowKeyTypes.get + } val partialType = new RowTypeInfo(allFieldTypes) partialType } diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetSessionWindowAggregateCombineGroupFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetSessionWindowAggregateCombineGroupFunction.scala new file mode 100644 index 00000000000000..32387916c9f9ab --- /dev/null +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetSessionWindowAggregateCombineGroupFunction.scala @@ -0,0 +1,128 @@ +/* + * 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.api.table.runtime.aggregate + +import java.lang.Iterable + +import org.apache.flink.api.common.functions.RichGroupCombineFunction +import org.apache.flink.api.common.typeinfo.TypeInformation +import org.apache.flink.api.java.typeutils.ResultTypeQueryable +import org.apache.flink.api.table.Row +import org.apache.flink.configuration.Configuration +import org.apache.flink.util.{Collector, Preconditions} + +import scala.collection.JavaConversions._ + +/** + * This wraps the aggregate logic inside of + * [[org.apache.flink.api.java.operators.GroupCombineOperator]]. + */ +class DataSetSessionWindowAggregateCombineGroupFunction( + aggregates: Array[Aggregate[_ <: Any]], + groupingKeys: Array[Int], + intermediateRowArity: Int, + gap: Long, + @transient returnType: TypeInformation[Row]) + extends RichGroupCombineFunction[Row,Row] + with ResultTypeQueryable[Row] { + + private var aggregateBuffer: Row = _ + private var rowTimePos = 0 + + override def open(config: Configuration) { + Preconditions.checkNotNull(aggregates) + Preconditions.checkNotNull(groupingKeys) + aggregateBuffer = new Row(intermediateRowArity) + rowTimePos = intermediateRowArity - 2 + } + + /** + * For sub-grouped intermediate aggregate Rows, divide window based on the row-time + * (current'row-time - previous’row-time > gap), and then merge data (within a unified window) + * into an aggregate buffer. + * + * @param records Sub-grouped intermediate aggregate Rows . + * @return Combined intermediate aggregate Row. + * + */ + override def combine( + records: Iterable[Row], + out: Collector[Row]): Unit = { + + var head:Row = null + var lastRowTime: Option[Long] = None + var currentRowTime: Option[Long] = None + + records.foreach( + (record) => { + currentRowTime = Some(record.productElement(rowTimePos).asInstanceOf[Long]) + + // initial traversal or new window open. + // the session window end is equal to last row-time + gap . + if (lastRowTime.isEmpty || + (lastRowTime.isDefined && (currentRowTime.get > (lastRowTime.get + gap)))) { + + // calculate the current window and open a new window. + if (lastRowTime.isDefined) { + // emit the current window's merged data + doCollect(out, head, lastRowTime.get) + }else{ + // set group keys to aggregateBuffer. + for (i <- 0 until groupingKeys.length) { + aggregateBuffer.setField(i, record.productElement(i)) + } + } + + // initiate intermediate aggregate value. + aggregates.foreach(_.initiate(aggregateBuffer)) + head = record + } + + // merge intermediate aggregate value to the buffered value. + aggregates.foreach(_.merge(record, aggregateBuffer)) + + // the current row-time is the last row-time of the next calculation. + lastRowTime = currentRowTime + }) + // emit the merged data of the current window. + doCollect(out, head, lastRowTime.get) + } + + def doCollect( + out: Collector[Row], + head: Row, + lastRowTime: Long): Unit = { + + // the window's start attribute value is the min (row-time) of all rows in the window. + val windowStart = head.productElement(rowTimePos).asInstanceOf[Long] + + // the window's end property value is max (row-time) + gap for all rows in the window. + val windowEnd = lastRowTime + gap + + // intermediate Row WindowStartPos is row-time pos . + aggregateBuffer.setField(rowTimePos, windowStart) + // intermediate Row WindowEndPos is row-time pos + 1 . + aggregateBuffer.setField(rowTimePos + 1, windowEnd) + + out.collect(aggregateBuffer) + } + + override def getProducedType: TypeInformation[Row] = { + returnType + } +} diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetSessionWindowAggregateReduceGroupFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetSessionWindowAggregateReduceGroupFunction.scala new file mode 100644 index 00000000000000..816f9c0b9f85d0 --- /dev/null +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetSessionWindowAggregateReduceGroupFunction.scala @@ -0,0 +1,159 @@ +/* + * 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.api.table.runtime.aggregate + +import java.lang.Iterable + +import org.apache.flink.api.common.functions.RichGroupReduceFunction +import org.apache.flink.api.table.Row +import org.apache.flink.configuration.Configuration +import org.apache.flink.util.{Collector, Preconditions} + +import scala.collection.JavaConversions._ +import org.apache.flink.streaming.api.windowing.windows.TimeWindow +/** + * This wraps the aggregate logic inside of + * [[org.apache.flink.api.java.operators.GroupReduceOperator]]. + * + * @param aggregates The aggregate functions. + * @param groupKeysMapping The index mapping of group keys between intermediate aggregate Row + * and output Row. + * @param aggregateMapping The index mapping between aggregate function list and aggregated value + * index in output Row. + */ +class DataSetSessionWindowAggregateReduceGroupFunction( + aggregates: Array[Aggregate[_ <: Any]], + groupKeysMapping: Array[(Int, Int)], + aggregateMapping: Array[(Int, Int)], + intermediateRowArity: Int, + finalRowArity: Int, + finalRowWindowStartPos: Option[Int], + finalRowWindowEndPos: Option[Int], + gap:Long) + extends RichGroupReduceFunction[Row, Row] { + + private var aggregateBuffer: Row = _ + private var output: Row = _ + private var collector: TimeWindowPropertyCollector = _ + private var intermediateRowWindowStartPos = 0 + private var intermediateRowWindowEndPos = 0 + + override def open(config: Configuration) { + Preconditions.checkNotNull(aggregates) + Preconditions.checkNotNull(groupKeysMapping) + aggregateBuffer = new Row(intermediateRowArity) + intermediateRowWindowStartPos = intermediateRowArity - 2 + intermediateRowWindowEndPos = intermediateRowArity - 1 + output = new Row(finalRowArity) + if (finalRowWindowStartPos.isDefined || finalRowWindowEndPos.isDefined) { + collector = new TimeWindowPropertyCollector(finalRowWindowStartPos, finalRowWindowEndPos) + } + } + + /** + * For grouped intermediate aggregate Rows, divide window according to the window-start + * and window-end, merge data (within a unified window) into an aggregate buffer, calculate + * aggregated values output from aggregate buffer, and then set them into output + * Row based on the mapping relationship between intermediate aggregate data and output data. + * + * @param records Grouped intermediate aggregate Rows iterator. + * @param out The collector to hand results to. + * + */ + override def reduce(records: Iterable[Row], out: Collector[Row]): Unit = { + + var last: Row = null + var head: Row = null + var lastWindowEnd: Option[Long] = None + var currentWindowStart: Option[Long] = None + + records.foreach( + (record) => { + currentWindowStart = + Some(record.productElement(intermediateRowWindowStartPos).asInstanceOf[Long]) + // initial traversal or new window open + if (lastWindowEnd.isEmpty || + (lastWindowEnd.isDefined && currentWindowStart.get > lastWindowEnd.get)) { + + // calculate the current window and open a new window + if (lastWindowEnd.isDefined) { + + // evaluate and emit the current window's result. + doEvaluateAndCollect(out, last, head) + } + // initiate intermediate aggregate value. + aggregates.foreach(_.initiate(aggregateBuffer)) + head = record + } + + aggregates.foreach(_.merge(record, aggregateBuffer)) + last = record + lastWindowEnd = Some(getWindowEnd(last)) + }) + + doEvaluateAndCollect(out, last, head) + + } + + def doEvaluateAndCollect( + out: Collector[Row], + last: Row, + head: Row): Unit = { + // set group keys value to final output. + groupKeysMapping.foreach { + case (after, previous) => + output.setField(after, last.productElement(previous)) + } + + // evaluate final aggregate value and set to output. + aggregateMapping.foreach { + case (after, previous) => + output.setField(after, aggregates(previous).evaluate(aggregateBuffer)) + } + + // adds TimeWindow properties to output then emit output + if (finalRowWindowStartPos.isDefined || finalRowWindowEndPos.isDefined) { + val start = + head.productElement(intermediateRowWindowStartPos).asInstanceOf[Long] + val end = getWindowEnd(last) + + collector.wrappedCollector = out + collector.timeWindow = new TimeWindow(start, end) + + collector.collect(output) + } else { + out.collect(output) + } + } + + def getWindowEnd(record: Row): Long = { + + // when partial aggregate is not supported, the input data structure of reduce is + // |groupKey1|groupKey2|sum1|count1|sum2|count2|rowTime| + if (record.productArity == intermediateRowWindowEndPos) { + //session window end is row-time + gap + record.productElement(intermediateRowWindowStartPos).asInstanceOf[Long] + gap + } + // when partial aggregate is supported, the input data structure of reduce is + // |groupKey1|groupKey2|sum1|count1|sum2|count2|windowStart|windowEnd| + else { + record.productElement(intermediateRowWindowEndPos).asInstanceOf[Long] + } + } + +} diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetWindowAggregateMapFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetWindowAggregateMapFunction.scala new file mode 100644 index 00000000000000..b8852576254c85 --- /dev/null +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/DataSetWindowAggregateMapFunction.scala @@ -0,0 +1,75 @@ +/* + * 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.api.table.runtime.aggregate + +import org.apache.flink.api.common.functions.RichMapFunction +import org.apache.flink.api.common.typeinfo.TypeInformation +import org.apache.flink.api.java.typeutils.ResultTypeQueryable +import org.apache.flink.api.table.Row +import org.apache.flink.api.table.runtime.aggregate.AggregateUtil.getTimestamp +import org.apache.flink.configuration.Configuration +import org.apache.flink.util.Preconditions + +/** + * Pre-process the aggregations and add row-time to the intermediate row. + * @param aggregates The aggregate functions. + */ +class DataSetWindowAggregateMapFunction[IN, OUT]( + aggregates: Array[Aggregate[_]], + aggFields: Array[Int], + groupingKeys: Array[Int], + rowTimeFieldPos: Int, + @transient returnType: TypeInformation[OUT]) + extends RichMapFunction[IN, OUT] + with ResultTypeQueryable[OUT] { + + private var output: Row = _ + + override def open(config: Configuration) { + Preconditions.checkNotNull(aggregates) + Preconditions.checkNotNull(aggFields) + Preconditions.checkArgument(aggregates.size == aggFields.size) + // add one arity used to store row-time. + val intermediateRowArity = groupingKeys.length + + aggregates.map(_.intermediateDataType.length).sum + 1 + output = new Row(intermediateRowArity) + } + + override def map(value: IN): OUT = { + + val input = value.asInstanceOf[Row] + + for (i <- 0 until aggregates.length) { + val fieldValue = input.productElement(aggFields(i)) + aggregates(i).prepare(fieldValue, output) + } + + for (i <- 0 until groupingKeys.length) { + output.setField(i, input.productElement(groupingKeys(i))) + } + + val rowTime = getTimestamp(input.productElement(rowTimeFieldPos)) + output.setField(output.productArity-1, rowTime) + + output.asInstanceOf[OUT] + } + + override def getProducedType: TypeInformation[OUT] = { + returnType + } +} diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala index c45e8719e1f4a8..267ab465da31e4 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala @@ -649,9 +649,6 @@ class Table( * @return A windowed table. */ def window(groupWindow: GroupWindow): GroupWindowedTable = { - if (tableEnv.isInstanceOf[BatchTableEnvironment]) { - throw new ValidationException(s"Windows on batch tables are currently not supported.") - } new GroupWindowedTable(this, Seq(), groupWindow) } } @@ -722,9 +719,6 @@ class GroupedTable( * @return A windowed table. */ def window(groupWindow: GroupWindow): GroupWindowedTable = { - if (table.tableEnv.isInstanceOf[BatchTableEnvironment]) { - throw new ValidationException(s"Windows on batch tables are currently not supported.") - } new GroupWindowedTable(table, groupKey, groupWindow) } } diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/AggregationsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/AggregationsITCase.scala index 16c8ececdb9821..4659b16ea9eb8e 100644 --- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/AggregationsITCase.scala +++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/AggregationsITCase.scala @@ -32,6 +32,7 @@ import org.junit.runner.RunWith import org.junit.runners.Parameterized import scala.collection.JavaConverters._ +import scala.collection.mutable @RunWith(classOf[Parameterized]) class AggregationsITCase( @@ -39,6 +40,13 @@ class AggregationsITCase( configMode: TableConfigMode) extends TableProgramsTestBase(mode, configMode) { + val data = new mutable.MutableList[(Long, String)] + data.+=((1L, "Hi")) + data.+=((4L, "Hello")) + data.+=((2L, "Hello")) + data.+=((17L, "Hello world")) + data.+=((8L, "Hello world")) + @Test def testAggregationTypes(): Unit = { @@ -400,5 +408,46 @@ class AggregationsITCase( TestBaseUtils.compareResultAsText(results.asJava, expected) } + @Test + def testEventTimeSessionGroupWindow(): Unit = { + val env = ExecutionEnvironment.getExecutionEnvironment + val tEnv = TableEnvironment.getTableEnvironment(env, config) + + val table = env.fromCollection(data).toTable(tEnv, 'rowtime, 'string) + val windowedTable = table + .groupBy('string) + .window(Session withGap 7.milli on 'rowtime as 'w) + .select('string, 'string.count, 'w.start, 'w.end) + + val results = windowedTable.toDataSet[Row].collect() + + val expected = "Hello world,1,1970-01-01 00:00:00.008,1970-01-01 00:00:00.015\nHello world,1," + + "1970-01-01 00:00:00.017,1970-01-01 00:00:00.024\nHello,2,1970-01-01 00:00:00.002," + + "1970-01-01 00:00:00.011\nHi,1,1970-01-01 00:00:00.001,1970-01-01 00:00:00.008" + TestBaseUtils.compareResultAsText(results.asJava, expected) + } + + @Test(expected = classOf[UnsupportedOperationException]) + def testAlldEventTimeSessionGroupWindow(): Unit = { + val env = ExecutionEnvironment.getExecutionEnvironment + val tEnv = TableEnvironment.getTableEnvironment(env, config) + val table = env.fromCollection(data).toTable(tEnv, 'rowtime, 'string) + val windowedTable =table + .window(Session withGap 7.milli on 'rowtime as 'w) + .select('string.count).toDataSet[Row].collect() + } + + @Test(expected = classOf[UnsupportedOperationException]) + def testProcessingTimeSessionGroupWindow(): Unit = { + val env = ExecutionEnvironment.getExecutionEnvironment + val tEnv = TableEnvironment.getTableEnvironment(env, config) + val table = env.fromCollection(data).toTable(tEnv, 'rowtime, 'string) + val windowedTable =table + .groupBy('string) + .window(Session withGap 7.milli as 'w) + .select('string.count).toDataSet[Row].collect() + } + } + diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/GroupWindowTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/GroupWindowTest.scala new file mode 100644 index 00000000000000..9a2bbf4432e031 --- /dev/null +++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/batch/table/GroupWindowTest.scala @@ -0,0 +1,70 @@ +/* + * 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.api.scala.batch.table + +import org.apache.flink.api.scala._ +import org.apache.flink.api.scala.table._ +import org.apache.flink.api.scala.table.Session +import org.apache.flink.api.table.ValidationException +import org.apache.flink.api.table.plan.logical._ +import org.apache.flink.api.table.utils.TableTestBase +import org.apache.flink.api.table.utils.TableTestUtil._ +import org.junit.Test + +class GroupWindowTest extends TableTestBase { + + @Test + def testEventTimeSessionGroupWindowOverTime(): Unit = { + val util = batchTestUtil() + val table = util.addTable[(Long, Int, String)]('long, 'int, 'string) + + val windowedTable = table + .groupBy('string) + .window(Session withGap 7.milli on 'long) + .select('string, 'int.count) + + val expected = unaryNode( + "DataSetWindowAggregate", + batchTableNode(0), + term("groupBy", "string"), + term("window", EventTimeSessionGroupWindow(None, 'long, 7.milli)), + term("select", "string", "COUNT(int) AS TMP_0") + ) + + util.verifyTable(windowedTable, expected) + } + + @Test + def testAllEventTimeSessionGroupWindowOverTime(): Unit = { + val util = batchTestUtil() + val table = util.addTable[(Long, Int, String)]('long, 'int, 'string) + + val windowedTable = table + .window(Session withGap 7.milli on 'long) + .select('int.count) + + val expected = unaryNode( + "DataSetWindowAggregate", + batchTableNode(0), + term("window", EventTimeSessionGroupWindow(None, 'long, 7.milli)), + term("select", "COUNT(int) AS TMP_0") + ) + + util.verifyTable(windowedTable, expected) + } +} diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/GroupWindowTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/GroupWindowTest.scala index b59b151d0934d2..cab7ca5132ad15 100644 --- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/GroupWindowTest.scala +++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/GroupWindowTest.scala @@ -29,17 +29,6 @@ import org.junit.{Ignore, Test} class GroupWindowTest extends TableTestBase { - // batch windows are not supported yet - @Test(expected = classOf[ValidationException]) - def testInvalidBatchWindow(): Unit = { - val util = batchTestUtil() - val table = util.addTable[(Long, Int, String)]('long, 'int, 'string) - - table - .groupBy('string) - .window(Session withGap 100.milli as 'string) - } - @Test(expected = classOf[ValidationException]) def testInvalidWindowProperty(): Unit = { val util = streamTestUtil()