[SPARK-19537] Move pendingPartitions to ShuffleMapStage.

[kayousterhout]

The pendingPartitions instance variable should be moved to ShuffleMapStage,
because it is only used by ShuffleMapStages. This change is purely refactoring
and does not change functionality.

I fixed this in an attempt to clarify some of the discussion around #16620, which I was having trouble reasoning about. I stole the helpful comment Imran wrote for pendingPartitions and used it here.

cc @squito@markhamstra@jinxing64

[markhamstra]

Makes good sense to me.

[kayousterhout]

Thanks for the quick look @markhamstra !

[markhamstra]

You're welcome -- but do be aware that I'm going to be extremely busy with non-Spark stuff for at least the next week, so for awhile my Spark code reviews are likely to be more cursory than they should be.

[jinxing64]

It's great to have pendingPartitions in ShuffleMapStage.

[SparkQA]

Test build #72662 has finished for PR 16876 at commit 3e9af42.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[squito]

I think including all the partitions could make this log line a little too long for an info message (what if there are 1k or 10k tasks). I understand not wanting to stick it in a debug message (you often dont' realize you want debug logging on until after something has failed), but I think this will be too much for regular runs.

Would it still be useful if you truncate to the first 25 tasks? tasks.take(25).map(...)

[kayousterhout]

Oh good point. 25 seemed like maybe too many for an info-level message so I split the difference with your original comment and did 15. I could also just leave the original logging if you think that's better.

[squito]

other than the logging issue, looks good. definitely a useful cleanup.

[squito]

lgtm when tests pass

[SparkQA]

Test build #72684 has finished for PR 16876 at commit 8fdecda.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[jinxing64]

@kayousterhout
It's great to give a definition of pendingPartitions in ShuffleMapStage.
May I ask few questions and make my understanding about pendingPartitions clear ? It means:

1. PartitionId gets removed from pendingPartitions because of real success(not including ones from failed executors);
2. pendingPartitions is the exact mirror(in reverse) of ShuffleMapStage 's output locations. Thus when pendingPartition is empty, all output locations are available;
3. When executor lost, we should check if there is need to add partitionId back to pendingPartitions accordingly.
4. No need to consider resubmit stage when pendingPartitions is empty.

Is my understanding correct ? It'd great if you can help rectify my wrong points.

[kayousterhout]

@jinxing64 I wrote a long comment on your other PR to attempt to explain (my current understanding of) these issues!

[kayousterhout]

Thanks all for the review; merged this into master