[SPARK-46480][CORE][SQL] Fix NPE when table cache task attempt #44445
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
yaooqinn
reviewed
Dec 21, 2023
core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala
Outdated
Show resolved
Hide resolved
|
which variable can be null? |
|
@cloud-fan according to the stack,the |
mridulm
reviewed
Dec 21, 2023
HyukjinKwon
reviewed
Dec 22, 2023
| /** | ||
| * Returns true if the task has failed. | ||
| */ | ||
| def isFailed(): Boolean |
There was a problem hiding this comment.
Would be great if we can add this into PySpark side as well ...
There was a problem hiding this comment.
It seems python taskcontext does not have isCompleted and isInterrupted method. Is there any reason ?
There was a problem hiding this comment.
We should have them all :-).
There was a problem hiding this comment.
I don't mind doing separately in a differnet PR
There was a problem hiding this comment.
I will create a new pr for it later
There was a problem hiding this comment.
@HyukjinKwon does pyspark support update status from jvm side ? It seems we can only send a snapshot status to python side ?
cloud-fan
reviewed
Dec 22, 2023
| @@ -275,6 +275,8 @@ private[spark] class TaskContextImpl( | |||
| @GuardedBy("this") | |||
| override def isCompleted(): Boolean = synchronized(completed) | |||
|
|
|||
| override def isFailed(): Boolean = synchronized(failureCauseOpt.isDefined) | |||
|
|
|||
| override def isInterrupted(): Boolean = reasonIfKilled.isDefined | |||
There was a problem hiding this comment.
I'm trying to reason about the relationship between these 3 flags:
isCompletedis false only when the task is interrupted? Or when the task is still running?isFailedis true only if the task completes and fails?isInterruptedcan be true before the task completes?
There was a problem hiding this comment.
isCompleted includes all the state success, failure and cancellation. If isCompleted is true, then the task won't be in running any more.
ulysses-you
force-pushed
the
fix-cache
branch
from
f853fbd
to
5273fb8
Compare
|
@mridulm yes, to make sure the cached rdd is materialized after all the tasks are succeeded |
cloud-fan
approved these changes
Dec 22, 2023
|
Thanks, merged to master according to the affect versions in JIRA |
ulysses-you
added a commit
to ulysses-you/spark
that referenced
this pull request
Dec 22, 2023
This pr adds a check: we only mark the cached partition is materialized if the task is not failed and not interrupted. And adds a new method `isFailed` in `TaskContext`. Before this pr, when do cache, task failure can cause NPE in other tasks ``` java.lang.NullPointerException at java.nio.ByteBuffer.wrap(ByteBuffer.java:396) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificColumnarIterator.accessors1$(Unknown Source) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificColumnarIterator.hasNext(Unknown Source) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458) at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:155) at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) at org.apache.spark.scheduler.Task.run(Task.scala:131) at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:497) ``` yes, it's a bug fix add test no Closes apache#44445 from ulysses-you/fix-cache. Authored-by: ulysses-you <ulyssesyou18@gmail.com> Signed-off-by: Kent Yao <yao@apache.org>
ulysses-you
added a commit
that referenced
this pull request
Dec 22, 2023
This pr backports #44445 for branch-3.5 ### What changes were proposed in this pull request? This pr adds a check: we only mark the cached partition is materialized if the task is not failed and not interrupted. And adds a new method `isFailed` in `TaskContext`. ### Why are the changes needed? Before this pr, when do cache, task failure can cause NPE in other tasks ``` java.lang.NullPointerException at java.nio.ByteBuffer.wrap(ByteBuffer.java:396) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificColumnarIterator.accessors1$(Unknown Source) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificColumnarIterator.hasNext(Unknown Source) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458) at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:155) at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) at org.apache.spark.scheduler.Task.run(Task.scala:131) at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:497) ``` ### Does this PR introduce _any_ user-facing change? yes, it's a bug fix ### How was this patch tested? add test ### Was this patch authored or co-authored using generative AI tooling? no Closes #44457 from ulysses-you/fix-cache-3.5. Authored-by: ulysses-you <ulyssesyou18@gmail.com> Signed-off-by: youxiduo <youxiduo@corp.netease.com>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
This pr adds a check: we only mark the cached partition is materialized if the task is not failed and not interrupted. And adds a new method
isFailedinTaskContext.Why are the changes needed?
Before this pr, when do cache, task failure can cause NPE in other tasks
Does this PR introduce any user-facing change?
yes, it's a bug fix
How was this patch tested?
add test
Was this patch authored or co-authored using generative AI tooling?
no