flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [1/4] flink git commit: [FLINK-6614] [table] Fix translation of group auxiliary functions (e.g., TUMBLE_END).
Date Thu, 18 May 2017 19:22:56 GMT
Repository: flink
Updated Branches:
  refs/heads/master c995ebd29 -> 9fc42df68


[FLINK-6614] [table] Fix translation of group auxiliary functions (e.g., TUMBLE_END).

This closes #3930.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9fc42df6
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9fc42df6
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9fc42df6

Branch: refs/heads/master
Commit: 9fc42df68d746c633b0d3c8995e0031064bfd362
Parents: 3a65e5a
Author: Fabian Hueske <fhueske@apache.org>
Authored: Wed May 17 16:26:27 2017 +0200
Committer: Fabian Hueske <fhueske@apache.org>
Committed: Thu May 18 21:22:12 2017 +0200

----------------------------------------------------------------------
 .../common/WindowStartEndPropertiesRule.scala   | 39 ++++++++++++++------
 .../scala/stream/sql/WindowAggregateTest.scala  | 27 ++++++++++++++
 2 files changed, 54 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/9fc42df6/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/WindowStartEndPropertiesRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/WindowStartEndPropertiesRule.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/WindowStartEndPropertiesRule.scala
index 7577deb..14e9b21 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/WindowStartEndPropertiesRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/common/WindowStartEndPropertiesRule.scala
@@ -36,13 +36,21 @@ class WindowStartEndPropertiesRule
   override def matches(call: RelOptRuleCall): Boolean = {
     val project = call.rel(0).asInstanceOf[LogicalProject]
     // project includes at least on group auxiliary function
-    project.getProjects.exists {
-      case c: RexCall => c.getOperator.isGroupAuxiliary
-      case _ => false
+
+    def hasGroupAuxiliaries(node: RexNode): Boolean = {
+      node match {
+        case c: RexCall if c.getOperator.isGroupAuxiliary => true
+        case c: RexCall =>
+          c.operands.exists(hasGroupAuxiliaries)
+        case _ => false
+      }
     }
+
+    project.getProjects.exists(hasGroupAuxiliaries)
   }
 
   override def onMatch(call: RelOptRuleCall): Unit = {
+
     val project = call.rel(0).asInstanceOf[LogicalProject]
     val innerProject = call.rel(1).asInstanceOf[LogicalProject]
     val agg = call.rel(2).asInstanceOf[LogicalWindowAggregate]
@@ -62,20 +70,27 @@ class WindowStartEndPropertiesRule
     transformed.project(
       innerProject.getProjects ++ Seq(transformed.field("w$start"), transformed.field("w$end")))
 
-    // replace window auxiliary function by access to window properties
-    transformed.project(
-      project.getProjects.map{ x =>
-        if (WindowStartEndPropertiesRule.isWindowStart(x)) {
+    def replaceGroupAuxiliaries(node: RexNode): RexNode = {
+      node match {
+        case c: RexCall if WindowStartEndPropertiesRule.isWindowStart(c) =>
           // replace expression by access to window start
-          rexBuilder.makeCast(x.getType, transformed.field("w$start"), false)
-        } else if (WindowStartEndPropertiesRule.isWindowEnd(x)) {
+          rexBuilder.makeCast(c.getType, transformed.field("w$start"), false)
+        case c: RexCall if WindowStartEndPropertiesRule.isWindowEnd(c) =>
           // replace expression by access to window end
-          rexBuilder.makeCast(x.getType, transformed.field("w$end"), false)
-        } else {
+          rexBuilder.makeCast(c.getType, transformed.field("w$end"), false)
+        case c: RexCall =>
+          // replace expressions in children
+          val newOps = c.getOperands.map(replaceGroupAuxiliaries)
+          c.clone(c.getType, newOps)
+        case x =>
           // preserve expression
           x
-        }
       }
+    }
+
+    // replace window auxiliary function by access to window properties
+    transformed.project(
+      project.getProjects.map(replaceGroupAuxiliaries)
     )
     val res = transformed.build()
     call.transformTo(res)

http://git-wip-us.apache.org/repos/asf/flink/blob/9fc42df6/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/WindowAggregateTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/WindowAggregateTest.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/WindowAggregateTest.scala
index 2022db8..f95d0ab 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/WindowAggregateTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/WindowAggregateTest.scala
@@ -150,6 +150,33 @@ class WindowAggregateTest extends TableTestBase {
     streamUtil.verifySql(sql, expected)
   }
 
+  @Test
+  def testExpressionOnWindowAuxFunction() = {
+    val sql =
+      "SELECT " +
+        "  COUNT(*), " +
+        "  TUMBLE_END(rowtime, INTERVAL '15' MINUTE) + INTERVAL '1' MINUTE " +
+        "FROM MyTable " +
+        "GROUP BY TUMBLE(rowtime, INTERVAL '15' MINUTE)"
+    val expected =
+      unaryNode(
+        "DataStreamCalc",
+        unaryNode(
+          "DataStreamGroupWindowAggregate",
+          unaryNode(
+            "DataStreamCalc",
+            streamTableNode(0),
+            term("select", "rowtime")
+          ),
+          term("window", TumblingGroupWindow('w$, 'rowtime, 900000.millis)),
+          term("select", "COUNT(*) AS EXPR$0", "start('w$) AS w$start", "end('w$) AS w$end")
+        ),
+        term("select", "EXPR$0", "DATETIME_PLUS(w$end, 60000) AS $f1")
+      )
+
+    streamUtil.verifySql(sql, expected)
+  }
+
   @Test(expected = classOf[TableException])
   def testTumbleWindowNoOffset(): Unit = {
     val sqlQuery =


Mime
View raw message