You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "Ngone51 (via GitHub)" <gi...@apache.org> on 2025/01/15 07:00:42 UTC
[PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Ngone51 opened a new pull request, #49501:
URL: https://github.com/apache/spark/pull/49501
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
This PR adds a new mutable state `taskInterrupted` in `InputRDDCodegen`. The state indicates whether the task thread is interrupted or not. `InputRDDCodegen` would check the state each time when it produces a record and stops immediately if `taskInterrupted=true`. `taskInterrupted` would be updated by `TaskContext.isInterrupted()` per 1000 output rows processed.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
This facilitates the resource release when there is a task killing request.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Added unit test.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2595245623
@cloud-fan is there a way to configure the parallelize for `Range`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2608007898
Gentle ping, @Ngone51 ~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #49501:
URL: https://github.com/apache/spark/pull/49501#discussion_r1994671983
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2048,6 +2048,16 @@ object SQLConf {
.checkValue(threshold => threshold > 0, "The threshold must be a positive integer.")
.createWithDefault(1024)
+ val INPUT_CODEGEN_RDD_INTERRUPT_ON_CANCEL =
+ buildConf("spark.sql.codegen.inputCodegenRDD.interruptOnCancel")
Review Comment:
Do we have a plan to introduce more configurations under `spark.sql.codegen.inputCodegenRDD.*` namespace? If not, maybe, we had better avoid introducing this new namespace, `spark.sql.codegen.inputCodegenRDD.*`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2608825042
> Then, numPartitions of spark.range(start: Long, end: Long, step: Long, numPartitions: Int) wasn't enough?
@dongjoon-hyun Thanks, this is what I'm looking for. Let me try it first.
(The test generates 2 tasks locally but 3 in GithubAction. So I'd liket explictly set the partitions to fix the stability of the test.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #49501: [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation
URL: https://github.com/apache/spark/pull/49501
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2592791859
@Ngone51 the new test failed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2612012707
The failure is not related to `numPartitions` but job cancelation. The Github Action Spark logs shows that job `SPARK-50806` can't be found in the active job list:
```log
05:13:22.837 dag-scheduler-event-loop INFO DAGScheduler: Got map stage job 0 ($anonfun$withThreadLocalCaptured$2 at FutureTask.java:264) with 2 output partitions
05:13:22.838 dag-scheduler-event-loop INFO DAGScheduler: Final stage: ShuffleMapStage 0 ($anonfun$withThreadLocalCaptured$2 at FutureTask.java:264)
05:13:22.838 dag-scheduler-event-loop INFO DAGScheduler: Parents of final stage: List()
05:13:22.838 dag-scheduler-event-loop INFO DAGScheduler: Missing parents: List()
05:13:22.838 dag-scheduler-event-loop INFO DAGScheduler: Submitting ShuffleMapStage 0 (MapPartitionsRDD[9] at $anonfun$withThreadLocalCaptured$2 at FutureTask.java:264), which has no missing parents
05:13:22.838 dag-scheduler-event-loop INFO MemoryStore: MemoryStore started with capacity 2.4 GiB
05:13:22.839 dag-scheduler-event-loop INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 46.7 KiB, free 2.4 GiB)
05:13:22.851 dag-scheduler-event-loop INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 20.2 KiB, free 2.4 GiB)
05:13:22.851 dag-scheduler-event-loop INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1656
05:13:22.851 dag-scheduler-event-loop INFO DAGScheduler: Submitting 2 missing tasks from ShuffleMapStage 0 (MapPartitionsRDD[9] at $anonfun$withThreadLocalCaptured$2 at FutureTask.java:264) (first 15 tasks are for partitions Vector(0, 1))
05:13:22.851 dag-scheduler-event-loop INFO TaskSchedulerImpl: Adding task set 0.0 with 2 tasks resource profile 0
05:13:22.852 dispatcher-event-loop-0 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0) (localhost,executor driver, partition 0, PROCESS_LOCAL, 9968 bytes)
05:13:22.852 dispatcher-event-loop-0 INFO TaskSetManager: Starting task 1.0 in stage 0.0 (TID 1) (localhost,executor driver, partition 1, PROCESS_LOCAL, 9968 bytes)
05:13:22.852 Executor task launch worker-0 INFO Executor: Starting executor with user classpath (userClassPathFirst = false): ''
05:13:22.852 Executor task launch worker-0 INFO Executor: Using REPL class URI: spark://localhost:35095/artifacts/69020f8b-80c6-45e1-9080-16fdff425ac4/classes/
05:13:22.852 Executor task launch worker-0 INFO Executor: Created or updated repl class loader org.apache.spark.executor.ExecutorClassLoader@36057c87 for 69020f8b-80c6-45e1-9080-16fdff425ac4.
05:13:22.852 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
05:13:22.852 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO Executor: Running task 1.0 in stage 0.0 (TID 1)
05:13:22.859 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO TransportClientFactory: Successfully created connection to localhost/127.0.0.1:35095 after 0 ms (0 ms spent in bootstraps)
05:13:22.864 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO CodeGenerator: Code generated in 7.28645 ms
05:13:22.867 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO CodeGenerator: Code generated in 2.50331 ms
05:13:22.883 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO CodeGenerator: Code generated in 13.913607 ms
05:13:22.891 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO CodeGenerator: Code generated in 6.30746 ms
05:13:22.892 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO SecurityManager: Changing view acls to: runner
05:13:22.892 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO SecurityManager: Changing modify acls to: runner
05:13:22.892 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO SecurityManager: Changing view acls groups to: runner
05:13:22.892 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO SecurityManager: Changing modify acls groups to: runner
05:13:22.892 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: runner groups with view permissions: EMPTY; users with modify permissions: runner; groups with modify permissions: EMPTY; RPC SSL disabled
05:13:22.904 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO CodeGenerator: Code generated in 2.286265 ms
05:13:22.911 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO CodeGenerator: Code generated in 3.089951 ms
05:13:27.820 pool-1-thread-1-ScalaTest-running-HashAggregateCodegenInterruptionSuite INFO DAGScheduler: Asked to cancel job group SPARK-50806 with cancelFutureJobs=false
05:13:27.820 dag-scheduler-event-loop WARN DAGScheduler: Failed to cancel job group SPARK-50806. Cannot find active jobs for it.
```
However, this is the local Spark logs, where the job `SPARK-50806` is canceled succesfully:
```log
56:32.453 dag-scheduler-event-loop INFO DAGScheduler: Got map stage job 0 ($anonfun$withThreadLocalCaptured$2 at FutureTask.java:264) with 2 output partitions
15:56:32.455 dag-scheduler-event-loop INFO DAGScheduler: Final stage: ShuffleMapStage 0 ($anonfun$withThreadLocalCaptured$2 at FutureTask.java:264)
15:56:32.455 dag-scheduler-event-loop INFO DAGScheduler: Parents of final stage: List()
15:56:32.456 dag-scheduler-event-loop INFO DAGScheduler: Missing parents: List()
15:56:32.458 dag-scheduler-event-loop INFO DAGScheduler: Submitting ShuffleMapStage 0 (MapPartitionsRDD[9] at $anonfun$withThreadLocalCaptured$2 at FutureTask.java:264), which has no missing parents
15:56:32.474 dag-scheduler-event-loop INFO MemoryStore: MemoryStore started with capacity 2.4 GiB
15:56:32.482 dag-scheduler-event-loop INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 46.8 KiB, free 2.4 GiB)
15:56:33.001 dag-scheduler-event-loop INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 20.2 KiB, free 2.4 GiB)
15:56:33.008 dag-scheduler-event-loop INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1656
15:56:33.021 dag-scheduler-event-loop INFO DAGScheduler: Submitting 2 missing tasks from ShuffleMapStage 0 (MapPartitionsRDD[9] at $anonfun$withThreadLocalCaptured$2 at FutureTask.java:264) (first 15 tasks are for partitions Vector(0, 1))
15:56:33.022 dag-scheduler-event-loop INFO TaskSchedulerImpl: Adding task set 0.0 with 2 tasks resource profile 0
15:56:33.040 dispatcher-event-loop-1 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0) (192.168.1.5,executor driver, partition 0, PROCESS_LOCAL, 9970 bytes)
15:56:33.042 dispatcher-event-loop-1 INFO TaskSetManager: Starting task 1.0 in stage 0.0 (TID 1) (192.168.1.5,executor driver, partition 1, PROCESS_LOCAL, 9970 bytes)
15:56:33.047 Executor task launch worker-0 INFO Executor: Starting executor with user classpath (userClassPathFirst = false): ''
15:56:33.048 Executor task launch worker-0 INFO Executor: Using REPL class URI: spark://192.168.1.5:58016/artifacts/dd408fe5-bacf-4bc5-a02f-0a1b6a038442/classes/
15:56:33.050 Executor task launch worker-0 INFO Executor: Created or updated repl class loader org.apache.spark.executor.ExecutorClassLoader@14e50195 for dd408fe5-bacf-4bc5-a02f-0a1b6a038442.
15:56:33.052 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
15:56:33.052 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO Executor: Running task 1.0 in stage 0.0 (TID 1)
15:56:33.121 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO TransportClientFactory: Successfully created connection to /192.168.1.5:58016 after 16 ms (0 ms spent in bootstraps)
15:56:33.143 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO CodeGenerator: Code generated in 46.866584 ms
15:56:33.160 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO CodeGenerator: Code generated in 6.688333 ms
15:56:33.195 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO CodeGenerator: Code generated in 28.92075 ms
15:56:33.215 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO CodeGenerator: Code generated in 12.730916 ms
15:56:33.220 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO SecurityManager: Changing view acls to: yi.wu
15:56:33.220 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO SecurityManager: Changing modify acls to: yi.wu
15:56:33.220 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO SecurityManager: Changing view acls groups to: yi.wu
15:56:33.220 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO SecurityManager: Changing modify acls groups to: yi.wu
15:56:33.220 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: yi.wu groups with view permissions: EMPTY; users with modify permissions: yi.wu; groups with modify permissions: EMPTY; RPC SSL disabled
15:56:33.234 Executor task launch worker for task 0.0 in stage 0.0 (TID 0) INFO CodeGenerator: Code generated in 4.3335 ms
15:56:33.264 Executor task launch worker for task 1.0 in stage 0.0 (TID 1) INFO CodeGenerator: Code generated in 5.507833 ms
15:56:37.204 pool-1-thread-1-ScalaTest-running-HashAggregateCodegenInterruptionSuite INFO DAGScheduler: Asked to cancel job group SPARK-50806 with cancelFutureJobs=false
15:56:37.284 dag-scheduler-event-loop INFO TaskSchedulerImpl: Canceling stage 0
15:56:37.284 dag-scheduler-event-loop INFO TaskSchedulerImpl: Killing all running tasks in stage 0: [SPARK_JOB_CANCELLED] Job 0 cancelled part of cancelled job group SPARK-50806 SQLSTATE: XXKDA
15:56:37.286 dag-scheduler-event-loop INFO TaskSchedulerImpl: Stage 0.0 was cancelled
```
The logs looks almost the same. It's confused why the job can't be found in Github Action's case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2611636963
Maybe, could you rebase this PR to `master` branch once more, @Ngone51 ? The failure is a fixed one in the master branch.
- https://github.com/apache/spark/pull/49621
```
[info] - SPARK-29442 Set `default` mode should override the existing mode *** FAILED *** (6 milliseconds)
[info] java.lang.NoSuchFieldException: mode
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #49501:
URL: https://github.com/apache/spark/pull/49501#discussion_r1994669214
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2048,6 +2048,16 @@ object SQLConf {
.checkValue(threshold => threshold > 0, "The threshold must be a positive integer.")
.createWithDefault(1024)
+ val INPUT_CODEGEN_RDD_INTERRUPT_ON_CANCEL =
+ buildConf("spark.sql.codegen.inputCodegenRDD.interruptOnCancel")
+ .internal()
+ .doc("Whether to interrupt InputCodegenRDD on the task cancellation. Interrupt " +
+ "InputCodegenRDD would be useful to stop task and release resource quickly, " +
+ "and also reduce the task killing timeout failures by task reaper.")
+ .version("3.5.0")
Review Comment:
Could you update this value, @Ngone51 ? Maybe, `4.1.0` because this is an `Improvement`?
<img width="706" alt="Screenshot 2025-03-13 at 20 53 34" src="https://github.com/user-attachments/assets/c732cd86-17c0-4cb2-9db1-d46a071bd108" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2735543085
@dongjoon-hyun Thanks, addressed the comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2611647823
@dongjoon-hyun Thanks, will updated. But this PR's test failure still exits. I'll take a closer look.
```java
2025-01-23T05:14:27.8423862Z [0m[[0m[0minfo[0m] [0m[0m[31m- SPARK-50806: HashAggregate codegen should be interrupted on task cancellation *** FAILED *** (1 minute, 5 seconds)[0m[0m
2025-01-23T05:14:27.8426704Z [0m[[0m[0minfo[0m] [0m[0m[31m The code passed to eventually never returned normally. Attempted 3956 times over 1.0001835922166666 minutes. Last failure message: 3 did not equal 2. (HashAggregateCodegenInterruptionSuite.scala:91)[0m[0m
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2611662438
Oh. Got it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #49501:
URL: https://github.com/apache/spark/pull/49501#discussion_r1916114001
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala:
##########
@@ -485,8 +487,16 @@ trait InputRDDCodegen extends CodegenSupport {
""
}
s"""
- | while ($limitNotReachedCond $input.hasNext()) {
+ | long $numOutputRows = 0;
+ | boolean $taskInterrupted = false;
+ | while ($limitNotReachedCond !$taskInterrupted && $input.hasNext()) {
| InternalRow $row = (InternalRow) $input.next();
+ | $numOutputRows++;
+ | if ($numOutputRows % 1000 == 0) {
Review Comment:
Yea, it's by intuition. And yes we expect 1000 rows to finish quickly soon.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2596065184
Do you mean `spark.range(..)` in your test case? Then, `numPartitions` of `spark.range(start: Long, end: Long, step: Long, numPartitions: Int)` wasn't enough?
> is there a way to configure the parallelize for `Range`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #49501:
URL: https://github.com/apache/spark/pull/49501#discussion_r1916107165
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/HashAggregateCodegenInterruptionSuite.scala:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution
+
+import scala.collection.mutable
+import scala.concurrent.Future
+import scala.concurrent.duration._
+
+import org.apache.logging.log4j.Level
+
+import org.apache.spark.{SparkConf, TaskEndReason, TaskKilled}
+import org.apache.spark.scheduler.{SparkListenerJobEnd, SparkListenerTaskEnd, SparkListenerTaskStart}
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.functions.sum
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+class HashAggregateCodegenInterruptionSuite extends QueryTest with SharedSparkSession {
+
+ override def sparkConf: SparkConf = {
+ super
+ .sparkConf
+ .set("spark.task.reaper.enabled", "true")
+ .set("spark.task.reaper.killTimeout", "10s")
+ }
+
+ test("SPARK-50806: HashAggregate codegen should be interrupted on task cancellation") {
+ import testImplicits._
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> true.toString,
+ SQLConf.INTERRUPT_ON_CANCEL.key -> false.toString) {
+ var taskId = -1L
+ var isJobEnded = false
+ var taskEndReasons = new mutable.ArrayBuffer[TaskEndReason]
+ spark.sparkContext.addSparkListener(new org.apache.spark.scheduler.SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ taskId = taskStart.taskInfo.taskId
+ }
+
+ override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
+ taskEndReasons += taskEnd.reason
+ }
+
+ override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = {
+ isJobEnded = true
+ }
+ })
+
+ val logAppender = new LogAppender("")
+ withLogAppender(logAppender, level = Some(Level.INFO)) {
+ spark.sparkContext.setJobGroup("SPARK-50806", "SPARK-50806", false)
+ val slowDF = spark.range(1, 100000).rdd.mapPartitions { iter =>
Review Comment:
Could you add a comment that this test case assumption: this should be `100000`(1k) because we check it per 1k rows?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #49501:
URL: https://github.com/apache/spark/pull/49501#discussion_r1916105392
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala:
##########
@@ -485,8 +487,16 @@ trait InputRDDCodegen extends CodegenSupport {
""
}
s"""
- | while ($limitNotReachedCond $input.hasNext()) {
+ | long $numOutputRows = 0;
+ | boolean $taskInterrupted = false;
+ | while ($limitNotReachedCond !$taskInterrupted && $input.hasNext()) {
| InternalRow $row = (InternalRow) $input.next();
+ | $numOutputRows++;
+ | if ($numOutputRows % 1000 == 0) {
Review Comment:
`1000` looks like a magic number. So, we don't need to interrupt when the number of rows is less than 1000 because it will finish quickly in any way?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #49501:
URL: https://github.com/apache/spark/pull/49501#discussion_r1916107165
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/HashAggregateCodegenInterruptionSuite.scala:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution
+
+import scala.collection.mutable
+import scala.concurrent.Future
+import scala.concurrent.duration._
+
+import org.apache.logging.log4j.Level
+
+import org.apache.spark.{SparkConf, TaskEndReason, TaskKilled}
+import org.apache.spark.scheduler.{SparkListenerJobEnd, SparkListenerTaskEnd, SparkListenerTaskStart}
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.functions.sum
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+class HashAggregateCodegenInterruptionSuite extends QueryTest with SharedSparkSession {
+
+ override def sparkConf: SparkConf = {
+ super
+ .sparkConf
+ .set("spark.task.reaper.enabled", "true")
+ .set("spark.task.reaper.killTimeout", "10s")
+ }
+
+ test("SPARK-50806: HashAggregate codegen should be interrupted on task cancellation") {
+ import testImplicits._
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> true.toString,
+ SQLConf.INTERRUPT_ON_CANCEL.key -> false.toString) {
+ var taskId = -1L
+ var isJobEnded = false
+ var taskEndReasons = new mutable.ArrayBuffer[TaskEndReason]
+ spark.sparkContext.addSparkListener(new org.apache.spark.scheduler.SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ taskId = taskStart.taskInfo.taskId
+ }
+
+ override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
+ taskEndReasons += taskEnd.reason
+ }
+
+ override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = {
+ isJobEnded = true
+ }
+ })
+
+ val logAppender = new LogAppender("")
+ withLogAppender(logAppender, level = Some(Level.INFO)) {
+ spark.sparkContext.setJobGroup("SPARK-50806", "SPARK-50806", false)
+ val slowDF = spark.range(1, 100000).rdd.mapPartitions { iter =>
Review Comment:
Could you add a comment that this test case assumption: this should be `100000`(>1k) because we check it per 1k rows?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #49501:
URL: https://github.com/apache/spark/pull/49501#discussion_r1916125326
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/HashAggregateCodegenInterruptionSuite.scala:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution
+
+import scala.collection.mutable
+import scala.concurrent.Future
+import scala.concurrent.duration._
+
+import org.apache.logging.log4j.Level
+
+import org.apache.spark.{SparkConf, TaskEndReason, TaskKilled}
+import org.apache.spark.scheduler.{SparkListenerJobEnd, SparkListenerTaskEnd, SparkListenerTaskStart}
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.functions.sum
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+class HashAggregateCodegenInterruptionSuite extends QueryTest with SharedSparkSession {
+
+ override def sparkConf: SparkConf = {
+ super
+ .sparkConf
+ .set("spark.task.reaper.enabled", "true")
+ .set("spark.task.reaper.killTimeout", "10s")
+ }
+
+ test("SPARK-50806: HashAggregate codegen should be interrupted on task cancellation") {
+ import testImplicits._
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> true.toString,
+ SQLConf.INTERRUPT_ON_CANCEL.key -> false.toString) {
+ var taskId = -1L
+ var isJobEnded = false
+ var taskEndReasons = new mutable.ArrayBuffer[TaskEndReason]
+ spark.sparkContext.addSparkListener(new org.apache.spark.scheduler.SparkListener {
+ override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+ taskId = taskStart.taskInfo.taskId
+ }
+
+ override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
+ taskEndReasons += taskEnd.reason
+ }
+
+ override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = {
+ isJobEnded = true
+ }
+ })
+
+ val logAppender = new LogAppender("")
+ withLogAppender(logAppender, level = Some(Level.INFO)) {
+ spark.sparkContext.setJobGroup("SPARK-50806", "SPARK-50806", false)
+ val slowDF = spark.range(1, 100000).rdd.mapPartitions { iter =>
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50806][SQL] Support InputRDDCodegen interruption on task cancellation [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49501:
URL: https://github.com/apache/spark/pull/49501#issuecomment-2723111433
@cloud-fan @dongjoon-hyun I have updated the PR. Could you take another look? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org