You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/03/20 11:06:56 UTC
[17/53] [abbrv] flink git commit: [optimizer] Rename optimizer
project to "flink-optimizer" (previously flink-compiler)
http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SingleInputNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SingleInputNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SingleInputNode.java
new file mode 100644
index 0000000..cc12bb8
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SingleInputNode.java
@@ -0,0 +1,518 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.dag;
+
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE_AND_DAM;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.NOT_FOUND;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.flink.api.common.ExecutionMode;
+import org.apache.flink.api.common.operators.Operator;
+import org.apache.flink.api.common.operators.SemanticProperties;
+import org.apache.flink.api.common.operators.SingleInputOperator;
+import org.apache.flink.api.common.operators.util.FieldSet;
+import org.apache.flink.optimizer.CompilerException;
+import org.apache.flink.optimizer.Optimizer;
+import org.apache.flink.optimizer.costs.CostEstimator;
+import org.apache.flink.optimizer.dataproperties.GlobalProperties;
+import org.apache.flink.optimizer.dataproperties.InterestingProperties;
+import org.apache.flink.optimizer.dataproperties.LocalProperties;
+import org.apache.flink.optimizer.dataproperties.RequestedGlobalProperties;
+import org.apache.flink.optimizer.dataproperties.RequestedLocalProperties;
+import org.apache.flink.optimizer.operators.OperatorDescriptorSingle;
+import org.apache.flink.optimizer.plan.Channel;
+import org.apache.flink.optimizer.plan.NamedChannel;
+import org.apache.flink.optimizer.plan.PlanNode;
+import org.apache.flink.optimizer.plan.SingleInputPlanNode;
+import org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport;
+import org.apache.flink.optimizer.util.NoOpUnaryUdfOp;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.DataExchangeMode;
+import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
+import org.apache.flink.util.Visitor;
+
+import com.google.common.collect.Sets;
+
+/**
+ * A node in the optimizer's program representation for an operation with a single input.
+ *
+ * This class contains all the generic logic for handling branching flows, as well as to
+ * enumerate candidate execution plans. The subclasses for specific operators simply add logic
+ * for cost estimates and specify possible strategies for their execution.
+ */
+public abstract class SingleInputNode extends OptimizerNode {
+
+ protected final FieldSet keys; // The set of key fields
+
+ protected DagConnection inConn; // the input of the node
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Creates a new node with a single input for the optimizer plan.
+ *
+ * @param programOperator The PACT that the node represents.
+ */
+ protected SingleInputNode(SingleInputOperator<?, ?, ?> programOperator) {
+ super(programOperator);
+
+ int[] k = programOperator.getKeyColumns(0);
+ this.keys = k == null || k.length == 0 ? null : new FieldSet(k);
+ }
+
+ protected SingleInputNode(FieldSet keys) {
+ super(NoOpUnaryUdfOp.INSTANCE);
+ this.keys = keys;
+ }
+
+ protected SingleInputNode() {
+ super(NoOpUnaryUdfOp.INSTANCE);
+ this.keys = null;
+ }
+
+ protected SingleInputNode(SingleInputNode toCopy) {
+ super(toCopy);
+
+ this.keys = toCopy.keys;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public SingleInputOperator<?, ?, ?> getOperator() {
+ return (SingleInputOperator<?, ?, ?>) super.getOperator();
+ }
+
+ /**
+ * Gets the input of this operator.
+ *
+ * @return The input.
+ */
+ public DagConnection getIncomingConnection() {
+ return this.inConn;
+ }
+
+ /**
+ * Sets the connection through which this node receives its input.
+ *
+ * @param inConn The input connection to set.
+ */
+ public void setIncomingConnection(DagConnection inConn) {
+ this.inConn = inConn;
+ }
+
+ /**
+ * Gets the predecessor of this node.
+ *
+ * @return The predecessor of this node.
+ */
+ public OptimizerNode getPredecessorNode() {
+ if (this.inConn != null) {
+ return this.inConn.getSource();
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public List<DagConnection> getIncomingConnections() {
+ return Collections.singletonList(this.inConn);
+ }
+
+
+ @Override
+ public SemanticProperties getSemanticProperties() {
+ return getOperator().getSemanticProperties();
+ }
+
+
+ @Override
+ public void setInput(Map<Operator<?>, OptimizerNode> contractToNode, ExecutionMode defaultExchangeMode)
+ throws CompilerException
+ {
+ // see if an internal hint dictates the strategy to use
+ final Configuration conf = getOperator().getParameters();
+ final String shipStrategy = conf.getString(Optimizer.HINT_SHIP_STRATEGY, null);
+ final ShipStrategyType preSet;
+
+ if (shipStrategy != null) {
+ if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_REPARTITION_HASH)) {
+ preSet = ShipStrategyType.PARTITION_HASH;
+ } else if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_REPARTITION_RANGE)) {
+ preSet = ShipStrategyType.PARTITION_RANGE;
+ } else if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_FORWARD)) {
+ preSet = ShipStrategyType.FORWARD;
+ } else if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_REPARTITION)) {
+ preSet = ShipStrategyType.PARTITION_RANDOM;
+ } else {
+ throw new CompilerException("Unrecognized ship strategy hint: " + shipStrategy);
+ }
+ } else {
+ preSet = null;
+ }
+
+ // get the predecessor node
+ Operator<?> children = ((SingleInputOperator<?, ?, ?>) getOperator()).getInput();
+
+ OptimizerNode pred;
+ DagConnection conn;
+ if (children == null) {
+ throw new CompilerException("Error: Node for '" + getOperator().getName() + "' has no input.");
+ } else {
+ pred = contractToNode.get(children);
+ conn = new DagConnection(pred, this, defaultExchangeMode);
+ if (preSet != null) {
+ conn.setShipStrategy(preSet);
+ }
+ }
+
+ // create the connection and add it
+ setIncomingConnection(conn);
+ pred.addOutgoingConnection(conn);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Properties and Optimization
+ // --------------------------------------------------------------------------------------------
+
+ protected abstract List<OperatorDescriptorSingle> getPossibleProperties();
+
+ @Override
+ public void computeInterestingPropertiesForInputs(CostEstimator estimator) {
+ // get what we inherit and what is preserved by our user code
+ final InterestingProperties props = getInterestingProperties().filterByCodeAnnotations(this, 0);
+
+ // add all properties relevant to this node
+ for (OperatorDescriptorSingle dps : getPossibleProperties()) {
+ for (RequestedGlobalProperties gp : dps.getPossibleGlobalProperties()) {
+
+ if (gp.getPartitioning().isPartitionedOnKey()) {
+ // make sure that among the same partitioning types, we do not push anything down that has fewer key fields
+
+ for (RequestedGlobalProperties contained : props.getGlobalProperties()) {
+ if (contained.getPartitioning() == gp.getPartitioning() && gp.getPartitionedFields().isValidSubset(contained.getPartitionedFields())) {
+ props.getGlobalProperties().remove(contained);
+ break;
+ }
+ }
+ }
+
+ props.addGlobalProperties(gp);
+ }
+
+ for (RequestedLocalProperties lp : dps.getPossibleLocalProperties()) {
+ props.addLocalProperties(lp);
+ }
+ }
+ this.inConn.setInterestingProperties(props);
+
+ for (DagConnection conn : getBroadcastConnections()) {
+ conn.setInterestingProperties(new InterestingProperties());
+ }
+ }
+
+
+ @Override
+ public List<PlanNode> getAlternativePlans(CostEstimator estimator) {
+ // check if we have a cached version
+ if (this.cachedPlans != null) {
+ return this.cachedPlans;
+ }
+
+ boolean childrenSkippedDueToReplicatedInput = false;
+
+ // calculate alternative sub-plans for predecessor
+ final List<? extends PlanNode> subPlans = getPredecessorNode().getAlternativePlans(estimator);
+ final Set<RequestedGlobalProperties> intGlobal = this.inConn.getInterestingProperties().getGlobalProperties();
+
+ // calculate alternative sub-plans for broadcast inputs
+ final List<Set<? extends NamedChannel>> broadcastPlanChannels = new ArrayList<Set<? extends NamedChannel>>();
+ List<DagConnection> broadcastConnections = getBroadcastConnections();
+ List<String> broadcastConnectionNames = getBroadcastConnectionNames();
+
+ for (int i = 0; i < broadcastConnections.size(); i++ ) {
+ DagConnection broadcastConnection = broadcastConnections.get(i);
+ String broadcastConnectionName = broadcastConnectionNames.get(i);
+ List<PlanNode> broadcastPlanCandidates = broadcastConnection.getSource().getAlternativePlans(estimator);
+
+ // wrap the plan candidates in named channels
+ HashSet<NamedChannel> broadcastChannels = new HashSet<NamedChannel>(broadcastPlanCandidates.size());
+ for (PlanNode plan: broadcastPlanCandidates) {
+ NamedChannel c = new NamedChannel(broadcastConnectionName, plan);
+ DataExchangeMode exMode = DataExchangeMode.select(broadcastConnection.getDataExchangeMode(),
+ ShipStrategyType.BROADCAST, broadcastConnection.isBreakingPipeline());
+ c.setShipStrategy(ShipStrategyType.BROADCAST, exMode);
+ broadcastChannels.add(c);
+ }
+ broadcastPlanChannels.add(broadcastChannels);
+ }
+
+ final RequestedGlobalProperties[] allValidGlobals;
+ {
+ Set<RequestedGlobalProperties> pairs = new HashSet<RequestedGlobalProperties>();
+ for (OperatorDescriptorSingle ods : getPossibleProperties()) {
+ pairs.addAll(ods.getPossibleGlobalProperties());
+ }
+ allValidGlobals = pairs.toArray(new RequestedGlobalProperties[pairs.size()]);
+ }
+ final ArrayList<PlanNode> outputPlans = new ArrayList<PlanNode>();
+
+ final ExecutionMode executionMode = this.inConn.getDataExchangeMode();
+
+ final int dop = getParallelism();
+ final int inDop = getPredecessorNode().getParallelism();
+ final boolean dopChange = inDop != dop;
+
+ final boolean breaksPipeline = this.inConn.isBreakingPipeline();
+
+ // create all candidates
+ for (PlanNode child : subPlans) {
+
+ if (child.getGlobalProperties().isFullyReplicated()) {
+ // fully replicated input is always locally forwarded if DOP is not changed
+ if (dopChange) {
+ // can not continue with this child
+ childrenSkippedDueToReplicatedInput = true;
+ continue;
+ } else {
+ this.inConn.setShipStrategy(ShipStrategyType.FORWARD);
+ }
+ }
+
+ if (this.inConn.getShipStrategy() == null) {
+ // pick the strategy ourselves
+ for (RequestedGlobalProperties igps: intGlobal) {
+ final Channel c = new Channel(child, this.inConn.getMaterializationMode());
+ igps.parameterizeChannel(c, dopChange, executionMode, breaksPipeline);
+
+ // if the DOP changed, make sure that we cancel out properties, unless the
+ // ship strategy preserves/establishes them even under changing DOPs
+ if (dopChange && !c.getShipStrategy().isNetworkStrategy()) {
+ c.getGlobalProperties().reset();
+ }
+
+ // check whether we meet any of the accepted properties
+ // we may remove this check, when we do a check to not inherit
+ // requested global properties that are incompatible with all possible
+ // requested properties
+ for (RequestedGlobalProperties rgps: allValidGlobals) {
+ if (rgps.isMetBy(c.getGlobalProperties())) {
+ c.setRequiredGlobalProps(rgps);
+ addLocalCandidates(c, broadcastPlanChannels, igps, outputPlans, estimator);
+ break;
+ }
+ }
+ }
+ } else {
+ // hint fixed the strategy
+ final Channel c = new Channel(child, this.inConn.getMaterializationMode());
+ final ShipStrategyType shipStrategy = this.inConn.getShipStrategy();
+ final DataExchangeMode exMode = DataExchangeMode.select(executionMode, shipStrategy, breaksPipeline);
+
+ if (this.keys != null) {
+ c.setShipStrategy(shipStrategy, this.keys.toFieldList(), exMode);
+ } else {
+ c.setShipStrategy(shipStrategy, exMode);
+ }
+
+ if (dopChange) {
+ c.adjustGlobalPropertiesForFullParallelismChange();
+ }
+
+ // check whether we meet any of the accepted properties
+ for (RequestedGlobalProperties rgps: allValidGlobals) {
+ if (rgps.isMetBy(c.getGlobalProperties())) {
+ addLocalCandidates(c, broadcastPlanChannels, rgps, outputPlans, estimator);
+ break;
+ }
+ }
+ }
+ }
+
+ if(outputPlans.isEmpty()) {
+ if(childrenSkippedDueToReplicatedInput) {
+ throw new CompilerException("No plan meeting the requirements could be created @ " + this + ". Most likely reason: Invalid use of replicated input.");
+ } else {
+ throw new CompilerException("No plan meeting the requirements could be created @ " + this + ". Most likely reason: Too restrictive plan hints.");
+ }
+ }
+
+ // cost and prune the plans
+ for (PlanNode node : outputPlans) {
+ estimator.costOperator(node);
+ }
+ prunePlanAlternatives(outputPlans);
+ outputPlans.trimToSize();
+
+ this.cachedPlans = outputPlans;
+ return outputPlans;
+ }
+
+ protected void addLocalCandidates(Channel template, List<Set<? extends NamedChannel>> broadcastPlanChannels, RequestedGlobalProperties rgps,
+ List<PlanNode> target, CostEstimator estimator)
+ {
+ for (RequestedLocalProperties ilp : this.inConn.getInterestingProperties().getLocalProperties()) {
+ final Channel in = template.clone();
+ ilp.parameterizeChannel(in);
+
+ // instantiate a candidate, if the instantiated local properties meet one possible local property set
+ outer:
+ for (OperatorDescriptorSingle dps: getPossibleProperties()) {
+ for (RequestedLocalProperties ilps : dps.getPossibleLocalProperties()) {
+ if (ilps.isMetBy(in.getLocalProperties())) {
+ in.setRequiredLocalProps(ilps);
+ instantiateCandidate(dps, in, broadcastPlanChannels, target, estimator, rgps, ilp);
+ break outer;
+ }
+ }
+ }
+ }
+ }
+
+ protected void instantiateCandidate(OperatorDescriptorSingle dps, Channel in, List<Set<? extends NamedChannel>> broadcastPlanChannels,
+ List<PlanNode> target, CostEstimator estimator, RequestedGlobalProperties globPropsReq, RequestedLocalProperties locPropsReq)
+ {
+ final PlanNode inputSource = in.getSource();
+
+ for (List<NamedChannel> broadcastChannelsCombination: Sets.cartesianProduct(broadcastPlanChannels)) {
+
+ boolean validCombination = true;
+ boolean requiresPipelinebreaker = false;
+
+ // check whether the broadcast inputs use the same plan candidate at the branching point
+ for (int i = 0; i < broadcastChannelsCombination.size(); i++) {
+ NamedChannel nc = broadcastChannelsCombination.get(i);
+ PlanNode bcSource = nc.getSource();
+
+ // check branch compatibility against input
+ if (!areBranchCompatible(bcSource, inputSource)) {
+ validCombination = false;
+ break;
+ }
+
+ // check branch compatibility against all other broadcast variables
+ for (int k = 0; k < i; k++) {
+ PlanNode otherBcSource = broadcastChannelsCombination.get(k).getSource();
+
+ if (!areBranchCompatible(bcSource, otherBcSource)) {
+ validCombination = false;
+ break;
+ }
+ }
+
+ // check if there is a common predecessor and whether there is a dam on the way to all common predecessors
+ if (this.hereJoinedBranches != null) {
+ for (OptimizerNode brancher : this.hereJoinedBranches) {
+ PlanNode candAtBrancher = in.getSource().getCandidateAtBranchPoint(brancher);
+
+ if (candAtBrancher == null) {
+ // closed branch between two broadcast variables
+ continue;
+ }
+
+ SourceAndDamReport res = in.getSource().hasDamOnPathDownTo(candAtBrancher);
+ if (res == NOT_FOUND) {
+ throw new CompilerException("Bug: Tracing dams for deadlock detection is broken.");
+ } else if (res == FOUND_SOURCE) {
+ requiresPipelinebreaker = true;
+ break;
+ } else if (res == FOUND_SOURCE_AND_DAM) {
+ // good
+ } else {
+ throw new CompilerException();
+ }
+ }
+ }
+ }
+
+ if (!validCombination) {
+ continue;
+ }
+
+ if (requiresPipelinebreaker) {
+ in.setTempMode(in.getTempMode().makePipelineBreaker());
+ }
+
+ final SingleInputPlanNode node = dps.instantiate(in, this);
+ node.setBroadcastInputs(broadcastChannelsCombination);
+
+ // compute how the strategy affects the properties
+ GlobalProperties gProps = in.getGlobalProperties().clone();
+ LocalProperties lProps = in.getLocalProperties().clone();
+ gProps = dps.computeGlobalProperties(gProps);
+ lProps = dps.computeLocalProperties(lProps);
+
+ SemanticProperties props = this.getSemanticProperties();
+ // filter by the user code field copies
+ gProps = gProps.filterBySemanticProperties(props, 0);
+ lProps = lProps.filterBySemanticProperties(props, 0);
+
+ // apply
+ node.initProperties(gProps, lProps);
+ node.updatePropertiesWithUniqueSets(getUniqueFields());
+ target.add(node);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Branch Handling
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public void computeUnclosedBranchStack() {
+ if (this.openBranches != null) {
+ return;
+ }
+
+ addClosedBranches(getPredecessorNode().closedBranchingNodes);
+ List<UnclosedBranchDescriptor> fromInput = getPredecessorNode().getBranchesForParent(this.inConn);
+
+ // handle the data flow branching for the broadcast inputs
+ List<UnclosedBranchDescriptor> result = computeUnclosedBranchStackForBroadcastInputs(fromInput);
+
+ this.openBranches = (result == null || result.isEmpty()) ? Collections.<UnclosedBranchDescriptor>emptyList() : result;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Miscellaneous
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public void accept(Visitor<OptimizerNode> visitor) {
+ if (visitor.preVisit(this)) {
+ if (getPredecessorNode() != null) {
+ getPredecessorNode().accept(visitor);
+ } else {
+ throw new CompilerException();
+ }
+ for (DagConnection connection : getBroadcastConnections()) {
+ connection.getSource().accept(visitor);
+ }
+ visitor.postVisit(this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SinkJoiner.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SinkJoiner.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SinkJoiner.java
new file mode 100644
index 0000000..40725ba
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SinkJoiner.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.dag;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.flink.api.common.ExecutionMode;
+import org.apache.flink.api.common.typeinfo.NothingTypeInfo;
+import org.apache.flink.optimizer.DataStatistics;
+import org.apache.flink.optimizer.operators.OperatorDescriptorDual;
+import org.apache.flink.optimizer.operators.UtilSinkJoinOpDescriptor;
+import org.apache.flink.optimizer.util.NoOpBinaryUdfOp;
+import org.apache.flink.types.Nothing;
+
+/**
+ * This class represents a utility node that is not part of the actual plan.
+ * It is used for plans with multiple data sinks to transform it into a plan with
+ * a single root node. That way, the code that makes sure no costs are double-counted and that
+ * candidate selection works correctly with nodes that have multiple outputs is transparently reused.
+ */
+public class SinkJoiner extends TwoInputNode {
+
+ public SinkJoiner(OptimizerNode input1, OptimizerNode input2) {
+ super(new NoOpBinaryUdfOp<Nothing>(new NothingTypeInfo()));
+
+ DagConnection conn1 = new DagConnection(input1, this, null, ExecutionMode.PIPELINED);
+ DagConnection conn2 = new DagConnection(input2, this, null, ExecutionMode.PIPELINED);
+
+ this.input1 = conn1;
+ this.input2 = conn2;
+
+ setDegreeOfParallelism(1);
+ }
+
+ @Override
+ public String getName() {
+ return "Internal Utility Node";
+ }
+
+ @Override
+ public List<DagConnection> getOutgoingConnections() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void computeUnclosedBranchStack() {
+ if (this.openBranches != null) {
+ return;
+ }
+
+ addClosedBranches(getFirstPredecessorNode().closedBranchingNodes);
+ addClosedBranches(getSecondPredecessorNode().closedBranchingNodes);
+
+ List<UnclosedBranchDescriptor> pred1branches = getFirstPredecessorNode().openBranches;
+ List<UnclosedBranchDescriptor> pred2branches = getSecondPredecessorNode().openBranches;
+
+ // if the predecessors do not have branches, then we have multiple sinks that do not originate from
+ // a common data flow.
+ if (pred1branches == null || pred1branches.isEmpty()) {
+
+ this.openBranches = (pred2branches == null || pred2branches.isEmpty()) ?
+ Collections.<UnclosedBranchDescriptor>emptyList() : // both empty - disconnected flow
+ pred2branches;
+ }
+ else if (pred2branches == null || pred2branches.isEmpty()) {
+ this.openBranches = pred1branches;
+ }
+ else {
+ // copy the lists and merge
+ List<UnclosedBranchDescriptor> result1 = new ArrayList<UnclosedBranchDescriptor>(pred1branches);
+ List<UnclosedBranchDescriptor> result2 = new ArrayList<UnclosedBranchDescriptor>(pred2branches);
+
+ ArrayList<UnclosedBranchDescriptor> result = new ArrayList<UnclosedBranchDescriptor>();
+ mergeLists(result1, result2, result, false);
+
+ this.openBranches = result.isEmpty() ? Collections.<UnclosedBranchDescriptor>emptyList() : result;
+ }
+ }
+
+ @Override
+ protected List<OperatorDescriptorDual> getPossibleProperties() {
+ return Collections.<OperatorDescriptorDual>singletonList(new UtilSinkJoinOpDescriptor());
+ }
+
+ @Override
+ public void computeOutputEstimates(DataStatistics statistics) {
+ // nothing to be done here
+ }
+
+ @Override
+ protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) {
+ // no estimates needed at this point
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SolutionSetNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SolutionSetNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SolutionSetNode.java
new file mode 100644
index 0000000..1292cf5
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SolutionSetNode.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.dag;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.flink.api.common.operators.base.DeltaIterationBase.SolutionSetPlaceHolder;
+import org.apache.flink.optimizer.DataStatistics;
+import org.apache.flink.optimizer.dataproperties.GlobalProperties;
+import org.apache.flink.optimizer.dataproperties.LocalProperties;
+import org.apache.flink.optimizer.plan.Channel;
+import org.apache.flink.optimizer.plan.PlanNode;
+import org.apache.flink.optimizer.plan.SolutionSetPlanNode;
+
+/**
+ * The optimizer's internal representation of the solution set of a workset iteration.
+ */
+public class SolutionSetNode extends AbstractPartialSolutionNode {
+
+ private final WorksetIterationNode iterationNode;
+
+
+ public SolutionSetNode(SolutionSetPlaceHolder<?> psph, WorksetIterationNode iterationNode) {
+ super(psph);
+ this.iterationNode = iterationNode;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public void setCandidateProperties(GlobalProperties gProps, LocalProperties lProps, Channel initialInput) {
+ this.cachedPlans = Collections.<PlanNode>singletonList(new SolutionSetPlanNode(this, "SolutionSet("+this.getOperator().getName()+")", gProps, lProps, initialInput));
+ }
+
+ public SolutionSetPlanNode getCurrentSolutionSetPlanNode() {
+ if (this.cachedPlans != null) {
+ return (SolutionSetPlanNode) this.cachedPlans.get(0);
+ } else {
+ throw new IllegalStateException();
+ }
+ }
+
+ public WorksetIterationNode getIterationNode() {
+ return this.iterationNode;
+ }
+
+ @Override
+ public void computeOutputEstimates(DataStatistics statistics) {
+ copyEstimates(this.iterationNode.getInitialSolutionSetPredecessorNode());
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Gets the contract object for this data source node.
+ *
+ * @return The contract.
+ */
+ @Override
+ public SolutionSetPlaceHolder<?> getOperator() {
+ return (SolutionSetPlaceHolder<?>) super.getOperator();
+ }
+
+ @Override
+ public String getName() {
+ return "Solution Set";
+ }
+
+ @Override
+ public void computeUnclosedBranchStack() {
+ if (this.openBranches != null) {
+ return;
+ }
+
+ DagConnection solutionSetInput = this.iterationNode.getFirstIncomingConnection();
+ OptimizerNode solutionSetSource = solutionSetInput.getSource();
+
+ addClosedBranches(solutionSetSource.closedBranchingNodes);
+ List<UnclosedBranchDescriptor> fromInput = solutionSetSource.getBranchesForParent(solutionSetInput);
+ this.openBranches = (fromInput == null || fromInput.isEmpty()) ? Collections.<UnclosedBranchDescriptor>emptyList() : fromInput;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SortPartitionNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SortPartitionNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SortPartitionNode.java
new file mode 100644
index 0000000..83bc39a
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/SortPartitionNode.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.dag;
+
+import org.apache.flink.api.common.operators.Ordering;
+import org.apache.flink.api.common.operators.SemanticProperties;
+import org.apache.flink.api.common.operators.SingleInputSemanticProperties;
+import org.apache.flink.api.common.operators.base.SortPartitionOperatorBase;
+import org.apache.flink.optimizer.DataStatistics;
+import org.apache.flink.optimizer.dataproperties.GlobalProperties;
+import org.apache.flink.optimizer.dataproperties.LocalProperties;
+import org.apache.flink.optimizer.dataproperties.RequestedGlobalProperties;
+import org.apache.flink.optimizer.dataproperties.RequestedLocalProperties;
+import org.apache.flink.optimizer.operators.OperatorDescriptorSingle;
+import org.apache.flink.optimizer.plan.Channel;
+import org.apache.flink.optimizer.plan.SingleInputPlanNode;
+import org.apache.flink.runtime.operators.DriverStrategy;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * The optimizer's internal representation of a <i>SortPartition</i> operator node.
+ */
+public class SortPartitionNode extends SingleInputNode {
+
+ private final List<OperatorDescriptorSingle> possibleProperties;
+
+ public SortPartitionNode(SortPartitionOperatorBase<?> operator) {
+ super(operator);
+
+ OperatorDescriptorSingle descr = new SortPartitionDescriptor(operator.getPartitionOrdering());
+ this.possibleProperties = Collections.singletonList(descr);
+ }
+
+ @Override
+ public SortPartitionOperatorBase<?> getOperator() {
+ return (SortPartitionOperatorBase<?>) super.getOperator();
+ }
+
+ @Override
+ public String getName() {
+ return "Sort-Partition";
+ }
+
+ @Override
+ protected List<OperatorDescriptorSingle> getPossibleProperties() {
+ return this.possibleProperties;
+ }
+
+ @Override
+ protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) {
+ // sorting does not change the number of records
+ this.estimatedNumRecords = getPredecessorNode().getEstimatedNumRecords();
+ this.estimatedOutputSize = getPredecessorNode().getEstimatedOutputSize();
+ }
+
+ @Override
+ public SemanticProperties getSemanticProperties() {
+ return new SingleInputSemanticProperties.AllFieldsForwardedProperties();
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static class SortPartitionDescriptor extends OperatorDescriptorSingle {
+
+ private Ordering partitionOrder;
+
+ public SortPartitionDescriptor(Ordering partitionOrder) {
+ this.partitionOrder = partitionOrder;
+ }
+
+ @Override
+ public DriverStrategy getStrategy() {
+ return DriverStrategy.UNARY_NO_OP;
+ }
+
+ @Override
+ public SingleInputPlanNode instantiate(Channel in, SingleInputNode node) {
+ return new SingleInputPlanNode(node, "Sort-Partition", in, DriverStrategy.UNARY_NO_OP);
+ }
+
+ @Override
+ protected List<RequestedGlobalProperties> createPossibleGlobalProperties() {
+ // sort partition does not require any global property
+ return Collections.singletonList(new RequestedGlobalProperties());
+ }
+
+ @Override
+ protected List<RequestedLocalProperties> createPossibleLocalProperties() {
+ // set partition order as required local property
+ RequestedLocalProperties rlp = new RequestedLocalProperties();
+ rlp.setOrdering(this.partitionOrder);
+
+ return Collections.singletonList(rlp);
+ }
+
+ @Override
+ public GlobalProperties computeGlobalProperties(GlobalProperties gProps) {
+ // sort partition is a no-operation operation, such that all global properties are preserved.
+ return gProps;
+ }
+
+ @Override
+ public LocalProperties computeLocalProperties(LocalProperties lProps) {
+ // sort partition is a no-operation operation, such that all global properties are preserved.
+ return lProps;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/TempMode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/TempMode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/TempMode.java
new file mode 100644
index 0000000..0d1dfc9
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/TempMode.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.dag;
+
+/**
+ * Enumeration to indicate the mode of temporarily materializing the data that flows across a connection.
+ * Introducing such an artificial dam is sometimes necessary to avoid that a certain data flows deadlock
+ * themselves, or as a cache to replay an intermediate result.
+ */
+public enum TempMode {
+
+ NONE(false, false),
+ PIPELINE_BREAKER(false, true),
+ CACHED(true, false),
+ CACHING_PIPELINE_BREAKER(true, true);
+
+ // --------------------------------------------------------------------------------------------
+
+ private final boolean cached;
+
+ private final boolean breaksPipeline;
+
+
+ private TempMode(boolean cached, boolean breaksPipeline) {
+ this.cached = cached;
+ this.breaksPipeline = breaksPipeline;
+ }
+
+ public boolean isCached() {
+ return cached;
+ }
+
+ public boolean breaksPipeline() {
+ return breaksPipeline;
+ }
+
+ public TempMode makePipelineBreaker() {
+ if (this == NONE) {
+ return PIPELINE_BREAKER;
+ } else if (this == CACHED) {
+ return CACHING_PIPELINE_BREAKER;
+ } else {
+ return this;
+ }
+ }
+
+ public TempMode makeCached() {
+ if (this == NONE) {
+ return CACHED;
+ } else if (this == PIPELINE_BREAKER) {
+ return CACHING_PIPELINE_BREAKER;
+ } else {
+ return this;
+ }
+ }
+
+
+ public TempMode makeNonCached() {
+ if (this == CACHED) {
+ return NONE;
+ } else if (this == CACHING_PIPELINE_BREAKER) {
+ return PIPELINE_BREAKER;
+ } else {
+ return this;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/TwoInputNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/TwoInputNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/TwoInputNode.java
new file mode 100644
index 0000000..39da165
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/TwoInputNode.java
@@ -0,0 +1,747 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.optimizer.dag;
+
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.FOUND_SOURCE_AND_DAM;
+import static org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport.NOT_FOUND;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.flink.api.common.ExecutionMode;
+import org.apache.flink.api.common.operators.DualInputOperator;
+import org.apache.flink.api.common.operators.Operator;
+import org.apache.flink.api.common.operators.SemanticProperties;
+import org.apache.flink.api.common.operators.util.FieldList;
+import org.apache.flink.optimizer.CompilerException;
+import org.apache.flink.optimizer.Optimizer;
+import org.apache.flink.optimizer.costs.CostEstimator;
+import org.apache.flink.optimizer.dataproperties.GlobalProperties;
+import org.apache.flink.optimizer.dataproperties.InterestingProperties;
+import org.apache.flink.optimizer.dataproperties.LocalProperties;
+import org.apache.flink.optimizer.dataproperties.RequestedGlobalProperties;
+import org.apache.flink.optimizer.dataproperties.RequestedLocalProperties;
+import org.apache.flink.optimizer.operators.OperatorDescriptorDual;
+import org.apache.flink.optimizer.operators.OperatorDescriptorDual.GlobalPropertiesPair;
+import org.apache.flink.optimizer.operators.OperatorDescriptorDual.LocalPropertiesPair;
+import org.apache.flink.optimizer.plan.Channel;
+import org.apache.flink.optimizer.plan.DualInputPlanNode;
+import org.apache.flink.optimizer.plan.NamedChannel;
+import org.apache.flink.optimizer.plan.PlanNode;
+import org.apache.flink.optimizer.plan.PlanNode.SourceAndDamReport;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.DataExchangeMode;
+import org.apache.flink.runtime.operators.DamBehavior;
+import org.apache.flink.runtime.operators.DriverStrategy;
+import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
+import org.apache.flink.util.Visitor;
+
+import com.google.common.collect.Sets;
+
+/**
+ * A node in the optimizer plan that represents a PACT with a two different inputs, such as MATCH or CROSS.
+ * The two inputs are not substitutable in their sides.
+ */
+public abstract class TwoInputNode extends OptimizerNode {
+
+ protected final FieldList keys1; // The set of key fields for the first input
+
+ protected final FieldList keys2; // The set of key fields for the second input
+
+ protected DagConnection input1; // The first input edge
+
+ protected DagConnection input2; // The second input edge
+
+ private List<OperatorDescriptorDual> cachedDescriptors;
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Creates a new node with a single input for the optimizer plan.
+ *
+ * @param pactContract
+ * The PACT that the node represents.
+ */
+ public TwoInputNode(DualInputOperator<?, ?, ?, ?> pactContract) {
+ super(pactContract);
+
+ int[] k1 = pactContract.getKeyColumns(0);
+ int[] k2 = pactContract.getKeyColumns(1);
+
+ this.keys1 = k1 == null || k1.length == 0 ? null : new FieldList(k1);
+ this.keys2 = k2 == null || k2.length == 0 ? null : new FieldList(k2);
+
+ if (this.keys1 != null) {
+ if (this.keys2 != null) {
+ if (this.keys1.size() != this.keys2.size()) {
+ throw new CompilerException("Unequal number of key fields on the two inputs.");
+ }
+ } else {
+ throw new CompilerException("Keys are set on first input, but not on second.");
+ }
+ } else if (this.keys2 != null) {
+ throw new CompilerException("Keys are set on second input, but not on first.");
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public DualInputOperator<?, ?, ?, ?> getOperator() {
+ return (DualInputOperator<?, ?, ?, ?>) super.getOperator();
+ }
+
+ /**
+ * Gets the <tt>PactConnection</tt> through which this node receives its <i>first</i> input.
+ *
+ * @return The first input connection.
+ */
+ public DagConnection getFirstIncomingConnection() {
+ return this.input1;
+ }
+
+ /**
+ * Gets the <tt>PactConnection</tt> through which this node receives its <i>second</i> input.
+ *
+ * @return The second input connection.
+ */
+ public DagConnection getSecondIncomingConnection() {
+ return this.input2;
+ }
+
+ public OptimizerNode getFirstPredecessorNode() {
+ if(this.input1 != null) {
+ return this.input1.getSource();
+ } else {
+ return null;
+ }
+ }
+
+ public OptimizerNode getSecondPredecessorNode() {
+ if(this.input2 != null) {
+ return this.input2.getSource();
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public List<DagConnection> getIncomingConnections() {
+ ArrayList<DagConnection> inputs = new ArrayList<DagConnection>(2);
+ inputs.add(input1);
+ inputs.add(input2);
+ return inputs;
+ }
+
+
+ @Override
+ public void setInput(Map<Operator<?>, OptimizerNode> contractToNode, ExecutionMode defaultExecutionMode) {
+ // see if there is a hint that dictates which shipping strategy to use for BOTH inputs
+ final Configuration conf = getOperator().getParameters();
+ ShipStrategyType preSet1 = null;
+ ShipStrategyType preSet2 = null;
+
+ String shipStrategy = conf.getString(Optimizer.HINT_SHIP_STRATEGY, null);
+ if (shipStrategy != null) {
+ if (Optimizer.HINT_SHIP_STRATEGY_FORWARD.equals(shipStrategy)) {
+ preSet1 = preSet2 = ShipStrategyType.FORWARD;
+ } else if (Optimizer.HINT_SHIP_STRATEGY_BROADCAST.equals(shipStrategy)) {
+ preSet1 = preSet2 = ShipStrategyType.BROADCAST;
+ } else if (Optimizer.HINT_SHIP_STRATEGY_REPARTITION_HASH.equals(shipStrategy)) {
+ preSet1 = preSet2 = ShipStrategyType.PARTITION_HASH;
+ } else if (Optimizer.HINT_SHIP_STRATEGY_REPARTITION_RANGE.equals(shipStrategy)) {
+ preSet1 = preSet2 = ShipStrategyType.PARTITION_RANGE;
+ } else if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_REPARTITION)) {
+ preSet1 = preSet2 = ShipStrategyType.PARTITION_RANDOM;
+ } else {
+ throw new CompilerException("Unknown hint for shipping strategy: " + shipStrategy);
+ }
+ }
+
+ // see if there is a hint that dictates which shipping strategy to use for the FIRST input
+ shipStrategy = conf.getString(Optimizer.HINT_SHIP_STRATEGY_FIRST_INPUT, null);
+ if (shipStrategy != null) {
+ if (Optimizer.HINT_SHIP_STRATEGY_FORWARD.equals(shipStrategy)) {
+ preSet1 = ShipStrategyType.FORWARD;
+ } else if (Optimizer.HINT_SHIP_STRATEGY_BROADCAST.equals(shipStrategy)) {
+ preSet1 = ShipStrategyType.BROADCAST;
+ } else if (Optimizer.HINT_SHIP_STRATEGY_REPARTITION_HASH.equals(shipStrategy)) {
+ preSet1 = ShipStrategyType.PARTITION_HASH;
+ } else if (Optimizer.HINT_SHIP_STRATEGY_REPARTITION_RANGE.equals(shipStrategy)) {
+ preSet1 = ShipStrategyType.PARTITION_RANGE;
+ } else if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_REPARTITION)) {
+ preSet1 = ShipStrategyType.PARTITION_RANDOM;
+ } else {
+ throw new CompilerException("Unknown hint for shipping strategy of input one: " + shipStrategy);
+ }
+ }
+
+ // see if there is a hint that dictates which shipping strategy to use for the SECOND input
+ shipStrategy = conf.getString(Optimizer.HINT_SHIP_STRATEGY_SECOND_INPUT, null);
+ if (shipStrategy != null) {
+ if (Optimizer.HINT_SHIP_STRATEGY_FORWARD.equals(shipStrategy)) {
+ preSet2 = ShipStrategyType.FORWARD;
+ } else if (Optimizer.HINT_SHIP_STRATEGY_BROADCAST.equals(shipStrategy)) {
+ preSet2 = ShipStrategyType.BROADCAST;
+ } else if (Optimizer.HINT_SHIP_STRATEGY_REPARTITION_HASH.equals(shipStrategy)) {
+ preSet2 = ShipStrategyType.PARTITION_HASH;
+ } else if (Optimizer.HINT_SHIP_STRATEGY_REPARTITION_RANGE.equals(shipStrategy)) {
+ preSet2 = ShipStrategyType.PARTITION_RANGE;
+ } else if (shipStrategy.equalsIgnoreCase(Optimizer.HINT_SHIP_STRATEGY_REPARTITION)) {
+ preSet2 = ShipStrategyType.PARTITION_RANDOM;
+ } else {
+ throw new CompilerException("Unknown hint for shipping strategy of input two: " + shipStrategy);
+ }
+ }
+
+ // get the predecessors
+ DualInputOperator<?, ?, ?, ?> contr = getOperator();
+
+ Operator<?> leftPred = contr.getFirstInput();
+ Operator<?> rightPred = contr.getSecondInput();
+
+ OptimizerNode pred1;
+ DagConnection conn1;
+ if (leftPred == null) {
+ throw new CompilerException("Error: Node for '" + getOperator().getName() + "' has no input set for first input.");
+ } else {
+ pred1 = contractToNode.get(leftPred);
+ conn1 = new DagConnection(pred1, this, defaultExecutionMode);
+ if (preSet1 != null) {
+ conn1.setShipStrategy(preSet1);
+ }
+ }
+
+ // create the connection and add it
+ this.input1 = conn1;
+ pred1.addOutgoingConnection(conn1);
+
+ OptimizerNode pred2;
+ DagConnection conn2;
+ if (rightPred == null) {
+ throw new CompilerException("Error: Node for '" + getOperator().getName() + "' has no input set for second input.");
+ } else {
+ pred2 = contractToNode.get(rightPred);
+ conn2 = new DagConnection(pred2, this, defaultExecutionMode);
+ if (preSet2 != null) {
+ conn2.setShipStrategy(preSet2);
+ }
+ }
+
+ // create the connection and add it
+ this.input2 = conn2;
+ pred2.addOutgoingConnection(conn2);
+ }
+
+ protected abstract List<OperatorDescriptorDual> getPossibleProperties();
+
+ private List<OperatorDescriptorDual> getProperties() {
+ if (this.cachedDescriptors == null) {
+ this.cachedDescriptors = getPossibleProperties();
+ }
+ return this.cachedDescriptors;
+ }
+
+ @Override
+ public void computeInterestingPropertiesForInputs(CostEstimator estimator) {
+ // get what we inherit and what is preserved by our user code
+ final InterestingProperties props1 = getInterestingProperties().filterByCodeAnnotations(this, 0);
+ final InterestingProperties props2 = getInterestingProperties().filterByCodeAnnotations(this, 1);
+
+ // add all properties relevant to this node
+ for (OperatorDescriptorDual dpd : getProperties()) {
+ for (GlobalPropertiesPair gp : dpd.getPossibleGlobalProperties()) {
+ // input 1
+ props1.addGlobalProperties(gp.getProperties1());
+
+ // input 2
+ props2.addGlobalProperties(gp.getProperties2());
+ }
+ for (LocalPropertiesPair lp : dpd.getPossibleLocalProperties()) {
+ // input 1
+ props1.addLocalProperties(lp.getProperties1());
+
+ // input 2
+ props2.addLocalProperties(lp.getProperties2());
+ }
+ }
+ this.input1.setInterestingProperties(props1);
+ this.input2.setInterestingProperties(props2);
+
+ for (DagConnection conn : getBroadcastConnections()) {
+ conn.setInterestingProperties(new InterestingProperties());
+ }
+ }
+
+ @Override
+ public List<PlanNode> getAlternativePlans(CostEstimator estimator) {
+ // check if we have a cached version
+ if (this.cachedPlans != null) {
+ return this.cachedPlans;
+ }
+
+ boolean childrenSkippedDueToReplicatedInput = false;
+
+ // step down to all producer nodes and calculate alternative plans
+ final List<? extends PlanNode> subPlans1 = getFirstPredecessorNode().getAlternativePlans(estimator);
+ final List<? extends PlanNode> subPlans2 = getSecondPredecessorNode().getAlternativePlans(estimator);
+
+ // calculate alternative sub-plans for predecessor
+ final Set<RequestedGlobalProperties> intGlobal1 = this.input1.getInterestingProperties().getGlobalProperties();
+ final Set<RequestedGlobalProperties> intGlobal2 = this.input2.getInterestingProperties().getGlobalProperties();
+
+ // calculate alternative sub-plans for broadcast inputs
+ final List<Set<? extends NamedChannel>> broadcastPlanChannels = new ArrayList<Set<? extends NamedChannel>>();
+ List<DagConnection> broadcastConnections = getBroadcastConnections();
+ List<String> broadcastConnectionNames = getBroadcastConnectionNames();
+
+ for (int i = 0; i < broadcastConnections.size(); i++ ) {
+ DagConnection broadcastConnection = broadcastConnections.get(i);
+ String broadcastConnectionName = broadcastConnectionNames.get(i);
+ List<PlanNode> broadcastPlanCandidates = broadcastConnection.getSource().getAlternativePlans(estimator);
+
+ // wrap the plan candidates in named channels
+ HashSet<NamedChannel> broadcastChannels = new HashSet<NamedChannel>(broadcastPlanCandidates.size());
+ for (PlanNode plan: broadcastPlanCandidates) {
+ final NamedChannel c = new NamedChannel(broadcastConnectionName, plan);
+ DataExchangeMode exMode = DataExchangeMode.select(broadcastConnection.getDataExchangeMode(),
+ ShipStrategyType.BROADCAST, broadcastConnection.isBreakingPipeline());
+ c.setShipStrategy(ShipStrategyType.BROADCAST, exMode);
+ broadcastChannels.add(c);
+ }
+ broadcastPlanChannels.add(broadcastChannels);
+ }
+
+ final GlobalPropertiesPair[] allGlobalPairs;
+ final LocalPropertiesPair[] allLocalPairs;
+ {
+ Set<GlobalPropertiesPair> pairsGlob = new HashSet<GlobalPropertiesPair>();
+ Set<LocalPropertiesPair> pairsLoc = new HashSet<LocalPropertiesPair>();
+ for (OperatorDescriptorDual ods : getProperties()) {
+ pairsGlob.addAll(ods.getPossibleGlobalProperties());
+ pairsLoc.addAll(ods.getPossibleLocalProperties());
+ }
+ allGlobalPairs = pairsGlob.toArray(new GlobalPropertiesPair[pairsGlob.size()]);
+ allLocalPairs = pairsLoc.toArray(new LocalPropertiesPair[pairsLoc.size()]);
+ }
+
+ final ArrayList<PlanNode> outputPlans = new ArrayList<PlanNode>();
+
+ final ExecutionMode input1Mode = this.input1.getDataExchangeMode();
+ final ExecutionMode input2Mode = this.input2.getDataExchangeMode();
+
+ final int dop = getParallelism();
+ final int inDop1 = getFirstPredecessorNode().getParallelism();
+ final int inDop2 = getSecondPredecessorNode().getParallelism();
+
+ final boolean dopChange1 = dop != inDop1;
+ final boolean dopChange2 = dop != inDop2;
+
+ final boolean input1breaksPipeline = this.input1.isBreakingPipeline();
+ final boolean input2breaksPipeline = this.input2.isBreakingPipeline();
+
+ // enumerate all pairwise combination of the children's plans together with
+ // all possible operator strategy combination
+
+ // create all candidates
+ for (PlanNode child1 : subPlans1) {
+
+ if (child1.getGlobalProperties().isFullyReplicated()) {
+ // fully replicated input is always locally forwarded if DOP is not changed
+ if (dopChange1) {
+ // can not continue with this child
+ childrenSkippedDueToReplicatedInput = true;
+ continue;
+ } else {
+ this.input1.setShipStrategy(ShipStrategyType.FORWARD);
+ }
+ }
+
+ for (PlanNode child2 : subPlans2) {
+
+ if (child2.getGlobalProperties().isFullyReplicated()) {
+ // fully replicated input is always locally forwarded if DOP is not changed
+ if (dopChange2) {
+ // can not continue with this child
+ childrenSkippedDueToReplicatedInput = true;
+ continue;
+ } else {
+ this.input2.setShipStrategy(ShipStrategyType.FORWARD);
+ }
+ }
+
+ // check that the children go together. that is the case if they build upon the same
+ // candidate at the joined branch plan.
+ if (!areBranchCompatible(child1, child2)) {
+ continue;
+ }
+
+ for (RequestedGlobalProperties igps1: intGlobal1) {
+ // create a candidate channel for the first input. mark it cached, if the connection says so
+ final Channel c1 = new Channel(child1, this.input1.getMaterializationMode());
+ if (this.input1.getShipStrategy() == null) {
+ // free to choose the ship strategy
+ igps1.parameterizeChannel(c1, dopChange1, input1Mode, input1breaksPipeline);
+
+ // if the DOP changed, make sure that we cancel out properties, unless the
+ // ship strategy preserves/establishes them even under changing DOPs
+ if (dopChange1 && !c1.getShipStrategy().isNetworkStrategy()) {
+ c1.getGlobalProperties().reset();
+ }
+ }
+ else {
+ // ship strategy fixed by compiler hint
+ ShipStrategyType shipType = this.input1.getShipStrategy();
+ DataExchangeMode exMode = DataExchangeMode.select(input1Mode, shipType, input1breaksPipeline);
+ if (this.keys1 != null) {
+ c1.setShipStrategy(shipType, this.keys1.toFieldList(), exMode);
+ }
+ else {
+ c1.setShipStrategy(shipType, exMode);
+ }
+
+ if (dopChange1) {
+ c1.adjustGlobalPropertiesForFullParallelismChange();
+ }
+ }
+
+ for (RequestedGlobalProperties igps2: intGlobal2) {
+ // create a candidate channel for the first input. mark it cached, if the connection says so
+ final Channel c2 = new Channel(child2, this.input2.getMaterializationMode());
+ if (this.input2.getShipStrategy() == null) {
+ // free to choose the ship strategy
+ igps2.parameterizeChannel(c2, dopChange2, input2Mode, input2breaksPipeline);
+
+ // if the DOP changed, make sure that we cancel out properties, unless the
+ // ship strategy preserves/establishes them even under changing DOPs
+ if (dopChange2 && !c2.getShipStrategy().isNetworkStrategy()) {
+ c2.getGlobalProperties().reset();
+ }
+ } else {
+ // ship strategy fixed by compiler hint
+ ShipStrategyType shipType = this.input2.getShipStrategy();
+ DataExchangeMode exMode = DataExchangeMode.select(input2Mode, shipType, input2breaksPipeline);
+ if (this.keys2 != null) {
+ c2.setShipStrategy(shipType, this.keys2.toFieldList(), exMode);
+ } else {
+ c2.setShipStrategy(shipType, exMode);
+ }
+
+ if (dopChange2) {
+ c2.adjustGlobalPropertiesForFullParallelismChange();
+ }
+ }
+
+ /* ********************************************************************
+ * NOTE: Depending on how we proceed with different partitioning,
+ * we might at some point need a compatibility check between
+ * the pairs of global properties.
+ * *******************************************************************/
+
+ outer:
+ for (GlobalPropertiesPair gpp : allGlobalPairs) {
+ if (gpp.getProperties1().isMetBy(c1.getGlobalProperties()) &&
+ gpp.getProperties2().isMetBy(c2.getGlobalProperties()) )
+ {
+ for (OperatorDescriptorDual desc : getProperties()) {
+ if (desc.areCompatible(gpp.getProperties1(), gpp.getProperties2(),
+ c1.getGlobalProperties(), c2.getGlobalProperties()))
+ {
+ Channel c1Clone = c1.clone();
+ c1Clone.setRequiredGlobalProps(gpp.getProperties1());
+ c2.setRequiredGlobalProps(gpp.getProperties2());
+
+ // we form a valid combination, so create the local candidates
+ // for this
+ addLocalCandidates(c1Clone, c2, broadcastPlanChannels, igps1, igps2,
+ outputPlans, allLocalPairs, estimator);
+ break outer;
+ }
+ }
+ }
+ }
+
+ // break the loop over input2's possible global properties, if the property
+ // is fixed via a hint. All the properties are overridden by the hint anyways,
+ // so we can stop after the first
+ if (this.input2.getShipStrategy() != null) {
+ break;
+ }
+ }
+
+ // break the loop over input1's possible global properties, if the property
+ // is fixed via a hint. All the properties are overridden by the hint anyways,
+ // so we can stop after the first
+ if (this.input1.getShipStrategy() != null) {
+ break;
+ }
+ }
+ }
+ }
+
+ if(outputPlans.isEmpty()) {
+ if(childrenSkippedDueToReplicatedInput) {
+ throw new CompilerException("No plan meeting the requirements could be created @ " + this
+ + ". Most likely reason: Invalid use of replicated input.");
+ } else {
+ throw new CompilerException("No plan meeting the requirements could be created @ " + this
+ + ". Most likely reason: Too restrictive plan hints.");
+ }
+ }
+
+ // cost and prune the plans
+ for (PlanNode node : outputPlans) {
+ estimator.costOperator(node);
+ }
+ prunePlanAlternatives(outputPlans);
+ outputPlans.trimToSize();
+
+ this.cachedPlans = outputPlans;
+ return outputPlans;
+ }
+
+ protected void addLocalCandidates(Channel template1, Channel template2, List<Set<? extends NamedChannel>> broadcastPlanChannels,
+ RequestedGlobalProperties rgps1, RequestedGlobalProperties rgps2,
+ List<PlanNode> target, LocalPropertiesPair[] validLocalCombinations, CostEstimator estimator)
+ {
+ for (RequestedLocalProperties ilp1 : this.input1.getInterestingProperties().getLocalProperties()) {
+ final Channel in1 = template1.clone();
+ ilp1.parameterizeChannel(in1);
+
+ for (RequestedLocalProperties ilp2 : this.input2.getInterestingProperties().getLocalProperties()) {
+ final Channel in2 = template2.clone();
+ ilp2.parameterizeChannel(in2);
+
+ for (OperatorDescriptorDual dps: getProperties()) {
+ for (LocalPropertiesPair lpp : dps.getPossibleLocalProperties()) {
+ if (lpp.getProperties1().isMetBy(in1.getLocalProperties()) &&
+ lpp.getProperties2().isMetBy(in2.getLocalProperties()) )
+ {
+ // valid combination
+ // for non trivial local properties, we need to check that they are co compatible
+ // (such as when some sort order is requested, that both are the same sort order
+ if (dps.areCoFulfilled(lpp.getProperties1(), lpp.getProperties2(),
+ in1.getLocalProperties(), in2.getLocalProperties()))
+ {
+ // copy, because setting required properties and instantiation may
+ // change the channels and should not affect prior candidates
+ Channel in1Copy = in1.clone();
+ in1Copy.setRequiredLocalProps(lpp.getProperties1());
+
+ Channel in2Copy = in2.clone();
+ in2Copy.setRequiredLocalProps(lpp.getProperties2());
+
+ // all right, co compatible
+ instantiate(dps, in1Copy, in2Copy, broadcastPlanChannels, target, estimator, rgps1, rgps2, ilp1, ilp2);
+ break;
+ }
+ // else cannot use this pair, fall through the loop and try the next one
+ }
+ }
+ }
+ }
+ }
+ }
+
+ protected void instantiate(OperatorDescriptorDual operator, Channel in1, Channel in2,
+ List<Set<? extends NamedChannel>> broadcastPlanChannels, List<PlanNode> target, CostEstimator estimator,
+ RequestedGlobalProperties globPropsReq1, RequestedGlobalProperties globPropsReq2,
+ RequestedLocalProperties locPropsReq1, RequestedLocalProperties locPropsReq2)
+ {
+ final PlanNode inputSource1 = in1.getSource();
+ final PlanNode inputSource2 = in2.getSource();
+
+ for (List<NamedChannel> broadcastChannelsCombination: Sets.cartesianProduct(broadcastPlanChannels)) {
+
+ boolean validCombination = true;
+
+ // check whether the broadcast inputs use the same plan candidate at the branching point
+ for (int i = 0; i < broadcastChannelsCombination.size(); i++) {
+ NamedChannel nc = broadcastChannelsCombination.get(i);
+ PlanNode bcSource = nc.getSource();
+
+ if (!(areBranchCompatible(bcSource, inputSource1) || areBranchCompatible(bcSource, inputSource2))) {
+ validCombination = false;
+ break;
+ }
+
+ // check branch compatibility against all other broadcast variables
+ for (int k = 0; k < i; k++) {
+ PlanNode otherBcSource = broadcastChannelsCombination.get(k).getSource();
+
+ if (!areBranchCompatible(bcSource, otherBcSource)) {
+ validCombination = false;
+ break;
+ }
+ }
+ }
+
+ if (!validCombination) {
+ continue;
+ }
+
+ placePipelineBreakersIfNecessary(operator.getStrategy(), in1, in2);
+
+ DualInputPlanNode node = operator.instantiate(in1, in2, this);
+ node.setBroadcastInputs(broadcastChannelsCombination);
+
+ SemanticProperties props = this.getSemanticProperties();
+ GlobalProperties gp1 = in1.getGlobalProperties().clone().filterBySemanticProperties(props, 0);
+ GlobalProperties gp2 = in2.getGlobalProperties().clone().filterBySemanticProperties(props, 1);
+ GlobalProperties combined = operator.computeGlobalProperties(gp1, gp2);
+
+ LocalProperties lp1 = in1.getLocalProperties().clone().filterBySemanticProperties(props, 0);
+ LocalProperties lp2 = in2.getLocalProperties().clone().filterBySemanticProperties(props, 1);
+ LocalProperties locals = operator.computeLocalProperties(lp1, lp2);
+
+ node.initProperties(combined, locals);
+ node.updatePropertiesWithUniqueSets(getUniqueFields());
+ target.add(node);
+ }
+ }
+
+ protected void placePipelineBreakersIfNecessary(DriverStrategy strategy, Channel in1, Channel in2) {
+ // before we instantiate, check for deadlocks by tracing back to the open branches and checking
+ // whether either no input, or all of them have a dam
+ if (this.hereJoinedBranches != null && this.hereJoinedBranches.size() > 0) {
+ boolean someDamOnLeftPaths = false;
+ boolean damOnAllLeftPaths = true;
+ boolean someDamOnRightPaths = false;
+ boolean damOnAllRightPaths = true;
+
+ if (strategy.firstDam() == DamBehavior.FULL_DAM || in1.getLocalStrategy().dams() || in1.getTempMode().breaksPipeline()) {
+ someDamOnLeftPaths = true;
+ } else {
+ for (OptimizerNode brancher : this.hereJoinedBranches) {
+ PlanNode candAtBrancher = in1.getSource().getCandidateAtBranchPoint(brancher);
+
+ // not all candidates are found, because this list includes joined branched from both regular inputs and broadcast vars
+ if (candAtBrancher == null) {
+ continue;
+ }
+
+ SourceAndDamReport res = in1.getSource().hasDamOnPathDownTo(candAtBrancher);
+ if (res == NOT_FOUND) {
+ throw new CompilerException("Bug: Tracing dams for deadlock detection is broken.");
+ } else if (res == FOUND_SOURCE) {
+ damOnAllLeftPaths = false;
+ } else if (res == FOUND_SOURCE_AND_DAM) {
+ someDamOnLeftPaths = true;
+ } else {
+ throw new CompilerException();
+ }
+ }
+ }
+
+ if (strategy.secondDam() == DamBehavior.FULL_DAM || in2.getLocalStrategy().dams() || in2.getTempMode().breaksPipeline()) {
+ someDamOnRightPaths = true;
+ } else {
+ for (OptimizerNode brancher : this.hereJoinedBranches) {
+ PlanNode candAtBrancher = in2.getSource().getCandidateAtBranchPoint(brancher);
+
+ // not all candidates are found, because this list includes joined branched from both regular inputs and broadcast vars
+ if (candAtBrancher == null) {
+ continue;
+ }
+
+ SourceAndDamReport res = in2.getSource().hasDamOnPathDownTo(candAtBrancher);
+ if (res == NOT_FOUND) {
+ throw new CompilerException("Bug: Tracing dams for deadlock detection is broken.");
+ } else if (res == FOUND_SOURCE) {
+ damOnAllRightPaths = false;
+ } else if (res == FOUND_SOURCE_AND_DAM) {
+ someDamOnRightPaths = true;
+ } else {
+ throw new CompilerException();
+ }
+ }
+ }
+
+ // okay combinations are both all dam or both no dam
+ if ( (damOnAllLeftPaths & damOnAllRightPaths) | (!someDamOnLeftPaths & !someDamOnRightPaths) ) {
+ // good, either both materialize already on the way, or both fully pipeline
+ } else {
+ if (someDamOnLeftPaths & !damOnAllRightPaths) {
+ // right needs a pipeline breaker
+ in2.setTempMode(in2.getTempMode().makePipelineBreaker());
+ }
+
+ if (someDamOnRightPaths & !damOnAllLeftPaths) {
+ // right needs a pipeline breaker
+ in1.setTempMode(in1.getTempMode().makePipelineBreaker());
+ }
+ }
+ }
+ }
+
+ @Override
+ public void computeUnclosedBranchStack() {
+ if (this.openBranches != null) {
+ return;
+ }
+
+ // handle the data flow branching for the regular inputs
+ addClosedBranches(getFirstPredecessorNode().closedBranchingNodes);
+ addClosedBranches(getSecondPredecessorNode().closedBranchingNodes);
+
+ List<UnclosedBranchDescriptor> result1 = getFirstPredecessorNode().getBranchesForParent(getFirstIncomingConnection());
+ List<UnclosedBranchDescriptor> result2 = getSecondPredecessorNode().getBranchesForParent(getSecondIncomingConnection());
+
+ ArrayList<UnclosedBranchDescriptor> inputsMerged = new ArrayList<UnclosedBranchDescriptor>();
+ mergeLists(result1, result2, inputsMerged, true);
+
+ // handle the data flow branching for the broadcast inputs
+ List<UnclosedBranchDescriptor> result = computeUnclosedBranchStackForBroadcastInputs(inputsMerged);
+
+ this.openBranches = (result == null || result.isEmpty()) ? Collections.<UnclosedBranchDescriptor>emptyList() : result;
+ }
+
+ @Override
+ public SemanticProperties getSemanticProperties() {
+ return getOperator().getSemanticProperties();
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Miscellaneous
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public void accept(Visitor<OptimizerNode> visitor) {
+ if (visitor.preVisit(this)) {
+ if (this.input1 == null || this.input2 == null) {
+ throw new CompilerException();
+ }
+
+ getFirstPredecessorNode().accept(visitor);
+ getSecondPredecessorNode().accept(visitor);
+
+ for (DagConnection connection : getBroadcastConnections()) {
+ connection.getSource().accept(visitor);
+ }
+
+ visitor.postVisit(this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/UnaryOperatorNode.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/UnaryOperatorNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/UnaryOperatorNode.java
new file mode 100644
index 0000000..45ecdac
--- /dev/null
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/UnaryOperatorNode.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.dag;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.flink.api.common.operators.SemanticProperties;
+import org.apache.flink.api.common.operators.SingleInputSemanticProperties;
+import org.apache.flink.api.common.operators.util.FieldSet;
+import org.apache.flink.optimizer.DataStatistics;
+import org.apache.flink.optimizer.operators.OperatorDescriptorSingle;
+
+
+public class UnaryOperatorNode extends SingleInputNode {
+
+ private final List<OperatorDescriptorSingle> operator;
+
+ private final String name;
+
+
+
+ public UnaryOperatorNode(String name, FieldSet keys, OperatorDescriptorSingle ... operators) {
+ this(name, keys, Arrays.asList(operators));
+ }
+
+ public UnaryOperatorNode(String name, FieldSet keys, List<OperatorDescriptorSingle> operators) {
+ super(keys);
+
+ this.operator = operators;
+ this.name = name;
+ }
+
+ @Override
+ protected List<OperatorDescriptorSingle> getPossibleProperties() {
+ return this.operator;
+ }
+
+ @Override
+ public String getName() {
+ return this.name;
+ }
+
+ @Override
+ public SemanticProperties getSemanticProperties() {
+ return new SingleInputSemanticProperties.AllFieldsForwardedProperties();
+ }
+
+ @Override
+ protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) {
+ // we have no estimates by default
+ }
+}