You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2018/11/08 04:14:42 UTC
[flink] branch master updated: [FLINK-10676][table] Change the over
window preceding clause from required to optional.
This is an automated email from the ASF dual-hosted git repository.
jincheng 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 483507a [FLINK-10676][table] Change the over window preceding clause from required to optional.
483507a is described below
commit 483507a65c7547347eaafb21a24967c470f94ed6
Author: hequn8128 <ch...@gmail.com>
AuthorDate: Sat Oct 27 15:04:35 2018 +0800
[FLINK-10676][table] Change the over window preceding clause from required to optional.
This closes #6949
---
docs/dev/table/tableApi.md | 4 ++-
.../org/apache/flink/table/api/java/windows.scala | 20 +++++++++++++-
.../org/apache/flink/table/api/scala/windows.scala | 22 ++++++++++++---
.../table/api/stream/sql/OverWindowTest.scala | 15 +++++++++++
.../table/api/stream/table/OverWindowTest.scala | 15 +++++++++--
.../OverWindowStringExpressionTest.scala | 31 +++++++++++++++++++++-
6 files changed, 99 insertions(+), 8 deletions(-)
diff --git a/docs/dev/table/tableApi.md b/docs/dev/table/tableApi.md
index b17753c..80930ca 100644
--- a/docs/dev/table/tableApi.md
+++ b/docs/dev/table/tableApi.md
@@ -1571,13 +1571,15 @@ The `OverWindow` defines a range of rows over which aggregates are computed. `Ov
</tr>
<tr>
<td><code>preceding</code></td>
- <td>Required</td>
+ <td>Optional</td>
<td>
<p>Defines the interval of rows that are included in the window and precede the current row. The interval can either be specified as time or row-count interval.</p>
<p><a href="tableApi.html#bounded-over-windows">Bounded over windows</a> are specified with the size of the interval, e.g., <code>10.minutes</code> for a time interval or <code>10.rows</code> for a row-count interval.</p>
<p><a href="tableApi.html#unbounded-over-windows">Unbounded over windows</a> are specified using a constant, i.e., <code>UNBOUNDED_RANGE</code> for a time interval or <code>UNBOUNDED_ROW</code> for a row-count interval. Unbounded over windows start with the first row of a partition.</p>
+
+ <p>If the <code>preceding</code> clause is omitted, <code>UNBOUNDED_RANGE</code> and <code>CURRENT_RANGE</code> are used as the default <code>preceding</code> and <code>following</code> for the window.</p>
</td>
</tr>
<tr>
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/java/windows.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/java/windows.scala
index f326f6f..121aab8 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/java/windows.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/java/windows.scala
@@ -18,7 +18,8 @@
package org.apache.flink.table.api.java
-import org.apache.flink.table.api.{TumbleWithSize, OverWindowWithPreceding, SlideWithSize, SessionWithGap}
+import org.apache.flink.table.api.scala.{CURRENT_RANGE, UNBOUNDED_RANGE}
+import org.apache.flink.table.api.{OverWindow, TumbleWithSize, OverWindowWithPreceding, SlideWithSize, SessionWithGap}
import org.apache.flink.table.expressions.{Expression, ExpressionParser}
/**
@@ -144,4 +145,21 @@ class OverWindowWithOrderBy(
new OverWindowWithPreceding(partitionByExpr, orderByExpr, precedingExpr)
}
+ /**
+ * Assigns an alias for this window that the following `select()` clause can refer to.
+ *
+ * @param alias alias for this over window
+ * @return over window
+ */
+ def as(alias: String): OverWindow = as(ExpressionParser.parseExpression(alias))
+
+ /**
+ * Assigns an alias for this window that the following `select()` clause can refer to.
+ *
+ * @param alias alias for this over window
+ * @return over window
+ */
+ def as(alias: Expression): OverWindow = {
+ OverWindow(alias, partitionByExpr, orderByExpr, UNBOUNDED_RANGE, CURRENT_RANGE)
+ }
}
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/windows.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/windows.scala
index 91bf1a6..2f88248 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/windows.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/scala/windows.scala
@@ -18,8 +18,8 @@
package org.apache.flink.table.api.scala
-import org.apache.flink.table.api.{TumbleWithSize, OverWindowWithPreceding, SlideWithSize, SessionWithGap}
-import org.apache.flink.table.expressions.Expression
+import org.apache.flink.table.api.{OverWindow, TumbleWithSize, OverWindowWithPreceding, SlideWithSize, SessionWithGap}
+import org.apache.flink.table.expressions.{Expression, ExpressionParser}
/**
* Helper object for creating a tumbling window. Tumbling windows are consecutive, non-overlapping
@@ -127,7 +127,6 @@ case class PartitionedOver(partitionBy: Array[Expression]) {
case class OverWindowWithOrderBy(partitionBy: Seq[Expression], orderBy: Expression) {
-
/**
* Set the preceding offset (based on time or row-count intervals) for over window.
*
@@ -138,4 +137,21 @@ case class OverWindowWithOrderBy(partitionBy: Seq[Expression], orderBy: Expressi
new OverWindowWithPreceding(partitionBy, orderBy, preceding)
}
+ /**
+ * Assigns an alias for this window that the following `select()` clause can refer to.
+ *
+ * @param alias alias for this over window
+ * @return over window
+ */
+ def as(alias: String): OverWindow = as(ExpressionParser.parseExpression(alias))
+
+ /**
+ * Assigns an alias for this window that the following `select()` clause can refer to.
+ *
+ * @param alias alias for this over window
+ * @return over window
+ */
+ def as(alias: Expression): OverWindow = {
+ OverWindow(alias, partitionBy, orderBy, UNBOUNDED_RANGE, CURRENT_RANGE)
+ }
}
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/OverWindowTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/OverWindowTest.scala
index 1be5810..eca53c0 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/OverWindowTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/OverWindowTest.scala
@@ -287,7 +287,15 @@ class OverWindowTest extends TableTestBase {
"sum(a) OVER (PARTITION BY c ORDER BY proctime RANGE UNBOUNDED preceding) as cnt2 " +
"FROM MyTable " +
"WINDOW w AS (PARTITION BY c ORDER BY proctime RANGE UNBOUNDED preceding)"
+
+ val sql3 = "SELECT " +
+ "c, " +
+ "count(a) OVER (PARTITION BY c ORDER BY proctime) as cnt1, " +
+ "sum(a) OVER (PARTITION BY c ORDER BY proctime) as cnt2 " +
+ "from MyTable"
+
streamUtil.verifySqlPlansIdentical(sql, sql2)
+ streamUtil.verifySqlPlansIdentical(sql, sql3)
val expected =
unaryNode(
@@ -523,6 +531,13 @@ class OverWindowTest extends TableTestBase {
"sum(a) OVER (PARTITION BY c ORDER BY rowtime RANGE UNBOUNDED preceding) as cnt2 " +
"from MyTable"
+ val sql1 = "SELECT " +
+ "c, " +
+ "count(a) OVER (PARTITION BY c ORDER BY rowtime) as cnt1, " +
+ "sum(a) OVER (PARTITION BY c ORDER BY rowtime) as cnt2 " +
+ "from MyTable"
+ streamUtil.verifySqlPlansIdentical(sql, sql1)
+
val expected =
unaryNode(
"DataStreamCalc",
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/OverWindowTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/OverWindowTest.scala
index 55e3ecb..eeb7d5f 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/OverWindowTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/OverWindowTest.scala
@@ -194,10 +194,15 @@ class OverWindowTest extends TableTestBase {
val weightedAvg = new WeightedAvgWithRetract
val result = table
- .window(Over partitionBy 'c orderBy 'proctime preceding UNBOUNDED_RANGE following
- CURRENT_RANGE as 'w)
+ .window(Over partitionBy 'c orderBy 'proctime preceding UNBOUNDED_RANGE as 'w)
+ .select('a, 'c, 'a.count over 'w, weightedAvg('c, 'a) over 'w)
+
+ val result2 = table
+ .window(Over partitionBy 'c orderBy 'proctime as 'w)
.select('a, 'c, 'a.count over 'w, weightedAvg('c, 'a) over 'w)
+ streamUtil.verify2Tables(result, result2)
+
val expected =
unaryNode(
"DataStreamCalc",
@@ -459,6 +464,12 @@ class OverWindowTest extends TableTestBase {
CURRENT_RANGE as 'w)
.select('a, 'c, 'a.count over 'w, weightedAvg('c, 'a) over 'w as 'wAvg)
+ val result2 = table
+ .window(Over partitionBy 'c orderBy 'rowtime as 'w)
+ .select('a, 'c, 'a.count over 'w, weightedAvg('c, 'a) over 'w as 'wAvg)
+
+ streamUtil.verify2Tables(result, result2)
+
val expected =
unaryNode(
"DataStreamCalc",
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/OverWindowStringExpressionTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/OverWindowStringExpressionTest.scala
index 3e757da..99114b8 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/OverWindowStringExpressionTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/stringexpr/OverWindowStringExpressionTest.scala
@@ -120,7 +120,7 @@ class OverWindowStringExpressionTest extends TableTestBase {
}
@Test
- def testUnboundedOverRange(): Unit = {
+ def testRowTimeUnboundedOverRange(): Unit = {
val util = streamTestUtil()
val t = util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'rowtime.rowtime)
@@ -134,8 +134,37 @@ class OverWindowStringExpressionTest extends TableTestBase {
.window(
JOver.orderBy("rowtime").preceding("unbounded_range").following("current_range").as("w"))
.select("a, SUM(b) OVER w, weightAvgFun(a, b) over w as myCnt")
+ val resJava2 = t
+ .window(
+ JOver.orderBy("rowtime").as("w"))
+ .select("a, SUM(b) OVER w, weightAvgFun(a, b) over w as myCnt")
+
+ verifyTableEquals(resScala, resJava)
+ verifyTableEquals(resScala, resJava2)
+ }
+
+ @Test
+ def testProcTimeUnboundedOverRange(): Unit = {
+ val util = streamTestUtil()
+ val t = util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'proctime.proctime)
+
+ val weightAvgFun = new WeightedAvg
+ util.tableEnv.registerFunction("weightAvgFun", weightAvgFun)
+
+ val resScala = t
+ .window(SOver orderBy 'proctime preceding UNBOUNDED_RANGE following CURRENT_RANGE as 'w)
+ .select('a, 'b.sum over 'w, weightAvgFun('a, 'b) over 'w as 'myCnt)
+ val resJava = t
+ .window(
+ JOver.orderBy("proctime").preceding("unbounded_range").following("current_range").as("w"))
+ .select("a, SUM(b) OVER w, weightAvgFun(a, b) over w as myCnt")
+ val resJava2 = t
+ .window(
+ JOver.orderBy("proctime").as("w"))
+ .select("a, SUM(b) OVER w, weightAvgFun(a, b) over w as myCnt")
verifyTableEquals(resScala, resJava)
+ verifyTableEquals(resScala, resJava2)
}
@Test