http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BulkIterationNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BulkIterationNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BulkIterationNode.java new file mode 100644 index 0000000..55b8785 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BulkIterationNode.java @@ -0,0 +1,390 @@ +/* + * 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.Iterator; +import java.util.List; +import java.util.Set; + +import org.apache.flink.api.common.ExecutionMode; +import org.apache.flink.api.common.operators.SemanticProperties; +import org.apache.flink.api.common.operators.SemanticProperties.EmptySemanticProperties; +import org.apache.flink.api.common.operators.base.BulkIterationBase; +import org.apache.flink.api.common.operators.util.FieldList; +import org.apache.flink.optimizer.CompilerException; +import org.apache.flink.optimizer.DataStatistics; +import org.apache.flink.optimizer.traversals.InterestingPropertyVisitor; +import org.apache.flink.optimizer.costs.CostEstimator; +import org.apache.flink.optimizer.dag.WorksetIterationNode.SingleRootJoiner; +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.NoOpDescriptor; +import org.apache.flink.optimizer.operators.OperatorDescriptorSingle; +import org.apache.flink.optimizer.plan.BulkIterationPlanNode; +import org.apache.flink.optimizer.plan.BulkPartialSolutionPlanNode; +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.FeedbackPropertiesMeetRequirementsReport; +import org.apache.flink.runtime.operators.DriverStrategy; +import org.apache.flink.util.Visitor; + +/** + * A node in the optimizer's program representation for a bulk iteration. + */ +public class BulkIterationNode extends SingleInputNode implements IterationNode { + + private BulkPartialSolutionNode partialSolution; + + private OptimizerNode terminationCriterion; + + private OptimizerNode nextPartialSolution; + + private DagConnection rootConnection; // connection out of the next partial solution + + private DagConnection terminationCriterionRootConnection; // connection out of the term. criterion + + private OptimizerNode singleRoot; + + private final int costWeight; + + // -------------------------------------------------------------------------------------------- + + /** + * Creates a new node for the bulk iteration. + * + * @param iteration The bulk iteration the node represents. + */ + public BulkIterationNode(BulkIterationBase<?> iteration) { + super(iteration); + + if (iteration.getMaximumNumberOfIterations() <= 0) { + throw new CompilerException("BulkIteration must have a maximum number of iterations specified."); + } + + int numIters = iteration.getMaximumNumberOfIterations(); + + this.costWeight = (numIters > 0 && numIters < OptimizerNode.MAX_DYNAMIC_PATH_COST_WEIGHT) ? + numIters : OptimizerNode.MAX_DYNAMIC_PATH_COST_WEIGHT; + } + + // -------------------------------------------------------------------------------------------- + + public BulkIterationBase<?> getIterationContract() { + return (BulkIterationBase<?>) getOperator(); + } + + /** + * Gets the partialSolution from this BulkIterationNode. + * + * @return The partialSolution. + */ + public BulkPartialSolutionNode getPartialSolution() { + return partialSolution; + } + + /** + * Sets the partialSolution for this BulkIterationNode. + * + * @param partialSolution The partialSolution to set. + */ + public void setPartialSolution(BulkPartialSolutionNode partialSolution) { + this.partialSolution = partialSolution; + } + + + /** + * Gets the nextPartialSolution from this BulkIterationNode. + * + * @return The nextPartialSolution. + */ + public OptimizerNode getNextPartialSolution() { + return nextPartialSolution; + } + + /** + * Sets the nextPartialSolution for this BulkIterationNode. + * + * @param nextPartialSolution The nextPartialSolution to set. + */ + public void setNextPartialSolution(OptimizerNode nextPartialSolution, OptimizerNode terminationCriterion) { + + // check if the root of the step function has the same DOP as the iteration + // or if the step function has any operator at all + if (nextPartialSolution.getParallelism() != getParallelism() || + nextPartialSolution == partialSolution || nextPartialSolution instanceof BinaryUnionNode) + { + // add a no-op to the root to express the re-partitioning + NoOpNode noop = new NoOpNode(); + noop.setDegreeOfParallelism(getParallelism()); + + DagConnection noOpConn = new DagConnection(nextPartialSolution, noop, ExecutionMode.PIPELINED); + noop.setIncomingConnection(noOpConn); + nextPartialSolution.addOutgoingConnection(noOpConn); + + nextPartialSolution = noop; + } + + this.nextPartialSolution = nextPartialSolution; + this.terminationCriterion = terminationCriterion; + + if (terminationCriterion == null) { + this.singleRoot = nextPartialSolution; + this.rootConnection = new DagConnection(nextPartialSolution, ExecutionMode.PIPELINED); + } + else { + // we have a termination criterion + SingleRootJoiner singleRootJoiner = new SingleRootJoiner(); + this.rootConnection = new DagConnection(nextPartialSolution, singleRootJoiner, ExecutionMode.PIPELINED); + this.terminationCriterionRootConnection = new DagConnection(terminationCriterion, singleRootJoiner, + ExecutionMode.PIPELINED); + + singleRootJoiner.setInputs(this.rootConnection, this.terminationCriterionRootConnection); + + this.singleRoot = singleRootJoiner; + + // add connection to terminationCriterion for interesting properties visitor + terminationCriterion.addOutgoingConnection(terminationCriterionRootConnection); + + } + + nextPartialSolution.addOutgoingConnection(rootConnection); + } + + public int getCostWeight() { + return this.costWeight; + } + + public OptimizerNode getSingleRootOfStepFunction() { + return this.singleRoot; + } + + // -------------------------------------------------------------------------------------------- + + @Override + public String getName() { + return "Bulk Iteration"; + } + + @Override + public SemanticProperties getSemanticProperties() { + return new EmptySemanticProperties(); + } + + protected void readStubAnnotations() {} + + @Override + protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) { + this.estimatedOutputSize = getPredecessorNode().getEstimatedOutputSize(); + this.estimatedNumRecords = getPredecessorNode().getEstimatedNumRecords(); + } + + // -------------------------------------------------------------------------------------------- + // Properties and Optimization + // -------------------------------------------------------------------------------------------- + + protected List<OperatorDescriptorSingle> getPossibleProperties() { + return Collections.<OperatorDescriptorSingle>singletonList(new NoOpDescriptor()); + } + + @Override + public void computeInterestingPropertiesForInputs(CostEstimator estimator) { + final InterestingProperties intProps = getInterestingProperties().clone(); + + if (this.terminationCriterion != null) { + // first propagate through termination Criterion. since it has no successors, it has no + // interesting properties + this.terminationCriterionRootConnection.setInterestingProperties(new InterestingProperties()); + this.terminationCriterion.accept(new InterestingPropertyVisitor(estimator)); + } + + // we need to make 2 interesting property passes, because the root of the step function needs also + // the interesting properties as generated by the partial solution + + // give our own interesting properties (as generated by the iterations successors) to the step function and + // make the first pass + this.rootConnection.setInterestingProperties(intProps); + this.nextPartialSolution.accept(new InterestingPropertyVisitor(estimator)); + + // take the interesting properties of the partial solution and add them to the root interesting properties + InterestingProperties partialSolutionIntProps = this.partialSolution.getInterestingProperties(); + intProps.getGlobalProperties().addAll(partialSolutionIntProps.getGlobalProperties()); + intProps.getLocalProperties().addAll(partialSolutionIntProps.getLocalProperties()); + + // clear all interesting properties to prepare the second traversal + // this clears only the path down from the next partial solution. The paths down + // from the termination criterion (before they meet the paths down from the next partial solution) + // remain unaffected by this step + this.rootConnection.clearInterestingProperties(); + this.nextPartialSolution.accept(InterestingPropertiesClearer.INSTANCE); + + // 2nd pass + this.rootConnection.setInterestingProperties(intProps); + this.nextPartialSolution.accept(new InterestingPropertyVisitor(estimator)); + + // now add the interesting properties of the partial solution to the input + final InterestingProperties inProps = this.partialSolution.getInterestingProperties().clone(); + inProps.addGlobalProperties(new RequestedGlobalProperties()); + inProps.addLocalProperties(new RequestedLocalProperties()); + this.inConn.setInterestingProperties(inProps); + } + + @Override + public void clearInterestingProperties() { + super.clearInterestingProperties(); + + this.singleRoot.accept(InterestingPropertiesClearer.INSTANCE); + this.rootConnection.clearInterestingProperties(); + } + + @Override + public void computeUnclosedBranchStack() { + if (this.openBranches != null) { + return; + } + + // the resulting branches are those of the step function + // because the BulkPartialSolution takes the input's branches + addClosedBranches(getSingleRootOfStepFunction().closedBranchingNodes); + List<UnclosedBranchDescriptor> result = getSingleRootOfStepFunction().openBranches; + + this.openBranches = (result == null || result.isEmpty()) ? Collections.<UnclosedBranchDescriptor>emptyList() : result; + } + + + @Override + protected void instantiateCandidate(OperatorDescriptorSingle dps, Channel in, List<Set<? extends NamedChannel>> broadcastPlanChannels, + List<PlanNode> target, CostEstimator estimator, RequestedGlobalProperties globPropsReq, RequestedLocalProperties locPropsReq) + { + // NOTES ON THE ENUMERATION OF THE STEP FUNCTION PLANS: + // Whenever we instantiate the iteration, we enumerate new candidates for the step function. + // That way, we make sure we have an appropriate plan for each candidate for the initial partial solution, + // we have a fitting candidate for the step function (often, work is pushed out of the step function). + // Among the candidates of the step function, we keep only those that meet the requested properties of the + // current candidate initial partial solution. That makes sure these properties exist at the beginning of + // the successive iteration. + + // 1) Because we enumerate multiple times, we may need to clean the cached plans + // before starting another enumeration + this.nextPartialSolution.accept(PlanCacheCleaner.INSTANCE); + if (this.terminationCriterion != null) { + this.terminationCriterion.accept(PlanCacheCleaner.INSTANCE); + } + + // 2) Give the partial solution the properties of the current candidate for the initial partial solution + this.partialSolution.setCandidateProperties(in.getGlobalProperties(), in.getLocalProperties(), in); + final BulkPartialSolutionPlanNode pspn = this.partialSolution.getCurrentPartialSolutionPlanNode(); + + // 3) Get the alternative plans + List<PlanNode> candidates = this.nextPartialSolution.getAlternativePlans(estimator); + + // 4) Make sure that the beginning of the step function does not assume properties that + // are not also produced by the end of the step function. + + { + List<PlanNode> newCandidates = new ArrayList<PlanNode>(); + + for (Iterator<PlanNode> planDeleter = candidates.iterator(); planDeleter.hasNext(); ) { + PlanNode candidate = planDeleter.next(); + + GlobalProperties atEndGlobal = candidate.getGlobalProperties(); + LocalProperties atEndLocal = candidate.getLocalProperties(); + + FeedbackPropertiesMeetRequirementsReport report = candidate.checkPartialSolutionPropertiesMet(pspn, atEndGlobal, atEndLocal); + if (report == FeedbackPropertiesMeetRequirementsReport.NO_PARTIAL_SOLUTION) { + ; // depends only through broadcast variable on the partial solution + } + else if (report == FeedbackPropertiesMeetRequirementsReport.NOT_MET) { + // attach a no-op node through which we create the properties of the original input + Channel toNoOp = new Channel(candidate); + globPropsReq.parameterizeChannel(toNoOp, false, rootConnection.getDataExchangeMode(), false); + locPropsReq.parameterizeChannel(toNoOp); + + UnaryOperatorNode rebuildPropertiesNode = new UnaryOperatorNode("Rebuild Partial Solution Properties", FieldList.EMPTY_LIST); + rebuildPropertiesNode.setDegreeOfParallelism(candidate.getParallelism()); + + SingleInputPlanNode rebuildPropertiesPlanNode = new SingleInputPlanNode(rebuildPropertiesNode, "Rebuild Partial Solution Properties", toNoOp, DriverStrategy.UNARY_NO_OP); + rebuildPropertiesPlanNode.initProperties(toNoOp.getGlobalProperties(), toNoOp.getLocalProperties()); + estimator.costOperator(rebuildPropertiesPlanNode); + + GlobalProperties atEndGlobalModified = rebuildPropertiesPlanNode.getGlobalProperties(); + LocalProperties atEndLocalModified = rebuildPropertiesPlanNode.getLocalProperties(); + + if (!(atEndGlobalModified.equals(atEndGlobal) && atEndLocalModified.equals(atEndLocal))) { + FeedbackPropertiesMeetRequirementsReport report2 = candidate.checkPartialSolutionPropertiesMet(pspn, atEndGlobalModified, atEndLocalModified); + + if (report2 != FeedbackPropertiesMeetRequirementsReport.NOT_MET) { + newCandidates.add(rebuildPropertiesPlanNode); + } + } + + planDeleter.remove(); + } + } + } + + if (candidates.isEmpty()) { + return; + } + + // 5) Create a candidate for the Iteration Node for every remaining plan of the step function. + if (terminationCriterion == null) { + for (PlanNode candidate : candidates) { + BulkIterationPlanNode node = new BulkIterationPlanNode(this, "BulkIteration ("+this.getOperator().getName()+")", in, pspn, candidate); + GlobalProperties gProps = candidate.getGlobalProperties().clone(); + LocalProperties lProps = candidate.getLocalProperties().clone(); + node.initProperties(gProps, lProps); + target.add(node); + } + } + else if (candidates.size() > 0) { + List<PlanNode> terminationCriterionCandidates = this.terminationCriterion.getAlternativePlans(estimator); + + SingleRootJoiner singleRoot = (SingleRootJoiner) this.singleRoot; + + for (PlanNode candidate : candidates) { + for (PlanNode terminationCandidate : terminationCriterionCandidates) { + if (singleRoot.areBranchCompatible(candidate, terminationCandidate)) { + BulkIterationPlanNode node = new BulkIterationPlanNode(this, "BulkIteration ("+this.getOperator().getName()+")", in, pspn, candidate, terminationCandidate); + GlobalProperties gProps = candidate.getGlobalProperties().clone(); + LocalProperties lProps = candidate.getLocalProperties().clone(); + node.initProperties(gProps, lProps); + target.add(node); + + } + } + } + + } + } + + // -------------------------------------------------------------------------------------------- + // Iteration Specific Traversals + // -------------------------------------------------------------------------------------------- + + public void acceptForStepFunction(Visitor<OptimizerNode> visitor) { + this.singleRoot.accept(visitor); + } +}
http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BulkPartialSolutionNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BulkPartialSolutionNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BulkPartialSolutionNode.java new file mode 100644 index 0000000..25a7eef --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/BulkPartialSolutionNode.java @@ -0,0 +1,103 @@ +/* + * 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.BulkIterationBase.PartialSolutionPlaceHolder; +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.BulkPartialSolutionPlanNode; +import org.apache.flink.optimizer.plan.Channel; +import org.apache.flink.optimizer.plan.PlanNode; + +/** + * The optimizer's internal representation of the partial solution that is input to a bulk iteration. + */ +public class BulkPartialSolutionNode extends AbstractPartialSolutionNode { + + private final BulkIterationNode iterationNode; + + + public BulkPartialSolutionNode(PartialSolutionPlaceHolder<?> psph, BulkIterationNode iterationNode) { + super(psph); + this.iterationNode = iterationNode; + } + + // -------------------------------------------------------------------------------------------- + + public void setCandidateProperties(GlobalProperties gProps, LocalProperties lProps, Channel initialInput) { + if (this.cachedPlans != null) { + throw new IllegalStateException(); + } else { + this.cachedPlans = Collections.<PlanNode>singletonList(new BulkPartialSolutionPlanNode(this, + "PartialSolution ("+this.getOperator().getName()+")", gProps, lProps, initialInput)); + } + } + + public BulkPartialSolutionPlanNode getCurrentPartialSolutionPlanNode() { + if (this.cachedPlans != null) { + return (BulkPartialSolutionPlanNode) this.cachedPlans.get(0); + } else { + throw new IllegalStateException(); + } + } + + public BulkIterationNode getIterationNode() { + return this.iterationNode; + } + + @Override + public void computeOutputEstimates(DataStatistics statistics) { + copyEstimates(this.iterationNode.getPredecessorNode()); + } + + // -------------------------------------------------------------------------------------------- + + /** + * Gets the operator (here the {@link PartialSolutionPlaceHolder}) that is represented by this + * optimizer node. + * + * @return The operator represented by this optimizer node. + */ + @Override + public PartialSolutionPlaceHolder<?> getOperator() { + return (PartialSolutionPlaceHolder<?>) super.getOperator(); + } + + @Override + public String getName() { + return "Bulk Partial Solution"; + } + + @Override + public void computeUnclosedBranchStack() { + if (this.openBranches != null) { + return; + } + + OptimizerNode inputToIteration = this.iterationNode.getPredecessorNode(); + + addClosedBranches(inputToIteration.closedBranchingNodes); + List<UnclosedBranchDescriptor> fromInput = inputToIteration.getBranchesForParent(this.iterationNode.getIncomingConnection()); + 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/CoGroupNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CoGroupNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CoGroupNode.java new file mode 100644 index 0000000..92076c3 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CoGroupNode.java @@ -0,0 +1,106 @@ +/* + * 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.functions.Partitioner; +import org.apache.flink.api.common.operators.Ordering; +import org.apache.flink.api.common.operators.base.CoGroupOperatorBase; +import org.apache.flink.optimizer.DataStatistics; +import org.apache.flink.optimizer.operators.CoGroupDescriptor; +import org.apache.flink.optimizer.operators.CoGroupWithSolutionSetFirstDescriptor; +import org.apache.flink.optimizer.operators.CoGroupWithSolutionSetSecondDescriptor; +import org.apache.flink.optimizer.operators.OperatorDescriptorDual; + +/** + * The Optimizer representation of a <i>CoGroup</i> operator. + */ +public class CoGroupNode extends TwoInputNode { + + private List<OperatorDescriptorDual> dataProperties; + + public CoGroupNode(CoGroupOperatorBase<?, ?, ?, ?> operator) { + super(operator); + this.dataProperties = initializeDataProperties(operator.getCustomPartitioner()); + } + + // -------------------------------------------------------------------------------------------- + + /** + * Gets the operator for this CoGroup node. + * + * @return The CoGroup operator. + */ + @Override + public CoGroupOperatorBase<?, ?, ?, ?> getOperator() { + return (CoGroupOperatorBase<?, ?, ?, ?>) super.getOperator(); + } + + @Override + public String getName() { + return "CoGroup"; + } + + @Override + protected List<OperatorDescriptorDual> getPossibleProperties() { + return this.dataProperties; + } + + public void makeCoGroupWithSolutionSet(int solutionsetInputIndex) { + OperatorDescriptorDual op; + if (solutionsetInputIndex == 0) { + op = new CoGroupWithSolutionSetFirstDescriptor(keys1, keys2); + } else if (solutionsetInputIndex == 1) { + op = new CoGroupWithSolutionSetSecondDescriptor(keys1, keys2); + } else { + throw new IllegalArgumentException(); + } + this.dataProperties = Collections.<OperatorDescriptorDual>singletonList(op); + } + + @Override + protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) { + // for CoGroup, we currently make no reasonable default estimates + } + + private List<OperatorDescriptorDual> initializeDataProperties(Partitioner<?> customPartitioner) { + Ordering groupOrder1 = null; + Ordering groupOrder2 = null; + + CoGroupOperatorBase<?, ?, ?, ?> cgc = getOperator(); + groupOrder1 = cgc.getGroupOrderForInputOne(); + groupOrder2 = cgc.getGroupOrderForInputTwo(); + + if (groupOrder1 != null && groupOrder1.getNumberOfFields() == 0) { + groupOrder1 = null; + } + if (groupOrder2 != null && groupOrder2.getNumberOfFields() == 0) { + groupOrder2 = null; + } + + CoGroupDescriptor descr = new CoGroupDescriptor(this.keys1, this.keys2, groupOrder1, groupOrder2); + if (customPartitioner != null) { + descr.setCustomPartitioner(customPartitioner); + } + + return Collections.<OperatorDescriptorDual>singletonList(descr); + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CollectorMapNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CollectorMapNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CollectorMapNode.java new file mode 100644 index 0000000..93be1e4 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CollectorMapNode.java @@ -0,0 +1,62 @@ +/* + * 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.SingleInputOperator; +import org.apache.flink.optimizer.DataStatistics; +import org.apache.flink.optimizer.operators.CollectorMapDescriptor; +import org.apache.flink.optimizer.operators.OperatorDescriptorSingle; + +/** + * The optimizer's internal representation of a <i>Map</i> operator node. + */ +public class CollectorMapNode extends SingleInputNode { + + private final List<OperatorDescriptorSingle> possibleProperties; + + + public CollectorMapNode(SingleInputOperator<?, ?, ?> operator) { + super(operator); + + this.possibleProperties = Collections.<OperatorDescriptorSingle>singletonList(new CollectorMapDescriptor()); + } + + @Override + public String getName() { + return "Map"; + } + + @Override + protected List<OperatorDescriptorSingle> getPossibleProperties() { + return this.possibleProperties; + } + + /** + * Computes the estimates for the Map operator. Map takes one value and transforms it into another value. + * The cardinality consequently stays the same. + */ + @Override + protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) { + this.estimatedNumRecords = getPredecessorNode().getEstimatedNumRecords(); + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CrossNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CrossNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CrossNode.java new file mode 100644 index 0000000..8de67e8 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/CrossNode.java @@ -0,0 +1,138 @@ +/* + * 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.operators.base.CrossOperatorBase; +import org.apache.flink.api.common.operators.base.CrossOperatorBase.CrossHint; +import org.apache.flink.optimizer.CompilerException; +import org.apache.flink.optimizer.DataStatistics; +import org.apache.flink.optimizer.Optimizer; +import org.apache.flink.optimizer.operators.CrossBlockOuterFirstDescriptor; +import org.apache.flink.optimizer.operators.CrossBlockOuterSecondDescriptor; +import org.apache.flink.optimizer.operators.CrossStreamOuterFirstDescriptor; +import org.apache.flink.optimizer.operators.CrossStreamOuterSecondDescriptor; +import org.apache.flink.optimizer.operators.OperatorDescriptorDual; +import org.apache.flink.configuration.Configuration; + +/** + * The Optimizer representation of a <i>Cross</i> (Cartesian product) operator. + */ +public class CrossNode extends TwoInputNode { + + private final List<OperatorDescriptorDual> dataProperties; + + /** + * Creates a new CrossNode for the given operator. + * + * @param operation The Cross operator object. + */ + public CrossNode(CrossOperatorBase<?, ?, ?, ?> operation) { + super(operation); + + Configuration conf = operation.getParameters(); + String localStrategy = conf.getString(Optimizer.HINT_LOCAL_STRATEGY, null); + + CrossHint hint = operation.getCrossHint(); + + if (localStrategy != null) { + + final boolean allowBCfirst = hint != CrossHint.SECOND_IS_SMALL; + final boolean allowBCsecond = hint != CrossHint.FIRST_IS_SMALL; + + final OperatorDescriptorDual fixedDriverStrat; + if (Optimizer.HINT_LOCAL_STRATEGY_NESTEDLOOP_BLOCKED_OUTER_FIRST.equals(localStrategy)) { + fixedDriverStrat = new CrossBlockOuterFirstDescriptor(allowBCfirst, allowBCsecond); + } else if (Optimizer.HINT_LOCAL_STRATEGY_NESTEDLOOP_BLOCKED_OUTER_SECOND.equals(localStrategy)) { + fixedDriverStrat = new CrossBlockOuterSecondDescriptor(allowBCfirst, allowBCsecond); + } else if (Optimizer.HINT_LOCAL_STRATEGY_NESTEDLOOP_STREAMED_OUTER_FIRST.equals(localStrategy)) { + fixedDriverStrat = new CrossStreamOuterFirstDescriptor(allowBCfirst, allowBCsecond); + } else if (Optimizer.HINT_LOCAL_STRATEGY_NESTEDLOOP_STREAMED_OUTER_SECOND.equals(localStrategy)) { + fixedDriverStrat = new CrossStreamOuterSecondDescriptor(allowBCfirst, allowBCsecond); + } else { + throw new CompilerException("Invalid local strategy hint for cross contract: " + localStrategy); + } + + this.dataProperties = Collections.singletonList(fixedDriverStrat); + } + else if (hint == CrossHint.SECOND_IS_SMALL) { + ArrayList<OperatorDescriptorDual> list = new ArrayList<OperatorDescriptorDual>(); + list.add(new CrossBlockOuterSecondDescriptor(false, true)); + list.add(new CrossStreamOuterFirstDescriptor(false, true)); + this.dataProperties = list; + } + else if (hint == CrossHint.FIRST_IS_SMALL) { + ArrayList<OperatorDescriptorDual> list = new ArrayList<OperatorDescriptorDual>(); + list.add(new CrossBlockOuterFirstDescriptor(true, false)); + list.add(new CrossStreamOuterSecondDescriptor(true, false)); + this.dataProperties = list; + } + else { + ArrayList<OperatorDescriptorDual> list = new ArrayList<OperatorDescriptorDual>(); + list.add(new CrossBlockOuterFirstDescriptor()); + list.add(new CrossBlockOuterSecondDescriptor()); + list.add(new CrossStreamOuterFirstDescriptor()); + list.add(new CrossStreamOuterSecondDescriptor()); + this.dataProperties = list; + } + } + + // ------------------------------------------------------------------------ + + @Override + public CrossOperatorBase<?, ?, ?, ?> getOperator() { + return (CrossOperatorBase<?, ?, ?, ?>) super.getOperator(); + } + + @Override + public String getName() { + return "Cross"; + } + + @Override + protected List<OperatorDescriptorDual> getPossibleProperties() { + return this.dataProperties; + } + + /** + * We assume that the cardinality is the product of the input cardinalities + * and that the result width is the sum of the input widths. + * + * @param statistics The statistics object to optionally access. + */ + @Override + protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) { + long card1 = getFirstPredecessorNode().getEstimatedNumRecords(); + long card2 = getSecondPredecessorNode().getEstimatedNumRecords(); + this.estimatedNumRecords = (card1 < 0 || card2 < 0) ? -1 : card1 * card2; + + if (this.estimatedNumRecords >= 0) { + float width1 = getFirstPredecessorNode().getEstimatedAvgWidthPerOutputRecord(); + float width2 = getSecondPredecessorNode().getEstimatedAvgWidthPerOutputRecord(); + float width = (width1 <= 0 || width2 <= 0) ? -1 : width1 + width2; + + if (width > 0) { + this.estimatedOutputSize = (long) (width * this.estimatedNumRecords); + } + } + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DagConnection.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DagConnection.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DagConnection.java new file mode 100644 index 0000000..4e65976 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DagConnection.java @@ -0,0 +1,290 @@ +/* + * 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.ExecutionMode; +import org.apache.flink.optimizer.dataproperties.InterestingProperties; +import org.apache.flink.optimizer.plandump.DumpableConnection; +import org.apache.flink.runtime.operators.shipping.ShipStrategyType; + +/** + * A connection between to operators. Represents an intermediate result + * and a data exchange between the two operators. + * + * The data exchange has a mode in which it performs (batch / pipelined). + * + * The data exchange strategy may be set on this connection, in which case + * it is fixed and will not be determined during candidate plan enumeration. + * + * During the enumeration of interesting properties, this connection also holds + * all interesting properties generated by the successor operator. + */ +public class DagConnection implements EstimateProvider, DumpableConnection<OptimizerNode> { + + private final OptimizerNode source; // The source node of the connection + + private final OptimizerNode target; // The target node of the connection. + + private final ExecutionMode dataExchangeMode; // defines whether to use batch or pipelined data exchange + + private InterestingProperties interestingProps; // local properties that succeeding nodes are interested in + + private ShipStrategyType shipStrategy; // The data shipping strategy, if predefined. + + private TempMode materializationMode = TempMode.NONE; // the materialization mode + + private int maxDepth = -1; + + private boolean breakPipeline; // whether this connection should break the pipeline due to potential deadlocks + + /** + * Creates a new Connection between two nodes. The shipping strategy is by default <tt>NONE</tt>. + * The temp mode is by default <tt>NONE</tt>. + * + * @param source + * The source node. + * @param target + * The target node. + */ + public DagConnection(OptimizerNode source, OptimizerNode target, ExecutionMode exchangeMode) { + this(source, target, null, exchangeMode); + } + + /** + * Creates a new Connection between two nodes. + * + * @param source + * The source node. + * @param target + * The target node. + * @param shipStrategy + * The shipping strategy. + * @param exchangeMode + * The data exchange mode (pipelined / batch / batch only for shuffles / ... ) + */ + public DagConnection(OptimizerNode source, OptimizerNode target, + ShipStrategyType shipStrategy, ExecutionMode exchangeMode) + { + if (source == null || target == null) { + throw new NullPointerException("Source and target must not be null."); + } + this.source = source; + this.target = target; + this.shipStrategy = shipStrategy; + this.dataExchangeMode = exchangeMode; + } + + /** + * Constructor to create a result from an operator that is not + * consumed by another operator. + * + * @param source + * The source node. + */ + public DagConnection(OptimizerNode source, ExecutionMode exchangeMode) { + if (source == null) { + throw new NullPointerException("Source and target must not be null."); + } + this.source = source; + this.target = null; + this.shipStrategy = ShipStrategyType.NONE; + this.dataExchangeMode = exchangeMode; + } + + /** + * Gets the source of the connection. + * + * @return The source Node. + */ + public OptimizerNode getSource() { + return this.source; + } + + /** + * Gets the target of the connection. + * + * @return The target node. + */ + public OptimizerNode getTarget() { + return this.target; + } + + /** + * Gets the shipping strategy for this connection. + * + * @return The connection's shipping strategy. + */ + public ShipStrategyType getShipStrategy() { + return this.shipStrategy; + } + + /** + * Sets the shipping strategy for this connection. + * + * @param strategy + * The shipping strategy to be applied to this connection. + */ + public void setShipStrategy(ShipStrategyType strategy) { + this.shipStrategy = strategy; + } + + /** + * Gets the data exchange mode to use for this connection. + * + * @return The data exchange mode to use for this connection. + */ + public ExecutionMode getDataExchangeMode() { + if (dataExchangeMode == null) { + throw new IllegalStateException("This connection does not have the data exchange mode set"); + } + return dataExchangeMode; + } + + /** + * Marks that this connection should do a decoupled data exchange (such as batched) + * rather then pipeline data. Connections are marked as pipeline breakers to avoid + * deadlock situations. + */ + public void markBreaksPipeline() { + this.breakPipeline = true; + } + + /** + * Checks whether this connection is marked to break the pipeline. + * + * @return True, if this connection is marked to break the pipeline, false otherwise. + */ + public boolean isBreakingPipeline() { + return this.breakPipeline; + } + + /** + * Gets the interesting properties object for this pact connection. + * If the interesting properties for this connections have not yet been set, + * this method returns null. + * + * @return The collection of all interesting properties, or null, if they have not yet been set. + */ + public InterestingProperties getInterestingProperties() { + return this.interestingProps; + } + + /** + * Sets the interesting properties for this pact connection. + * + * @param props The interesting properties. + */ + public void setInterestingProperties(InterestingProperties props) { + if (this.interestingProps == null) { + this.interestingProps = props; + } else { + throw new IllegalStateException("Interesting Properties have already been set."); + } + } + + public void clearInterestingProperties() { + this.interestingProps = null; + } + + public void initMaxDepth() { + + if (this.maxDepth == -1) { + this.maxDepth = this.source.getMaxDepth() + 1; + } else { + throw new IllegalStateException("Maximum path depth has already been initialized."); + } + } + + public int getMaxDepth() { + if (this.maxDepth != -1) { + return this.maxDepth; + } else { + throw new IllegalStateException("Maximum path depth has not been initialized."); + } + } + + // -------------------------------------------------------------------------------------------- + // Estimates + // -------------------------------------------------------------------------------------------- + + @Override + public long getEstimatedOutputSize() { + return this.source.getEstimatedOutputSize(); + } + + @Override + public long getEstimatedNumRecords() { + return this.source.getEstimatedNumRecords(); + } + + @Override + public float getEstimatedAvgWidthPerOutputRecord() { + return this.source.getEstimatedAvgWidthPerOutputRecord(); + } + + // -------------------------------------------------------------------------------------------- + + + public TempMode getMaterializationMode() { + return this.materializationMode; + } + + public void setMaterializationMode(TempMode materializationMode) { + this.materializationMode = materializationMode; + } + + public boolean isOnDynamicPath() { + return this.source.isOnDynamicPath(); + } + + public int getCostWeight() { + return this.source.getCostWeight(); + } + + // -------------------------------------------------------------------------------------------- + + public String toString() { + StringBuilder buf = new StringBuilder(50); + buf.append("Connection: "); + + if (this.source == null) { + buf.append("null"); + } else { + buf.append(this.source.getOperator().getName()); + buf.append('(').append(this.source.getName()).append(')'); + } + + buf.append(" -> "); + + if (this.shipStrategy != null) { + buf.append('['); + buf.append(this.shipStrategy.name()); + buf.append(']').append(' '); + } + + if (this.target == null) { + buf.append("null"); + } else { + buf.append(this.target.getOperator().getName()); + buf.append('(').append(this.target.getName()).append(')'); + } + + return buf.toString(); + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSinkNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSinkNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSinkNode.java new file mode 100644 index 0000000..dbe04f4 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSinkNode.java @@ -0,0 +1,266 @@ +/* + * 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 java.util.Map; + +import org.apache.flink.api.common.ExecutionMode; +import org.apache.flink.api.common.distributions.DataDistribution; +import org.apache.flink.api.common.operators.GenericDataSinkBase; +import org.apache.flink.api.common.operators.Operator; +import org.apache.flink.api.common.operators.Ordering; +import org.apache.flink.api.common.operators.SemanticProperties; +import org.apache.flink.api.common.operators.SemanticProperties.EmptySemanticProperties; +import org.apache.flink.optimizer.CompilerException; +import org.apache.flink.optimizer.DataStatistics; +import org.apache.flink.optimizer.costs.CostEstimator; +import org.apache.flink.optimizer.dataproperties.InterestingProperties; +import org.apache.flink.optimizer.dataproperties.RequestedGlobalProperties; +import org.apache.flink.optimizer.dataproperties.RequestedLocalProperties; +import org.apache.flink.optimizer.plan.Channel; +import org.apache.flink.optimizer.plan.PlanNode; +import org.apache.flink.optimizer.plan.SinkPlanNode; +import org.apache.flink.util.Visitor; + +/** + * The Optimizer representation of a data sink. + */ +public class DataSinkNode extends OptimizerNode { + + protected DagConnection input; // The input edge + + /** + * Creates a new DataSinkNode for the given sink operator. + * + * @param sink The data sink contract object. + */ + public DataSinkNode(GenericDataSinkBase<?> sink) { + super(sink); + } + + // -------------------------------------------------------------------------------------- + + /** + * Gets the input of the sink. + * + * @return The input connection. + */ + public DagConnection getInputConnection() { + return this.input; + } + + /** + * Gets the predecessor of this node. + * + * @return The predecessor, or null, if no predecessor has been set. + */ + public OptimizerNode getPredecessorNode() { + if(this.input != null) { + return input.getSource(); + } else { + return null; + } + } + + /** + * Gets the operator for which this optimizer sink node was created. + * + * @return The node's underlying operator. + */ + @Override + public GenericDataSinkBase<?> getOperator() { + return (GenericDataSinkBase<?>) super.getOperator(); + } + + @Override + public String getName() { + return "Data Sink"; + } + + @Override + public List<DagConnection> getIncomingConnections() { + return Collections.singletonList(this.input); + } + + /** + * Gets all outgoing connections, which is an empty set for the data sink. + * + * @return An empty list. + */ + @Override + public List<DagConnection> getOutgoingConnections() { + return Collections.emptyList(); + } + + @Override + public void setInput(Map<Operator<?>, OptimizerNode> contractToNode, ExecutionMode defaultExchangeMode) { + Operator<?> children = getOperator().getInput(); + + final OptimizerNode pred; + final DagConnection conn; + + pred = contractToNode.get(children); + conn = new DagConnection(pred, this, defaultExchangeMode); + + // create the connection and add it + this.input = conn; + pred.addOutgoingConnection(conn); + } + + /** + * Computes the estimated outputs for the data sink. Since the sink does not modify anything, it simply + * copies the output estimates from its direct predecessor. + */ + @Override + protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) { + this.estimatedNumRecords = getPredecessorNode().getEstimatedNumRecords(); + this.estimatedOutputSize = getPredecessorNode().getEstimatedOutputSize(); + } + + @Override + public void computeInterestingPropertiesForInputs(CostEstimator estimator) { + final InterestingProperties iProps = new InterestingProperties(); + + { + final Ordering partitioning = getOperator().getPartitionOrdering(); + final DataDistribution dataDist = getOperator().getDataDistribution(); + final RequestedGlobalProperties partitioningProps = new RequestedGlobalProperties(); + if (partitioning != null) { + if(dataDist != null) { + partitioningProps.setRangePartitioned(partitioning, dataDist); + } else { + partitioningProps.setRangePartitioned(partitioning); + } + iProps.addGlobalProperties(partitioningProps); + } + iProps.addGlobalProperties(partitioningProps); + } + + { + final Ordering localOrder = getOperator().getLocalOrder(); + final RequestedLocalProperties orderProps = new RequestedLocalProperties(); + if (localOrder != null) { + orderProps.setOrdering(localOrder); + } + iProps.addLocalProperties(orderProps); + } + + this.input.setInterestingProperties(iProps); + } + + // -------------------------------------------------------------------------------------------- + // Branch Handling + // -------------------------------------------------------------------------------------------- + + @Override + public void computeUnclosedBranchStack() { + if (this.openBranches != null) { + return; + } + + // we need to track open branches even in the sinks, because they get "closed" when + // we build a single "root" for the data flow plan + addClosedBranches(getPredecessorNode().closedBranchingNodes); + this.openBranches = getPredecessorNode().getBranchesForParent(this.input); + } + + @Override + protected List<UnclosedBranchDescriptor> getBranchesForParent(DagConnection parent) { + // return our own stack of open branches, because nothing is added + return this.openBranches; + } + + // -------------------------------------------------------------------------------------------- + // Recursive Optimization + // -------------------------------------------------------------------------------------------- + + @Override + public List<PlanNode> getAlternativePlans(CostEstimator estimator) { + // check if we have a cached version + if (this.cachedPlans != null) { + return this.cachedPlans; + } + + // calculate alternative sub-plans for predecessor + List<? extends PlanNode> subPlans = getPredecessorNode().getAlternativePlans(estimator); + List<PlanNode> outputPlans = new ArrayList<PlanNode>(); + + final int dop = getParallelism(); + final int inDop = getPredecessorNode().getParallelism(); + + final ExecutionMode executionMode = this.input.getDataExchangeMode(); + final boolean dopChange = dop != inDop; + final boolean breakPipeline = this.input.isBreakingPipeline(); + + InterestingProperties ips = this.input.getInterestingProperties(); + for (PlanNode p : subPlans) { + for (RequestedGlobalProperties gp : ips.getGlobalProperties()) { + for (RequestedLocalProperties lp : ips.getLocalProperties()) { + Channel c = new Channel(p); + gp.parameterizeChannel(c, dopChange, executionMode, breakPipeline); + lp.parameterizeChannel(c); + c.setRequiredLocalProps(lp); + c.setRequiredGlobalProps(gp); + + // no need to check whether the created properties meet what we need in case + // of ordering or global ordering, because the only interesting properties we have + // are what we require + outputPlans.add(new SinkPlanNode(this, "DataSink ("+this.getOperator().getName()+")" ,c)); + } + } + } + + // cost and prune the plans + for (PlanNode node : outputPlans) { + estimator.costOperator(node); + } + prunePlanAlternatives(outputPlans); + + this.cachedPlans = outputPlans; + return outputPlans; + } + + // -------------------------------------------------------------------------------------------- + // Function Annotation Handling + // -------------------------------------------------------------------------------------------- + + @Override + public SemanticProperties getSemanticProperties() { + return new EmptySemanticProperties(); + } + + // -------------------------------------------------------------------------------------------- + // Miscellaneous + // -------------------------------------------------------------------------------------------- + + @Override + public void accept(Visitor<OptimizerNode> visitor) { + if (visitor.preVisit(this)) { + if (getPredecessorNode() != null) { + getPredecessorNode().accept(visitor); + } else { + throw new CompilerException(); + } + visitor.postVisit(this); + } + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSourceNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSourceNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSourceNode.java new file mode 100644 index 0000000..e4b35b7 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/DataSourceNode.java @@ -0,0 +1,306 @@ +/* + * 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 java.util.Map; + +import org.apache.flink.api.common.ExecutionMode; +import org.apache.flink.api.common.functions.Partitioner; +import org.apache.flink.api.common.io.FileInputFormat; +import org.apache.flink.api.common.io.InputFormat; +import org.apache.flink.api.common.io.NonParallelInput; +import org.apache.flink.api.common.io.ReplicatingInputFormat; +import org.apache.flink.api.common.io.statistics.BaseStatistics; +import org.apache.flink.api.common.operators.GenericDataSourceBase; +import org.apache.flink.api.common.operators.GenericDataSourceBase.SplitDataProperties; +import org.apache.flink.api.common.operators.Operator; +import org.apache.flink.api.common.operators.Ordering; +import org.apache.flink.api.common.operators.SemanticProperties; +import org.apache.flink.api.common.operators.SemanticProperties.EmptySemanticProperties; +import org.apache.flink.api.common.operators.util.FieldList; +import org.apache.flink.optimizer.DataStatistics; +import org.apache.flink.optimizer.Optimizer; +import org.apache.flink.optimizer.costs.CostEstimator; +import org.apache.flink.optimizer.costs.Costs; +import org.apache.flink.optimizer.dataproperties.GlobalProperties; +import org.apache.flink.optimizer.dataproperties.LocalProperties; +import org.apache.flink.optimizer.plan.PlanNode; +import org.apache.flink.optimizer.plan.SourcePlanNode; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.util.Visitor; + +/** + * The optimizer's internal representation of a data source. + */ +public class DataSourceNode extends OptimizerNode { + + private final boolean sequentialInput; + + private final boolean replicatedInput; + + private GlobalProperties gprops; + + private LocalProperties lprops; + + /** + * Creates a new DataSourceNode for the given contract. + * + * @param pactContract + * The data source contract object. + */ + public DataSourceNode(GenericDataSourceBase<?, ?> pactContract) { + super(pactContract); + + if (pactContract.getUserCodeWrapper().getUserCodeClass() == null) { + throw new IllegalArgumentException("Input format has not been set."); + } + + if (NonParallelInput.class.isAssignableFrom(pactContract.getUserCodeWrapper().getUserCodeClass())) { + setDegreeOfParallelism(1); + this.sequentialInput = true; + } else { + this.sequentialInput = false; + } + + this.replicatedInput = ReplicatingInputFormat.class.isAssignableFrom( + pactContract.getUserCodeWrapper().getUserCodeClass()); + + this.gprops = new GlobalProperties(); + this.lprops = new LocalProperties(); + + SplitDataProperties<?> splitProps = pactContract.getSplitDataProperties(); + + if(replicatedInput) { + this.gprops.setFullyReplicated(); + this.lprops = new LocalProperties(); + } else if (splitProps != null) { + // configure data properties of data source using split properties + setDataPropertiesFromSplitProperties(splitProps); + } + + } + + /** + * Gets the contract object for this data source node. + * + * @return The contract. + */ + @Override + public GenericDataSourceBase<?, ?> getOperator() { + return (GenericDataSourceBase<?, ?>) super.getOperator(); + } + + @Override + public String getName() { + return "Data Source"; + } + + @Override + public void setDegreeOfParallelism(int degreeOfParallelism) { + // if unsplittable, parallelism remains at 1 + if (!this.sequentialInput) { + super.setDegreeOfParallelism(degreeOfParallelism); + } + } + + @Override + public List<DagConnection> getIncomingConnections() { + return Collections.<DagConnection>emptyList(); + } + + @Override + public void setInput(Map<Operator<?>, OptimizerNode> contractToNode, ExecutionMode defaultDataExchangeMode) {} + + @Override + protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) { + // see, if we have a statistics object that can tell us a bit about the file + if (statistics != null) { + // instantiate the input format, as this is needed by the statistics + InputFormat<?, ?> format; + String inFormatDescription = "<unknown>"; + + try { + format = getOperator().getFormatWrapper().getUserCodeObject(); + Configuration config = getOperator().getParameters(); + format.configure(config); + } + catch (Throwable t) { + if (Optimizer.LOG.isWarnEnabled()) { + Optimizer.LOG.warn("Could not instantiate InputFormat to obtain statistics." + + " Limited statistics will be available.", t); + } + return; + } + try { + inFormatDescription = format.toString(); + } + catch (Throwable t) { + // we can ignore this error, as it only prevents us to use a cosmetic string + } + + // first of all, get the statistics from the cache + final String statisticsKey = getOperator().getStatisticsKey(); + final BaseStatistics cachedStatistics = statistics.getBaseStatistics(statisticsKey); + + BaseStatistics bs = null; + try { + bs = format.getStatistics(cachedStatistics); + } + catch (Throwable t) { + if (Optimizer.LOG.isWarnEnabled()) { + Optimizer.LOG.warn("Error obtaining statistics from input format: " + t.getMessage(), t); + } + } + + if (bs != null) { + final long len = bs.getTotalInputSize(); + if (len == BaseStatistics.SIZE_UNKNOWN) { + if (Optimizer.LOG.isInfoEnabled()) { + Optimizer.LOG.info("Compiler could not determine the size of input '" + inFormatDescription + "'. Using default estimates."); + } + } + else if (len >= 0) { + this.estimatedOutputSize = len; + } + + final long card = bs.getNumberOfRecords(); + if (card != BaseStatistics.NUM_RECORDS_UNKNOWN) { + this.estimatedNumRecords = card; + } + } + } + } + + @Override + public void computeInterestingPropertiesForInputs(CostEstimator estimator) { + // no children, so nothing to compute + } + + @Override + public void computeUnclosedBranchStack() { + // because there are no inputs, there are no unclosed branches. + this.openBranches = Collections.emptyList(); + } + + @Override + public List<PlanNode> getAlternativePlans(CostEstimator estimator) { + if (this.cachedPlans != null) { + return this.cachedPlans; + } + + SourcePlanNode candidate = new SourcePlanNode(this, "DataSource ("+this.getOperator().getName()+")", + this.gprops, this.lprops); + + if(!replicatedInput) { + candidate.updatePropertiesWithUniqueSets(getUniqueFields()); + + final Costs costs = new Costs(); + if (FileInputFormat.class.isAssignableFrom(getOperator().getFormatWrapper().getUserCodeClass()) && + this.estimatedOutputSize >= 0) { + estimator.addFileInputCost(this.estimatedOutputSize, costs); + } + candidate.setCosts(costs); + } else { + // replicated input + final Costs costs = new Costs(); + InputFormat<?,?> inputFormat = + ((ReplicatingInputFormat<?,?>) getOperator().getFormatWrapper().getUserCodeObject()).getReplicatedInputFormat(); + if (FileInputFormat.class.isAssignableFrom(inputFormat.getClass()) && + this.estimatedOutputSize >= 0) { + estimator.addFileInputCost(this.estimatedOutputSize * this.getParallelism(), costs); + } + candidate.setCosts(costs); + } + + // since there is only a single plan for the data-source, return a list with that element only + List<PlanNode> plans = new ArrayList<PlanNode>(1); + plans.add(candidate); + + this.cachedPlans = plans; + return plans; + } + + @Override + public SemanticProperties getSemanticProperties() { + return new EmptySemanticProperties(); + } + + @Override + public void accept(Visitor<OptimizerNode> visitor) { + if (visitor.preVisit(this)) { + visitor.postVisit(this); + } + } + + private void setDataPropertiesFromSplitProperties(SplitDataProperties splitProps) { + + // set global properties + int[] partitionKeys = splitProps.getSplitPartitionKeys(); + Partitioner<?> partitioner = splitProps.getSplitPartitioner(); + + if(partitionKeys != null && partitioner != null) { + this.gprops.setCustomPartitioned(new FieldList(partitionKeys), partitioner); + } + else if(partitionKeys != null) { + this.gprops.setAnyPartitioning(new FieldList(partitionKeys)); + } + // set local properties + int[] groupingKeys = splitProps.getSplitGroupKeys(); + Ordering ordering = splitProps.getSplitOrder(); + + // more than one split per source tasks possible. + // adapt split grouping and sorting + if(ordering != null) { + + // sorting falls back to grouping because a source can read multiple, + // randomly assigned splits + groupingKeys = ordering.getFieldPositions(); + } + + if(groupingKeys != null && partitionKeys != null) { + // check if grouping is also valid across splits, i.e., whether grouping keys are + // valid superset of partition keys + boolean allFieldsIncluded = true; + for(int i : partitionKeys) { + boolean fieldIncluded = false; + for(int j : groupingKeys) { + if(i == j) { + fieldIncluded = true; + break; + } + } + if(!fieldIncluded) { + allFieldsIncluded = false; + break; + } + } + if (allFieldsIncluded) { + this.lprops = LocalProperties.forGrouping(new FieldList(groupingKeys)); + } else { + this.lprops = new LocalProperties(); + } + + } else { + this.lprops = new LocalProperties(); + } + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/EstimateProvider.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/EstimateProvider.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/EstimateProvider.java new file mode 100644 index 0000000..482951b --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/EstimateProvider.java @@ -0,0 +1,47 @@ +/* + * 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; + +/** + * Methods for operators / connections that provide estimated about data size and + * characteristics. + */ +public interface EstimateProvider { + + /** + * Gets the estimated output size from this node. + * + * @return The estimated output size. + */ + long getEstimatedOutputSize(); + + /** + * Gets the estimated number of records in the output of this node. + * + * @return The estimated number of records. + */ + long getEstimatedNumRecords(); + + /** + * Gets the estimated number of bytes per record. + * + * @return The estimated number of bytes per record. + */ + float getEstimatedAvgWidthPerOutputRecord(); +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/FilterNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/FilterNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/FilterNode.java new file mode 100644 index 0000000..118ddc8 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/FilterNode.java @@ -0,0 +1,72 @@ +/* + * 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.SemanticProperties; +import org.apache.flink.api.common.operators.SingleInputSemanticProperties; +import org.apache.flink.api.common.operators.base.FilterOperatorBase; +import org.apache.flink.optimizer.DataStatistics; +import org.apache.flink.optimizer.operators.FilterDescriptor; +import org.apache.flink.optimizer.operators.OperatorDescriptorSingle; + +/** + * The optimizer's internal representation of a <i>FlatMap</i> operator node. + */ +public class FilterNode extends SingleInputNode { + + private final List<OperatorDescriptorSingle> possibleProperties; + + public FilterNode(FilterOperatorBase<?, ?> operator) { + super(operator); + this.possibleProperties = Collections.<OperatorDescriptorSingle>singletonList(new FilterDescriptor()); + } + + @Override + public FilterOperatorBase<?, ?> getOperator() { + return (FilterOperatorBase<?, ?>) super.getOperator(); + } + + @Override + public String getName() { + return "Filter"; + } + + @Override + public SemanticProperties getSemanticProperties() { + return new SingleInputSemanticProperties.AllFieldsForwardedProperties(); + } + + @Override + protected List<OperatorDescriptorSingle> getPossibleProperties() { + return this.possibleProperties; + } + + /** + * Computes the estimates for the Filter operator. Since it applies a filter on the data we assume a cardinality + * decrease. To give the system a hint at data decrease, we use a default magic number to indicate a 0.5 decrease. + */ + @Override + protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) { + this.estimatedNumRecords = (long) (getPredecessorNode().getEstimatedNumRecords() * 0.5); + this.estimatedOutputSize = (long) (getPredecessorNode().getEstimatedOutputSize() * 0.5); + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/FlatMapNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/FlatMapNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/FlatMapNode.java new file mode 100644 index 0000000..f713d56 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/FlatMapNode.java @@ -0,0 +1,65 @@ +/* + * 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.FlatMapOperatorBase; +import org.apache.flink.optimizer.DataStatistics; +import org.apache.flink.optimizer.operators.FlatMapDescriptor; +import org.apache.flink.optimizer.operators.OperatorDescriptorSingle; + +/** + * The optimizer's internal representation of a <i>FlatMap</i> operator node. + */ +public class FlatMapNode extends SingleInputNode { + + private final List<OperatorDescriptorSingle> possibleProperties; + + public FlatMapNode(FlatMapOperatorBase<?, ?, ?> operator) { + super(operator); + + this.possibleProperties = Collections.<OperatorDescriptorSingle>singletonList(new FlatMapDescriptor()); + } + + @Override + public FlatMapOperatorBase<?, ?, ?> getOperator() { + return (FlatMapOperatorBase<?, ?, ?>) super.getOperator(); + } + + @Override + public String getName() { + return "FlatMap"; + } + + @Override + protected List<OperatorDescriptorSingle> getPossibleProperties() { + return this.possibleProperties; + } + + /** + * Computes the estimates for the FlatMap operator. Since it un-nests, we assume a cardinality + * increase. To give the system a hint at data increase, we take a default magic number of a 5 times increase. + */ + @Override + protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) { + this.estimatedNumRecords = getPredecessorNode().getEstimatedNumRecords() * 5; + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/GroupCombineNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/GroupCombineNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/GroupCombineNode.java new file mode 100644 index 0000000..564c0d3 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/GroupCombineNode.java @@ -0,0 +1,100 @@ +/* + * 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.base.GroupCombineOperatorBase; +import org.apache.flink.optimizer.DataStatistics; +import org.apache.flink.optimizer.operators.AllGroupCombineProperties; +import org.apache.flink.optimizer.operators.GroupCombineProperties; +import org.apache.flink.optimizer.operators.OperatorDescriptorSingle; + +import java.util.Collections; +import java.util.List; + +/** + * The optimizer representation of a <i>GroupCombineNode</i> operation. + */ +public class GroupCombineNode extends SingleInputNode { + + private final List<OperatorDescriptorSingle> possibleProperties; + + /** + * Creates a new optimizer node for the given operator. + * + * @param operator The reduce operation. + */ + public GroupCombineNode(GroupCombineOperatorBase<?, ?, ?> operator) { + super(operator); + + if (this.keys == null) { + // case of a key-less reducer. force a parallelism of 1 + setDegreeOfParallelism(1); + } + + this.possibleProperties = initPossibleProperties(); + } + + private List<OperatorDescriptorSingle> initPossibleProperties() { + + // check if we can work with a grouping (simple reducer), or if we need ordering because of a group order + Ordering groupOrder = getOperator().getGroupOrder(); + if (groupOrder != null && groupOrder.getNumberOfFields() == 0) { + groupOrder = null; + } + + OperatorDescriptorSingle props = (this.keys == null ? + new AllGroupCombineProperties() : + new GroupCombineProperties(this.keys, groupOrder)); + + return Collections.singletonList(props); + } + + // ------------------------------------------------------------------------ + + /** + * Gets the operator represented by this optimizer node. + * + * @return The operator represented by this optimizer node. + */ + @Override + public GroupCombineOperatorBase<?, ?, ?> getOperator() { + return (GroupCombineOperatorBase<?, ?, ?>) super.getOperator(); + } + + @Override + public String getName() { + return "GroupCombine"; + } + + @Override + protected List<OperatorDescriptorSingle> getPossibleProperties() { + return this.possibleProperties; + } + + // -------------------------------------------------------------------------------------------- + // Estimates + // -------------------------------------------------------------------------------------------- + + @Override + protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) { + // no real estimates possible for a reducer. + } + +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/GroupReduceNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/GroupReduceNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/GroupReduceNode.java new file mode 100644 index 0000000..77acae5 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/GroupReduceNode.java @@ -0,0 +1,158 @@ +/* + * 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.functions.Partitioner; +import org.apache.flink.api.common.operators.Ordering; +import org.apache.flink.api.common.operators.base.GroupReduceOperatorBase; +import org.apache.flink.optimizer.CompilerException; +import org.apache.flink.optimizer.DataStatistics; +import org.apache.flink.optimizer.Optimizer; +import org.apache.flink.optimizer.operators.AllGroupReduceProperties; +import org.apache.flink.optimizer.operators.AllGroupWithPartialPreGroupProperties; +import org.apache.flink.optimizer.operators.GroupReduceProperties; +import org.apache.flink.optimizer.operators.GroupReduceWithCombineProperties; +import org.apache.flink.optimizer.operators.OperatorDescriptorSingle; +import org.apache.flink.configuration.Configuration; + +/** + * The optimizer representation of a <i>GroupReduce</i> operation. + */ +public class GroupReduceNode extends SingleInputNode { + + private final List<OperatorDescriptorSingle> possibleProperties; + + private GroupReduceNode combinerUtilityNode; + + /** + * Creates a new optimizer node for the given operator. + * + * @param operator The reduce operation. + */ + public GroupReduceNode(GroupReduceOperatorBase<?, ?, ?> operator) { + super(operator); + + if (this.keys == null) { + // case of a key-less reducer. force a parallelism of 1 + setDegreeOfParallelism(1); + } + + this.possibleProperties = initPossibleProperties(operator.getCustomPartitioner()); + } + + public GroupReduceNode(GroupReduceNode reducerToCopyForCombiner) { + super(reducerToCopyForCombiner); + + this.possibleProperties = Collections.emptyList(); + } + + private List<OperatorDescriptorSingle> initPossibleProperties(Partitioner<?> customPartitioner) { + // see if an internal hint dictates the strategy to use + final Configuration conf = getOperator().getParameters(); + final String localStrategy = conf.getString(Optimizer.HINT_LOCAL_STRATEGY, null); + + final boolean useCombiner; + if (localStrategy != null) { + if (Optimizer.HINT_LOCAL_STRATEGY_SORT.equals(localStrategy)) { + useCombiner = false; + } + else if (Optimizer.HINT_LOCAL_STRATEGY_COMBINING_SORT.equals(localStrategy)) { + if (!isCombineable()) { + Optimizer.LOG.warn("Strategy hint for GroupReduce '" + getOperator().getName() + + "' requires combinable reduce, but user function is not marked combinable."); + } + useCombiner = true; + } else { + throw new CompilerException("Invalid local strategy hint for match contract: " + localStrategy); + } + } else { + useCombiner = isCombineable(); + } + + // check if we can work with a grouping (simple reducer), or if we need ordering because of a group order + Ordering groupOrder = null; + if (getOperator() instanceof GroupReduceOperatorBase) { + groupOrder = ((GroupReduceOperatorBase<?, ?, ?>) getOperator()).getGroupOrder(); + if (groupOrder != null && groupOrder.getNumberOfFields() == 0) { + groupOrder = null; + } + } + + OperatorDescriptorSingle props = useCombiner ? + (this.keys == null ? new AllGroupWithPartialPreGroupProperties() : new GroupReduceWithCombineProperties(this.keys, groupOrder, customPartitioner)) : + (this.keys == null ? new AllGroupReduceProperties() : new GroupReduceProperties(this.keys, groupOrder, customPartitioner)); + + return Collections.singletonList(props); + } + + // ------------------------------------------------------------------------ + + /** + * Gets the operator represented by this optimizer node. + * + * @return The operator represented by this optimizer node. + */ + @Override + public GroupReduceOperatorBase<?, ?, ?> getOperator() { + return (GroupReduceOperatorBase<?, ?, ?>) super.getOperator(); + } + + /** + * Checks, whether a combiner function has been given for the function encapsulated + * by this reduce contract. + * + * @return True, if a combiner has been given, false otherwise. + */ + public boolean isCombineable() { + return getOperator().isCombinable(); + } + + @Override + public String getName() { + return "GroupReduce"; + } + + @Override + protected List<OperatorDescriptorSingle> getPossibleProperties() { + return this.possibleProperties; + } + + // -------------------------------------------------------------------------------------------- + // Estimates + // -------------------------------------------------------------------------------------------- + + @Override + protected void computeOperatorSpecificDefaultEstimates(DataStatistics statistics) { + // no real estimates possible for a reducer. + } + + public GroupReduceNode getCombinerUtilityNode() { + if (this.combinerUtilityNode == null) { + this.combinerUtilityNode = new GroupReduceNode(this); + + // we conservatively assume the combiner returns the same data size as it consumes + this.combinerUtilityNode.estimatedOutputSize = getPredecessorNode().getEstimatedOutputSize(); + this.combinerUtilityNode.estimatedNumRecords = getPredecessorNode().getEstimatedNumRecords(); + } + return this.combinerUtilityNode; + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/InterestingPropertiesClearer.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/InterestingPropertiesClearer.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/InterestingPropertiesClearer.java new file mode 100644 index 0000000..1fdae51 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/InterestingPropertiesClearer.java @@ -0,0 +1,39 @@ +/* + * 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.util.Visitor; + +final class InterestingPropertiesClearer implements Visitor<OptimizerNode> { + + static final InterestingPropertiesClearer INSTANCE = new InterestingPropertiesClearer(); + + @Override + public boolean preVisit(OptimizerNode visitable) { + if (visitable.getInterestingProperties() != null) { + visitable.clearInterestingProperties(); + return true; + } else { + return false; + } + } + + @Override + public void postVisit(OptimizerNode visitable) {} +} http://git-wip-us.apache.org/repos/asf/flink/blob/633b0d6a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/IterationNode.java ---------------------------------------------------------------------- diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/IterationNode.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/IterationNode.java new file mode 100644 index 0000000..5d28043 --- /dev/null +++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/IterationNode.java @@ -0,0 +1,30 @@ +/* + * 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.util.Visitor; + +/** + * + */ +public interface IterationNode { + + void acceptForStepFunction(Visitor<OptimizerNode> visitor); + +}
