You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by ala <gi...@git.apache.org> on 2017/02/09 13:13:50 UTC
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
GitHub user ala opened a pull request:
https://github.com/apache/spark/pull/16872
[SPARK-19514] Making range interruptible.
## What changes were proposed in this pull request?
Previously range operator could not be interrupted. For example, using DAGScheduler.cancelStage(...) on a query with range might have been ineffective.
This change adds periodic checks of TaskContext.isInterrupted to codegen version, and InterruptibleOperator to non-codegen version.
## How was this patch tested?
Adds a unit test.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/ala/spark SPARK-19514b
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/16872.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #16872
----
commit 7d1c014f3ac6db889173ef385b9c0b73b6d69d9f
Author: Ala Luszczak <al...@databricks.com>
Date: 2017-02-09T12:53:54Z
[SPARK-19514] Making range interruptible.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16872
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/72643/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100309681
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala ---
@@ -443,6 +443,10 @@ case class RangeExec(range: org.apache.spark.sql.catalyst.plans.logical.Range)
| if (shouldStop()) return;
| }
|
+ | if (TaskContext.get().isInterrupted()) {
--- End diff --
You could also move TaskContext.get() out of the loop. It should not change during evaluation.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16872
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/72647/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100350259
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
cc @zsxwing do you have a suggestion to do this nicely?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by ala <gi...@git.apache.org>.
Github user ala commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100825479
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
OK, so how about this:
```scala
test("Cancelling stage in a query with Range.") {
val listener = new SparkListener {
override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
eventually(timeout(10.seconds)) {
assert(DataFrameRangeSuite.stageToKill > 0)
}
sparkContext.cancelStage(DataFrameRangeSuite.stageToKill)
}
}
sparkContext.addSparkListener(listener)
for (codegen <- Seq(true, false)) {
withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString()) {
DataFrameRangeSuite.stageToKill = -1
val ex = intercept[SparkException] {
spark.range(1000000000L).map { x =>
DataFrameRangeSuite.stageToKill = TaskContext.get().stageId()
x
}.toDF("id").agg(sum("id")).collect()
}
ex.getCause() match {
case null =>
assert(ex.getMessage().contains("cancelled"))
case cause: SparkException =>
assert(cause.getMessage().contains("cancelled"))
case cause: Throwable =>
fail("Expected the cause to be SparkException, got " + cause.toString() + " instead.")
}
}
eventually(timeout(20.seconds)) {
assert(sparkContext.statusTracker.getExecutorInfos.map(_.numRunningTasks()).sum == 0)
}
}
}
```
Now we should be 100% sure what exactly is getting killed. I just checked that it fails without the changes from this PR.
Do I need a new issue in Apache Jira for this?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by ala <gi...@git.apache.org>.
Github user ala commented on the issue:
https://github.com/apache/spark/pull/16872
@hvanhovell @rxin
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/16872
LGTM
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100386073
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
What's the purpose of this listener? Blocking `onTaskStart` only slows down the listener bus thread. It should not block other events from being emitted (It may drop events if the queue is full though). If it does happen, then it's a bug.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/16872
I'm going to merge this in master. If we find a way to optimize the test we can do a follow-up pr.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100310307
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
+ sparkContext.cancelStage(taskStart.stageId)
+ }
+ }
+
+ sparkContext.addSparkListener(listener)
+ for (codegen <- Seq(true, false)) {
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString()) {
+ val ex = intercept[SparkException] {
+ spark.range(100000L).crossJoin(spark.range(100000L))
+ .toDF("a", "b").agg(sum("a"), sum("b")).collect()
+ }
+ val msg = if (ex.getCause() != null) ex.getCause().getMessage() else ex.getMessage()
--- End diff --
Which exception do we actually expect here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100827652
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
Just create another PR under the same ticket.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16872
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16872
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16872
**[Test build #72643 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/72643/testReport)** for PR 16872 at commit [`7d1c014`](https://github.com/apache/spark/commit/7d1c014f3ac6db889173ef385b9c0b73b6d69d9f).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by ala <gi...@git.apache.org>.
Github user ala commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100764417
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
Thanks, Ryan. Worked like a charm.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100389578
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
The problem is that we are trying to kill a query once it actually started executing. If we do not call `sleep(...)` the query gets killed before it starts executing. The other option is to wait for `onExecutorMetricsUpdate(..)` but this is tied to the heartbeat (10s), and makes the test much slower.
Should we move to the other approach, or do you have any other suggestions?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/16872
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by ala <gi...@git.apache.org>.
Github user ala commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100347811
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
The only other thing in the SparkListener interface that seems to fit here seems to be `onExecutorMetricsUpdate()`. However, the metrics are send with the heartbeats, and the heartbeats are send around 10s apart. Hence, using this method makes the tests much slower (4s vs 20s).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100399822
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
Sorry that I didn't see `sparkContext.cancelStage(taskStart.stageId)`. How about this:
```Scala
class DataFrameRangeSuite {
...
test("Cancelling stage in a query with Range.") {
DataFrameRangeSuite.isTaskStarted = false
val listener = new SparkListener {
override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
eventually(timeout(20.seconds)) {
assert(DataFrameRangeSuite.isTaskStarted)
}
sparkContext.cancelStage(taskStart.stageId)
}
}
sparkContext.addSparkListener(listener)
for (codegen <- Seq(true, false)) {
withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString()) {
import testImplicits._
val ex = intercept[SparkException] {
spark.range(100000L).map { i =>
DataFrameRangeSuite.isTaskStarted = true
i
}.crossJoin(spark.range(100000L))
.toDF("a", "b").agg(sum("a"), sum("b")).collect()
}
ex.getCause() match {
case null =>
assert(ex.getMessage().contains("cancelled"))
case cause: SparkException =>
assert(cause.getMessage().contains("cancelled"))
case cause: Throwable =>
fail("Expected the casue to be SparkException, got " + cause.toString() + " instead.")
}
}
eventually(timeout(20.seconds)) {
assert(sparkContext.statusTracker.getExecutorInfos.map(_.numRunningTasks()).sum == 0)
}
}
}
}
object DataFrameRangeSuite {
@volatile var isTaskStarted = false
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100396033
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
Please don't make this test 10s ...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16872
**[Test build #72647 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/72647/testReport)** for PR 16872 at commit [`bb5a1da`](https://github.com/apache/spark/commit/bb5a1dafdb23b892f94c0eccf6e3ad1b9ca9f795).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16872
**[Test build #72647 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/72647/testReport)** for PR 16872 at commit [`bb5a1da`](https://github.com/apache/spark/commit/bb5a1dafdb23b892f94c0eccf6e3ad1b9ca9f795).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100308080
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
Can we do something a little less coarse (it probably also blocks other events from being emitted), and with less potential for flakyness?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by ala <gi...@git.apache.org>.
Github user ala commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100316320
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
+ sparkContext.cancelStage(taskStart.stageId)
+ }
+ }
+
+ sparkContext.addSparkListener(listener)
+ for (codegen <- Seq(true, false)) {
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString()) {
+ val ex = intercept[SparkException] {
+ spark.range(100000L).crossJoin(spark.range(100000L))
+ .toDF("a", "b").agg(sum("a"), sum("b")).collect()
+ }
+ val msg = if (ex.getCause() != null) ex.getCause().getMessage() else ex.getMessage()
--- End diff --
Either this:
`org.apache.spark.SparkException: Job 43 cancelled because Stage 85 was cancelled`
or the same wrapped by this:
https://github.com/apache/spark/blob/fb3081d3b38a50aa5e023c603e1b191e57f7c876/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala#L205
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16872: [SPARK-19514] Making range interruptible.
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/16872#discussion_r100789955
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameRangeSuite.scala ---
@@ -127,4 +133,28 @@ class DataFrameRangeSuite extends QueryTest with SharedSQLContext {
}
}
}
+
+ test("Cancelling stage in a query with Range.") {
+ val listener = new SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ Thread.sleep(100)
--- End diff --
Actually there is an issue with the test case -- how do we know that we are cancelling the original tasks, and not the ones in shuffle? We need to add another flag to block the range operator from proceeding.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16872: [SPARK-19514] Making range interruptible.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16872
**[Test build #72643 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/72643/testReport)** for PR 16872 at commit [`7d1c014`](https://github.com/apache/spark/commit/7d1c014f3ac6db889173ef385b9c0b73b6d69d9f).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org