You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2021/07/28 02:04:41 UTC
[flink] branch master updated: [FLINK-19739][table-runtime] Fix
compile exception for window aggregate in batch jobs
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 941e8b8 [FLINK-19739][table-runtime] Fix compile exception for window aggregate in batch jobs
941e8b8 is described below
commit 941e8b81cac761fab748c83b361ae8a2b35ce1cd
Author: tsreaper <ts...@gmail.com>
AuthorDate: Wed Jul 28 10:04:12 2021 +0800
[FLINK-19739][table-runtime] Fix compile exception for window aggregate in batch jobs
This closes #16591
---
.../agg/batch/HashWindowCodeGenerator.scala | 11 ++++++-
.../codegen/agg/batch/WindowCodeGenerator.scala | 12 ++-----
.../runtime/batch/sql/agg/GroupWindowITCase.scala | 37 ++++++++++++++++++++++
3 files changed, 49 insertions(+), 11 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashWindowCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashWindowCodeGenerator.scala
index 864e71b..cee7229 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashWindowCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashWindowCodeGenerator.scala
@@ -23,8 +23,9 @@ import org.apache.flink.api.java.typeutils.ListTypeInfo
import org.apache.flink.runtime.operators.sort.QuickSort
import org.apache.flink.streaming.api.operators.OneInputStreamOperator
import org.apache.flink.table.api.Types
-import org.apache.flink.table.data.RowData
+import org.apache.flink.table.data.{GenericRowData, RowData}
import org.apache.flink.table.data.binary.BinaryRowData
+import org.apache.flink.table.data.utils.JoinedRowData
import org.apache.flink.table.planner.codegen.CodeGenUtils.{BINARY_ROW, newName}
import org.apache.flink.table.planner.codegen.OperatorCodeGenerator.generateCollect
import org.apache.flink.table.planner.codegen._
@@ -612,6 +613,14 @@ class HashWindowCodeGenerator(
""".stripMargin
}
+ private def getOutputRowClass: Class[_ <: RowData] = {
+ if (namedProperties.isEmpty && grouping.isEmpty && isFinal) {
+ classOf[GenericRowData]
+ } else {
+ classOf[JoinedRowData]
+ }
+ }
+
private def genOutputDirectly(
windowSize: Long,
inputTerm: String,
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala
index a712e71..f0b5b39 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala
@@ -22,12 +22,12 @@ 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
-import org.apache.flink.table.data.{GenericRowData, RowData}
+import org.apache.flink.table.data.GenericRowData
import org.apache.flink.table.expressions.ExpressionUtils.extractValue
import org.apache.flink.table.expressions.{Expression, ValueLiteralExpression}
import org.apache.flink.table.functions.AggregateFunction
import org.apache.flink.table.planner.JLong
-import org.apache.flink.table.planner.expressions.PlannerNamedWindowProperty;
+import org.apache.flink.table.planner.expressions.PlannerNamedWindowProperty
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.codegen.CodeGenUtils.{BINARY_ROW, TIMESTAMP_DATA, boxedTypeTermForType, newName}
import org.apache.flink.table.planner.codegen.GenerateUtils.generateFieldAccess
@@ -97,14 +97,6 @@ abstract class WindowCodeGenerator(
(groupKeyRowType.getChildren :+ timestampInternalType).toArray,
(groupKeyRowType.getFieldNames :+ "assignedTs$").toArray)
- def getOutputRowClass: Class[_ <: RowData] = {
- if (namedProperties.isEmpty && grouping.isEmpty) {
- classOf[GenericRowData]
- } else {
- classOf[JoinedRowData]
- }
- }
-
private[flink] def getWindowsGroupingElementInfo(
enablePreAccumulate: Boolean = true): RowType = {
if (enablePreAccumulate) {
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/GroupWindowITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/GroupWindowITCase.scala
index fa717d6..27016d0 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/GroupWindowITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/GroupWindowITCase.scala
@@ -22,14 +22,18 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo.{INT_TYPE_INFO, LONG_T
import org.apache.flink.api.common.typeinfo.LocalTimeTypeInfo.LOCAL_DATE_TIME
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.api.scala._
+import org.apache.flink.table.planner.factories.TestValuesTableFactory
import org.apache.flink.table.planner.runtime.utils.BatchTestBase
import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
import org.apache.flink.table.planner.runtime.utils.TestData._
import org.apache.flink.table.planner.utils.DateTimeTestUtil.localDateTime
import org.apache.flink.table.planner.utils.{CountAggFunction, IntAvgAggFunction, IntSumAggFunction}
+import org.apache.flink.types.Row
import org.junit.{Before, Test}
+import java.time.LocalDateTime
+
class GroupWindowITCase extends BatchTestBase {
@Before
@@ -641,4 +645,37 @@ class GroupWindowITCase extends BatchTestBase {
checkResult(sqlQuery, Seq())
}
+
+ @Test
+ def testLocalGlobalWindowAggregateWithoutGroupingAndNamedProperties(): Unit = {
+ val data: Seq[Row] = Seq(
+ row(1, LocalDateTime.of(2021, 7, 26, 0, 0, 0)),
+ row(2, LocalDateTime.of(2021, 7, 26, 0, 0, 3)),
+ row(3, LocalDateTime.of(2021, 7, 26, 0, 0, 6)),
+ row(4, LocalDateTime.of(2021, 7, 26, 0, 0, 4)),
+ row(5, LocalDateTime.of(2021, 7, 26, 0, 0, 5)),
+ row(6, LocalDateTime.of(2021, 7, 26, 0, 0, 8)),
+ row(7, LocalDateTime.of(2021, 7, 26, 0, 0, 10)))
+ val dataId = TestValuesTableFactory.registerData(data)
+ val ddl =
+ s"""
+ |CREATE TABLE MyTable (
+ | a INT,
+ | ts TIMESTAMP
+ |) WITH (
+ | 'connector' = 'values',
+ | 'data-id' = '$dataId',
+ | 'bounded' = 'true'
+ |)
+ |""".stripMargin
+
+ tEnv.executeSql(ddl)
+ checkResult(
+ """
+ |SELECT sum(a) FROM MyTable
+ |GROUP BY
+ |TUMBLE(ts, interval '5' seconds)
+ |""".stripMargin,
+ Seq(row(14), row(7), row(7)))
+ }
}