-
Notifications
You must be signed in to change notification settings - Fork 28.5k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[SPARK-28699][Core] Cache an indeterminate RDD could lead to incorrect result while stage rerun #25420
Conversation
(cherry picked from commit b543fab) Signed-off-by: Yuanjian Li <[email protected]>
@@ -2710,7 +2710,7 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with TimeLi | |||
assert(countSubmittedMapStageAttempts() === 2) | |||
} | |||
|
|||
test("SPARK-23207: retry all the succeeding stages when the map stage is indeterminate") { | |||
ignore("SPARK-23207: retry all the succeeding stages when the map stage is indeterminate") { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ignore this for the behavior change, as the approach now, we need to abort the stage of the current mapStage.
As we will finally support stage rerun, I suggest to skip this behavior, we can directly support the cache scenario after SPARK-25341 merged.
@@ -131,7 +131,7 @@ case class CachedRDDBuilder( | |||
|
|||
def hasNext: Boolean = rowIterator.hasNext | |||
} | |||
}.persist(storageLevel) | |||
}, isOrderSensitive = true).persist(storageLevel) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't quite get it. The isOrderSensitive
flag is used to describe the map function. Why the map function is order sensitive?
Test build #108978 has finished for PR 25420 at commit
|
@@ -86,7 +86,7 @@ case class CachedRDDBuilder( | |||
|
|||
private def buildBuffers(): RDD[CachedBatch] = { | |||
val output = cachedPlan.output | |||
val cached = cachedPlan.execute().mapPartitionsInternal { rowIterator => | |||
val cached = cachedPlan.execute().mapPartitionsWithIndexInternal({ (_, rowIterator) => |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think we only run cached plan once. I think it should be determinate? If so, this map should not be order sensitive effectively?
Retest this please. |
@xuanyuanking Can we have a test case to prevent a future regression on this? |
Test build #108993 has finished for PR 25420 at commit
|
Sorry for the late. Thanks for the comment and review! After taking a further investigation, I found the root cause for this issue is about radix sort in ShuffleExchangeExec, please review it in #25491. |
What changes were proposed in this pull request?
It's another case for the indeterminate stage/RDD rerun while stage rerun happened. In the CachedRDDBuilder, we miss tracking the
isOrderSensitive
characteristic to the newly created MapPartitionsRDD.This patch just a safeguard, if we need the support for stage rerunning, it should be done after #24892.
How was this patch tested?
Integrated test with an exception, instead of the wrong answer.