update

Texera · Dec 16, 2024 · 713bebb · 713bebb
1 parent f866205
commit 713bebb
Showing 1 changed file with 6 additions and 10 deletions.
diff --git a/...r/src/main/scala/edu/uci/ics/amber/engine/architecture/scheduling/ScheduleGenerator.scala b/...r/src/main/scala/edu/uci/ics/amber/engine/architecture/scheduling/ScheduleGenerator.scala
@@ -153,7 +153,7 @@ abstract class ScheduleGenerator(
     // create cache writer and link
     val matWriterInputSchema = fromOp.outputPorts(fromPortId)._3.toOption.get
     val matWriterPhysicalOp: PhysicalOp =
-      createMatWriter(physicalLink, Array(matWriterInputSchema), workflowContext.workflowId)
+      createMatWriter(physicalLink, Array(matWriterInputSchema))
     val sourceToWriterLink =
       PhysicalLink(
         fromOp.id,
@@ -186,15 +186,15 @@ abstract class ScheduleGenerator(
       matWriterLogicalOpId: OperatorIdentity,
       physicalLink: PhysicalLink
   ): PhysicalOp = {
-    println("Creating mat reader!!!")
+    val opResultStorage =  ResultStorage.getOpResultStorage(workflowContext.workflowId)
     PhysicalOp
       .sourcePhysicalOp(
         workflowContext.workflowId,
         workflowContext.executionId,
         OperatorIdentity(s"cacheSource_${getMatIdFromPhysicalLink(physicalLink)}"),
         OpExecInitInfo((_, _) =>
           new CacheSourceOpExec(
-            ResultStorage.getOpResultStorage(workflowContext.workflowId).get(matWriterLogicalOpId)
+            opResultStorage.get(matWriterLogicalOpId)
           )
         )
       )
@@ -203,10 +203,7 @@ abstract class ScheduleGenerator(
       .withPropagateSchema(
         SchemaPropagationFunc(_ =>
           Map(
-            OutputPort().id -> ResultStorage
-              .getOpResultStorage(workflowContext.workflowId)
-              .getSchema(matWriterLogicalOpId)
-              .get
+            OutputPort().id -> opResultStorage.getSchema(matWriterLogicalOpId).get
           )
         )
       )
@@ -216,16 +213,15 @@ abstract class ScheduleGenerator(
 
   private def createMatWriter(
       physicalLink: PhysicalLink,
-      inputSchema: Array[Schema],
-      workflowIdentity: WorkflowIdentity
+      inputSchema: Array[Schema]
   ): PhysicalOp = {
     val matWriter = new ProgressiveSinkOpDesc()
     matWriter.setContext(workflowContext)
     matWriter.setOperatorId(s"materialized_${getMatIdFromPhysicalLink(physicalLink)}")
     // expect exactly one input port and one output port
     val schema = matWriter.getOutputSchema(inputSchema)
     ResultStorage
-      .getOpResultStorage(workflowIdentity)
+      .getOpResultStorage(workflowContext.workflowId)
       .create(
         key = matWriter.operatorIdentifier,
         mode = OpResultStorage.defaultStorageMode,