You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2020/02/04 08:33:14 UTC
[flink] branch release-1.10 updated:
[FLINK-15494][table-planner-blink] Fix incorrect time field index in
LogicalWindowAggregateRuleBase
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.10 by this push:
new 2f6d138 [FLINK-15494][table-planner-blink] Fix incorrect time field index in LogicalWindowAggregateRuleBase
2f6d138 is described below
commit 2f6d13818baf6e783eb2028fda131b89ea823cc2
Author: Benchao Li <li...@gmail.com>
AuthorDate: Tue Feb 4 13:29:52 2020 +0800
[FLINK-15494][table-planner-blink] Fix incorrect time field index in LogicalWindowAggregateRuleBase
This closes #10784
---
.../logical/BatchLogicalWindowAggregateRule.scala | 32 ++++++----------------
.../logical/LogicalWindowAggregateRuleBase.scala | 6 ++--
.../logical/StreamLogicalWindowAggregateRule.scala | 32 +++++++++-------------
.../batch/sql/agg/WindowAggregateITCase.scala | 23 ++++++++++++++++
.../runtime/stream/sql/WindowAggregateITCase.scala | 29 ++++++++++++++++++++
5 files changed, 77 insertions(+), 45 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/BatchLogicalWindowAggregateRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/BatchLogicalWindowAggregateRule.scala
index 35dcf31..50099b0 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/BatchLogicalWindowAggregateRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/BatchLogicalWindowAggregateRule.scala
@@ -27,7 +27,6 @@ import org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter.fromLog
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.logical.{LogicalAggregate, LogicalProject}
import org.apache.calcite.rex._
-import org.apache.calcite.sql.SqlKind
import _root_.java.math.{BigDecimal => JBigDecimal}
@@ -56,33 +55,20 @@ class BatchLogicalWindowAggregateRule
private[table] override def getTimeFieldReference(
operand: RexNode,
- windowExprIdx: Int,
+ timeAttributeIndex: Int,
rowType: RelDataType): FieldReferenceExpression = {
if (FlinkTypeFactory.isProctimeIndicatorType(operand.getType)) {
throw new ValidationException("Window can not be defined over "
+ "a proctime attribute column for batch mode")
}
- operand match {
- case c: RexCall if c.getKind == SqlKind.CAST =>
- getTimeFieldReference(c.getOperands.get(0), windowExprIdx, rowType)
- // match TUMBLE_ROWTIME and TUMBLE_PROCTIME
- case c: RexCall if c.getOperands.size() == 1 &&
- FlinkTypeFactory.isTimeIndicatorType(c.getType) =>
- new FieldReferenceExpression(
- rowType.getFieldList.get(windowExprIdx).getName,
- fromLogicalTypeToDataType(toLogicalType(c.getType)),
- 0, // only one input, should always be 0
- windowExprIdx)
- case ref: RexInputRef =>
- // resolve field name of window attribute
- val fieldName = rowType.getFieldList.get(ref.getIndex).getName
- val fieldType = rowType.getFieldList.get(ref.getIndex).getType
- new FieldReferenceExpression(
- fieldName,
- fromLogicalTypeToDataType(toLogicalType(fieldType)),
- 0, // only one input, should always be 0
- windowExprIdx)
- }
+
+ val fieldName = rowType.getFieldList.get(timeAttributeIndex).getName
+ val fieldType = rowType.getFieldList.get(timeAttributeIndex).getType
+ new FieldReferenceExpression(
+ fieldName,
+ fromLogicalTypeToDataType(toLogicalType(fieldType)),
+ 0,
+ timeAttributeIndex)
}
def getOperandAsLong(call: RexCall, idx: Int): Long =
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalWindowAggregateRuleBase.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalWindowAggregateRuleBase.scala
index f9260af..2a5d6a4 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalWindowAggregateRuleBase.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalWindowAggregateRuleBase.scala
@@ -40,10 +40,8 @@ import org.apache.calcite.rex._
import org.apache.calcite.sql.`type`.SqlTypeUtil
import org.apache.calcite.tools.RelBuilder
import org.apache.calcite.util.ImmutableBitSet
-import org.apache.calcite.util.{Pair => CPair}
import _root_.scala.collection.JavaConversions._
-import _root_.scala.collection.mutable.ArrayBuffer
/**
* Planner rule that transforms simple [[LogicalAggregate]] on a [[LogicalProject]]
@@ -75,7 +73,6 @@ abstract class LogicalWindowAggregateRuleBase(description: String)
val project: LogicalProject = rewriteProctimeWindows(call.rel(1), builder)
val (windowExpr, windowExprIdx) = getWindowExpressions(agg, project).head
- val window = translateWindow(windowExpr, windowExprIdx, project.getInput.getRowType)
val rexBuilder = agg.getCluster.getRexBuilder
@@ -88,6 +85,9 @@ abstract class LogicalWindowAggregateRuleBase(description: String)
.project(project.getChildExps.updated(windowExprIdx, inAggGroupExpression))
.build()
+ // translate window against newProject.
+ val window = translateWindow(windowExpr, windowExprIdx, newProject.getRowType)
+
// Currently, this rule removes the window from GROUP BY operation which may lead to changes
// of AggCall's type which brings fails on type checks.
// To solve the problem, we change the types to the inferred types in the Aggregate and then
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/StreamLogicalWindowAggregateRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/StreamLogicalWindowAggregateRule.scala
index 9334730..cd367d9 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/StreamLogicalWindowAggregateRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/StreamLogicalWindowAggregateRule.scala
@@ -47,7 +47,8 @@ class StreamLogicalWindowAggregateRule
val timeAttribute = windowExpression.operands.get(0)
if (!FlinkTypeFactory.isTimeIndicatorType(timeAttribute.getType)) {
- throw new TableException(s"Time attribute expected but ${timeAttribute.getType} encountered.")
+ throw new TableException(s"Window aggregate can only be defined over a " +
+ s"time attribute column, but ${timeAttribute.getType} encountered.")
}
timeAttribute
}
@@ -67,26 +68,19 @@ class StreamLogicalWindowAggregateRule
private[table] override def getTimeFieldReference(
operand: RexNode,
- windowExprIdx: Int,
+ timeAttributeIndex: Int,
rowType: RelDataType): FieldReferenceExpression = {
- operand match {
- // match TUMBLE_ROWTIME and TUMBLE_PROCTIME
- case c: RexCall if c.getOperands.size() == 1 &&
- FlinkTypeFactory.isTimeIndicatorType(c.getType) =>
- new FieldReferenceExpression(
- rowType.getFieldList.get(windowExprIdx).getName,
- fromLogicalTypeToDataType(toLogicalType(c.getType)),
- 0, // only one input, should always be 0
- windowExprIdx)
- case v: RexInputRef if FlinkTypeFactory.isTimeIndicatorType(v.getType) =>
- new FieldReferenceExpression(
- rowType.getFieldList.get(v.getIndex).getName,
- fromLogicalTypeToDataType(toLogicalType(v.getType)),
- 0, // only one input, should always be 0
- windowExprIdx)
- case _ =>
- throw new ValidationException("Window can only be defined over a time attribute column.")
+ if (!FlinkTypeFactory.isTimeIndicatorType(operand.getType)) {
+ throw new ValidationException("Window can only be defined over a time attribute column.")
}
+
+ val fieldName = rowType.getFieldList.get(timeAttributeIndex).getName
+ val fieldType = rowType.getFieldList.get(timeAttributeIndex).getType
+ new FieldReferenceExpression(
+ fieldName,
+ fromLogicalTypeToDataType(toLogicalType(fieldType)),
+ 0,
+ timeAttributeIndex)
}
def getOperandAsLong(call: RexCall, idx: Int): Long =
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/WindowAggregateITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/WindowAggregateITCase.scala
index 4cbd7ed..0bd467f 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/WindowAggregateITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/WindowAggregateITCase.scala
@@ -141,6 +141,29 @@ class WindowAggregateITCase extends BatchTestBase {
}
@Test
+ def testCascadingTumbleWindow(): Unit = {
+ checkResult(
+ s"""
+ |SELECT b, SUM(cnt)
+ |FROM (
+ | SELECT b, COUNT(1) AS cnt, TUMBLE_ROWTIME(ts, INTERVAL '30' SECOND) AS ts
+ | FROM Table3WithTimestamp
+ | GROUP BY a, b, TUMBLE(ts, INTERVAL '30' SECOND)
+ |)
+ |GROUP BY b, TUMBLE(ts, INTERVAL '30' SECOND)
+ |""".stripMargin,
+ Seq(
+ row(1, 1),
+ row(2, 2),
+ row(3, 3),
+ row(4, 4),
+ row(5, 5),
+ row(6, 6)
+ )
+ )
+ }
+
+ @Test
def testSlidingWindow(): Unit = {
// keyed; 2-phase; pre-accumulate with paned optimization;
checkResult(
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/WindowAggregateITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/WindowAggregateITCase.scala
index b11eb08..7eb00c8 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/WindowAggregateITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/WindowAggregateITCase.scala
@@ -101,6 +101,35 @@ class WindowAggregateITCase(mode: StateBackendMode)
}
@Test
+ def testCascadingTumbleWindow(): Unit = {
+ val stream = failingDataSource(data)
+ .assignTimestampsAndWatermarks(
+ new TimestampAndWatermarkWithOffset
+ [(Long, Int, Double, Float, BigDecimal, String, String)](10L))
+ val table = stream.toTable(tEnv,
+ 'rowtime.rowtime, 'int, 'double, 'float, 'bigdec, 'string, 'name)
+ tEnv.registerTable("T1", table)
+
+ val sql =
+ """
+ |SELECT SUM(cnt)
+ |FROM (
+ | SELECT COUNT(1) AS cnt, TUMBLE_ROWTIME(rowtime, INTERVAL '10' SECOND) AS ts
+ | FROM T1
+ | GROUP BY `int`, `string`, TUMBLE(rowtime, INTERVAL '10' SECOND)
+ |)
+ |GROUP BY TUMBLE(ts, INTERVAL '10' SECOND)
+ |""".stripMargin
+
+ val sink = new TestingAppendSink
+ tEnv.sqlQuery(sql).toAppendStream[Row].addSink(sink)
+ env.execute()
+
+ val expected = Seq("9")
+ assertEquals(expected.sorted, sink.getAppendResults.sorted)
+ }
+
+ @Test
def testEventTimeSessionWindow(): Unit = {
//To verify the "merge" functionality, we create this test with the following characteristics:
// 1. set the Parallelism to 1, and have the test data out of order