[SPARK-4961] [CORE] Put HadoopRDD.getPartitions forward to reduce DAGScheduler.JobSubmitted processing time

core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala

@@ -190,7 +190,8 @@ class HadoopRDD[K, V](
     newInputFormat
   }
 
-  override def getPartitions: Array[Partition] = {
+  protected def getThesePartitions() : Array[Partition] = {
+    val start = System.nanoTime
     val jobConf = getJobConf()
     // add the credentials here as this can be called before SparkContext initialized
     SparkHadoopUtil.get.addCredentials(jobConf)
@@ -203,9 +204,27 @@ class HadoopRDD[K, V](
     for (i <- 0 until inputSplits.size) {
       array(i) = new HadoopPartition(id, i, inputSplits(i))
     }
+    logDebug("Get these partitions took %f s".format((System.nanoTime - start) / 1e9))
     array
   }
 
+  @transient private var thesePartitions_ : Array[Partition] = {
+    try {
+      getThesePartitions()
+    } catch {
+      case e: Exception => 
+        logDebug("Error initializing HadoopRDD's partitions", e)
+        null
+    }
+  }
+
+  override def getPartitions: Array[Partition] = {
+    if (thesePartitions_ == null) {
+      thesePartitions_ = getThesePartitions()
+    }
+    thesePartitions_
+  }
+
   override def compute(theSplit: Partition, context: TaskContext): InterruptibleIterator[(K, V)] = {
     val iter = new NextIterator[(K, V)] {