fix storage key issue

Texera · Dec 19, 2024 · dffc702 · dffc702
1 parent 3dbfc33
commit dffc702
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 3 deletions.
diff --git a/core/amber/src/main/scala/edu/uci/ics/texera/web/service/ExecutionResultService.scala b/core/amber/src/main/scala/edu/uci/ics/texera/web/service/ExecutionResultService.scala
@@ -302,6 +302,7 @@ class ExecutionResultService(
         ResultStorage
           .getOpResultStorage(workflowIdentity)
           .getAllKeys
+          .filter(!_.id.startsWith("materialized_"))
           .map(storageKey => {
             val count = ResultStorage
               .getOpResultStorage(workflowIdentity)

diff --git a/core/amber/src/main/scala/edu/uci/ics/texera/workflow/WorkflowCompiler.scala b/core/amber/src/main/scala/edu/uci/ics/texera/workflow/WorkflowCompiler.scala
@@ -123,7 +123,7 @@ class WorkflowCompiler(
                   physicalOp.id,
                   outputPort.id,
                   sinkPhysicalOp.id,
-                  PortIdentity(internal = true)
+                  PortIdentity()
                 )
               physicalPlan = physicalPlan.addOperator(sinkPhysicalOp).addLink(sinkLink)
           })

diff --git a/...orkflow-operator/src/main/scala/edu/uci/ics/amber/operator/SpecialPhysicalOpFactory.scala b/...orkflow-operator/src/main/scala/edu/uci/ics/amber/operator/SpecialPhysicalOpFactory.scala
@@ -35,8 +35,8 @@ object SpecialPhysicalOpFactory {
           )
         )
       )
-      .withInputPorts(List(InputPort(PortIdentity(internal = true))))
-      .withOutputPorts(List(OutputPort(PortIdentity(internal = true))))
+      .withInputPorts(List(InputPort(PortIdentity())))
+      .withOutputPorts(List(OutputPort(PortIdentity())))
       .withPropagateSchema(
         SchemaPropagationFunc((inputSchemas: Map[PortIdentity, Schema]) => {
           // Get the first schema from inputSchemas