more prefix progress

src/main/scala/workflow/graph/DanglingNodeRemovalRule.scala

@@ -9,8 +9,8 @@ object DanglingNodeRemovalRule extends Rule {
       case (ancestors, sink) => ancestors ++ AnalysisUtils.getAncestors(plan, sink)
     }
 
-    val nodesToRemove = plan.nodes -- ancestorsOfSinks.collect{ case node: NodeId => node }
-    val sourcesToRemove = plan.sources -- ancestorsOfSinks.collect{ case source: SourceId => source }
+    val nodesToRemove = plan.nodes -- ancestorsOfSinks.collect { case node: NodeId => node }
+    val sourcesToRemove = plan.sources -- ancestorsOfSinks.collect { case source: SourceId => source }
 
     val afterSourceRemoval = sourcesToRemove.foldLeft(plan) {
       case (curPlan, sourceToRemove) => curPlan.removeSource(sourceToRemove)

src/main/scala/workflow/graph/DefaultOptimizer.scala

@@ -5,6 +5,7 @@ package workflow.graph
  */
 object DefaultOptimizer extends Optimizer {
   protected val batches: Seq[Batch] =
+    Batch("Load Saved State", Once, ExtractSaveablePrefixes, SavedStateLoadRule, DanglingNodeRemovalRule) ::
     Batch("Common Sub-expression Elimination", FixedPoint(Int.MaxValue), EquivalentNodeMergeRule) ::
       Nil
 }

src/main/scala/workflow/graph/ExtractSaveablePrefixes.scala

@@ -0,0 +1,11 @@
+package workflow.graph
+
+/**
+ * Extract the prefixes whose state we want to save (aka for any cacher or estimator node)
+ */
+object ExtractSaveablePrefixes extends Rule {
+  override def apply(plan: Graph, prefixes: Map[NodeId, Prefix]): (Graph, Map[NodeId, Prefix]) = {
+    // TODO FIXME
+    (plan, prefixes)
+  }
+}