Refactor Content.rec(..) function to support tail-optimisation

src/main/scala/com/microsoft/hyperspace/index/IndexLogEntry.scala

@@ -31,6 +31,7 @@ import org.apache.spark.sql.types.{DataType, StructType}
 
 import com.microsoft.hyperspace.HyperspaceException
 import com.microsoft.hyperspace.actions.Constants
+import com.microsoft.hyperspace.index.Content.recFilesApply
 import com.microsoft.hyperspace.util.{PathUtils, SchemaUtils}
 
 // IndexLogEntry-specific fingerprint to be temporarily used where fingerprint is not defined.
@@ -45,27 +46,17 @@ case class Content(root: Directory, fingerprint: NoOpFingerprint = NoOpFingerpri
   @JsonIgnore
   lazy val files: Seq[Path] = {
     // Recursively find files from directory tree.
-    rec(new Path(root.name), root, (f, prefix) => new Path(prefix, f.name))
+    recFilesApply(new Path(root.name), root, (f, prefix) => new Path(prefix, f.name))
   }
 
   @JsonIgnore
   lazy val fileInfos: Set[FileInfo] = {
-    rec(
+    recFilesApply(
       new Path(root.name),
       root,
       (f, prefix) =>
         FileInfo(new Path(prefix, f.name).toString, f.size, f.modifiedTime, f.id)).toSet
   }
-
-  private def rec[T](
-      prefixPath: Path,
-      directory: Directory,
-      func: (FileInfo, Path) => T): Seq[T] = {
-    val files = directory.files.map(f => func(f, prefixPath))
-    files ++ directory.subDirs.flatMap { dir =>
-      rec(new Path(prefixPath, dir.name), dir, func)
-    }
-  }
 }
 
 object Content {
@@ -111,6 +102,43 @@ object Content {
       None
     }
   }
+
+  /**
+   * Apply `func` to each file in directory recursively.
+   * @param prefixPath Root prefix
+   * @param directory Root directory
+   * @param func function which would apply to current prefix and file
+   * @tparam T
+   * @return
+   */
+  def recFilesApply[T](
+              prefixPath: Path,
+              directory: Directory,
+              func: (FileInfo, Path) => T): Seq[T] = {
+    @tailrec
+    def recAcc[A](
+                   dirMap: List[(Path, Seq[Directory])],
+                   func: (FileInfo, Path) => A,
+                   acc: Seq[A] = Seq.empty): Seq[A] = {
+      dirMap match {
+        case Nil => acc
+        case (curPrefixPath, curDirs) :: xs =>
+
+          val curAcc = for {
+            dir <- curDirs
+            file <- dir.files
+          } yield func(file, new Path(curPrefixPath, dir.name))
+
+          val newLevels = curDirs
+            .filter(_.subDirs.nonEmpty)
+            .map(dir => (new Path(curPrefixPath, dir.name), dir.subDirs))
+
+          recAcc(xs ++ newLevels, func, curAcc ++ acc)
+      }
+    }
+
+    recAcc(List(prefixPath -> Seq(directory)), func)
+  }
 }
 
 /**

src/test/scala/com/microsoft/hyperspace/index/IndexLogEntryTest.scala

@@ -240,6 +240,47 @@ class IndexLogEntryTest extends SparkFunSuite with SQLHelper with BeforeAndAfter
     assert(actual.sourceFilesSizeInBytes == 200L)
   }
 
+  test("Content.recFilesApply apply function to all files ") {
+    val directory = Directory("file:/",
+      files = Seq(FileInfo("f0", 0, 0, UNKNOWN_FILE_ID)),
+      subDirs = Seq(
+        Directory("a",
+          files = Seq(FileInfo("f1", 0, 0, UNKNOWN_FILE_ID), FileInfo("f2", 0, 0, UNKNOWN_FILE_ID)),
+          subDirs = Seq(
+            Directory("b",
+              files =
+                Seq(FileInfo("f3", 0, 0, UNKNOWN_FILE_ID), FileInfo("f4", 0, 0, UNKNOWN_FILE_ID)),
+              subDirs = Seq(Directory("c"))
+            ),
+            Directory("d"))
+        )))
+
+    val res = Content.recFilesApply(
+      new Path("file:/"),
+      directory,
+      (f, prefix) => new Path(prefix, f.name)
+    )
+
+    val expected =
+      Seq("file:/f0", "file:/a/f1", "file:/a/f2", "file:/a/b/f3", "file:/a/b/f4")
+        .map(new Path(_)).toSet
+
+    val actual = res.toSet
+    assert(actual.equals(expected))
+  }
+
+  test("Content.recFilesApply return empty list for directories without files ") {
+    val directory = Directory("file:/")
+
+    val res = Content.recFilesApply(
+      new Path("file:/"),
+      directory,
+      (f, prefix) => new Path(prefix, f.name)
+    )
+
+    assert(res.isEmpty)
+  }
+
   test("Content.files api lists all files from Content object.") {
     val content = Content(Directory("file:/", subDirs =
       Seq(