| /* |
| * 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.wayang.core.plan.wayangplan; |
| |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Optional; |
| import org.apache.logging.log4j.LogManager; |
| import org.apache.wayang.core.api.Configuration; |
| import org.apache.wayang.core.optimizer.OptimizationContext; |
| import org.apache.wayang.core.optimizer.cardinality.CardinalityEstimate; |
| import org.apache.wayang.core.optimizer.costs.LoadProfile; |
| import org.apache.wayang.core.optimizer.costs.LoadProfileEstimator; |
| import org.apache.wayang.core.optimizer.costs.LoadProfileEstimators; |
| import org.apache.wayang.core.plan.executionplan.Channel; |
| import org.apache.wayang.core.plan.executionplan.ExecutionTask; |
| import org.apache.wayang.core.platform.ChannelDescriptor; |
| import org.apache.wayang.core.platform.ChannelInstance; |
| import org.apache.wayang.core.platform.Executor; |
| import org.apache.wayang.core.platform.Platform; |
| import org.apache.wayang.core.platform.lineage.ExecutionLineageNode; |
| import org.apache.wayang.core.platform.lineage.LazyExecutionLineageNode; |
| import org.apache.wayang.core.util.Tuple; |
| |
| /** |
| * An execution operator is handled by a certain platform. |
| */ |
| public interface ExecutionOperator extends ElementaryOperator { |
| |
| /** |
| * @return the platform that can run this operator |
| */ |
| Platform getPlatform(); |
| |
| /** |
| * @return a copy of this instance; it's {@link Slot}s will not be connected |
| */ |
| ExecutionOperator copy(); |
| |
| /** |
| * @return this instance or, if it was derived via {@link #copy()}, the original instance |
| */ |
| ExecutionOperator getOriginal(); |
| |
| /** |
| * Developers of {@link ExecutionOperator}s can provide a default {@link LoadProfileEstimator} via this method. |
| * |
| * @param configuration in which the {@link LoadProfile} should be estimated. |
| * @return an {@link Optional} that might contain the {@link LoadProfileEstimator} (but {@link Optional#empty()} |
| * by default) |
| */ |
| default Optional<LoadProfileEstimator> createLoadProfileEstimator(Configuration configuration) { |
| Collection<String> configurationKeys = this.getLoadProfileEstimatorConfigurationKeys(); |
| LoadProfileEstimator mainEstimator = createLoadProfileEstimators(configuration, configurationKeys); |
| return Optional.ofNullable(mainEstimator); |
| } |
| |
| /** |
| * Creates a {@link LoadProfileEstimator} according to the {@link Configuration}. |
| * |
| * @param configuration the {@link Configuration} |
| * @param configurationKeys keys for the specification within the {@link Configuration} |
| * @return the {@link LoadProfileEstimator} or {@code null} if none could be created |
| */ |
| static LoadProfileEstimator createLoadProfileEstimators(Configuration configuration, Collection<String> configurationKeys) { |
| LoadProfileEstimator mainEstimator = null; |
| for (String configurationKey : configurationKeys) { |
| final LoadProfileEstimator loadProfileEstimator = LoadProfileEstimators.createFromSpecification(configurationKey, configuration); |
| if (mainEstimator == null) { |
| mainEstimator = loadProfileEstimator; |
| } else { |
| mainEstimator.nest(loadProfileEstimator); |
| } |
| } |
| return mainEstimator; |
| } |
| |
| /** |
| * Provide the {@link Configuration} keys for the {@link LoadProfileEstimator} specification of this instance. |
| * |
| * @return the {@link Configuration} keys |
| */ |
| default Collection<String> getLoadProfileEstimatorConfigurationKeys() { |
| final String singleKey = this.getLoadProfileEstimatorConfigurationKey(); |
| return singleKey == null ? |
| Collections.emptyList() : |
| Collections.singletonList(singleKey); |
| } |
| |
| /** |
| * @deprecated Use {@link #getLoadProfileEstimatorConfigurationKeys()} |
| */ |
| default String getLoadProfileEstimatorConfigurationKey() { |
| return null; |
| } |
| |
| /** |
| * Provides a base key for configuring input and output limits. |
| * @return the limit base key or {@code null} |
| */ |
| default String getLimitBaseKey() { |
| // By default, try to infer the key. |
| String loadProfileEstimatorConfigurationKey = this.getLoadProfileEstimatorConfigurationKey(); |
| if (loadProfileEstimatorConfigurationKey != null && loadProfileEstimatorConfigurationKey.endsWith(".load")) { |
| return loadProfileEstimatorConfigurationKey |
| .substring(0, loadProfileEstimatorConfigurationKey.length() - 5) |
| .concat(".limit"); |
| } |
| return null; |
| } |
| |
| /** |
| * Tells whether this instance should not be executed on the face of the given |
| * {@link OptimizationContext.OperatorContext}. For instance, when this instance |
| * might not be able to handle the amount of data. |
| * |
| * @param operatorContext an {@link OptimizationContext.OperatorContext} within which this instance might be |
| * executed |
| * @return whether this instance should <b>not</b> be used for execution |
| */ |
| default boolean isFiltered(OptimizationContext.OperatorContext operatorContext) { |
| assert operatorContext.getOperator() == this; |
| |
| // By default, we look for configuration keys formed like this: |
| // <my.operator.limit.key>.<input/output name> |
| // If such a key exists, we compare it to values in the operatorContext. |
| String limitBaseKey = this.getLimitBaseKey(); |
| if (limitBaseKey != null) { |
| Configuration configuration = operatorContext.getOptimizationContext().getConfiguration(); |
| |
| // Check the inputs. |
| for (InputSlot<?> input : this.getAllInputs()) { |
| String key = limitBaseKey + "." + input.getName(); |
| long limit = configuration.getLongProperty(key, -1); |
| if (limit >= 0) { |
| CardinalityEstimate cardinality = operatorContext.getInputCardinality(input.getIndex()); |
| if (cardinality != null && cardinality.getGeometricMeanEstimate() > limit) return true; |
| } |
| } |
| |
| // Check the outputs. |
| for (OutputSlot<?> output : this.getAllOutputs()) { |
| String key = limitBaseKey + "." + output.getName(); |
| long limit = configuration.getLongProperty(key, -1); |
| if (limit >= 0) { |
| CardinalityEstimate cardinality = operatorContext.getOutputCardinality(output.getIndex()); |
| if (cardinality != null && cardinality.getGeometricMeanEstimate() > limit) return true; |
| } |
| } |
| } |
| |
| return false; |
| } |
| |
| /** |
| * Display the supported {@link Channel}s for a certain {@link InputSlot}. |
| * |
| * @param index the index of the {@link InputSlot} |
| * @return an {@link List} of {@link Channel}s' {@link Class}es, ordered by their preference of use |
| */ |
| List<ChannelDescriptor> getSupportedInputChannels(int index); |
| |
| /** |
| * Display the supported {@link Channel}s for a certain {@link OutputSlot}. |
| * |
| * @param index the index of the {@link OutputSlot} |
| * @return an {@link List} of {@link Channel}s' {@link Class}es, ordered by their preference of use |
| * @see #getOutputChannelDescriptor(int) |
| * @deprecated {@link ExecutionOperator}s should only support a single {@link ChannelDescriptor} |
| */ |
| @Deprecated |
| List<ChannelDescriptor> getSupportedOutputChannels(int index); |
| |
| /** |
| * Display the {@link Channel} used to implement a certain {@link OutputSlot}. |
| * |
| * @param index index of the {@link OutputSlot} |
| * @return the {@link ChannelDescriptor} for the mentioned {@link Channel} |
| */ |
| default ChannelDescriptor getOutputChannelDescriptor(int index) { |
| final List<ChannelDescriptor> supportedOutputChannels = this.getSupportedOutputChannels(index); |
| assert !supportedOutputChannels.isEmpty() : String.format("No supported output channels for %s.", this); |
| if (supportedOutputChannels.size() > 1) { |
| LogManager.getLogger(this.getClass()).warn("Treat {} as the only supported channel for {}.", |
| supportedOutputChannels.get(0), this.getOutput(index) |
| ); |
| } |
| return supportedOutputChannels.get(0); |
| } |
| |
| /** |
| * Create output {@link ChannelInstance}s for this instance. |
| * |
| * @param task the {@link ExecutionTask} in which this instance is being wrapped |
| * @param producerOperatorContext the {@link OptimizationContext.OperatorContext} for this instance |
| * @param inputChannelInstances the input {@link ChannelInstance}s for the {@code task} |
| * @return |
| */ |
| default ChannelInstance[] createOutputChannelInstances(Executor executor, |
| ExecutionTask task, |
| OptimizationContext.OperatorContext producerOperatorContext, |
| List<ChannelInstance> inputChannelInstances) { |
| |
| assert task.getOperator() == this; |
| ChannelInstance[] channelInstances = new ChannelInstance[task.getNumOuputChannels()]; |
| for (int outputIndex = 0; outputIndex < channelInstances.length; outputIndex++) { |
| final Channel outputChannel = task.getOutputChannel(outputIndex); |
| final ChannelInstance outputChannelInstance = outputChannel.createInstance(executor, producerOperatorContext, outputIndex); |
| channelInstances[outputIndex] = outputChannelInstance; |
| } |
| return channelInstances; |
| } |
| |
| /** |
| * Models eager execution by marking all {@link LazyExecutionLineageNode}s as executed and collecting all marked ones. |
| * |
| * @param inputs the input {@link ChannelInstance}s |
| * @param outputs the output {@link ChannelInstance}s |
| * @param operatorContext the executed {@link OptimizationContext.OperatorContext} |
| * @return the executed {@link OptimizationContext.OperatorContext} and produced {@link ChannelInstance}s |
| */ |
| static Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> modelEagerExecution( |
| ChannelInstance[] inputs, |
| ChannelInstance[] outputs, |
| OptimizationContext.OperatorContext operatorContext) { |
| |
| final ExecutionLineageNode executionLineageNode = new ExecutionLineageNode(operatorContext); |
| executionLineageNode.addAtomicExecutionFromOperatorContext(); |
| LazyExecutionLineageNode.connectAll(inputs, executionLineageNode, outputs); |
| |
| final Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> collectors; |
| if (outputs.length == 0) { |
| collectors = executionLineageNode.collectAndMark(); |
| } else { |
| collectors = new Tuple<>(new LinkedList<>(), new LinkedList<>()); |
| for (ChannelInstance output : outputs) { |
| output.getLineage().collectAndMark(collectors.getField0(), collectors.getField1()); |
| } |
| } |
| return collectors; |
| } |
| |
| /** |
| * Models eager execution by marking all {@link LazyExecutionLineageNode}s as executed and collecting all marked ones. |
| * However, the output {@link ChannelInstance}s are not yet produced. |
| * |
| * @param inputs the input {@link ChannelInstance}s |
| * @param outputs the output {@link ChannelInstance}s |
| * @param operatorContext the executed {@link OptimizationContext.OperatorContext} |
| * @return the executed {@link OptimizationContext.OperatorContext} and produced {@link ChannelInstance}s |
| */ |
| static Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> modelQuasiEagerExecution( |
| ChannelInstance[] inputs, |
| ChannelInstance[] outputs, |
| OptimizationContext.OperatorContext operatorContext) { |
| |
| final ExecutionLineageNode executionLineageNode = new ExecutionLineageNode(operatorContext); |
| executionLineageNode.addAtomicExecutionFromOperatorContext(); |
| LazyExecutionLineageNode.connectAll(inputs, executionLineageNode, outputs); |
| |
| return executionLineageNode.collectAndMark(); |
| } |
| |
| /** |
| * Models lazy execution by not marking any {@link LazyExecutionLineageNode}s. |
| * |
| * @param inputs the input {@link ChannelInstance}s |
| * @param outputs the output {@link ChannelInstance}s |
| * @param operatorContext the executed {@link OptimizationContext.OperatorContext} |
| * @return the executed {@link OptimizationContext.OperatorContext} and produced {@link ChannelInstance}s |
| */ |
| static Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> |
| modelLazyExecution(ChannelInstance[] inputs, |
| ChannelInstance[] outputs, |
| OptimizationContext.OperatorContext operatorContext) { |
| |
| final ExecutionLineageNode executionLineageNode = new ExecutionLineageNode(operatorContext); |
| executionLineageNode.addAtomicExecutionFromOperatorContext(); |
| LazyExecutionLineageNode.connectAll(inputs, executionLineageNode, outputs); |
| |
| return new Tuple<>(Collections.emptyList(), Collections.emptyList()); |
| } |
| |
| } |
