| /* |
| * 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.api; |
| |
| import org.apache.commons.io.IOUtils; |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.wayang.core.api.configuration.CollectionProvider; |
| import org.apache.wayang.core.api.configuration.ConstantValueProvider; |
| import org.apache.wayang.core.api.configuration.ExplicitCollectionProvider; |
| import org.apache.wayang.core.api.configuration.FunctionalCollectionProvider; |
| import org.apache.wayang.core.api.configuration.FunctionalKeyValueProvider; |
| import org.apache.wayang.core.api.configuration.FunctionalValueProvider; |
| import org.apache.wayang.core.api.configuration.KeyValueProvider; |
| import org.apache.wayang.core.api.configuration.MapBasedKeyValueProvider; |
| import org.apache.wayang.core.api.configuration.ValueProvider; |
| import org.apache.wayang.core.api.exception.WayangException; |
| import org.apache.wayang.core.function.FlatMapDescriptor; |
| import org.apache.wayang.core.function.FunctionDescriptor; |
| import org.apache.wayang.core.function.MapPartitionsDescriptor; |
| import org.apache.wayang.core.function.PredicateDescriptor; |
| import org.apache.wayang.core.mapping.Mapping; |
| import org.apache.wayang.core.optimizer.ProbabilisticDoubleInterval; |
| import org.apache.wayang.core.optimizer.cardinality.CardinalityEstimate; |
| import org.apache.wayang.core.optimizer.cardinality.CardinalityEstimator; |
| import org.apache.wayang.core.optimizer.cardinality.FallbackCardinalityEstimator; |
| import org.apache.wayang.core.optimizer.channels.ChannelConversion; |
| import org.apache.wayang.core.optimizer.costs.IntervalLoadEstimator; |
| import org.apache.wayang.core.optimizer.costs.LoadProfileEstimator; |
| import org.apache.wayang.core.optimizer.costs.LoadProfileToTimeConverter; |
| import org.apache.wayang.core.optimizer.costs.LoadToTimeConverter; |
| import org.apache.wayang.core.optimizer.costs.NestableLoadProfileEstimator; |
| import org.apache.wayang.core.optimizer.costs.TimeToCostConverter; |
| import org.apache.wayang.core.optimizer.enumeration.PlanEnumerationPruningStrategy; |
| import org.apache.wayang.core.plan.wayangplan.ElementaryOperator; |
| import org.apache.wayang.core.plan.wayangplan.ExecutionOperator; |
| import org.apache.wayang.core.plan.wayangplan.OutputSlot; |
| import org.apache.wayang.core.platform.Platform; |
| import org.apache.wayang.core.plugin.Plugin; |
| import org.apache.wayang.core.profiling.InstrumentationStrategy; |
| import org.apache.wayang.core.profiling.OutboundInstrumentationStrategy; |
| import org.apache.wayang.core.util.Actions; |
| import org.apache.wayang.core.util.ReflectionUtils; |
| import org.apache.wayang.core.util.fs.FileSystem; |
| import org.apache.wayang.core.util.fs.FileSystems; |
| import org.apache.logging.log4j.LogManager; |
| import org.apache.logging.log4j.Logger; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.net.URL; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.LinkedList; |
| import java.util.Map; |
| import java.util.Optional; |
| import java.util.OptionalDouble; |
| import java.util.OptionalLong; |
| import java.util.Properties; |
| import java.util.Random; |
| import java.util.function.ToDoubleFunction; |
| |
| import static org.apache.wayang.core.util.ReflectionUtils.instantiateDefault; |
| |
| /** |
| * Describes both the configuration of a {@link WayangContext} and {@link Job}s. |
| */ |
| public class Configuration { |
| |
| private static final Logger logger = LogManager.getLogger(Configuration.class); |
| |
| private static final String DEFAULT_CONFIGURATION_FILE = "wayang-core-defaults.properties"; |
| |
| private static final Configuration defaultConfiguration = new Configuration((Configuration) null); |
| |
| static { |
| defaultConfiguration.name = "default"; |
| Actions.doSafe(() -> bootstrapCardinalityEstimationProvider(defaultConfiguration)); |
| Actions.doSafe(() -> bootstrapSelectivityProviders(defaultConfiguration)); |
| Actions.doSafe(() -> bootstrapLoadAndTimeEstimatorProviders(defaultConfiguration)); |
| Actions.doSafe(() -> bootstrapPruningProviders(defaultConfiguration)); |
| Actions.doSafe(() -> bootstrapProperties(defaultConfiguration)); |
| Actions.doSafe(() -> bootstrapPlugins(defaultConfiguration)); |
| } |
| |
| private static final String BASIC_PLUGIN = "org.apache.wayang.basic.WayangBasics.defaultPlugin()"; |
| |
| private String name = "(no name)"; |
| |
| private final Configuration parent; |
| |
| private KeyValueProvider<OutputSlot<?>, CardinalityEstimator> cardinalityEstimatorProvider; |
| |
| private KeyValueProvider<FunctionDescriptor, ProbabilisticDoubleInterval> udfSelectivityProvider; |
| |
| private KeyValueProvider<ExecutionOperator, LoadProfileEstimator> operatorLoadProfileEstimatorProvider; |
| |
| private KeyValueProvider<FunctionDescriptor, LoadProfileEstimator> functionLoadProfileEstimatorProvider; |
| |
| private MapBasedKeyValueProvider<String, LoadProfileEstimator> loadProfileEstimatorCache; |
| |
| private KeyValueProvider<Platform, LoadProfileToTimeConverter> loadProfileToTimeConverterProvider; |
| |
| private KeyValueProvider<Platform, TimeToCostConverter> timeToCostConverterProvider; |
| |
| private ValueProvider<ToDoubleFunction<ProbabilisticDoubleInterval>> costSquasherProvider; |
| |
| private KeyValueProvider<Platform, Long> platformStartUpTimeProvider; |
| |
| private ExplicitCollectionProvider<Platform> platformProvider; |
| |
| private ExplicitCollectionProvider<Mapping> mappingProvider; |
| |
| private ExplicitCollectionProvider<ChannelConversion> channelConversionProvider; |
| |
| private CollectionProvider<Class<PlanEnumerationPruningStrategy>> pruningStrategyClassProvider; |
| |
| private ValueProvider<InstrumentationStrategy> instrumentationStrategyProvider; |
| |
| private KeyValueProvider<String, String> properties; |
| |
| /** |
| * Creates a new top-level instance that bases directly from the default instance. Will try to load the |
| * user configuration file. |
| * |
| * @see #getDefaultConfiguration() |
| */ |
| public Configuration() { |
| this(findUserConfigurationFile()); |
| } |
| |
| /** |
| * Creates a new top-level instance that bases directly from the default instance and loads the specified |
| * configuration file. |
| * |
| * @see #getDefaultConfiguration() |
| * @see #load(String) |
| */ |
| public Configuration(String configurationFileUrl) { |
| this(getDefaultConfiguration()); |
| if (configurationFileUrl != null) { |
| this.name = configurationFileUrl; |
| this.load(configurationFileUrl); |
| } |
| } |
| |
| /** |
| * Basic constructor. |
| */ |
| private Configuration(Configuration parent) { |
| this.parent = parent; |
| |
| if (this.parent != null) { |
| // Providers for platforms. |
| this.platformProvider = new ExplicitCollectionProvider<>(this, this.parent.platformProvider); |
| this.mappingProvider = new ExplicitCollectionProvider<>(this, this.parent.mappingProvider); |
| this.channelConversionProvider = new ExplicitCollectionProvider<>(this, this.parent.channelConversionProvider); |
| |
| // Providers for cardinality estimation. |
| this.cardinalityEstimatorProvider = |
| new MapBasedKeyValueProvider<>(this.parent.cardinalityEstimatorProvider, this); |
| this.udfSelectivityProvider = |
| new MapBasedKeyValueProvider<>(this.parent.udfSelectivityProvider, this); |
| |
| // Providers for cost functions. |
| this.operatorLoadProfileEstimatorProvider = |
| new MapBasedKeyValueProvider<>(this.parent.operatorLoadProfileEstimatorProvider, this); |
| this.functionLoadProfileEstimatorProvider = |
| new MapBasedKeyValueProvider<>(this.parent.functionLoadProfileEstimatorProvider, this); |
| this.loadProfileEstimatorCache = |
| new MapBasedKeyValueProvider<>(this.parent.loadProfileEstimatorCache, this); |
| this.loadProfileToTimeConverterProvider = |
| new MapBasedKeyValueProvider<>(this.parent.loadProfileToTimeConverterProvider, this); |
| this.timeToCostConverterProvider = |
| new MapBasedKeyValueProvider<>(this.parent.timeToCostConverterProvider, this); |
| this.platformStartUpTimeProvider = |
| new MapBasedKeyValueProvider<>(this.parent.platformStartUpTimeProvider, this); |
| this.costSquasherProvider = |
| new ConstantValueProvider<>(this, this.parent.costSquasherProvider); |
| |
| // Providers for plan enumeration. |
| this.pruningStrategyClassProvider = new ExplicitCollectionProvider<>(this, this.parent.pruningStrategyClassProvider); |
| this.instrumentationStrategyProvider = new ConstantValueProvider<>(this, this.parent.instrumentationStrategyProvider); |
| |
| // Properties. |
| this.properties = new MapBasedKeyValueProvider<>(this.parent.properties, this); |
| |
| } |
| } |
| |
| private static String findUserConfigurationFile() { |
| final String systemProperty = System.getProperty("wayang.configuration"); |
| if (systemProperty != null) { |
| logger.info("Using configuration at {}.", systemProperty); |
| return systemProperty; |
| } |
| |
| final URL classPathResource = ReflectionUtils.getResourceURL("wayang.properties"); |
| if (classPathResource != null) { |
| logger.info("Using configuration at {}.", classPathResource); |
| return classPathResource.toString(); |
| } |
| |
| logger.info("Using blank configuration."); |
| return null; |
| } |
| |
| /** |
| * Adjusts this instance to the properties specified in the given file. |
| * |
| * @param configurationUrl URL to the configuration file |
| */ |
| public void load(String configurationUrl) { |
| final Optional<FileSystem> fileSystem = FileSystems.getFileSystem(configurationUrl); |
| if (!fileSystem.isPresent()) { |
| throw new WayangException(String.format("Could not access %s.", configurationUrl)); |
| } |
| try (InputStream configInputStream = fileSystem.get().open(configurationUrl)) { |
| this.load(configInputStream); |
| } catch (Exception e) { |
| throw new WayangException(String.format("Could not load configuration from %s.", configurationUrl), e); |
| } |
| } |
| |
| /** |
| * Adjusts this instance to the properties specified in the given file. |
| * |
| * @param configInputStream of the file |
| */ |
| public void load(InputStream configInputStream) { |
| try { |
| final Properties properties = new Properties(); |
| properties.load(configInputStream); |
| for (Map.Entry<Object, Object> propertyEntry : properties.entrySet()) { |
| final String key = propertyEntry.getKey().toString(); |
| final String value = propertyEntry.getValue().toString(); |
| this.handleConfigurationFileEntry(key, value); |
| } |
| } catch (IOException e) { |
| throw new WayangException("Could not load configuration.", e); |
| } finally { |
| IOUtils.closeQuietly(configInputStream); |
| } |
| } |
| |
| /** |
| * Handle a just loaded property. |
| * |
| * @param key the property's key |
| * @param value the property's value |
| */ |
| private void handleConfigurationFileEntry(String key, String value) { |
| switch (key) { |
| case "wayang.core.optimizer.cost.squash": |
| if (!(this.costSquasherProvider instanceof ConstantValueProvider)) { |
| logger.warn("Cannot update cost estimate provider."); |
| } else if ("expectation".equals(value)) { |
| ((ConstantValueProvider<ToDoubleFunction<ProbabilisticDoubleInterval>>) this.costSquasherProvider).setValue( |
| ProbabilisticDoubleInterval::getGeometricMeanEstimate |
| ); |
| } else if ("random".equals(value)) { |
| final int salt = new Random().nextInt(); |
| ((ConstantValueProvider<ToDoubleFunction<ProbabilisticDoubleInterval>>) this.costSquasherProvider).setValue( |
| cost -> cost.hashCode() * salt + cost.hashCode() |
| ); |
| } else { |
| logger.warn("Cannot set unknown cost comparator \"{}\".", value); |
| } |
| break; |
| default: |
| this.setProperty(key, value); |
| break; |
| } |
| } |
| |
| |
| /** |
| * Returns the global default instance. It will be the fallback for all other instances and should only modified |
| * to provide default values. |
| */ |
| public static Configuration getDefaultConfiguration() { |
| return defaultConfiguration; |
| } |
| |
| private static void bootstrapPlugins(Configuration configuration) { |
| configuration.setPlatformProvider(new ExplicitCollectionProvider<>(configuration)); |
| configuration.setMappingProvider(new ExplicitCollectionProvider<>(configuration)); |
| configuration.setChannelConversionProvider(new ExplicitCollectionProvider<>(configuration)); |
| try { |
| Plugin basicPlugin = ReflectionUtils.evaluate(BASIC_PLUGIN); |
| basicPlugin.configure(configuration); |
| } catch (Exception e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Could not load basic plugin.", e); |
| } else { |
| logger.warn("Could not load basic plugin."); |
| } |
| } |
| } |
| |
| private static void bootstrapCardinalityEstimationProvider(final Configuration configuration) { |
| // Safety net: provide a fallback estimator. |
| KeyValueProvider<OutputSlot<?>, CardinalityEstimator> fallbackProvider = |
| new FunctionalKeyValueProvider<OutputSlot<?>, CardinalityEstimator>( |
| outputSlot -> new FallbackCardinalityEstimator(), |
| configuration |
| ).withSlf4jWarning("Creating fallback cardinality estimator for {}."); |
| |
| // Default option: Implementations define their estimators. |
| KeyValueProvider<OutputSlot<?>, CardinalityEstimator> defaultProvider = |
| new FunctionalKeyValueProvider<>(fallbackProvider, (outputSlot, requestee) -> { |
| assert outputSlot.getOwner().isElementary() |
| : String.format("Cannot provide estimator for composite %s.", outputSlot.getOwner()); |
| final ElementaryOperator operator = (ElementaryOperator) outputSlot.getOwner(); |
| // Instance-level estimator? |
| if (operator.getCardinalityEstimator(outputSlot.getIndex()) != null) { |
| return operator.getCardinalityEstimator(outputSlot.getIndex()); |
| } |
| // Type-level estimator? |
| return operator |
| .createCardinalityEstimator(outputSlot.getIndex(), configuration) |
| .orElse(null); |
| }); |
| |
| // Customizable layer: Users can override manually. |
| KeyValueProvider<OutputSlot<?>, CardinalityEstimator> overrideProvider = |
| new MapBasedKeyValueProvider<>(defaultProvider); |
| |
| configuration.setCardinalityEstimatorProvider(overrideProvider); |
| } |
| |
| private static void bootstrapSelectivityProviders(Configuration configuration) { |
| // Selectivity of UDFs |
| { |
| // Safety net: provide a fallback selectivity. |
| KeyValueProvider<FunctionDescriptor, ProbabilisticDoubleInterval> fallbackProvider = |
| new FunctionalKeyValueProvider<FunctionDescriptor, ProbabilisticDoubleInterval>( |
| functionDescriptor -> { |
| if (functionDescriptor instanceof PredicateDescriptor) { |
| return new ProbabilisticDoubleInterval(0.1, 1, 0.9d); |
| } else if (functionDescriptor instanceof FlatMapDescriptor) { |
| return new ProbabilisticDoubleInterval(0.1, 1, 0.9d); |
| } else if (functionDescriptor instanceof MapPartitionsDescriptor) { |
| return new ProbabilisticDoubleInterval(0.1, 1, 0.9d); |
| } else { |
| throw new WayangException("Cannot provide fallback selectivity for " + functionDescriptor); |
| } |
| }, |
| configuration |
| ).withSlf4jWarning("Using fallback selectivity for {}."); |
| |
| // Built-in option: Let the PredicateDescriptor provide its selectivity. |
| KeyValueProvider<FunctionDescriptor, ProbabilisticDoubleInterval> builtInProvider = |
| new FunctionalKeyValueProvider<>( |
| fallbackProvider, |
| functionDescriptor -> FunctionDescriptor.getSelectivity(functionDescriptor).orElse(null) |
| ); |
| |
| // Customizable layer: Users can override manually. |
| KeyValueProvider<FunctionDescriptor, ProbabilisticDoubleInterval> overrideProvider = |
| new MapBasedKeyValueProvider<>(builtInProvider); |
| |
| configuration.setUdfSelectivityProvider(overrideProvider); |
| } |
| } |
| |
| private static void bootstrapLoadAndTimeEstimatorProviders(Configuration configuration) { |
| { |
| // Safety net: provide a fallback selectivity. |
| KeyValueProvider<ExecutionOperator, LoadProfileEstimator> fallbackProvider = |
| new FunctionalKeyValueProvider<ExecutionOperator, LoadProfileEstimator>( |
| (operator, requestee) -> { |
| final Configuration conf = requestee.getConfiguration(); |
| return new NestableLoadProfileEstimator( |
| IntervalLoadEstimator.createIOLinearEstimator( |
| null, |
| conf.getLongProperty("wayang.core.fallback.udf.cpu.lower"), |
| conf.getLongProperty("wayang.core.fallback.udf.cpu.upper"), |
| conf.getDoubleProperty("wayang.core.fallback.udf.cpu.confidence"), |
| CardinalityEstimate.EMPTY_ESTIMATE |
| ), |
| IntervalLoadEstimator.createIOLinearEstimator( |
| null, |
| conf.getLongProperty("wayang.core.fallback.udf.ram.lower"), |
| conf.getLongProperty("wayang.core.fallback.udf.ram.upper"), |
| conf.getDoubleProperty("wayang.core.fallback.udf.ram.confidence"), |
| CardinalityEstimate.EMPTY_ESTIMATE |
| ) |
| ); |
| }, |
| configuration |
| ).withSlf4jWarning("Creating fallback load estimator for {}."); |
| |
| // Built-in option: let the ExecutionOperators provide the LoadProfileEstimator. |
| KeyValueProvider<ExecutionOperator, LoadProfileEstimator> builtInProvider = |
| new FunctionalKeyValueProvider<>( |
| fallbackProvider, |
| (operator, requestee) -> operator.createLoadProfileEstimator(requestee.getConfiguration()).orElse(null) |
| ); |
| |
| // Customizable layer: Users can override manually. |
| KeyValueProvider<ExecutionOperator, LoadProfileEstimator> overrideProvider = |
| new MapBasedKeyValueProvider<>(builtInProvider); |
| |
| configuration.setOperatorLoadProfileEstimatorProvider(overrideProvider); |
| } |
| { |
| // Safety net: provide a fallback selectivity. |
| KeyValueProvider<FunctionDescriptor, LoadProfileEstimator> fallbackProvider = |
| new FunctionalKeyValueProvider<FunctionDescriptor, LoadProfileEstimator>( |
| (operator, requestee) -> { |
| final Configuration conf = requestee.getConfiguration(); |
| return new NestableLoadProfileEstimator( |
| IntervalLoadEstimator.createIOLinearEstimator( |
| null, |
| conf.getLongProperty("wayang.core.fallback.operator.cpu.lower"), |
| conf.getLongProperty("wayang.core.fallback.operator.cpu.upper"), |
| conf.getDoubleProperty("wayang.core.fallback.operator.cpu.confidence"), |
| CardinalityEstimate.EMPTY_ESTIMATE |
| ), |
| IntervalLoadEstimator.createIOLinearEstimator( |
| null, |
| conf.getLongProperty("wayang.core.fallback.operator.ram.lower"), |
| conf.getLongProperty("wayang.core.fallback.operator.ram.upper"), |
| conf.getDoubleProperty("wayang.core.fallback.operator.ram.confidence"), |
| CardinalityEstimate.EMPTY_ESTIMATE |
| ) |
| ); |
| }, |
| configuration |
| ).withSlf4jWarning("Creating fallback load estimator for {}."); |
| |
| // Built-in layer: let the FunctionDescriptors provide the LoadProfileEstimators themselves. |
| KeyValueProvider<FunctionDescriptor, LoadProfileEstimator> builtInProvider = |
| new FunctionalKeyValueProvider<>( |
| fallbackProvider, |
| functionDescriptor -> functionDescriptor.getLoadProfileEstimator().orElse(null) |
| ); |
| |
| // Customizable layer: Users can override manually. |
| KeyValueProvider<FunctionDescriptor, LoadProfileEstimator> overrideProvider = |
| new MapBasedKeyValueProvider<>(builtInProvider); |
| |
| configuration.setFunctionLoadProfileEstimatorProvider(overrideProvider); |
| } |
| { |
| // Safety net: provide a fallback start up costs. |
| final KeyValueProvider<Platform, Long> builtinProvider = new FunctionalKeyValueProvider<>( |
| (platform, requestee) -> platform.getInitializeMillis(requestee.getConfiguration()), |
| configuration |
| ); |
| |
| // Override layer. |
| KeyValueProvider<Platform, Long> overrideProvider = new MapBasedKeyValueProvider<>(builtinProvider); |
| configuration.setPlatformStartUpTimeProvider(overrideProvider); |
| } |
| { |
| // Safety net: provide a fallback start up costs. |
| final KeyValueProvider<Platform, LoadProfileToTimeConverter> fallbackProvider = |
| new FunctionalKeyValueProvider<Platform, LoadProfileToTimeConverter>( |
| platform -> LoadProfileToTimeConverter.createDefault( |
| LoadToTimeConverter.createLinearCoverter(0.0000005), // 1 CPU with 2 GHz |
| LoadToTimeConverter.createLinearCoverter(0.00001), // 10 ms to read/write 1 MB |
| LoadToTimeConverter.createLinearCoverter(0.00001), // 10 ms to receive/send 1 MB |
| (cpuEstimate, diskEstimate, networkEstimate) -> cpuEstimate.plus(diskEstimate).plus(networkEstimate) |
| ), |
| configuration |
| ) |
| .withSlf4jWarning("Using fallback load-to-time converter for {}."); |
| final KeyValueProvider<Platform, LoadProfileToTimeConverter> defaultProvider = |
| new FunctionalKeyValueProvider<>( |
| fallbackProvider, |
| (platform, requestee) -> platform.createLoadProfileToTimeConverter( |
| requestee.getConfiguration() |
| ) |
| ); |
| final KeyValueProvider<Platform, LoadProfileToTimeConverter> overrideProvider = |
| new MapBasedKeyValueProvider<>(defaultProvider, false); |
| configuration.setLoadProfileToTimeConverterProvider(overrideProvider); |
| } |
| { |
| // Safety net: provide a fallback start up costs. |
| final KeyValueProvider<Platform, TimeToCostConverter> fallbackProvider = |
| new FunctionalKeyValueProvider<Platform, TimeToCostConverter>( |
| platform -> new TimeToCostConverter(0d, 1d), |
| configuration |
| ).withSlf4jWarning("Using fallback time-to-cost converter for {}."); |
| final KeyValueProvider<Platform, TimeToCostConverter> builtInProvider = |
| new FunctionalKeyValueProvider<>( |
| fallbackProvider, |
| (platform, requestee) -> platform.createTimeToCostConverter( |
| requestee.getConfiguration() |
| ) |
| ); |
| final KeyValueProvider<Platform, TimeToCostConverter> overrideProvider = |
| new MapBasedKeyValueProvider<>(builtInProvider, false); |
| configuration.setTimeToCostConverterProvider(overrideProvider); |
| } |
| { |
| configuration.setLoadProfileEstimatorCache(new MapBasedKeyValueProvider<>(configuration, true)); |
| } |
| } |
| |
| private static void bootstrapPruningProviders(Configuration configuration) { |
| { |
| // By default, load pruning from the wayang.core.optimizer.pruning.strategies property. |
| CollectionProvider<Class<PlanEnumerationPruningStrategy>> propertyBasedProvider = |
| new FunctionalCollectionProvider<>( |
| config -> { |
| final String strategyClassNames = config.getStringProperty("wayang.core.optimizer.pruning.strategies"); |
| if (strategyClassNames == null || strategyClassNames.isEmpty()) { |
| return Collections.emptySet(); |
| } |
| Collection<Class<PlanEnumerationPruningStrategy>> strategyClasses = new LinkedList<>(); |
| for (String strategyClassName : strategyClassNames.split(",")) { |
| try { |
| @SuppressWarnings("unchecked") |
| final Class<PlanEnumerationPruningStrategy> strategyClass = (Class<PlanEnumerationPruningStrategy>) Class.forName(strategyClassName); |
| strategyClasses.add(strategyClass); |
| } catch (ClassNotFoundException e) { |
| logger.warn("Illegal pruning strategy class name: \"{}\".", strategyClassName); |
| } |
| } |
| return strategyClasses; |
| }, |
| configuration |
| ); |
| CollectionProvider<Class<PlanEnumerationPruningStrategy>> overrideProvider = |
| new ExplicitCollectionProvider<>(configuration, propertyBasedProvider); |
| configuration.setPruningStrategyClassProvider(overrideProvider); |
| } |
| { |
| ValueProvider<ToDoubleFunction<ProbabilisticDoubleInterval>> defaultProvider = |
| new ConstantValueProvider<>(ProbabilisticDoubleInterval::getGeometricMeanEstimate, configuration); |
| ValueProvider<ToDoubleFunction<ProbabilisticDoubleInterval>> overrideProvider = |
| new ConstantValueProvider<>(defaultProvider); |
| configuration.setCostSquasherProvider(overrideProvider); |
| } |
| { |
| ValueProvider<InstrumentationStrategy> defaultProvider = |
| new ConstantValueProvider<>(new OutboundInstrumentationStrategy(), configuration); |
| ValueProvider<InstrumentationStrategy> configProvider = |
| new FunctionalValueProvider<>( |
| requestee -> { |
| Optional<String> optInstrumentationtStrategyClass = |
| requestee.getConfiguration().getOptionalStringProperty("wayang.core.optimizer.instrumentation"); |
| if (!optInstrumentationtStrategyClass.isPresent()) { |
| return null; |
| } |
| return instantiateDefault(optInstrumentationtStrategyClass.get()); |
| }, |
| defaultProvider |
| ); |
| ValueProvider<InstrumentationStrategy> overrideProvider = new ConstantValueProvider<>(configProvider); |
| configuration.setInstrumentationStrategyProvider(overrideProvider); |
| } |
| } |
| |
| private static void bootstrapProperties(Configuration configuration) { |
| // Here, we could put some default values. |
| final KeyValueProvider<String, String> defaultProperties = new MapBasedKeyValueProvider<>(configuration, false); |
| configuration.setProperties(defaultProperties); |
| configuration.load(ReflectionUtils.loadResource(DEFAULT_CONFIGURATION_FILE)); |
| |
| // Set some dynamic properties. |
| configuration.setProperty("wayang.core.log.cardinalities", StringUtils.join( |
| Arrays.asList(System.getProperty("user.home"), ".wayang", "cardinalities.json"), |
| File.separator |
| )); |
| configuration.setProperty("wayang.core.log.executions", StringUtils.join( |
| Arrays.asList(System.getProperty("user.home"), ".wayang", "executions.json"), |
| File.separator |
| )); |
| |
| // Supplement with a customizable layer. |
| final KeyValueProvider<String, String> customizableProperties = new MapBasedKeyValueProvider<>(defaultProperties); |
| configuration.setProperties(customizableProperties); |
| |
| |
| } |
| |
| /** |
| * Creates a child instance. |
| */ |
| public Configuration fork() { |
| return new Configuration(this); |
| } |
| |
| /** |
| * Creates a child instance. |
| * |
| * @param name for the child instance |
| */ |
| public Configuration fork(String name) { |
| final Configuration configuration = new Configuration(this); |
| configuration.name = name; |
| return configuration; |
| } |
| |
| |
| public KeyValueProvider<OutputSlot<?>, CardinalityEstimator> getCardinalityEstimatorProvider() { |
| return this.cardinalityEstimatorProvider; |
| } |
| |
| public void setCardinalityEstimatorProvider( |
| KeyValueProvider<OutputSlot<?>, CardinalityEstimator> cardinalityEstimatorProvider) { |
| this.cardinalityEstimatorProvider = cardinalityEstimatorProvider; |
| } |
| |
| public KeyValueProvider<FunctionDescriptor, ProbabilisticDoubleInterval> getUdfSelectivityProvider() { |
| return this.udfSelectivityProvider; |
| } |
| |
| public void setUdfSelectivityProvider( |
| KeyValueProvider<FunctionDescriptor, ProbabilisticDoubleInterval> udfSelectivityProvider) { |
| this.udfSelectivityProvider = udfSelectivityProvider; |
| } |
| |
| public KeyValueProvider<ExecutionOperator, LoadProfileEstimator> getOperatorLoadProfileEstimatorProvider() { |
| return this.operatorLoadProfileEstimatorProvider; |
| } |
| |
| public void setOperatorLoadProfileEstimatorProvider(KeyValueProvider<ExecutionOperator, LoadProfileEstimator> operatorLoadProfileEstimatorProvider) { |
| this.operatorLoadProfileEstimatorProvider = operatorLoadProfileEstimatorProvider; |
| } |
| |
| public KeyValueProvider<FunctionDescriptor, LoadProfileEstimator> getFunctionLoadProfileEstimatorProvider() { |
| return this.functionLoadProfileEstimatorProvider; |
| } |
| |
| public void setFunctionLoadProfileEstimatorProvider(KeyValueProvider<FunctionDescriptor, LoadProfileEstimator> functionLoadProfileEstimatorProvider) { |
| this.functionLoadProfileEstimatorProvider = functionLoadProfileEstimatorProvider; |
| } |
| |
| public MapBasedKeyValueProvider<String, LoadProfileEstimator> getLoadProfileEstimatorCache() { |
| return this.loadProfileEstimatorCache; |
| } |
| |
| public void setLoadProfileEstimatorCache(MapBasedKeyValueProvider<String, LoadProfileEstimator> loadProfileEstimatorCache) { |
| this.loadProfileEstimatorCache = loadProfileEstimatorCache; |
| } |
| |
| public ExplicitCollectionProvider<Platform> getPlatformProvider() { |
| return this.platformProvider; |
| } |
| |
| public void setPlatformProvider(ExplicitCollectionProvider<Platform> platformProvider) { |
| this.platformProvider = platformProvider; |
| } |
| |
| public ExplicitCollectionProvider<Mapping> getMappingProvider() { |
| return mappingProvider; |
| } |
| |
| public void setMappingProvider(ExplicitCollectionProvider<Mapping> mappingProvider) { |
| this.mappingProvider = mappingProvider; |
| } |
| |
| public ExplicitCollectionProvider<ChannelConversion> getChannelConversionProvider() { |
| return channelConversionProvider; |
| } |
| |
| public void setChannelConversionProvider(ExplicitCollectionProvider<ChannelConversion> channelConversionProvider) { |
| this.channelConversionProvider = channelConversionProvider; |
| } |
| |
| public CollectionProvider<Class<PlanEnumerationPruningStrategy>> getPruningStrategyClassProvider() { |
| return this.pruningStrategyClassProvider; |
| } |
| |
| |
| public void setPruningStrategyClassProvider(CollectionProvider<Class<PlanEnumerationPruningStrategy>> pruningStrategyClassProvider) { |
| this.pruningStrategyClassProvider = pruningStrategyClassProvider; |
| } |
| |
| public ValueProvider<InstrumentationStrategy> getInstrumentationStrategyProvider() { |
| return this.instrumentationStrategyProvider; |
| } |
| |
| public void setInstrumentationStrategyProvider(ValueProvider<InstrumentationStrategy> instrumentationStrategyProvider) { |
| this.instrumentationStrategyProvider = instrumentationStrategyProvider; |
| } |
| |
| public KeyValueProvider<Platform, Long> getPlatformStartUpTimeProvider() { |
| return this.platformStartUpTimeProvider; |
| } |
| |
| public void setPlatformStartUpTimeProvider(KeyValueProvider<Platform, Long> platformStartUpTimeProvider) { |
| this.platformStartUpTimeProvider = platformStartUpTimeProvider; |
| } |
| |
| public void setProperties(KeyValueProvider<String, String> properties) { |
| this.properties = properties; |
| } |
| |
| public KeyValueProvider<String, String> getProperties() { |
| return this.properties; |
| } |
| |
| public void setProperty(String key, String value) { |
| this.properties.set(key, value); |
| } |
| |
| public String getStringProperty(String key) { |
| return this.properties.provideFor(key); |
| } |
| |
| public Optional<String> getOptionalStringProperty(String key) { |
| return this.properties.optionallyProvideFor(key); |
| } |
| |
| public String getStringProperty(String key, String fallback) { |
| return this.getOptionalStringProperty(key).orElse(fallback); |
| } |
| |
| public KeyValueProvider<Platform, LoadProfileToTimeConverter> getLoadProfileToTimeConverterProvider() { |
| return this.loadProfileToTimeConverterProvider; |
| } |
| |
| public void setLoadProfileToTimeConverterProvider(KeyValueProvider<Platform, LoadProfileToTimeConverter> loadProfileToTimeConverterProvider) { |
| this.loadProfileToTimeConverterProvider = loadProfileToTimeConverterProvider; |
| } |
| |
| public KeyValueProvider<Platform, TimeToCostConverter> getTimeToCostConverterProvider() { |
| return timeToCostConverterProvider; |
| } |
| |
| public void setTimeToCostConverterProvider(KeyValueProvider<Platform, TimeToCostConverter> timeToCostConverterProvider) { |
| this.timeToCostConverterProvider = timeToCostConverterProvider; |
| } |
| |
| public ValueProvider<ToDoubleFunction<ProbabilisticDoubleInterval>> getCostSquasherProvider() { |
| return this.costSquasherProvider; |
| } |
| |
| public void setCostSquasherProvider(ValueProvider<ToDoubleFunction<ProbabilisticDoubleInterval>> costSquasherProvider) { |
| this.costSquasherProvider = costSquasherProvider; |
| } |
| |
| public OptionalLong getOptionalLongProperty(String key) { |
| final Optional<String> longValue = this.properties.optionallyProvideFor(key); |
| if (longValue.isPresent()) { |
| return OptionalLong.of(Long.valueOf(longValue.get())); |
| } else { |
| return OptionalLong.empty(); |
| } |
| } |
| |
| public long getLongProperty(String key) { |
| final OptionalLong optionalLongProperty = this.getOptionalLongProperty(key); |
| if (!optionalLongProperty.isPresent()) { |
| throw new WayangException(String.format("No value for \"%s\".", key)); |
| } |
| return optionalLongProperty.getAsLong(); |
| } |
| |
| public long getLongProperty(String key, long fallback) { |
| return this.getOptionalLongProperty(key).orElse(fallback); |
| } |
| |
| public OptionalDouble getOptionalDoubleProperty(String key) { |
| final Optional<String> optionalDouble = this.properties.optionallyProvideFor(key); |
| if (optionalDouble.isPresent()) { |
| return OptionalDouble.of(Double.valueOf(optionalDouble.get())); |
| } else { |
| return OptionalDouble.empty(); |
| } |
| } |
| |
| public double getDoubleProperty(String key) { |
| return this.getOptionalDoubleProperty(key).getAsDouble(); |
| } |
| |
| public double getDoubleProperty(String key, double fallback) { |
| return this.getOptionalDoubleProperty(key).orElse(fallback); |
| } |
| |
| public Optional<Boolean> getOptionalBooleanProperty(String key) { |
| return this.properties.optionallyProvideFor(key).map(Boolean::valueOf); |
| } |
| |
| public boolean getBooleanProperty(String key) { |
| return this.getOptionalBooleanProperty(key).get(); |
| } |
| |
| public boolean getBooleanProperty(String key, boolean fallback) { |
| return this.getOptionalBooleanProperty(key).orElse(fallback); |
| } |
| |
| public Configuration getParent() { |
| return parent; |
| } |
| |
| @Override |
| public String toString() { |
| return String.format("%s[%s]", this.getClass().getSimpleName(), this.name); |
| } |
| |
| public String getName() { |
| return this.name; |
| } |
| } |
