Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Introduce IF operator #3090

Open
wants to merge 31 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
31 commits
Select commit Hold shift + click to select a range
0245900
init
aglinxinyuan Nov 20, 2024
2f9f1a7
update
aglinxinyuan Nov 20, 2024
247e43e
fix fmt
aglinxinyuan Nov 20, 2024
a5f1e01
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Nov 21, 2024
b961031
rename op
aglinxinyuan Nov 21, 2024
ce9a777
rename op
aglinxinyuan Nov 21, 2024
d080335
rename op
aglinxinyuan Nov 21, 2024
8ba8e95
Remove "Data" the name
aglinxinyuan Nov 22, 2024
f573953
Remove (State) in the port name
aglinxinyuan Nov 22, 2024
09a4a0d
update
aglinxinyuan Nov 22, 2024
c44bb31
update
aglinxinyuan Nov 22, 2024
112965c
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Nov 22, 2024
29d4975
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Nov 22, 2024
3bb3d04
fix fmt
aglinxinyuan Nov 22, 2024
d052a43
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Nov 23, 2024
70c749a
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Nov 23, 2024
c140c83
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Nov 25, 2024
b87d37e
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Nov 27, 2024
2c95f98
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Dec 11, 2024
f4f9eb9
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Dec 12, 2024
a6b0a26
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Dec 15, 2024
cff5998
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Dec 16, 2024
c601b0d
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Dec 17, 2024
3dfa0f9
update
aglinxinyuan Dec 17, 2024
58b1650
update
aglinxinyuan Dec 17, 2024
412e728
update
aglinxinyuan Dec 17, 2024
adda4d6
update
aglinxinyuan Dec 17, 2024
e5a7a52
update
aglinxinyuan Dec 17, 2024
2749a1d
update
aglinxinyuan Dec 17, 2024
f229ae5
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Dec 17, 2024
da4abc9
Merge branch 'master' into xinyuan-if-statement
aglinxinyuan Dec 17, 2024
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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)
)
}
Original file line number Diff line number Diff line change
Expand Up @@ -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.IfOpDesc
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
Expand Down Expand Up @@ -159,6 +160,7 @@ trait StateTransferFunc
)
@JsonSubTypes(
Array(
new Type(value = classOf[IfOpDesc], name = "IfStatement"),
new Type(value = classOf[SankeyDiagramOpDesc], name = "SankeyDiagram"),
new Type(value = classOf[IcicleChartOpDesc], name = "IcicleChart"),
new Type(value = classOf[CSVScanSourceOpDesc], name = "CSVFileScan"),
Expand Down
Original file line number Diff line number Diff line change
@@ -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 IfOpDesc 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 IfOpExec(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",
"If",
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))
}
Original file line number Diff line number Diff line change
@@ -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 IfOpExec(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
}
aglinxinyuan marked this conversation as resolved.
Show resolved Hide resolved

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] = ???
}
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Loading