From e4e8f1a5097a1a90163705a7b99dc33e5f32d925 Mon Sep 17 00:00:00 2001
From: Burak Yavuz <brkyvz@gmail.com>
Date: Mon, 19 Sep 2016 13:31:42 -0700
Subject: [PATCH] ready for review

---
 .../execution/datasources/ListingFileCatalog.scala   | 12 ++++++++++--
 .../sql/execution/datasources/FileCatalogSuite.scala | 11 +++++++++++
 2 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ListingFileCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ListingFileCatalog.scala
index 60742bdbed204..32532084236cf 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ListingFileCatalog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ListingFileCatalog.scala
@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.execution.datasources
 
+import java.io.FileNotFoundException
+
 import scala.collection.mutable
 
 import org.apache.hadoop.fs.{FileStatus, LocatedFileStatus, Path}
@@ -97,8 +99,14 @@ class ListingFileCatalog(
         logTrace(s"Listing $path on driver")
 
         val childStatuses = {
-          val stats = fs.listStatus(path)
-          if (pathFilter != null) stats.filter(f => pathFilter.accept(f.getPath)) else stats
+          try {
+            val stats = fs.listStatus(path)
+            if (pathFilter != null) stats.filter(f => pathFilter.accept(f.getPath)) else stats
+          } catch {
+            case _: FileNotFoundException =>
+              logWarning(s"The directory $path was not found. Was it deleted very recently?")
+              Array.empty[FileStatus]
+          }
         }
 
         childStatuses.map {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileCatalogSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileCatalogSuite.scala
index 0d9ea512729bd..5c8d3226e9e26 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileCatalogSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileCatalogSuite.scala
@@ -67,4 +67,15 @@ class FileCatalogSuite extends SharedSQLContext {
 
     }
   }
+
+  test("ListingFileCatalog: folders that don't exist don't throw exceptions") {
+    withTempDir { dir =>
+      val deletedFolder = new File(dir, "deleted")
+      assert(!deletedFolder.exists())
+      val catalog1 = new ListingFileCatalog(
+        spark, Seq(new Path(deletedFolder.getCanonicalPath)), Map.empty, None)
+      // doesn't throw an exception
+      assert(catalog1.listLeafFiles(catalog1.paths).isEmpty)
+    }
+  }
 }