...
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
package de.uniol.inf.is.odysseus.core.server.logicaloperator;
import java.util.List;
import de.uniol.inf.is.odysseus.core.predicate.IPredicate;
import de.uniol.inf.is.odysseus.core.server.logicaloperator.annotations.LogicalOperator;
import de.uniol.inf.is.odysseus.core.server.logicaloperator.annotations.Parameter;
import de.uniol.inf.is.odysseus.core.server.logicaloperator.builder.PredicateParameter;
@LogicalOperator(name="ROUTE", minInputPorts=1, maxInputPorts=1, doc = "My operator is doing ...", url = "http://example.com/MyOperator.html", category = { "Category" })
public class RouteAO extends UnaryLogicalOp {
private static final long serialVersionUID = -8015847502104587689L;
public RouteAO(){
super();
}
public RouteAO(RouteAO routeAO){
super(routeAO);
}
@Override
@Parameter(type=PredicateParameter.class, isList=true, optional = false, doc = "This parameter sets the selection predicate")
public void setPredicates(List<IPredicate<?>> predicates) {
super.setPredicates(predicates);
}
@Override
public AbstractLogicalOperator clone() {
return new RouteAO(this);
}
} |
Every operator needs to provide an output schema, i.e. what is the schema of the elements that are produced. As a default implementation AbstractLogicalOperator delivers the input schema as the output schema. Route does not change the input schema, so this implementation is sufficient. If the input schema is not the same as the output schema, the class needs to implement getOutputSchemaIntern(int port), where port is the output port of the operator.
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
@Override
public OutputMode getOutputMode() {
return OutputMode.INPUT;
} |
Important: For operators with more than one input the processing must be synchronized. Especially, process_next and processPunctuation!
- process_next(R input, int port)
- Process new input element on input port, a new created element can be send to the next operator with the transfer() method
- Process new input element on input port, a new created element can be send to the next operator with the transfer() method
...
- processPunctuation(Punctuation punctuation, int port)
- This method is needed to process punctuations (e.g.heartbeat, Heartbeats allow determining the progress of time without the need to produce new elements). It is important, that punctuations and elements are timely ordered, i.e. if a punctuation is send, no other element is allowed to be send with an older timestamp anymore! If the operator has no state, it is enough to call sendPunctuation.
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
@Override
public void processPunctuation(PointInTime timestamp, int port) {
sendPunctuation(timestamp);
} |
- process_isSemanticallyEqual(IPhysicalOperator ipo)
- This method is needed for query sharing. Return true, if the operator given and the current operator are equivalent and can be replaced by each other.
| Code Block | ||||
|---|---|---|---|---|
| ||||
@Override
public boolean process_isSemanticallyEqual(IPhysicalOperator ipo) {
if(!(ipo instanceof RoutePO)) {
return false;
}
RoutePO spo = (RoutePO) ipo;
if(this.hasSameSources(spo) &&
this.predicates.size() == spo.predicates.size()) {
for(int i = 0; i<this.predicates.size(); i++) {
if(!this.predicates.get(i).equals(spo.predicates.get(i))) {
return false;
}
}
return true;
}
return false;
} |
...
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
package de.uniol.inf.is.odysseus.transform.rules;
import de.uniol.inf.is.odysseus.core.server.logicaloperator.RouteAO;
import de.uniol.inf.is.odysseus.core.server.physicaloperator.RoutePO;
import de.uniol.inf.is.odysseus.core.server.planmanagement.TransformationConfiguration;
import de.uniol.inf.is.odysseus.ruleengine.ruleflow.IRuleFlowGroup;
import de.uniol.inf.is.odysseus.transform.flow.TransformRuleFlowGroup;
import de.uniol.inf.is.odysseus.transform.rule.AbstractTransformationRule;
@SuppressWarnings({"unchecked","rawtypes"})
public class TRouteAORule extends AbstractTransformationRule<RouteAO> {
@Override
public int getPriority() {
return 0;
}
@Override
public void execute(RouteAO routeAO, TransformationConfiguration config) {
defaultExecute(routeAO, new RoutePO(routeAO.getPredicates()), config, true, true);
}
@Override
public boolean isExecutable(RouteAO operator, TransformationConfiguration transformConfig) {
return operator.isAllPhysicalInputSet();
}
@Override
public String getName() {
return "RouteAO -> RoutePO";
}
@Override
public IRuleFlowGroup getRuleFlowGroup() {
return TransformRuleFlowGroup.TRANSFORMATION;
}
@Override
public Class<? super RouteAO> getConditionClass() {
return RouteAO.class;
}
}
|
...