comments - functional global hash computation + comments fixes

Author: nikhil-zlai

orchestration/src/main/scala/ai/chronon/orchestration/RepoIndex.scala

@@ -52,7 +52,7 @@ class RepoIndex[T >: Null](proc: ConfProcessor[T]) extends Logging {
 
   /**
     * @param fileHashes map of all the file names to their hashes in the branch.
-    * @param newNodes nodes that aren't present in the index's [[fileHashToContent]] table. New nodes typically.
+    * @param newNodes nodes that aren't already present in the index. De-duped across branches.
     * @param branch branch on which the user is making the changes.
     * @param dryRun when true shows potential versions that will be assigned to nodes without modifying the index.
     * @return
@@ -74,8 +74,7 @@ class RepoIndex[T >: Null](proc: ConfProcessor[T]) extends Logging {
       *
       * we use memoization to avoid recomputing global hashes via the [[globalHashes]] map
       */
-    val globalHashes = mutable.Map.empty[Name, GlobalHash]
-    def computeGlobalHash(name: Name): GlobalHash = {
+    def computeGlobalHash(name: Name, globalHashes: mutable.Map[Name, GlobalHash]): GlobalHash = {
 
       if (globalHashes.contains(name)) return globalHashes(name)
 
@@ -107,7 +106,7 @@ class RepoIndex[T >: Null](proc: ConfProcessor[T]) extends Logging {
       // recursively compute parent hashes
       val parentHashes = parents
         .map { parent =>
-          val parentHash = computeGlobalHash(parent).hash
+          val parentHash = computeGlobalHash(parent, globalHashes).hash
           s"${parent.name}:$parentHash"
         }
         .mkString(",")
@@ -123,7 +122,9 @@ class RepoIndex[T >: Null](proc: ConfProcessor[T]) extends Logging {
       globalHash
     }
 
-    fileHashes.foreach { case (name, _) => computeGlobalHash(name) }
+    val globalHashes = mutable.Map.empty[Name, GlobalHash]
+    // this line fills global hashes
+    fileHashes.foreach { case (name, _) => computeGlobalHash(name, globalHashes) }
 
     val existingVersions = branchVersionIndex.getOrElse(branch, mutable.Map.empty)
     val mainVersions = branchVersionIndex.getOrElse(Branch.main, mutable.Map.empty)
@@ -146,19 +147,20 @@ class RepoIndex[T >: Null](proc: ConfProcessor[T]) extends Logging {
       branchToFileHash.update(branch, enrichedFileHashes)
       branchVersionIndex.update(branch, newVersions)
 
-      return VersionUpdate.join(newVersions, existingVersions, mainVersions)
-    }
+      VersionUpdate.join(newVersions, existingVersions, mainVersions)
 
-    // dry run - don't insert into any members of the index
-    val newVersions = mutable.Map.empty[Name, Version]
-    globalHashes.foreach {
-      case (name, globalHash) =>
-        val versionIndex = versionSequencer.potentialIndex(name, globalHash)
-        newVersions.update(name, Version("v" + versionIndex.toString))
-    }
+    } else {
+
+      // dry run - don't insert into any members of the index
+      val newVersions = mutable.Map.empty[Name, Version]
+      globalHashes.foreach {
+        case (name, globalHash) =>
+          val versionIndex = versionSequencer.potentialIndex(name, globalHash)
+          newVersions.update(name, Version("v" + versionIndex.toString))
+      }
 
-    val versionUpdates = VersionUpdate.join(newVersions, existingVersions, mainVersions)
-    versionUpdates
+      VersionUpdate.join(newVersions, existingVersions, mainVersions)
+    }
   }
 
   def addFiles(fileHashes: mutable.Map[Name, FileHash], updatedFiles: Map[String, String], branch: Branch): Unit = {
@@ -171,7 +173,9 @@ class RepoIndex[T >: Null](proc: ConfProcessor[T]) extends Logging {
     addNodes(fileHashes, nodes, branch)
   }
 
-  // returns the contents of the files not present in the index
+  /**
+    * returns the names of the files whose contents haven't been found in the index across any of the versions
+    */
   def diff(incomingFileHashes: mutable.Map[Name, FileHash]): Seq[Name] = {
 
     incomingFileHashes

orchestration/src/test/scala/ai/chronon/orchestration/test/RepoIndexSpec.scala

@@ -164,6 +164,19 @@ class RepoIndexSpec extends AnyFlatSpec with Matchers with Logging {
     )
 
     updateIndex(branchConfs5, Branch.main, "new sq3 and gb3, un-deleted gb2")
+
+    val branchConfs6 = Seq(
+      TestConf("sq1", "v1", "4g", Seq.empty, Seq("t1"))                       -> "v0",
+      TestConf("sq3", "v1", "4g", Seq.empty, Seq("t3"))                       -> "v0",
+      TestConf("gb1", "v1", "4g", Seq("t1"))                                  -> "v0",
+      TestConf("gb3", "v1", "4g", Seq("t3"))                                  -> "v0",
+      TestConf("gb2", "v1", "8g", Seq("t2"))                                  -> "v0",
+      TestConf("j1",  "v1", "4g", Seq("gb1", "gb2", "gb3"), Seq("table_j1"))  -> "v3",
+      TestConf("m1", "v1", "4g", Seq("j1"), Seq("table_m1"))                  -> "v3",
+      TestConf("m2", "v1", "4g", Seq("j1"), Seq("table_m2"))                  -> "v0",
+    )
+
+    updateIndex(branchConfs6, Branch.main, "m2 is added")
   }