<commit_before>package ducttape.workflow.builder

import WorkflowBuilder.CONFIG_TASK_DEF
import WorkflowBuilder.InputMode
import WorkflowBuilder.ParamMode
import WorkflowBuilder.ResolveMode
import ducttape.hyperdag.meta.MetaHyperDagBuilder
import ducttape.hyperdag.PackedVertex
import ducttape.syntax.AbstractSyntaxTree.ASTType
import ducttape.syntax.AbstractSyntaxTree.BranchGraft
import ducttape.syntax.AbstractSyntaxTree.BranchPointDef
import ducttape.syntax.AbstractSyntaxTree.BranchPointRef
import ducttape.syntax.AbstractSyntaxTree.Comments
import ducttape.syntax.AbstractSyntaxTree.ConfigAssignment
import ducttape.syntax.AbstractSyntaxTree.ConfigVariable
import ducttape.syntax.AbstractSyntaxTree.CrossProduct
import ducttape.syntax.AbstractSyntaxTree.Literal
import ducttape.syntax.AbstractSyntaxTree.LiteralSpec
import ducttape.syntax.AbstractSyntaxTree.PlanDefinition
import ducttape.syntax.AbstractSyntaxTree.Sequence
import ducttape.syntax.AbstractSyntaxTree.SequentialBranchPoint
import ducttape.syntax.AbstractSyntaxTree.Spec
import ducttape.syntax.AbstractSyntaxTree.SubmitterDef
import ducttape.syntax.AbstractSyntaxTree.TaskDef
import ducttape.syntax.AbstractSyntaxTree.TaskHeader
import ducttape.syntax.AbstractSyntaxTree.TaskVariable
import ducttape.syntax.AbstractSyntaxTree.Unbound
import ducttape.syntax.AbstractSyntaxTree.VersionerDef
import ducttape.syntax.AbstractSyntaxTree.WorkflowDefinition
import ducttape.syntax.AbstractSyntaxTreeException
import ducttape.syntax.BashCode
import ducttape.syntax.FileFormatException
import ducttape.workflow.Branch
import ducttape.workflow.BranchFactory
import ducttape.workflow.BranchInfo
import ducttape.workflow.BranchPoint
import ducttape.workflow.BranchPointFactory
import ducttape.workflow.HyperWorkflow
import ducttape.workflow.NoSuchBranchException
import ducttape.workflow.NoSuchBranchPointException
import ducttape.workflow.RealizationPlan
import ducttape.workflow.Task
import ducttape.workflow.TaskTemplate
import scala.collection.Seq
import scala.collection.Set
import scala.collection.Map
import scala.collection.mutable

private[builder] class WorkflowResolver(
    wd: WorkflowDefinition,
    confSpecs: Map[String,Spec],
    dag: MetaHyperDagBuilder[TaskTemplate,BranchPoint,BranchInfo,Seq[Spec]],
    branchPointFactory: BranchPointFactory,
    branchFactory: BranchFactory) {
  
  def findTasks(): FoundTasks = {
    
    val taskMap = new mutable.HashMap[String,TaskDef]
    for (t: TaskDef <- wd.tasks) {
      taskMap += t.name -> t
    }

    val parents = new mutable.HashMap[TaskTemplate,Map[BranchInfo,mutable.Set[Option[TaskDef]]]]
    val vertices = new mutable.HashMap[String,PackedVertex[TaskTemplate]]
    val branchPoints = new mutable.ArrayBuffer[BranchPoint]
    val branchPointsByTask = new mutable.HashMap[TaskDef,mutable.Set[BranchPoint]]
    
    for (taskDef <- wd.tasks) {
       // TODO: Check for all inputs/outputs/param names being unique in this step
       if (vertices.contains(taskDef.name)) {
         val prev: TaskTemplate = vertices(taskDef.name).value
         throw new FileFormatException("Duplicate task name: %s".format(taskDef.name),
                                       List(taskDef, prev.taskDef))
       }

       val parentsByBranch = new mutable.HashMap[BranchInfo,mutable.Set[Option[TaskDef]]]

       // parameters are different than file dependencies in that they do not
       // add any temporal dependencies between tasks and therefore do not
       // add any edges in the MetaHyperDAG
       val paramVals = new mutable.ArrayBuffer[(Spec,Map[Branch,(LiteralSpec,TaskDef)])](taskDef.params.size)
       for (paramSpec: Spec <- taskDef.params) {
         resolveBranchPoint(taskDef, paramSpec, taskMap, branchPoints, branchPointsByTask, paramVals,
           // params have no effect on temporal ordering, but can affect derivation of branches
           // therefore, params are *always* rooted at phantom vertices, no matter what (hence, None)
           recordParentsFunc = (branchInfo: BranchInfo, taskDef: Option[TaskDef])
             => {parentsByBranch.getOrElseUpdate(branchInfo, {new mutable.HashSet}) += None},
           resolveVarFunc=resolveParam)
       }

       val inputVals = new mutable.ArrayBuffer[(Spec,Map[Branch,(Spec,TaskDef)])](taskDef.inputs.size)
       for (inSpec: Spec <- taskDef.inputs) {
         resolveBranchPoint(taskDef, inSpec, taskMap, branchPoints, branchPointsByTask, inputVals,
             // make a note of what edges we'll need to add later
             recordParentsFunc = (branchInfo: BranchInfo, srcTaskDef: Option[TaskDef])
               => {parentsByBranch.getOrElseUpdate(branchInfo, {new mutable.HashSet}) += srcTaskDef},
             resolveVarFunc=resolveInput)
       }

       if (paramVals.size == 0 && inputVals.size == 0) {
         // we MUST inherit from at least the baseline branch point
         branchPointsByTask.getOrElseUpdate(taskDef, {new mutable.HashSet}) += Task.NO_BRANCH_POINT
       }

       val task = new TaskTemplate(taskDef, branchPointsByTask(taskDef).toSeq, inputVals, paramVals)
       parents += task -> parentsByBranch
       vertices += task.name -> dag.addVertex(task)
     }
    new FoundTasks(parents, vertices, branchPoints, branchPointsByTask)
  }

   // the resolved Spec is guaranteed to be a literal for params
   private def resolveParam(taskDef: TaskDef, taskMap: Map[String,TaskDef], spec: Spec)
                           : (LiteralSpec, TaskDef, Seq[Branch]) = {
     resolveVar(ParamMode())(taskDef, taskMap, spec)().asInstanceOf[(LiteralSpec,TaskDef,Seq[Branch])]
   }
   
   private def resolveInput(taskDef: TaskDef, taskMap: Map[String,TaskDef], spec: Spec) = {
     resolveVar(InputMode())(taskDef, taskMap, spec)()
   }
   
   // helper for resolveVar
   def resolveTaskVar(mode: ResolveMode)
                     (taskDef: TaskDef, taskMap: Map[String,TaskDef], spec: Spec)
                     (curSpec: Spec, src: TaskDef, grafts: Seq[Branch])
                     (srcTaskName: String, srcOutName: String)
                     : (Spec, TaskDef, Seq[Branch]) = {
     
     taskMap.get(srcTaskName) match {
       case Some(srcDef: TaskDef) => {
         // determine where to search when resolving this variable's parent spec
         val specSet = mode match {
           case InputMode() => srcDef.outputs
           case ParamMode() => srcDef.params
         }
         // search for the parent spec
         specSet.find(outSpec => outSpec.name == srcOutName) match {
           case Some(srcSpec) => resolveVar(mode)(taskDef, taskMap, spec)(srcSpec, srcDef, grafts)
           case None => {
             throw new FileFormatException(
               "Output %s at source task %s for required by input %s at task %s not found. Candidate outputs are: %s".
                 format(srcOutName, srcTaskName, spec.name, taskDef.name, specSet.map(_.name).mkString(" ")),
               List(spec, srcDef))
           }
         }
       }
       case None => {
         throw new FileFormatException(
           "Source task %s for input %s at task %s not found".
           format(srcTaskName, spec.name, taskDef.name), spec)
       }
     }
   }

   // TODO: document what's going on here -- maybe move elsewhere
   // group parameters by (1) initial state and (2) items that change recursively
   private def resolveVar(mode: ResolveMode)
                         (taskDef: TaskDef, taskMap: Map[String,TaskDef], spec: Spec)
                         (curSpec: Spec = spec, src: TaskDef = taskDef, grafts: Seq[Branch] = Nil)
                         : (Spec, TaskDef, Seq[Branch]) = {
     
     curSpec.rval match {
       case Literal(litValue) => {
         // TODO: can we enforce this as part of the match instead of using a cast?
         val litSpec = curSpec.asInstanceOf[LiteralSpec] // guaranteed to succeed
         (litSpec, src, Nil) // literals will never have a use for grafts
       }
       case ConfigVariable(varName) => {
         confSpecs.get(varName) match {
           // TODO: Does this TaskDef break line numbering for error reporting?
           // TODO: Should we return? Or do we allow config files to point back into workflows?
           case Some(confSpec) => (confSpec, CONFIG_TASK_DEF, grafts)
           case None => throw new FileFormatException(
               "Config variable %s required by input %s at task %s not found in config file.".
                 format(varName, spec.name, taskDef.name),
               List(spec, src))
         }
       }
       case TaskVariable(srcTaskName, srcOutName) => {
         resolveTaskVar(mode)(taskDef, taskMap, spec)(curSpec, src, grafts)(srcTaskName, srcOutName)
       }
       case BranchGraft(srcOutName, srcTaskName, branchGraftElements) => {
         val (srcSpec, srcTask, prevGrafts) = {
           resolveTaskVar(mode)(taskDef, taskMap, spec)(curSpec, src, grafts)(srcTaskName, srcOutName)
         }
         val resultGrafts = prevGrafts ++ branchGraftElements.map{ e => try {
             branchFactory(e.branchName, e.branchPointName)
           } catch {
             case ex: NoSuchBranchException => throw new FileFormatException(ex.getMessage, e)
           }
         }
         (srcSpec, srcTask, resultGrafts)
       }
       case BranchPointDef(_,_) => throw new RuntimeException("Expected branches to be resolved by now")
       case SequentialBranchPoint(_,_) => {
         throw new RuntimeException("Expected branches to be resolved by now")
       }
       case Unbound() => {
         mode match {
           case InputMode() => {
             // make sure we didn't just refer to ourselves -- 
             // referring to an unbound output is fine though (and usual)
             if (src != taskDef || spec != curSpec) {
               (curSpec, src, grafts)
             } else {
               throw new FileFormatException("Unbound input variable: %s".format(curSpec.name),
                                             List(taskDef, src))
             }<commit_msg>Steamrolling Lane's changes broke the build (sorry). Now there's again a match-case for the branch grafting, which now has an Option. However, now it calls either resolveTaskVar or resolveConfigVar.<commit_after>package ducttape.workflow.builder

import WorkflowBuilder.CONFIG_TASK_DEF
import WorkflowBuilder.InputMode
import WorkflowBuilder.ParamMode
import WorkflowBuilder.ResolveMode
import ducttape.hyperdag.meta.MetaHyperDagBuilder
import ducttape.hyperdag.PackedVertex
import ducttape.syntax.AbstractSyntaxTree.ASTType
import ducttape.syntax.AbstractSyntaxTree.BranchGraft
import ducttape.syntax.AbstractSyntaxTree.BranchPointDef
import ducttape.syntax.AbstractSyntaxTree.BranchPointRef
import ducttape.syntax.AbstractSyntaxTree.Comments
import ducttape.syntax.AbstractSyntaxTree.ConfigAssignment
import ducttape.syntax.AbstractSyntaxTree.ConfigVariable
import ducttape.syntax.AbstractSyntaxTree.CrossProduct
import ducttape.syntax.AbstractSyntaxTree.Literal
import ducttape.syntax.AbstractSyntaxTree.LiteralSpec
import ducttape.syntax.AbstractSyntaxTree.PlanDefinition
import ducttape.syntax.AbstractSyntaxTree.Sequence
import ducttape.syntax.AbstractSyntaxTree.SequentialBranchPoint
import ducttape.syntax.AbstractSyntaxTree.Spec
import ducttape.syntax.AbstractSyntaxTree.SubmitterDef
import ducttape.syntax.AbstractSyntaxTree.TaskDef
import ducttape.syntax.AbstractSyntaxTree.TaskHeader
import ducttape.syntax.AbstractSyntaxTree.TaskVariable
import ducttape.syntax.AbstractSyntaxTree.Unbound
import ducttape.syntax.AbstractSyntaxTree.VersionerDef
import ducttape.syntax.AbstractSyntaxTree.WorkflowDefinition
import ducttape.syntax.AbstractSyntaxTreeException
import ducttape.syntax.BashCode
import ducttape.syntax.FileFormatException
import ducttape.workflow.Branch
import ducttape.workflow.BranchFactory
import ducttape.workflow.BranchInfo
import ducttape.workflow.BranchPoint
import ducttape.workflow.BranchPointFactory
import ducttape.workflow.HyperWorkflow
import ducttape.workflow.NoSuchBranchException
import ducttape.workflow.NoSuchBranchPointException
import ducttape.workflow.RealizationPlan
import ducttape.workflow.Task
import ducttape.workflow.TaskTemplate
import scala.collection.Seq
import scala.collection.Set
import scala.collection.Map
import scala.collection.mutable

private[builder] class WorkflowResolver(
    wd: WorkflowDefinition,
    confSpecs: Map[String,Spec],
    dag: MetaHyperDagBuilder[TaskTemplate,BranchPoint,BranchInfo,Seq[Spec]],
    branchPointFactory: BranchPointFactory,
    branchFactory: BranchFactory) {
  
  def findTasks(): FoundTasks = {
    
    val taskMap = new mutable.HashMap[String,TaskDef]
    for (t: TaskDef <- wd.tasks) {
      taskMap += t.name -> t
    }

    val parents = new mutable.HashMap[TaskTemplate,Map[BranchInfo,mutable.Set[Option[TaskDef]]]]
    val vertices = new mutable.HashMap[String,PackedVertex[TaskTemplate]]
    val branchPoints = new mutable.ArrayBuffer[BranchPoint]
    val branchPointsByTask = new mutable.HashMap[TaskDef,mutable.Set[BranchPoint]]
    
    for (taskDef <- wd.tasks) {
       // TODO: Check for all inputs/outputs/param names being unique in this step
       if (vertices.contains(taskDef.name)) {
         val prev: TaskTemplate = vertices(taskDef.name).value
         throw new FileFormatException("Duplicate task name: %s".format(taskDef.name),
                                       List(taskDef, prev.taskDef))
       }

       val parentsByBranch = new mutable.HashMap[BranchInfo,mutable.Set[Option[TaskDef]]]

       // parameters are different than file dependencies in that they do not
       // add any temporal dependencies between tasks and therefore do not
       // add any edges in the MetaHyperDAG
       val paramVals = new mutable.ArrayBuffer[(Spec,Map[Branch,(LiteralSpec,TaskDef)])](taskDef.params.size)
       for (paramSpec: Spec <- taskDef.params) {
         resolveBranchPoint(taskDef, paramSpec, taskMap, branchPoints, branchPointsByTask, paramVals,
           // params have no effect on temporal ordering, but can affect derivation of branches
           // therefore, params are *always* rooted at phantom vertices, no matter what (hence, None)
           recordParentsFunc = (branchInfo: BranchInfo, taskDef: Option[TaskDef])
             => {parentsByBranch.getOrElseUpdate(branchInfo, {new mutable.HashSet}) += None},
           resolveVarFunc=resolveParam)
       }

       val inputVals = new mutable.ArrayBuffer[(Spec,Map[Branch,(Spec,TaskDef)])](taskDef.inputs.size)
       for (inSpec: Spec <- taskDef.inputs) {
         resolveBranchPoint(taskDef, inSpec, taskMap, branchPoints, branchPointsByTask, inputVals,
             // make a note of what edges we'll need to add later
             recordParentsFunc = (branchInfo: BranchInfo, srcTaskDef: Option[TaskDef])
               => {parentsByBranch.getOrElseUpdate(branchInfo, {new mutable.HashSet}) += srcTaskDef},
             resolveVarFunc=resolveInput)
       }

       if (paramVals.size == 0 && inputVals.size == 0) {
         // we MUST inherit from at least the baseline branch point
         branchPointsByTask.getOrElseUpdate(taskDef, {new mutable.HashSet}) += Task.NO_BRANCH_POINT
       }

       val task = new TaskTemplate(taskDef, branchPointsByTask(taskDef).toSeq, inputVals, paramVals)
       parents += task -> parentsByBranch
       vertices += task.name -> dag.addVertex(task)
     }
    new FoundTasks(parents, vertices, branchPoints, branchPointsByTask)
  }

   // the resolved Spec is guaranteed to be a literal for params
   private def resolveParam(taskDef: TaskDef, taskMap: Map[String,TaskDef], spec: Spec)
                           : (LiteralSpec, TaskDef, Seq[Branch]) = {
     resolveVar(ParamMode())(taskDef, taskMap, spec)().asInstanceOf[(LiteralSpec,TaskDef,Seq[Branch])]
   }
   
   private def resolveInput(taskDef: TaskDef, taskMap: Map[String,TaskDef], spec: Spec) = {
     resolveVar(InputMode())(taskDef, taskMap, spec)()
   }
   
   // helper for resolveVar
   def resolveTaskVar(mode: ResolveMode)
                     (taskDef: TaskDef, taskMap: Map[String,TaskDef], spec: Spec)
                     (curSpec: Spec, src: TaskDef, grafts: Seq[Branch])
                     (srcTaskName: String, srcOutName: String)
                     : (Spec, TaskDef, Seq[Branch]) = {
     
     taskMap.get(srcTaskName) match {
       case Some(srcDef: TaskDef) => {
         // determine where to search when resolving this variable's parent spec
         val specSet = mode match {
           case InputMode() => srcDef.outputs
           case ParamMode() => srcDef.params
         }
         // search for the parent spec
         specSet.find(outSpec => outSpec.name == srcOutName) match {
           case Some(srcSpec) => resolveVar(mode)(taskDef, taskMap, spec)(srcSpec, srcDef, grafts)
           case None => {
             throw new FileFormatException(
               "Output %s at source task %s for required by input %s at task %s not found. Candidate outputs are: %s".
                 format(srcOutName, srcTaskName, spec.name, taskDef.name, specSet.map(_.name).mkString(" ")),
               List(spec, srcDef))
           }
         }
       }
       case None => {
         throw new FileFormatException(
           "Source task %s for input %s at task %s not found".
           format(srcTaskName, spec.name, taskDef.name), spec)
       }
     }
   }
   
   def resolveConfigVar(varName: String, taskDef: TaskDef, spec: Spec, src: TaskDef, grafts: Seq[Branch]) = {
     confSpecs.get(varName) match {
       // TODO: Does this TaskDef break line numbering for error reporting?
       // TODO: Should we return? Or do we allow config files to point back into workflows?
       case Some(confSpec) => (confSpec, CONFIG_TASK_DEF, grafts)
       case None => throw new FileFormatException(
           "Config variable %s required by input %s at task %s not found in config file.".
             format(varName, spec.name, taskDef.name),
           List(spec, src))
     }
   }

   // TODO: document what's going on here -- maybe move elsewhere
   // group parameters by (1) initial state and (2) items that change recursively
   private def resolveVar(mode: ResolveMode)
                         (taskDef: TaskDef, taskMap: Map[String,TaskDef], spec: Spec)
                         (curSpec: Spec = spec, src: TaskDef = taskDef, grafts: Seq[Branch] = Nil)
                         : (Spec, TaskDef, Seq[Branch]) = {
     
     curSpec.rval match {
       case Literal(litValue) => {
         // TODO: can we enforce this as part of the match instead of using a cast?
         val litSpec = curSpec.asInstanceOf[LiteralSpec] // guaranteed to succeed
         (litSpec, src, Nil) // literals will never have a use for grafts
       }
       case ConfigVariable(varName) => resolveConfigVar(varName, taskDef, spec, src, grafts)
       case TaskVariable(srcTaskName, srcOutName) => {
         resolveTaskVar(mode)(taskDef, taskMap, spec)(curSpec, src, grafts)(srcTaskName, srcOutName)
       }
       case BranchGraft(srcOutName, srcTaskNameOpt, branchGraftElements) => {
         val (srcSpec, srcTask, prevGrafts) = srcTaskNameOpt match {
           case Some(srcTaskName) => {
             resolveTaskVar(mode)(taskDef, taskMap, spec)(curSpec, src, grafts)(srcTaskName, srcOutName)
           }
           case None => {
             resolveConfigVar(srcOutName, taskDef, spec, src, grafts)
           }
         }
         val resultGrafts = prevGrafts ++ branchGraftElements.map{ e => try {
             branchFactory(e.branchName, e.branchPointName)
           } catch {
             case ex: NoSuchBranchException => throw new FileFormatException(ex.getMessage, e)
           }
         }
         (srcSpec, srcTask, resultGrafts)
       }
       case BranchPointDef(_,_) => throw new RuntimeException("Expected branches to be resolved by now")
       case SequentialBranchPoint(_,_) => {
         throw new RuntimeException("Expected branches to be resolved by now")
       }
       case Unbound() => {
         mode match {
           case InputMode() => {
             // make sure we didn't just refer to ourselves -- 
             // referring to an unbound output is fine though (and usual)
             if (src != taskDef || spec != curSpec) {
               (curSpec, src, grafts)
             } else {
               throw new FileFormatException("Unbound input variable: %s".format(curSpec.name),
                                             List(taskDef, src))
             }
{'fraction_non_alphanumeric': 0.09034267912772585, 'fraction_numerical': 0.0003894080996884735, 'mean_word_length': 3.0056541236108405, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3139301', 'n_tokens_mistral': 5822, 'n_tokens_neox': 5659, 'n_words': 1481}