Parallel calls

sumeet-db · Mar 22, 2024 · c01665b · c01665b
1 parent 27a894e
commit c01665b
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 4 deletions.
diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/SnapshotManagement.scala b/spark/src/main/scala/org/apache/spark/sql/delta/SnapshotManagement.scala
@@ -231,13 +231,15 @@ trait SnapshotManagement { self: DeltaLog =>
           maxDeltaVersionSeen < unbackfilledCommitsResponse.latestTableVersion}
     val needAdditionalFsListing =
       versionToLoad.forall(maxDeltaVersionSeen < _) && areDeltaFilesMissing
-    val additionalLogTuplesFromFsListingOpt =
-      Option.when(needAdditionalFsListing) {
+    val additionalLogTuplesFromFsListingOpt: Option[Array[(FileStatus, FileType.Value, Long)]] =
+      if (needAdditionalFsListing) {
         recordDeltaEvent(this, "delta.listDeltaAndCheckpointFiles.needAdditionalFsListing")
         listDeltaCompactedDeltaAndCheckpointFilesOpt(
           startVersionOverride = maxDeltaVersionSeen + 1)
-      }.flatten
-    val finalLogTuplesFromFsListingOpt =
+      } else {
+        None
+      }
+    val finalLogTuplesFromFsListingOpt: Option[Array[(FileStatus, FileType.Value, Long)]] =
       (initialLogTuplesFromFsListingOpt, additionalLogTuplesFromFsListingOpt) match {
         case (Some(initial), Some(additional)) => Some(initial ++ additional)
         case (Some(initial), None) => Some(initial)

diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/util/threads/DeltaThreadPool.scala b/spark/src/main/scala/org/apache/spark/sql/delta/util/threads/DeltaThreadPool.scala
@@ -31,6 +31,9 @@ import org.apache.spark.util.ThreadUtils.namedThreadFactory
 
 /** A wrapper for [[ThreadPoolExecutor]] whose tasks run with the caller's [[SparkSession]]. */
 private[delta] class DeltaThreadPool(tpe: ThreadPoolExecutor) {
+  def getActiveCount: Int = tpe.getActiveCount
+  def getMaximumPoolSize: Int = tpe.getMaximumPoolSize
+
   /** Submits a task for execution and returns a [[Future]] representing that task. */
   def submit[T](spark: SparkSession)(body: => T): Future[T] = {
     tpe.submit { () => spark.withActive(body) }