{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/81998716","pull_request_review_id":2936945,"id":81998716,"diff_hunk":"@@ -0,0 +1,340 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one\n+ * or more contributor license agreements.  See the NOTICE file\n+ * distributed with this work for additional information\n+ * regarding copyright ownership.  The ASF licenses this file\n+ * to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance\n+ * with the License.  You may obtain a copy of the License at\n+ *\n+ *     http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package org.apache.flink.api.table\n+\n+import org.apache.flink.api.table.expressions.{Expression, ExpressionParser}\n+import org.apache.flink.api.table.plan.logical._\n+\n+/**\n+  * Group-window specification. Group-windows allow aggregates which are computed for a group of\n+  * elements. A (time or row-count) window is required to bound the infinite input stream into a\n+  * finite group. Group-windows are evaluated once per group.\n+  */\n+trait GroupWindow {\n+\n+  /**\n+    * Converts an API class to a logical window for planning. This is an internal method.\n+    */\n+  private[flink] def toLogicalWindow: LogicalWindow\n+}\n+\n+/**\n+  * A group-window operating on event-time.\n+  *\n+  * @param timeField defines the time mode for streaming tables and acts as a time attribute for\n+  *                  batch tables over which the query is evaluated.\n+  */\n+abstract class EventTimeWindow(timeField: Expression) extends GroupWindow {\n+\n+  protected var name: Option[Expression] = None\n+\n+  /**\n+    * Assigns an alias for this window that the following `select()` clause can refer to in order\n+    * to access window properties such as window start or end time.\n+    *\n+    * @param alias alias for this window\n+    * @return this window\n+    */\n+  def as(alias: Expression): EventTimeWindow = {\n+    this.name = Some(alias)\n+    this\n+  }\n+\n+  /**\n+    * Assigns an alias for this window that the following `select()` clause can refer to in order\n+    * to access window properties such as window start or end time.\n+    *\n+    * @param alias alias for this window\n+    * @return this window\n+    */\n+  def as(alias: String): EventTimeWindow = as(ExpressionParser.parseExpression(alias))\n+}\n+\n+// ------------------------------------------------------------------------------------------------\n+// Tumbling group-windows\n+// ------------------------------------------------------------------------------------------------\n+\n+/**\n+  * Tumbling group-window. By default, it works on processing-time.\n+  *\n+  * @param size size of the window either as number of rows or interval of milliseconds\n+  */\n+class TumblingWindow(size: Expression) extends GroupWindow {\n+\n+  /**\n+    * Tumbling group-window. By default, it works on processing-time.\n+    *\n+    * @param size size of the window either as number of rows or interval of milliseconds\n+    */\n+  def this(size: String) = this(ExpressionParser.parseExpression(size))\n+\n+  private var alias: Option[Expression] = None\n+\n+  /**\n+    * Defines the time mode for streaming tables and specifies a time attribute for\n+    * batch tables over which the query is evaluated.\n+    *\n+    * @param timeField time mode for streaming tables and time attribute for batch tables\n+    * @return a tumbling group-window on event-time\n+    */\n+  def on(timeField: Expression): TumblingEventTimeWindow =\n+    new TumblingEventTimeWindow(alias, timeField, size)\n+\n+  /**\n+    * Defines the time mode for streaming tables and specifies a time attribute for\n+    * batch tables over which the query is evaluated.\n+    *\n+    * @param timeField time mode for streaming tables and time attribute for batch tables\n+    * @return a tumbling group-window on event-time\n+    */\n+  def on(timeField: String): TumblingEventTimeWindow =\n+    on(ExpressionParser.parseExpression(timeField))\n+\n+  /**\n+    * Assigns an alias for this window that the following `select()` clause can refer to in order\n+    * to access window properties such as window start or end time.\n+    *\n+    * @param alias alias for this window\n+    * @return this window\n+    */\n+  def as(alias: Expression): TumblingWindow = {\n+    this.alias = Some(alias)\n+    this\n+  }\n+\n+  /**\n+    * Assigns an alias for this window that the following `select()` clause can refer to in order\n+    * to access window properties such as window start or end time.\n+    *\n+    * @param alias alias for this window\n+    * @return this window\n+    */\n+  def as(alias: String): TumblingWindow = as(ExpressionParser.parseExpression(alias))\n+\n+  override private[flink] def toLogicalWindow: LogicalWindow =\n+    ProcessingTimeTumblingGroupWindow(alias, size)\n+}\n+\n+/**\n+  * Tumbling group-window on event-time.\n+  */\n+class TumblingEventTimeWindow(\n+    alias: Option[Expression],\n+    time: Expression,\n+    size: Expression)\n+  extends EventTimeWindow(time) {\n+\n+  override private[flink] def toLogicalWindow: LogicalWindow =\n+    EventTimeTumblingGroupWindow(name.orElse(alias), time, size)\n+}\n+\n+// ------------------------------------------------------------------------------------------------\n+// Sliding group windows\n+// ------------------------------------------------------------------------------------------------\n+\n+/**\n+  * Partially specified sliding window.\n+  *\n+  * @param size size of the window either as number of rows or interval of milliseconds\n+  */\n+class SlideWithSize(size: Expression) {\n+\n+  /**\n+    * Partially specified sliding window.\n+    *\n+    * @param size size of the window either as number of rows or interval of milliseconds\n+    */\n+  def this(size: String) = this(ExpressionParser.parseExpression(size))\n+\n+  /**\n+    * Specifies the window's slide. This is the size of the overlap.\n+    *\n+    * For example, you could have windows of size 10 minutes that slide by 5 minutes. With this\n+    * the following operation gets 10 minutes worth of elements and will be invoked for every 5","path":"flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/windows.scala","position":169,"original_position":169,"commit_id":"edbfe00cb0fd7ea8362c90988eb0860eb9ce6078","original_commit_id":"edbfe00cb0fd7ea8362c90988eb0860eb9ce6078","user":{"login":"twalthr","id":5746567,"avatar_url":"https://avatars.githubusercontent.com/u/5746567?v=3","gravatar_id":"","url":"https://api.github.com/users/twalthr","html_url":"https://github.com/twalthr","followers_url":"https://api.github.com/users/twalthr/followers","following_url":"https://api.github.com/users/twalthr/following{/other_user}","gists_url":"https://api.github.com/users/twalthr/gists{/gist_id}","starred_url":"https://api.github.com/users/twalthr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/twalthr/subscriptions","organizations_url":"https://api.github.com/users/twalthr/orgs","repos_url":"https://api.github.com/users/twalthr/repos","events_url":"https://api.github.com/users/twalthr/events{/privacy}","received_events_url":"https://api.github.com/users/twalthr/received_events","type":"User","site_admin":false},"body":"Done.","created_at":"2016-10-05T15:22:13Z","updated_at":"2016-10-05T15:22:13Z","html_url":"https://github.com/apache/flink/pull/2562#discussion_r81998716","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/2562","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/81998716"},"html":{"href":"https://github.com/apache/flink/pull/2562#discussion_r81998716"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/2562"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/2562","id":87107781,"html_url":"https://github.com/apache/flink/pull/2562","diff_url":"https://github.com/apache/flink/pull/2562.diff","patch_url":"https://github.com/apache/flink/pull/2562.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/2562","number":2562,"state":"open","locked":false,"title":"[FLINK-4691] [table] Add group-windows for streaming tables","user":{"login":"twalthr","id":5746567,"avatar_url":"https://avatars.githubusercontent.com/u/5746567?v=3","gravatar_id":"","url":"https://api.github.com/users/twalthr","html_url":"https://github.com/twalthr","followers_url":"https://api.github.com/users/twalthr/followers","following_url":"https://api.github.com/users/twalthr/following{/other_user}","gists_url":"https://api.github.com/users/twalthr/gists{/gist_id}","starred_url":"https://api.github.com/users/twalthr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/twalthr/subscriptions","organizations_url":"https://api.github.com/users/twalthr/orgs","repos_url":"https://api.github.com/users/twalthr/repos","events_url":"https://api.github.com/users/twalthr/events{/privacy}","received_events_url":"https://api.github.com/users/twalthr/received_events","type":"User","site_admin":false},"body":"This PR implements Tumble, Slide, Session group-windows for streaming tables as described in FLIP-11. It adds API, validation, logical representation, and runtime components.\r\n\r\nSome additional comments:\r\n\r\nI have not implemented the 'systemtime' keyword yet as this would cause more problems than it solves. Especially integrating it into the validation layer would be tricky. The resolution of those special fields happens within a WindowAggregate, however, the logical type of a window should already be known at this point. We are mixing logical operators and expressions which is not very nice. Furthermore, what happens in batch environment if 'systemtime' is used? It could also be a existing column but does not have to be one. That is not specified in the FLIP yet.\r\n\r\nThe aggregations are not very efficient yet. Currently this PR uses window functions that wrap the GroupReduce functions. We have to rework the aggregations first. Maybe we could use `WindowedStream#apply(R, FoldFunction<T,R>, WindowFunction<R,R,K,W>, TypeInformation<R>)` which means that `R` has to be created in the translation phase.\r\n\r\nThe tests are mainly ITCases yet, we might want to change that to unit tests once we have means (like new test bases) to do that.\r\n\r\nThe website documentation is missing yet.\r\n\r\nAllowed lateness is not part of this PR.","created_at":"2016-09-28T12:15:43Z","updated_at":"2016-10-05T15:22:13Z","closed_at":null,"merged_at":null,"merge_commit_sha":"aa1e3d1d3ef9763db3eb97560ac07dfb5485dec3","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/2562/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/2562/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/2562/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/edbfe00cb0fd7ea8362c90988eb0860eb9ce6078","head":{"label":"twalthr:FLINK-4691","ref":"FLINK-4691","sha":"edbfe00cb0fd7ea8362c90988eb0860eb9ce6078","user":{"login":"twalthr","id":5746567,"avatar_url":"https://avatars.githubusercontent.com/u/5746567?v=3","gravatar_id":"","url":"https://api.github.com/users/twalthr","html_url":"https://github.com/twalthr","followers_url":"https://api.github.com/users/twalthr/followers","following_url":"https://api.github.com/users/twalthr/following{/other_user}","gists_url":"https://api.github.com/users/twalthr/gists{/gist_id}","starred_url":"https://api.github.com/users/twalthr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/twalthr/subscriptions","organizations_url":"https://api.github.com/users/twalthr/orgs","repos_url":"https://api.github.com/users/twalthr/repos","events_url":"https://api.github.com/users/twalthr/events{/privacy}","received_events_url":"https://api.github.com/users/twalthr/received_events","type":"User","site_admin":false},"repo":{"id":20658760,"name":"flink","full_name":"twalthr/flink","owner":{"login":"twalthr","id":5746567,"avatar_url":"https://avatars.githubusercontent.com/u/5746567?v=3","gravatar_id":"","url":"https://api.github.com/users/twalthr","html_url":"https://github.com/twalthr","followers_url":"https://api.github.com/users/twalthr/followers","following_url":"https://api.github.com/users/twalthr/following{/other_user}","gists_url":"https://api.github.com/users/twalthr/gists{/gist_id}","starred_url":"https://api.github.com/users/twalthr/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/twalthr/subscriptions","organizations_url":"https://api.github.com/users/twalthr/orgs","repos_url":"https://api.github.com/users/twalthr/repos","events_url":"https://api.github.com/users/twalthr/events{/privacy}","received_events_url":"https://api.github.com/users/twalthr/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/twalthr/flink","description":"Mirror of Apache Flink","fork":true,"url":"https://api.github.com/repos/twalthr/flink","forks_url":"https://api.github.com/repos/twalthr/flink/forks","keys_url":"https://api.github.com/repos/twalthr/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/twalthr/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/twalthr/flink/teams","hooks_url":"https://api.github.com/repos/twalthr/flink/hooks","issue_events_url":"https://api.github.com/repos/twalthr/flink/issues/events{/number}","events_url":"https://api.github.com/repos/twalthr/flink/events","assignees_url":"https://api.github.com/repos/twalthr/flink/assignees{/user}","branches_url":"https://api.github.com/repos/twalthr/flink/branches{/branch}","tags_url":"https://api.github.com/repos/twalthr/flink/tags","blobs_url":"https://api.github.com/repos/twalthr/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/twalthr/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/twalthr/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/twalthr/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/twalthr/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/twalthr/flink/languages","stargazers_url":"https://api.github.com/repos/twalthr/flink/stargazers","contributors_url":"https://api.github.com/repos/twalthr/flink/contributors","subscribers_url":"https://api.github.com/repos/twalthr/flink/subscribers","subscription_url":"https://api.github.com/repos/twalthr/flink/subscription","commits_url":"https://api.github.com/repos/twalthr/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/twalthr/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/twalthr/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/twalthr/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/twalthr/flink/contents/{+path}","compare_url":"https://api.github.com/repos/twalthr/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/twalthr/flink/merges","archive_url":"https://api.github.com/repos/twalthr/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/twalthr/flink/downloads","issues_url":"https://api.github.com/repos/twalthr/flink/issues{/number}","pulls_url":"https://api.github.com/repos/twalthr/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/twalthr/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/twalthr/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/twalthr/flink/labels{/name}","releases_url":"https://api.github.com/repos/twalthr/flink/releases{/id}","deployments_url":"https://api.github.com/repos/twalthr/flink/deployments","created_at":"2014-06-09T19:08:51Z","updated_at":"2016-01-11T15:02:23Z","pushed_at":"2016-10-05T12:34:21Z","git_url":"git://github.com/twalthr/flink.git","ssh_url":"git@github.com:twalthr/flink.git","clone_url":"https://github.com/twalthr/flink.git","svn_url":"https://github.com/twalthr/flink","homepage":null,"size":93830,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"90902914ac4b11f9554b67ad49e0d697a0d02f93","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":20587599,"name":"flink","full_name":"apache/flink","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/flink","description":"Mirror of Apache Flink","fork":false,"url":"https://api.github.com/repos/apache/flink","forks_url":"https://api.github.com/repos/apache/flink/forks","keys_url":"https://api.github.com/repos/apache/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/flink/teams","hooks_url":"https://api.github.com/repos/apache/flink/hooks","issue_events_url":"https://api.github.com/repos/apache/flink/issues/events{/number}","events_url":"https://api.github.com/repos/apache/flink/events","assignees_url":"https://api.github.com/repos/apache/flink/assignees{/user}","branches_url":"https://api.github.com/repos/apache/flink/branches{/branch}","tags_url":"https://api.github.com/repos/apache/flink/tags","blobs_url":"https://api.github.com/repos/apache/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/flink/languages","stargazers_url":"https://api.github.com/repos/apache/flink/stargazers","contributors_url":"https://api.github.com/repos/apache/flink/contributors","subscribers_url":"https://api.github.com/repos/apache/flink/subscribers","subscription_url":"https://api.github.com/repos/apache/flink/subscription","commits_url":"https://api.github.com/repos/apache/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/flink/contents/{+path}","compare_url":"https://api.github.com/repos/apache/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/flink/merges","archive_url":"https://api.github.com/repos/apache/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/flink/downloads","issues_url":"https://api.github.com/repos/apache/flink/issues{/number}","pulls_url":"https://api.github.com/repos/apache/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/flink/labels{/name}","releases_url":"https://api.github.com/repos/apache/flink/releases{/id}","deployments_url":"https://api.github.com/repos/apache/flink/deployments","created_at":"2014-06-07T07:00:10Z","updated_at":"2016-10-05T12:56:31Z","pushed_at":"2016-10-05T15:12:46Z","git_url":"git://github.com/apache/flink.git","ssh_url":"git@github.com:apache/flink.git","clone_url":"https://github.com/apache/flink.git","svn_url":"https://github.com/apache/flink","homepage":null,"size":129761,"stargazers_count":1617,"watchers_count":1617,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1101,"mirror_url":"git://git.apache.org/flink.git","open_issues_count":153,"forks":1101,"open_issues":153,"watchers":1617,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/2562"},"html":{"href":"https://github.com/apache/flink/pull/2562"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/2562"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/2562/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/2562/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/2562/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/edbfe00cb0fd7ea8362c90988eb0860eb9ce6078"}}}}
{"actor":{"display_login":"twalthr"}}