Introduce State Related Operators

core/amber/src/main/scala/edu/uci/ics/texera/workflow/common/metadata/OperatorGroupConstants.scala

@@ -20,6 +20,7 @@ object OperatorGroupConstants {
   final val PYTHON_GROUP = "Python"
   final val JAVA_GROUP = "Java"
   final val R_GROUP = "R"
+  final val STATE_GROUP = "State"
   final val MACHINE_LEARNING_GENERAL_GROUP = "Machine Learning General"
 
   /**
@@ -43,7 +44,7 @@ object OperatorGroupConstants {
         GroupInfo(MACHINE_LEARNING_GENERAL_GROUP)
       )
     ),
-    GroupInfo(UTILITY_GROUP),
+    GroupInfo(UTILITY_GROUP, List(GroupInfo(STATE_GROUP))),
     GroupInfo(API_GROUP),
     GroupInfo(UDF_GROUP, List(GroupInfo(PYTHON_GROUP), GroupInfo(JAVA_GROUP), GroupInfo(R_GROUP))),
     GroupInfo(VISUALIZATION_GROUP)

core/amber/src/main/scala/edu/uci/ics/texera/workflow/common/operators/LogicalOp.scala

@@ -147,6 +147,7 @@ import edu.uci.ics.texera.workflow.operators.visualization.quiverPlot.QuiverPlot
 import edu.uci.ics.texera.workflow.operators.visualization.contourPlot.ContourPlotOpDesc
 import edu.uci.ics.texera.workflow.operators.visualization.figureFactoryTable.FigureFactoryTableOpDesc
 import edu.uci.ics.texera.workflow.operators.visualization.sankeyDiagram.SankeyDiagramOpDesc
+import edu.uci.ics.texera.workflow.operators.state.{DataToStateOpDesc, StateToDataOpDesc}
 import java.util.UUID
 import scala.collection.mutable
 import scala.util.Try
@@ -162,6 +163,8 @@ trait StateTransferFunc
 )
 @JsonSubTypes(
   Array(
+    new Type(value = classOf[StateToDataOpDesc], name = "StateToData"),
+    new Type(value = classOf[DataToStateOpDesc], name = "DataToState"),
     new Type(value = classOf[SankeyDiagramOpDesc], name = "SankeyDiagram"),
     new Type(value = classOf[IcicleChartOpDesc], name = "IcicleChart"),
     new Type(value = classOf[CSVScanSourceOpDesc], name = "CSVFileScan"),

core/amber/src/main/scala/edu/uci/ics/texera/workflow/operators/state/DataToStateOpDesc.scala

@@ -0,0 +1,54 @@
+package edu.uci.ics.texera.workflow.operators.state
+
+import com.fasterxml.jackson.annotation.JsonProperty
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize
+import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
+import edu.uci.ics.amber.engine.architecture.deploysemantics.{PhysicalOp, SchemaPropagationFunc}
+import edu.uci.ics.amber.engine.architecture.deploysemantics.layer.OpExecInitInfo
+import edu.uci.ics.amber.engine.common.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import edu.uci.ics.amber.engine.common.workflow.{InputPort, OutputPort, PortIdentity}
+import edu.uci.ics.texera.workflow.common.metadata.{OperatorGroupConstants, OperatorInfo}
+import edu.uci.ics.texera.workflow.common.operators.LogicalOp
+import edu.uci.ics.texera.workflow.common.tuple.schema.Schema
+
+class DataToStateOpDesc extends LogicalOp {
+  @JsonProperty(defaultValue = "false")
+  @JsonSchemaTitle("Pass To All Downstream")
+  @JsonDeserialize(contentAs = classOf[java.lang.Boolean])
+  var passToAllDownstream: Option[Boolean] = Option(false)
+
+  override def getPhysicalOp(
+      workflowId: WorkflowIdentity,
+      executionId: ExecutionIdentity
+  ): PhysicalOp = {
+    PhysicalOp
+      .oneToOnePhysicalOp(
+        workflowId,
+        executionId,
+        operatorIdentifier,
+        OpExecInitInfo((_, _) => {
+          new DataToStateOpExec(passToAllDownstream.get)
+        })
+      )
+      .withInputPorts(operatorInfo.inputPorts)
+      .withOutputPorts(operatorInfo.outputPorts)
+      .withPropagateSchema(
+        SchemaPropagationFunc(inputSchemas => Map(PortIdentity() -> inputSchemas(PortIdentity(1))))
+      )
+      .withSuggestedWorkerNum(1)
+  }
+
+  override def operatorInfo: OperatorInfo =
+    OperatorInfo(
+      "Data To State",
+      "Convert Data to State",
+      OperatorGroupConstants.STATE_GROUP,
+      inputPorts = List(
+        InputPort(PortIdentity(), "State"),
+        InputPort(PortIdentity(1), "Data", dependencies = List(PortIdentity()))
+      ),
+      outputPorts = List(OutputPort())
+    )
+
+  override def getOutputSchema(schemas: Array[Schema]): Schema = schemas(1)
+}

core/amber/src/main/scala/edu/uci/ics/texera/workflow/operators/state/DataToStateOpExec.scala

@@ -0,0 +1,29 @@
+package edu.uci.ics.texera.workflow.operators.state
+
+import edu.uci.ics.amber.engine.common.tuple.amber.TupleLike
+import edu.uci.ics.texera.workflow.common.State
+import edu.uci.ics.texera.workflow.common.operators.OperatorExecutor
+import edu.uci.ics.texera.workflow.common.tuple.Tuple
+
+import scala.collection.mutable
+
+class DataToStateOpExec(passToAllDownstream: Boolean) extends OperatorExecutor {
+  private val dataTuples = new mutable.ArrayBuffer[Tuple]()
+  private var stateTuple: Option[Tuple] = None
+
+  override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike] = {
+    port match {
+      case 0 =>
+        if (stateTuple.isEmpty)
+          stateTuple = Some(tuple)
+      case 1 =>
+        dataTuples += tuple
+    }
+    Iterator.empty
+  }
+
+  override def produceStateOnFinish(port: Int): Option[State] =
+    Some(State(stateTuple, passToAllDownstream))
+
+  override def onFinish(port: Int): Iterator[TupleLike] = dataTuples.iterator
+}

core/amber/src/main/scala/edu/uci/ics/texera/workflow/operators/state/StateToDataOpDesc.scala

@@ -0,0 +1,65 @@
+package edu.uci.ics.texera.workflow.operators.state
+
+import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
+import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
+import edu.uci.ics.amber.engine.architecture.deploysemantics.{PhysicalOp, SchemaPropagationFunc}
+import edu.uci.ics.amber.engine.architecture.deploysemantics.layer.OpExecInitInfo
+import edu.uci.ics.amber.engine.common.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import edu.uci.ics.amber.engine.common.workflow.{InputPort, OutputPort, PortIdentity}
+import edu.uci.ics.texera.workflow.common.metadata.{OperatorGroupConstants, OperatorInfo}
+import edu.uci.ics.texera.workflow.common.operators.LogicalOp
+import edu.uci.ics.texera.workflow.common.tuple.schema.{Attribute, AttributeType, Schema}
+
+class StateToDataOpDesc extends LogicalOp {
+  @JsonProperty
+  @JsonSchemaTitle("State output column(s)")
+  @JsonPropertyDescription(
+    "Name of the output columns that the first port will produce, if any"
+  )
+  var outputColumns: List[Attribute] = List()
+
+  override def getPhysicalOp(
+      workflowId: WorkflowIdentity,
+      executionId: ExecutionIdentity
+  ): PhysicalOp = {
+    PhysicalOp
+      .oneToOnePhysicalOp(
+        workflowId,
+        executionId,
+        operatorIdentifier,
+        OpExecInitInfo((_, _) => {
+          new StateToDataOpExec()
+        })
+      )
+      .withInputPorts(operatorInfo.inputPorts)
+      .withOutputPorts(operatorInfo.outputPorts)
+      .withPropagateSchema(
+        SchemaPropagationFunc(inputSchemas =>
+          getOutputSchemas(
+            operatorInfo.inputPorts.map(port => inputSchemas(port.id)).toArray
+          ).zipWithIndex.map {
+            case (schema, index) => PortIdentity(index) -> schema
+          }.toMap
+        )
+      )
+      .withSuggestedWorkerNum(1)
+  }
+
+  override def operatorInfo: OperatorInfo =
+    OperatorInfo(
+      "State To Data",
+      "Convert State to Data",
+      OperatorGroupConstants.STATE_GROUP,
+      inputPorts = List(InputPort()),
+      outputPorts = List(OutputPort(PortIdentity(), "State"), OutputPort(PortIdentity(1), "Data"))
+    )
+
+  override def getOutputSchema(schemas: Array[Schema]): Schema = throw new NotImplementedError()
+
+  override def getOutputSchemas(schemas: Array[Schema]): Array[Schema] =
+    Array(
+      Schema.builder().add("passToAllDownstream", AttributeType.BOOLEAN).add(outputColumns).build(),
+      schemas(0)
+    )
+
+}

core/amber/src/main/scala/edu/uci/ics/texera/workflow/operators/state/StateToDataOpExec.scala

@@ -0,0 +1,35 @@
+package edu.uci.ics.texera.workflow.operators.state
+
+import edu.uci.ics.amber.engine.common.tuple.amber.TupleLike
+import edu.uci.ics.amber.engine.common.workflow.PortIdentity
+import edu.uci.ics.texera.workflow.common.State
+import edu.uci.ics.texera.workflow.common.operators.OperatorExecutor
+import edu.uci.ics.texera.workflow.common.tuple.Tuple
+
+import scala.collection.mutable
+
+class StateToDataOpExec extends OperatorExecutor {
+  private val outputTuples = new mutable.ArrayBuffer[(Tuple, Option[PortIdentity])]()
+  private var stateTuple: Option[Tuple] = None
+
+  override def processState(state: State, port: Int): Option[State] = {
+    if (stateTuple.isEmpty) {
+      stateTuple = Some(state.toTuple)
+      None
+    } else {
+      Some(state)
+    }
+  }
+
+  override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike] = {
+    outputTuples += ((tuple, Some(PortIdentity(1))))
+    Iterator.empty
+  }
+
+  override def onFinishMultiPort(port: Int): Iterator[(TupleLike, Option[PortIdentity])] = {
+    if (stateTuple.isDefined) {
+      outputTuples += ((stateTuple.get, Some(PortIdentity())))
+    }
+    outputTuples.iterator
+  }
+}