[MAPR-30583] InMemoryFileIndex changed to getFileBlockLocations in parallel way (apache#221)

Author: rsotn-mapr

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InMemoryFileIndex.scala

@@ -294,10 +294,7 @@ object InMemoryFileIndex extends Logging {
       if (filter != null) allFiles.filter(f => filter.accept(f.getPath)) else allFiles
     }
 
-    val missingFiles = mutable.ArrayBuffer.empty[String]
-    val filteredLeafStatuses = allLeafStatuses.filterNot(
-      status => shouldFilterOut(status.getPath.getName))
-    val resolvedLeafStatuses = filteredLeafStatuses.flatMap {
+    allLeafStatuses.par.filterNot(status => shouldFilterOut(status.getPath.getName)).map {
       case f: LocatedFileStatus =>
         Some(f)
 
@@ -327,14 +324,8 @@ object InMemoryFileIndex extends Logging {
             missingFiles += f.getPath.toString
             None
         }
-    }
-
-    if (missingFiles.nonEmpty) {
-      logWarning(
-        s"the following files were missing during file scan:\n  ${missingFiles.mkString("\n  ")}")
-    }
-
-    resolvedLeafStatuses
+        lfs
+    }.seq
   }
 
   /** Checks if we should filter out this path name. */