{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/139578010","pull_request_review_id":63529741,"id":139578010,"diff_hunk":"@@ -0,0 +1,123 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements.  See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * 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.spark.sql.streaming\n+\n+import java.util.UUID\n+\n+import org.apache.spark.rdd.RDD\n+import org.apache.spark.sql.catalyst.InternalRow\n+import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute\n+import org.apache.spark.sql.catalyst.expressions.Attribute\n+import org.apache.spark.sql.catalyst.plans.physical._\n+import org.apache.spark.sql.execution.{SparkPlan, SparkPlanTest, UnaryExecNode}\n+import org.apache.spark.sql.execution.exchange.{Exchange, ShuffleExchange}\n+import org.apache.spark.sql.execution.streaming.{IncrementalExecution, OffsetSeqMetadata, StatefulOperator, StatefulOperatorStateInfo}\n+import org.apache.spark.sql.test.SharedSQLContext\n+\n+class IncrementalExecutionRulesSuite extends SparkPlanTest with SharedSQLContext {\n+\n+  import testImplicits._\n+  super.beforeAll()\n+\n+  private val baseDf = Seq((1, \"A\"), (2, \"b\")).toDF(\"num\", \"char\")\n+\n+  testEnsureStatefulOpPartitioning(\n+    \"ClusteredDistribution generates Exchange with HashPartitioning\",\n+    baseDf.queryExecution.sparkPlan,\n+    keys => ClusteredDistribution(keys),\n+    keys => HashPartitioning(keys, spark.sessionState.conf.numShufflePartitions),\n+    expectShuffle = true)\n+\n+  testEnsureStatefulOpPartitioning(\n+    \"ClusteredDistribution with coalesce(1) generates Exchange with HashPartitioning\",\n+    baseDf.coalesce(1).queryExecution.sparkPlan,\n+    keys => ClusteredDistribution(keys),\n+    keys => HashPartitioning(keys, spark.sessionState.conf.numShufflePartitions),\n+    expectShuffle = true)\n+\n+  testEnsureStatefulOpPartitioning(\n+    \"AllTuples generates Exchange with SinglePartition\",\n+    baseDf.queryExecution.sparkPlan,\n+    keys => AllTuples,\n+    keys => SinglePartition,\n+    expectShuffle = true)\n+\n+  testEnsureStatefulOpPartitioning(\n+    \"AllTuples with coalesce(1) doesn't need Exchange\",\n+    baseDf.coalesce(1).queryExecution.sparkPlan,\n+    keys => AllTuples,\n+    keys => SinglePartition,\n+    expectShuffle = false)\n+\n+  private def testEnsureStatefulOpPartitioning(\n+      testName: String,\n+      inputPlan: SparkPlan,\n+      requiredDistribution: Seq[Attribute] => Distribution,\n+      expectedPartitioning: Seq[Attribute] => Partitioning,\n+      expectShuffle: Boolean): Unit = {\n+    test(\"EnsureStatefulOpPartitioning - \" + testName) {\n+      val operator = TestOperator(inputPlan, requiredDistribution(inputPlan.output.take(1)))\n+      val executed = executePlan(operator, OutputMode.Complete())\n+      if (expectShuffle) {\n+        val exchange = executed.children.find(_.isInstanceOf[Exchange])\n+        if (exchange.isEmpty) {\n+          fail(s\"Was expecting an exchange but didn't get one in:\\n$executed\")\n+        }\n+        assert(exchange.get ===\n+          ShuffleExchange(expectedPartitioning(inputPlan.output.take(1)), inputPlan),\n+          s\"Exchange didn't have expected properties:\\n${exchange.get}\")\n+      } else {\n+        assert(!executed.children.exists(_.isInstanceOf[Exchange]),\n+          s\"Unexpected exchange found in:\\n$executed\")\n+      }\n+    }\n+  }\n+\n+  private def executePlan(\n+      p: SparkPlan,\n+      outputMode: OutputMode = OutputMode.Append()): SparkPlan = {\n+    val execution = new IncrementalExecution(\n+      spark,\n+      null,\n+      OutputMode.Complete(),\n+      \"chk\",\n+      UUID.randomUUID(),\n+      0L,\n+      OffsetSeqMetadata()) {\n+      override lazy val sparkPlan: SparkPlan = p transform {\n+        case plan: SparkPlan =>\n+          val inputMap = plan.children.flatMap(_.output).map(a => (a.name, a)).toMap\n+          plan transformExpressions {\n+            case UnresolvedAttribute(Seq(u)) =>\n+              inputMap.getOrElse(u,\n+                sys.error(s\"Invalid Test: Cannot resolve $u given input $inputMap\"))\n+          }\n+      }\n+    }\n+    execution.executedPlan\n+  }\n+}\n+\n+case class TestOperator(","path":"sql/core/src/test/scala/org/apache/spark/sql/streaming/IncrementalExecutionRulesSuite.scala","position":116,"original_position":116,"commit_id":"be3912552d200fced24f69c436b59a2c24639380","original_commit_id":"be3912552d200fced24f69c436b59a2c24639380","user":{"login":"tdas","id":663212,"avatar_url":"https://avatars1.githubusercontent.com/u/663212?v=4","gravatar_id":"","url":"https://api.github.com/users/tdas","html_url":"https://github.com/tdas","followers_url":"https://api.github.com/users/tdas/followers","following_url":"https://api.github.com/users/tdas/following{/other_user}","gists_url":"https://api.github.com/users/tdas/gists{/gist_id}","starred_url":"https://api.github.com/users/tdas/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tdas/subscriptions","organizations_url":"https://api.github.com/users/tdas/orgs","repos_url":"https://api.github.com/users/tdas/repos","events_url":"https://api.github.com/users/tdas/events{/privacy}","received_events_url":"https://api.github.com/users/tdas/received_events","type":"User","site_admin":false},"body":"Name it as TestStatefulOperator to make it more specific than just \"test operator\". And add docs saying what is it useful for.","created_at":"2017-09-19T01:15:30Z","updated_at":"2017-09-19T01:15:30Z","html_url":"https://github.com/apache/spark/pull/19196#discussion_r139578010","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/19196","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/139578010"},"html":{"href":"https://github.com/apache/spark/pull/19196#discussion_r139578010"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/19196"}},"in_reply_to_id":139577945},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/19196","id":140486503,"html_url":"https://github.com/apache/spark/pull/19196","diff_url":"https://github.com/apache/spark/pull/19196.diff","patch_url":"https://github.com/apache/spark/pull/19196.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/19196","number":19196,"state":"open","locked":false,"title":"[SPARK-21977] SinglePartition optimizations break certain Streaming Stateful Aggregation requirements","user":{"login":"brkyvz","id":5243515,"avatar_url":"https://avatars1.githubusercontent.com/u/5243515?v=4","gravatar_id":"","url":"https://api.github.com/users/brkyvz","html_url":"https://github.com/brkyvz","followers_url":"https://api.github.com/users/brkyvz/followers","following_url":"https://api.github.com/users/brkyvz/following{/other_user}","gists_url":"https://api.github.com/users/brkyvz/gists{/gist_id}","starred_url":"https://api.github.com/users/brkyvz/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/brkyvz/subscriptions","organizations_url":"https://api.github.com/users/brkyvz/orgs","repos_url":"https://api.github.com/users/brkyvz/repos","events_url":"https://api.github.com/users/brkyvz/events{/privacy}","received_events_url":"https://api.github.com/users/brkyvz/received_events","type":"User","site_admin":false},"body":"## What changes were proposed in this pull request?\r\n\r\nThis is a bit hard to explain as there are several issues here, I'll try my best. Here are the requirements:\r\n  1. A StructuredStreaming Source that can generate empty RDDs with 0 partitions\r\n  2. A StructuredStreaming query that uses the above source, performs a stateful aggregation \r\n     (mapGroupsWithState, groupBy.count, ...), and coalesce's by 1\r\n\r\nThe crux of the problem is that when a dataset has a `coalesce(1)` call, it receives a `SinglePartition` partitioning scheme. This scheme satisfies most required distributions used for aggregations such as HashAggregateExec. This causes a world of problems:\r\n  Symptom 1. If the input RDD has 0 partitions, the whole lineage will receive 0 partitions, nothing will be executed, the state store will not create any delta files. When this happens, the next trigger fails, because the StateStore fails to load the delta file for the previous trigger\r\n  Symptom 2. Let's say that there was data. Then in this case, if you stop your stream, and change `coalesce(1)` with `coalesce(2)`, then restart your stream, your stream will fail, because `spark.sql.shuffle.partitions - 1` number of StateStores will fail to find its delta files.\r\n\r\nTo fix the issues above, we must check that the partitioning of the child of a `StatefulOperator` satisfies:\r\nIf the grouping expressions are empty:\r\n  a) AllTuple distribution\r\n  b) Single physical partition\r\nIf the grouping expressions are non empty:\r\n  a) Clustered distribution\r\n  b) spark.sql.shuffle.partition # of partitions\r\nwhether or not `coalesce(1)` exists in the plan, and whether or not the input RDD for the trigger has any data.\r\n\r\nOnce you fix the above problem by adding an Exchange to the plan, you come across the following bug:\r\nIf you call `coalesce(1).groupBy().count()` on a Streaming DataFrame, and if you have a trigger with no data, `StateStoreRestoreExec` doesn't return the prior state. However, for this specific aggregation, `HashAggregateExec` after the restore returns a (0, 0) row, since we're performing a count, and there is no data. Then this data gets stored in `StateStoreSaveExec` causing the previous counts to be overwritten and lost.\r\n\r\n## How was this patch tested?\r\n\r\nRegression tests","created_at":"2017-09-12T04:35:04Z","updated_at":"2017-09-19T01:15:30Z","closed_at":null,"merged_at":null,"merge_commit_sha":"0f83a9f61833ad542692e68abcaf17db6c8c0210","assignee":null,"assignees":[],"requested_reviewers":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/19196/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/19196/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/19196/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/be3912552d200fced24f69c436b59a2c24639380","head":{"label":"brkyvz:sa-0","ref":"sa-0","sha":"be3912552d200fced24f69c436b59a2c24639380","user":{"login":"brkyvz","id":5243515,"avatar_url":"https://avatars1.githubusercontent.com/u/5243515?v=4","gravatar_id":"","url":"https://api.github.com/users/brkyvz","html_url":"https://github.com/brkyvz","followers_url":"https://api.github.com/users/brkyvz/followers","following_url":"https://api.github.com/users/brkyvz/following{/other_user}","gists_url":"https://api.github.com/users/brkyvz/gists{/gist_id}","starred_url":"https://api.github.com/users/brkyvz/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/brkyvz/subscriptions","organizations_url":"https://api.github.com/users/brkyvz/orgs","repos_url":"https://api.github.com/users/brkyvz/repos","events_url":"https://api.github.com/users/brkyvz/events{/privacy}","received_events_url":"https://api.github.com/users/brkyvz/received_events","type":"User","site_admin":false},"repo":{"id":21919521,"name":"spark","full_name":"brkyvz/spark","owner":{"login":"brkyvz","id":5243515,"avatar_url":"https://avatars1.githubusercontent.com/u/5243515?v=4","gravatar_id":"","url":"https://api.github.com/users/brkyvz","html_url":"https://github.com/brkyvz","followers_url":"https://api.github.com/users/brkyvz/followers","following_url":"https://api.github.com/users/brkyvz/following{/other_user}","gists_url":"https://api.github.com/users/brkyvz/gists{/gist_id}","starred_url":"https://api.github.com/users/brkyvz/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/brkyvz/subscriptions","organizations_url":"https://api.github.com/users/brkyvz/orgs","repos_url":"https://api.github.com/users/brkyvz/repos","events_url":"https://api.github.com/users/brkyvz/events{/privacy}","received_events_url":"https://api.github.com/users/brkyvz/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/brkyvz/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/brkyvz/spark","forks_url":"https://api.github.com/repos/brkyvz/spark/forks","keys_url":"https://api.github.com/repos/brkyvz/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/brkyvz/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/brkyvz/spark/teams","hooks_url":"https://api.github.com/repos/brkyvz/spark/hooks","issue_events_url":"https://api.github.com/repos/brkyvz/spark/issues/events{/number}","events_url":"https://api.github.com/repos/brkyvz/spark/events","assignees_url":"https://api.github.com/repos/brkyvz/spark/assignees{/user}","branches_url":"https://api.github.com/repos/brkyvz/spark/branches{/branch}","tags_url":"https://api.github.com/repos/brkyvz/spark/tags","blobs_url":"https://api.github.com/repos/brkyvz/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/brkyvz/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/brkyvz/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/brkyvz/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/brkyvz/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/brkyvz/spark/languages","stargazers_url":"https://api.github.com/repos/brkyvz/spark/stargazers","contributors_url":"https://api.github.com/repos/brkyvz/spark/contributors","subscribers_url":"https://api.github.com/repos/brkyvz/spark/subscribers","subscription_url":"https://api.github.com/repos/brkyvz/spark/subscription","commits_url":"https://api.github.com/repos/brkyvz/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/brkyvz/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/brkyvz/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/brkyvz/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/brkyvz/spark/contents/{+path}","compare_url":"https://api.github.com/repos/brkyvz/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/brkyvz/spark/merges","archive_url":"https://api.github.com/repos/brkyvz/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/brkyvz/spark/downloads","issues_url":"https://api.github.com/repos/brkyvz/spark/issues{/number}","pulls_url":"https://api.github.com/repos/brkyvz/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/brkyvz/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/brkyvz/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/brkyvz/spark/labels{/name}","releases_url":"https://api.github.com/repos/brkyvz/spark/releases{/id}","deployments_url":"https://api.github.com/repos/brkyvz/spark/deployments","created_at":"2014-07-16T21:51:06Z","updated_at":"2015-05-08T16:54:55Z","pushed_at":"2017-09-18T01:32:49Z","git_url":"git://github.com/brkyvz/spark.git","ssh_url":"git@github.com:brkyvz/spark.git","clone_url":"https://github.com/brkyvz/spark.git","svn_url":"https://github.com/brkyvz/spark","homepage":null,"size":251505,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_projects":true,"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":"dc74c0e67d96ed22210916b73dd896137f54bc37","user":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","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":17165658,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","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/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","deployments_url":"https://api.github.com/repos/apache/spark/deployments","created_at":"2014-02-25T08:00:08Z","updated_at":"2017-09-19T00:53:37Z","pushed_at":"2017-09-19T00:10:20Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":262788,"stargazers_count":14355,"watchers_count":14355,"language":"Scala","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":13429,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":392,"forks":13429,"open_issues":392,"watchers":14355,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/19196"},"html":{"href":"https://github.com/apache/spark/pull/19196"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/19196"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/19196/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/19196/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/19196/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/be3912552d200fced24f69c436b59a2c24639380"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"tdas"}}