Hi. Ok . did the final checkin. Pls feel free to review. Regards Asif
On Sat, Feb 15, 2025 at 6:42 PM Asif Shahid <asif.sha...@gmail.com> wrote: > Pls hold on reviewing the patch, as I need to do one more checkin. > I have still left a window of race , by releasing the read lock early, for > the case of first task being successful, and the failing task has yet to > acquire the write lock. > Regards > Asif > > On Thu, Feb 13, 2025 at 10:48 PM Asif Shahid <asif.sha...@gmail.com> > wrote: > >> Hi Wenchen, >> Apologies... I thought it was Santosh asking for the PR. and race >> condition... >> Yes the race condition causes only some partitions to be retried .. >> May be a checksum can be used as fallback mechanism? >> I suppose checksum would be implemented on executor sides..? Then there >> might be condition where shuffle files could be lost before >> driver/executors are communicated checksum ? >> Regards >> Asif >> >> >> On Thu, Feb 13, 2025 at 7:39 PM Asif Shahid <asif.sha...@gmail.com> >> wrote: >> >>> The bugrepro patch , when applied on current master, will show failure >>> with incorrect results. >>> While on the PR branch , it will pass. >>> The number of iterations in the test is 100. >>> >>> Regards >>> Asif >>> >>> On Thu, Feb 13, 2025 at 7:35 PM Asif Shahid <asif.sha...@gmail.com> >>> wrote: >>> >>>> Hi, >>>> Following up on this issue. >>>> The PR opened is: >>>> 1)https://github.com/apache/spark/pull/49708 >>>> The PR IMO fixes the 3 issues which I have described below. >>>> The race condition fix requires use of read write lock. >>>> >>>> 2) Jira is: >>>> https://issues.apache.org/jira/browse/SPARK-51016 >>>> >>>> A checksum approach would result in greater number of retries, I >>>> believe: >>>> There is already mechanism in Expression to identify indeterminism >>>> >>>> Anyways, the issues which I think are , multiple, which the opened PR >>>> addresses. >>>> 1) The stage.isIndeterminate is buggy as it is not able to identify >>>> inDeterminate, and the fix spans some files like ShuffleDependency, RDD >>>> etc. >>>> 2) The race condition in DagScheduler: >>>> a) Say a task1 corresponding to Partition P1 of an inDeterminate >>>> Stage Result fails with Fetch Failure >>>> b) The code rightly checks if no previous partitions of result stage >>>> are successful yet, instead of throwing Exception by aborting , it decides >>>> to retry. >>>> c) During this window of retry, to check for dependencies and >>>> making a new attempt initiated by Task1, Task2 corresponding to Partition 2 >>>> of result stage becomes successful. >>>> d) The new attempt, now sees instead of all missing partitions, sees >>>> n -1. and that creates Data corruption. as it is no longer retrying on >>>> all partitions. >>>> >>>> 3) The DagSchedulerSuite's test >>>> a) test("SPARK-25341: retry all the succeeding stages when the >>>> map stage is indeterminate") >>>> has following assertion: >>>> >>>> assert(failedStages.collect { >>>> case stage: ShuffleMapStage if stage.shuffleDep.shuffleId == >>>> shuffleId2 => stage >>>> }.head.findMissingPartitions() == Seq(0)) >>>> >>>> If my understanding is right and as seen from the outcome of the opened >>>> PR, as the ShuffleMapSage is turning out to be inDeterminate, all tasks >>>> should be retried: >>>> so the assertion should be , as expected value Seq(0, 1) >>>> >>>> b) If the above #a is valid, then the second test >>>> test("SPARK-32923: handle stage failure for indeterminate map >>>> stage with push-based shuffle") { >>>> which has assertion: >>>> >>>> assert(failedStages.collect { >>>> case stage: ShuffleMapStage if stage.shuffleDep.shuffleId == >>>> shuffleId2 => stage >>>> }.head.findMissingPartitions() == Seq(0)) >>>> >>>> should also be instead, Seq(0, 1) >>>> >>>> >>>> I am also attaching a patch which I was using a basis to debug and fix. >>>> The patch modifies the source code by using inline code interception >>>> and tweaks to reproduce the issue of data corruption using a unit test. >>>> >>>> And I am also attaching the BugTest which I used. >>>> >>>> The test is supposed to pass if either it encounters exception from >>>> spark layer, due to exception >>>> "Please eliminate the ndeterminacy by checkpointing the RDD before >>>> repartition and try again" >>>> or if the results of retry are as expected ( 12 rows along witrh data >>>> correctness check) >>>> >>>> Meanwhile I will try to productize the bug test. >>>> >>>> Would appreciate clarification on Test issue #3. And a review of PR, if >>>> done, would be appreciate much more. >>>> >>>> Sanotsh, >>>> I believe my PR would do a full partition retry ( if the Fetch failed >>>> happend prior to any other final task being successful, else it will abort >>>> query throwing Exception). No wrong result. I believe. >>>> Your review will be much appreciated. >>>> Regards >>>> Asif >>>> >>>> >>>> On Thu, Feb 13, 2025 at 4:54 PM Wenchen Fan <cloud0...@gmail.com> >>>> wrote: >>>> >>>>> > Have identified a race condition , in the current DagSchdeduler code >>>>> base >>>>> >>>>> Does it lead to the failure of full-stage retry? Can you also share >>>>> your PR so that we can take a closer look? >>>>> >>>>> BTW, I'm working with my colleagues on this runtime checksum idea. We >>>>> have something similar in our internal infra and I think we can create a >>>>> working solution shortly. >>>>> >>>>> On Thu, Feb 13, 2025 at 12:53 PM Santosh Pingale < >>>>> santosh.ping...@adyen.com> wrote: >>>>> >>>>>> > Maybe we should do it at runtime: if Spark retries a shuffle stage >>>>>> but the data becomes different (e.g. use checksum to check it), then >>>>>> Spark >>>>>> should retry all the partitions of this stage. >>>>>> >>>>>> Having bitten hard recently by this behavior in spark and after >>>>>> having gone down the rabbit hole to investigate, we should definitely try >>>>>> to refine and implement this approach. The non-deterministicness is >>>>>> dependent on several factors, including datatypes of the columns or use >>>>>> of >>>>>> repartition; it is not easy for the users the reason about this behavior. >>>>>> >>>>>> PS: The JIRA that I found and helped me confirm my understanding of >>>>>> the problem, plus educate internal users and platform team: >>>>>> https://issues.apache.org/jira/browse/SPARK-38388. Checksum approach >>>>>> was brought up in that JIRA too and I feel that is the balanced way to >>>>>> look >>>>>> at this problem. >>>>>> >>>>>> Kind regards >>>>>> Santosh >>>>>> >>>>>> On Thu, Feb 13, 2025, 12:41 AM Asif Shahid <asif.sha...@gmail.com> >>>>>> wrote: >>>>>> >>>>>>> Hi, >>>>>>> Following up on the issue with some information: >>>>>>> 1) have opened a PR related to fix the isInDeterminate method in >>>>>>> Stage , RDD etc. >>>>>>> 2) After tweaking the source code , have been able to reproduce the >>>>>>> data loss issue reliably, but cannot productize the test, as it >>>>>>> requires a >>>>>>> lot of inline interception and coaxing to induce test failure in a >>>>>>> single >>>>>>> VM unit test. will post a diff of the source code patch and bug test. >>>>>>> 3) Have identified a race condition , in the current >>>>>>> DagSchdeduler code base: >>>>>>> In the current DagScheduler code, function >>>>>>> *handleTaskCompletion, * when a task fails, while resubmitting the >>>>>>> failure as ResubmitFailedStage, prior to submitting, it checks if the >>>>>>> shuffle is inDeterminate. If it is , then stage is submitted ONLY IF >>>>>>> the >>>>>>> ResultStage has ZERO partitions processed. Which is right, But it is >>>>>>> not a >>>>>>> atomic operation till the point , where missingPartitions from stage is >>>>>>> invoked. As a result it is possible that one of the task from the >>>>>>> current >>>>>>> attempt, marks one of the partition as successful, and while the new >>>>>>> attempt is made, instead of getting missing partitions as total >>>>>>> partitions, it fetches only some of them. >>>>>>> Will be modifying the PR to fix the race. >>>>>>> Regards >>>>>>> Asif >>>>>>> >>>>>>> On Sun, Jan 26, 2025 at 11:19 PM Asif Shahid <asif.sha...@gmail.com> >>>>>>> wrote: >>>>>>> >>>>>>>> Shouldn't it be possible to determine with static data , if output >>>>>>>> will be deterministic ?. Expressions already have deterministic flag. >>>>>>>> So >>>>>>>> when an attribute is created from alias, it will be possible to know if >>>>>>>> attribute is pointing to an inDeterminate component. >>>>>>>> >>>>>>>> On Sun, Jan 26, 2025, 11:09 PM Wenchen Fan <cloud0...@gmail.com> >>>>>>>> wrote: >>>>>>>> >>>>>>>>> It looks like a hard problem to statically analyze the query plan >>>>>>>>> and decide whether a Spark stage is deterministic or not. When I added >>>>>>>>> RDD DeterministicLevel, I thought it was not a hard problem for the >>>>>>>>> callers >>>>>>>>> to specify it, but seems I was wrong. >>>>>>>>> >>>>>>>>> Maybe we should do it at runtime: if Spark retries a shuffle stage >>>>>>>>> but the data becomes different (e.g. use checksum to check it), then >>>>>>>>> Spark >>>>>>>>> should retry all the partitions of this stage. I'll look into this >>>>>>>>> repro >>>>>>>>> after I'm back from the national holiday. >>>>>>>>> >>>>>>>>> On Mon, Jan 27, 2025 at 2:16 PM Asif Shahid <asif.sha...@gmail.com> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>>> >>>>>>>>>> I am using below test. as a unit test though , it will pass, as >>>>>>>>>> to simulate executor lost in a single vm is difficult, but there is >>>>>>>>>> definitely a bug. >>>>>>>>>> Using debugger if you check, the ShuffleStage.isDeterminate is >>>>>>>>>> turning out to be true, though it clearly should not be. >>>>>>>>>> As result if you look at DagScheduler, TaskSchedulerImpl and >>>>>>>>>> TaskSet code, it will not retry the ShuffleStage fully, instead >>>>>>>>>> would only >>>>>>>>>> retry missing task. >>>>>>>>>> Which means on retry if the join used a random value which puts >>>>>>>>>> it in already completed shuffle task, the results will be missed. >>>>>>>>>> >>>>>>>>>>> package org.apache.spark.sql.vectorized >>>>>>>>>>> >>>>>>>>>>> import org.apache.spark.rdd.ZippedPartitionsRDD2 >>>>>>>>>>> import org.apache.spark.sql.{DataFrame, Encoders, QueryTest} >>>>>>>>>>> import org.apache.spark.sql.catalyst.expressions.Literal >>>>>>>>>>> import org.apache.spark.sql.execution.datasources.FileScanRDD >>>>>>>>>>> import org.apache.spark.sql.functions._ >>>>>>>>>>> import org.apache.spark.sql.internal.SQLConf >>>>>>>>>>> import org.apache.spark.sql.test.SharedSparkSession >>>>>>>>>>> import org.apache.spark.sql.types.LongType >>>>>>>>>>> >>>>>>>>>>> class BugTest extends QueryTest with SharedSparkSession { >>>>>>>>>>> import testImplicits._ >>>>>>>>>>> /* test("no retries") { >>>>>>>>>>> withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { >>>>>>>>>>> val baseDf = spark.createDataset( >>>>>>>>>>> Seq((1L, "a"), (2L, "b"), (3L, "c"), (null, "w"), (null, >>>>>>>>>>> "x"), (null, "y"), (null, "z")))( >>>>>>>>>>> Encoders.tupleEncoder(Encoders.LONG, >>>>>>>>>>> Encoders.STRING)).toDF("pkLeft", "strleft") >>>>>>>>>>> >>>>>>>>>>> val leftOuter = baseDf.select( >>>>>>>>>>> $"strleft", when(isnull($"pkLeft"), >>>>>>>>>>> monotonically_increasing_id() + Literal(100)). >>>>>>>>>>> otherwise($"pkLeft").as("pkLeft")) >>>>>>>>>>> leftOuter.show(10000) >>>>>>>>>>> >>>>>>>>>>> val innerRight = spark.createDataset( >>>>>>>>>>> Seq((1L, "11"), (2L, "22"), (3L, "33")))( >>>>>>>>>>> Encoders.tupleEncoder(Encoders.LONG, >>>>>>>>>>> Encoders.STRING)).toDF("pkRight", "strright") >>>>>>>>>>> >>>>>>>>>>> val innerjoin = leftOuter.join(innerRight, $"pkLeft" === >>>>>>>>>>> $"pkRight", "inner") >>>>>>>>>>> >>>>>>>>>>> innerjoin.show(1000) >>>>>>>>>>> >>>>>>>>>>> val outerjoin = leftOuter.join(innerRight, $"pkLeft" === >>>>>>>>>>> $"pkRight", "left_outer") >>>>>>>>>>> >>>>>>>>>>> outerjoin.show(1000) >>>>>>>>>>> } >>>>>>>>>>> } */ >>>>>>>>>>> >>>>>>>>>>> test("with retries") { >>>>>>>>>>> withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { >>>>>>>>>>> withTable("outer", "inner") { >>>>>>>>>>> createBaseTables() >>>>>>>>>>> val outerjoin: DataFrame = getOuterJoinDF >>>>>>>>>>> >>>>>>>>>>> println("Initial data") >>>>>>>>>>> outerjoin.show(1000) >>>>>>>>>>> val correctRows = outerjoin.collect() >>>>>>>>>>> for( i <- 0 until 100) { >>>>>>>>>>> FileScanRDD.throwException = false >>>>>>>>>>> ZippedPartitionsRDD2.throwException = true >>>>>>>>>>> val rowsAfterRetry = getOuterJoinDF.collect() >>>>>>>>>>> import scala.jdk.CollectionConverters._ >>>>>>>>>>> val temp = >>>>>>>>>>> spark.createDataFrame(rowsAfterRetry.toSeq.asJava, outerjoin.schema) >>>>>>>>>>> println("after retry data") >>>>>>>>>>> temp.show(1000) >>>>>>>>>>> assert(correctRows.length == rowsAfterRetry.length) >>>>>>>>>>> val retriedResults = rowsAfterRetry.toBuffer >>>>>>>>>>> correctRows.foreach(r => { >>>>>>>>>>> val index = retriedResults.indexWhere(x => >>>>>>>>>>> r.getString(0) == x.getString(0) && >>>>>>>>>>> (r.getLong(1) == x.getLong(1) || (x.isNullAt(2) && >>>>>>>>>>> r.isNullAt(2) && >>>>>>>>>>> x.isNullAt(3) && r.isNullAt(3))) && >>>>>>>>>>> ((r.isNullAt(2) && x.isNullAt(2)) || r.getLong(2) >>>>>>>>>>> == x.getLong(2)) && >>>>>>>>>>> ((r.isNullAt(3) && x.isNullAt(3)) || r.getString(3) >>>>>>>>>>> == x.getString(3))) >>>>>>>>>>> assert(index >= 0) >>>>>>>>>>> retriedResults.remove(index) >>>>>>>>>>> } >>>>>>>>>>> ) >>>>>>>>>>> assert(retriedResults.isEmpty) >>>>>>>>>>> } >>>>>>>>>>> >>>>>>>>>>> // Thread.sleep(10000000) >>>>>>>>>>> } >>>>>>>>>>> } >>>>>>>>>>> } >>>>>>>>>>> >>>>>>>>>>> private def createBaseTables(): Unit = { >>>>>>>>>>> /*val outerDf = spark.createDataset( >>>>>>>>>>> Seq((1L, "aa"), (null, "aa"), (2L, "bb"), (null, "bb"), (3L, >>>>>>>>>>> "cc"), (null, "cc")))( >>>>>>>>>>> Encoders.tupleEncoder(Encoders.LONG, >>>>>>>>>>> Encoders.STRING)).toDF("pkLeft", "strleft") >>>>>>>>>>> outerDf.write.format("parquet").saveAsTable("outer")*/ >>>>>>>>>>> >>>>>>>>>>> val outerDf = spark.createDataset( >>>>>>>>>>> Seq((1L, "aa"), (null, "aa"), (2L, "aa"), (null, "bb"), (3L, >>>>>>>>>>> "bb"), (null, "bb")))( >>>>>>>>>>> Encoders.tupleEncoder(Encoders.LONG, >>>>>>>>>>> Encoders.STRING)).toDF("pkLeftt", "strleft") >>>>>>>>>>> >>>>>>>>>>> outerDf.write.format("parquet").partitionBy("strleft").saveAsTable("outer") >>>>>>>>>>> >>>>>>>>>>> /*val innerDf = spark.createDataset( >>>>>>>>>>> Seq((1L, "11"), (2L, "22"), (3L, "33")))( >>>>>>>>>>> Encoders.tupleEncoder(Encoders.LONG, >>>>>>>>>>> Encoders.STRING)).toDF("pkRight", "strright")*/ >>>>>>>>>>> >>>>>>>>>>> val innerDf = spark.createDataset( >>>>>>>>>>> Seq((1L, "11"), (2L, "11"), (3L, "33")))( >>>>>>>>>>> Encoders.tupleEncoder(Encoders.LONG, >>>>>>>>>>> Encoders.STRING)).toDF("pkRight", "strright") >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> innerDf.write.format("parquet").partitionBy("strright").saveAsTable("inner") >>>>>>>>>>> >>>>>>>>>>> val innerInnerDf = spark.createDataset( >>>>>>>>>>> Seq((1L, "111"), (2L, "222"), (3L, "333")))( >>>>>>>>>>> Encoders.tupleEncoder(Encoders.LONG, >>>>>>>>>>> Encoders.STRING)).toDF("pkpkRight", "strstrright") >>>>>>>>>>> >>>>>>>>>>> // innerInnerDf.write.format("parquet").saveAsTable("innerinner") >>>>>>>>>>> } >>>>>>>>>>> >>>>>>>>>>> private def getOuterJoinDF = { >>>>>>>>>>> val leftOuter = spark.table("outer").select( >>>>>>>>>>> $"strleft", when(isnull($"pkLeftt"), floor(rand() * >>>>>>>>>>> Literal(10000000L)). >>>>>>>>>>> cast(LongType)). >>>>>>>>>>> otherwise($"pkLeftt").as("pkLeft")) >>>>>>>>>>> >>>>>>>>>>> val innerRight = spark.table("inner") >>>>>>>>>>> // val innerinnerRight = spark.table("innerinner") >>>>>>>>>>> >>>>>>>>>>> val outerjoin = leftOuter.hint("shuffle_hash"). >>>>>>>>>>> join(innerRight, $"pkLeft" === $"pkRight", "left_outer") >>>>>>>>>>> outerjoin >>>>>>>>>>> /* >>>>>>>>>>> val outerOuterJoin = outerjoin.hint("shuffle_hash"). >>>>>>>>>>> join(innerinnerRight, $"pkLeft" === $"pkpkRight", >>>>>>>>>>> "left_outer") >>>>>>>>>>> outerOuterJoin >>>>>>>>>>> >>>>>>>>>>> */ >>>>>>>>>>> } >>>>>>>>>>> } >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>> On Sun, Jan 26, 2025 at 10:05 PM Asif Shahid < >>>>>>>>>> asif.sha...@gmail.com> wrote: >>>>>>>>>> >>>>>>>>>>> Sure. I will send prototypical query tomorrow. Though its >>>>>>>>>>> difficult to simulate issue using unit test , but I think the issue >>>>>>>>>>> is >>>>>>>>>>> Rdd.isIndeterminate is not returning true for the query. As a >>>>>>>>>>> result, on retry, the shuffle stage is not reattempted fully. >>>>>>>>>>> And rdd is not returning inDeterminate as true , is due to >>>>>>>>>>> ShuffleRdd not taking into account , as the ShuffleDependency is >>>>>>>>>>> not taking >>>>>>>>>>> into account inDeterminate nature of HashPartitioner. >>>>>>>>>>> Moreover the attribute ref provided to HashPartitioner though is >>>>>>>>>>> pointing to an inDeterminate alias, is having deterministic flag as >>>>>>>>>>> true ( >>>>>>>>>>> which in itself is logically correct). >>>>>>>>>>> So I feel apart from ShuffleDependency code change, all >>>>>>>>>>> expressions should have a lazy Boolean say containsIndeterministic >>>>>>>>>>> component. Which will be true if the expression is indeterministic >>>>>>>>>>> or >>>>>>>>>>> contains any attribute ref which has is indeterministicComponent >>>>>>>>>>> true. >>>>>>>>>>> >>>>>>>>>>> And on personal note.. thanks for your interest.. this is very >>>>>>>>>>> rare attitude. >>>>>>>>>>> Regards >>>>>>>>>>> Asif >>>>>>>>>>> >>>>>>>>>>> On Sun, Jan 26, 2025, 9:45 PM Ángel < >>>>>>>>>>> angel.alvarez.pas...@gmail.com> wrote: >>>>>>>>>>> >>>>>>>>>>>> Hi Asif, >>>>>>>>>>>> >>>>>>>>>>>> Could you provide an example (code+dataset) to analize this? >>>>>>>>>>>> Looks interesting ... >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> Regards, >>>>>>>>>>>> Ángel >>>>>>>>>>>> >>>>>>>>>>>> El dom, 26 ene 2025 a las 20:58, Asif Shahid (< >>>>>>>>>>>> asif.sha...@gmail.com>) escribió: >>>>>>>>>>>> >>>>>>>>>>>>> Hi, >>>>>>>>>>>>> On further thoughts, I concur that leaf expressions like >>>>>>>>>>>>> AttributeRefs can always be considered to be deterministic, as , >>>>>>>>>>>>> as a java >>>>>>>>>>>>> variable the value contained in it per iteration is invariant ( >>>>>>>>>>>>> except when >>>>>>>>>>>>> changed by some deterministic logic). So in that sense what I >>>>>>>>>>>>> said in the >>>>>>>>>>>>> above mail as that an issue is incorrect. >>>>>>>>>>>>> But I think that AttributeRef should have a boolean method >>>>>>>>>>>>> which tells, whether the value it represents is from an >>>>>>>>>>>>> indeterminate >>>>>>>>>>>>> source or not. >>>>>>>>>>>>> Regards >>>>>>>>>>>>> Asif >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> On Fri, Jan 24, 2025 at 5:18 PM Asif Shahid < >>>>>>>>>>>>> asif.sha...@gmail.com> wrote: >>>>>>>>>>>>> >>>>>>>>>>>>>> Hi, >>>>>>>>>>>>>> While testing a use case where the query had an outer join >>>>>>>>>>>>>> such that joining key of left outer table either had a valid >>>>>>>>>>>>>> value or a >>>>>>>>>>>>>> random value( salting to avoid skew). >>>>>>>>>>>>>> The case was reported to have incorrect results in case of >>>>>>>>>>>>>> node failure, with retry. >>>>>>>>>>>>>> On debugging the code, have found following, which has left >>>>>>>>>>>>>> me confused as to what is spark's strategy for indeterministic >>>>>>>>>>>>>> fields. >>>>>>>>>>>>>> Some serious issues are : >>>>>>>>>>>>>> 1) All the leaf expressions, like AttributeReference are >>>>>>>>>>>>>> always considered deterministic. Which means if an attribute is >>>>>>>>>>>>>> pointing to >>>>>>>>>>>>>> an Alias which itself is indeterministic, the attribute will >>>>>>>>>>>>>> still be >>>>>>>>>>>>>> considered deterministic >>>>>>>>>>>>>> 2) In CheckAnalysis there is code which checks whether each >>>>>>>>>>>>>> Operator either supports indeterministic value or not . Join is >>>>>>>>>>>>>> not >>>>>>>>>>>>>> included in the list of supported, but it passes even if the >>>>>>>>>>>>>> joining key is >>>>>>>>>>>>>> pointing to an indeterministic alias. ( When I tried fixing it, >>>>>>>>>>>>>> found a >>>>>>>>>>>>>> plethora of operators failing Like DeserializedObject, >>>>>>>>>>>>>> LocalRelation etc >>>>>>>>>>>>>> which are not supposed to contain indeterministic attributes ( >>>>>>>>>>>>>> because they >>>>>>>>>>>>>> are not in the list of supporting operators). >>>>>>>>>>>>>> 3) The ShuffleDependency does not check for indeterministic >>>>>>>>>>>>>> nature of partitioner ( fixed it locally and then realized that >>>>>>>>>>>>>> there is >>>>>>>>>>>>>> the bug #1 which needs to be fixed too). >>>>>>>>>>>>>> >>>>>>>>>>>>>> The code in DagScheduler / TaskSet, TaskScheduler etc, seems >>>>>>>>>>>>>> to have been written , keeping in mind the indeterministic >>>>>>>>>>>>>> nature of the >>>>>>>>>>>>>> previous and current stages , so as to rexecute previous stages >>>>>>>>>>>>>> as a whole, >>>>>>>>>>>>>> instead of just missing tasks, but the above 3 points, do not >>>>>>>>>>>>>> seem to >>>>>>>>>>>>>> support the code of DagScheduler / TaskScheduler. >>>>>>>>>>>>>> >>>>>>>>>>>>>> Regards >>>>>>>>>>>>>> Asif >>>>>>>>>>>>>> >>>>>>>>>>>>>>
