Introduce IF operator

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

@@ -21,6 +21,7 @@ object OperatorGroupConstants {
   final val JAVA_GROUP = "Java"
   final val R_GROUP = "R"
   final val MACHINE_LEARNING_GENERAL_GROUP = "Machine Learning General"
+  final val CONTROL_GROUP = "Control Block"
 
   /**
     * The order of the groups to show up in the frontend operator panel.
@@ -46,6 +47,7 @@ object OperatorGroupConstants {
     GroupInfo(UTILITY_GROUP),
     GroupInfo(API_GROUP),
     GroupInfo(UDF_GROUP, List(GroupInfo(PYTHON_GROUP), GroupInfo(JAVA_GROUP), GroupInfo(R_GROUP))),
-    GroupInfo(VISUALIZATION_GROUP)
+    GroupInfo(VISUALIZATION_GROUP),
+    GroupInfo(CONTROL_GROUP)
   )
 }

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

@@ -22,6 +22,7 @@ import edu.uci.ics.texera.web.OPversion
 import edu.uci.ics.texera.workflow.common.metadata.{OperatorInfo, PropertyNameConstants}
 import edu.uci.ics.texera.workflow.operators.aggregate.AggregateOpDesc
 import edu.uci.ics.texera.workflow.operators.cartesianProduct.CartesianProductOpDesc
+import edu.uci.ics.texera.workflow.operators.controlBlock.ifStatement.IfStatementOpDesc
 import edu.uci.ics.texera.workflow.operators.dictionary.DictionaryMatcherOpDesc
 import edu.uci.ics.texera.workflow.operators.difference.DifferenceOpDesc
 import edu.uci.ics.texera.workflow.operators.distinct.DistinctOpDesc
@@ -159,6 +160,7 @@ trait StateTransferFunc
 )
 @JsonSubTypes(
   Array(
+    new Type(value = classOf[IfStatementOpDesc], name = "IfStatement"),
     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/controlBlock/ifStatement/IfStatementOpDesc.scala

@@ -0,0 +1,61 @@
+package edu.uci.ics.texera.workflow.operators.controlBlock.ifStatement
+
+import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
+import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
+import edu.uci.ics.amber.engine.architecture.deploysemantics.layer.OpExecInitInfo
+import edu.uci.ics.amber.engine.common.model.{PhysicalOp, SchemaPropagationFunc}
+import edu.uci.ics.amber.engine.common.model.tuple.Schema
+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
+
+class IfStatementOpDesc extends LogicalOp {
+  @JsonProperty(required = true)
+  @JsonSchemaTitle("Condition State")
+  @JsonPropertyDescription("name of the state variable to evaluate")
+  var conditionName: String = _
+
+  override def getPhysicalOp(
+      workflowId: WorkflowIdentity,
+      executionId: ExecutionIdentity
+  ): PhysicalOp = {
+    PhysicalOp
+      .oneToOnePhysicalOp(
+        workflowId,
+        executionId,
+        operatorIdentifier,
+        OpExecInitInfo((_, _) => {
+          new IfStatementOpExec(conditionName)
+        })
+      )
+      .withInputPorts(operatorInfo.inputPorts)
+      .withOutputPorts(operatorInfo.outputPorts)
+      .withParallelizable(false)
+      .withPropagateSchema(
+        SchemaPropagationFunc(inputSchemas =>
+          operatorInfo.outputPorts
+            .map(_.id)
+            .map(id => id -> inputSchemas(operatorInfo.inputPorts.last.id))
+            .toMap
+        )
+      )
+  }
+
+  override def operatorInfo: OperatorInfo =
+    OperatorInfo(
+      "If Statement",
+      "If Statement",
+      OperatorGroupConstants.CONTROL_GROUP,
+      inputPorts = List(
+        InputPort(PortIdentity(), "Condition(State)"),
+        InputPort(PortIdentity(1), "Data", dependencies = List(PortIdentity()))
+      ),
+      outputPorts = List(OutputPort(PortIdentity(), "True"), OutputPort(PortIdentity(1), "False"))
+    )
+
+  override def getOutputSchema(schemas: Array[Schema]): Schema = throw new NotImplementedError()
+
+  override def getOutputSchemas(schemas: Array[Schema]): Array[Schema] =
+    Array(schemas(1), schemas(1))
+}

core/amber/src/main/scala/edu/uci/ics/texera/workflow/operators/controlBlock/ifStatement/IfStatementOpExec.scala

@@ -0,0 +1,23 @@
+package edu.uci.ics.texera.workflow.operators.controlBlock.ifStatement
+
+import edu.uci.ics.amber.engine.common.executor.OperatorExecutor
+import edu.uci.ics.amber.engine.common.model.State
+import edu.uci.ics.amber.engine.common.model.tuple.{Tuple, TupleLike}
+import edu.uci.ics.amber.engine.common.workflow.PortIdentity
+
+class IfStatementOpExec(conditionName: String) extends OperatorExecutor {
+  private var output_port: Int = _
+
+  override def processState(state: State, port: Int): Option[State] = {
+    output_port = if (state.get(conditionName).asInstanceOf[Boolean]) 0 else 1
+    None
+  }
+
+  override def processTupleMultiPort(
+      tuple: Tuple,
+      port: Int
+  ): Iterator[(TupleLike, Option[PortIdentity])] =
+    Iterator((tuple, Some(PortIdentity(output_port))))
+
+  override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike] = ???
+}