[CARMEL-6061] Skip the storage detection for access-denied paths (#1013)

Author: fenzhu

sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala

@@ -245,30 +245,54 @@ case class DescribeDatabaseCommand(
     val databasePath = new Path(dbLocation)
     val fs = databasePath.getFileSystem(hadoopConf)
     val dbPath = fs.makeQualified(databasePath)
+    val tableList = catalog.listTables(databaseName)
+    val subDirList = fs.listStatus(dbPath)
+
+    val allSubStatus = subDirList.map(fss => (fss, getLen(fss, fs))).sortWith {
+      (t1, t2) => t1._2 > t2._2
+    }.partition(_._2 >= 0)
+    val subStatus = allSubStatus._1
+    val abnormalSubStatus = allSubStatus._2
+
     result += Row("## Database Storage Summary", "")
-    val summary = fs.getContentSummary(dbPath)
-    val subStatus = fs.listStatus(dbPath).sortWith {
-      (fss1, fss2) => getLen(fss1, fs) > getLen(fss2, fs)
-    }
-    val (spaceQuotaStr, spaceQuotaRemStr) = if (summary.getSpaceQuota > 0L) {
-      (formatSize(summary.getSpaceQuota),
-        formatSize(summary.getSpaceQuota - summary.getSpaceConsumed))
-    } else {
-      ("none", "inf")
+    val (spaceQuotaStr, spaceQuotaRemStr) = try {
+      val summary = fs.getContentSummary(dbPath)
+      if (summary.getSpaceQuota > 0L) {
+        (formatSize(summary.getSpaceQuota),
+          formatSize(summary.getSpaceQuota - summary.getSpaceConsumed))
+      } else {
+        ("none", "inf")
+      }
+    } catch {
+      case _: Exception => ("unknown (sub-directory access denied)", "unknown")
     }
-    result += Row("space quota", spaceQuotaStr)
+
+    result += Row("total space quota", spaceQuotaStr)
     result += Row("remaining space quota", spaceQuotaRemStr)
-    result += Row("total tables", String.valueOf(subStatus.length))
+    result += Row("total directories", String.valueOf(subDirList.length))
+    result += Row("total tables", String.valueOf(tableList.length))
 
     result += Row("", "")
+
     val k = Math.min(15, subStatus.length)
     result += Row(s"## Top $k Directories", "Size")
     for (i <- 0 until k) {
-      val len = getLen(subStatus(i), fs)
-      result += Row(subStatus(i).getPath.toString, formatSize(len))
+      val len = subStatus(i)._2
+      result += Row(subStatus(i)._1.getPath.toString, formatSize(len))
     }
 
-    val tableLocations = catalog.listTables(databaseName).map { t =>
+    if (allSubStatus._2.nonEmpty) {
+      result += Row("", "")
+      val ak = Math.min(15, abnormalSubStatus.length)
+      result += Row(s"## Top $ak Abnormal Directories (e.g., Access Denied)", "Size")
+      for (i <- 0 until ak) {
+        val len = abnormalSubStatus(i)._2
+        result += Row(abnormalSubStatus(i)._1.getPath.toString, formatSize(len))
+      }
+    }
+
+    result += Row("", "")
+    val tableLocations = tableList.map { t =>
       val meta = catalog.getTableMetadata(t)
       val loc = if (meta.storage.locationUri.isEmpty) {
         ""
@@ -280,10 +304,9 @@ case class DescribeDatabaseCommand(
     }.filter(_.nonEmpty).toSet
 
     val leakStatus = subStatus.filterNot { s =>
-      tableLocations.contains(fs.makeQualified(s.getPath).toString)
+      tableLocations.contains(fs.makeQualified(s._1.getPath).toString)
     }
 
-    result += Row("", "")
     val message = if (clear) {
       "try to clear"
     } else if (leakStatus.nonEmpty) {
@@ -304,14 +327,14 @@ case class DescribeDatabaseCommand(
 
     if (leakStatus.nonEmpty) {
       leakStatus.foreach { ls =>
-        val lp = ls.getPath.toString
+        val lp = ls._1.getPath.toString
         var state = lp
         if (clear) {
           val ok = performFakeDelete(lp, trashPath, fs)
           val prefix = if (ok) "success" else "failure"
           state = s"($prefix) " + state
         }
-        result += Row(state, formatSize(getLen(ls, fs)))
+        result += Row(state, formatSize(ls._2))
       }
     }
     result
@@ -321,7 +344,7 @@ case class DescribeDatabaseCommand(
     if (fss.isFile) {
       return fss.getLen
     }
-    var dirLength = 0L
+    var dirLength = -1L
     try {
       dirLength = fs.getContentSummary(fss.getPath).getLength
     } catch {