/*
 * 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.optimizer.enumeration;

import org.apache.commons.lang3.Validate;
import org.apache.wayang.core.api.Configuration;
import org.apache.wayang.core.optimizer.OptimizationContext;
import org.apache.wayang.core.optimizer.ProbabilisticDoubleInterval;
import org.apache.wayang.core.optimizer.costs.TimeEstimate;
import org.apache.wayang.core.optimizer.costs.TimeToCostConverter;
import org.apache.wayang.core.plan.executionplan.Channel;
import org.apache.wayang.core.plan.executionplan.ExecutionTask;
import org.apache.wayang.core.plan.wayangplan.ElementaryOperator;
import org.apache.wayang.core.plan.wayangplan.ExecutionOperator;
import org.apache.wayang.core.plan.wayangplan.InputSlot;
import org.apache.wayang.core.plan.wayangplan.LoopSubplan;
import org.apache.wayang.core.plan.wayangplan.Operator;
import org.apache.wayang.core.plan.wayangplan.OperatorAlternative;
import org.apache.wayang.core.plan.wayangplan.OutputSlot;
import org.apache.wayang.core.plan.wayangplan.WayangPlan;
import org.apache.wayang.core.plan.wayangplan.Slot;
import org.apache.wayang.core.platform.Junction;
import org.apache.wayang.core.platform.Platform;
import org.apache.wayang.core.util.Canonicalizer;
import org.apache.wayang.core.util.WayangCollections;
import org.apache.wayang.core.util.Tuple;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.ToDoubleFunction;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Represents a partial execution plan.
 */
public class PlanImplementation {

    private static final Logger logger = LogManager.getLogger(PlanImplementation.class);

    /**
     * {@link ExecutionOperator}s contained in this instance.
     */
    private final Canonicalizer<ExecutionOperator> operators;

    /**
     * Describes the {@link Channel}s that have been picked between {@link ExecutionOperator}s and how they are
     * implemented.
     */
    private final Map<OutputSlot<?>, Junction> junctions;

    /**
     * Defines how {@link LoopSubplan}s should be executed.
     */
    private final Map<LoopSubplan, LoopImplementation> loopImplementations = new HashMap<>();

    /**
     * An enumerated plan is mainly characterized by the {@link OperatorAlternative.Alternative}s that have
     * been picked so far. This member keeps track of them.
     */
    private final Map<OperatorAlternative, OperatorAlternative.Alternative> settledAlternatives =
            new HashMap<>();

    /**
     * The {@link PlanEnumeration} that hosts this instance. Can change over time.
     */
    // TODO: I think, we don't maintain this field properly. Also, its semantics blur inside of LoopImplementations. Can we remove it?
    private PlanEnumeration planEnumeration;

    /**
     * Keep track of the {@link Platform}s of our {@link #operators}.
     */
    private Set<Platform> platformCache;

    /**
     * {@link OptimizationContext} that provides estimates for the {@link #operators}.
     */
    private final OptimizationContext optimizationContext;

    /**
     * The squashed cost estimate to execute this instance. This will be used to select the best plan!
     */
    private double squashedCostEstimateCache = Double.NaN, squashedCostEstimateWithoutOverheadCache = Double.NaN;

    /**
     * The parallel cost estimate . This will store both calculated squashed cost and cost that will be used to select the best enumerated plan!
     */
    private Tuple<List<ProbabilisticDoubleInterval>, List<Double>> parallelCostEstimateCache = null;

    /**
     * This will be used to store the parallel cost of each operator.
     */
    private List<Tuple<Operator, Tuple<List<ProbabilisticDoubleInterval>, List<Double>>>> calculatedParallelOperatorCostCache = new ArrayList<>();

    /**
     * Create a new instance.
     */
    PlanImplementation(
            PlanEnumeration planEnumeration,
            Map<OutputSlot<?>, Junction> junctions,
            Collection<ExecutionOperator> operators,
            OptimizationContext optimizationContext) {
        this(planEnumeration, junctions, new Canonicalizer<>(operators), optimizationContext);
    }

    /**
     * Creates new instance.
     */
    PlanImplementation(PlanEnumeration planEnumeration,
                       Map<OutputSlot<?>, Junction> junctions,
                       OptimizationContext optimizationContext) {
        this(planEnumeration, junctions, new Canonicalizer<>(), optimizationContext);
    }

    /**
     * Copy constructor.
     */
    public PlanImplementation(PlanImplementation original) {
        this.planEnumeration = original.planEnumeration;
        this.junctions = new HashMap<>(original.junctions);
        this.operators = new Canonicalizer<>(original.getOperators());
        this.settledAlternatives.putAll(original.settledAlternatives);
        this.loopImplementations.putAll(original.loopImplementations);
        this.optimizationContext = original.optimizationContext;
    }

    /**
     * Base constructor.
     */
    private PlanImplementation(PlanEnumeration planEnumeration,
                               Map<OutputSlot<?>, Junction> junctions,
                               Canonicalizer<ExecutionOperator> operators,
                               OptimizationContext optimizationContext) {
        this.planEnumeration = planEnumeration;
        this.junctions = junctions;
        this.operators = operators;
        this.optimizationContext = optimizationContext;

        assert this.planEnumeration != null;
    }


    /**
     * @return the {@link PlanEnumeration} this instance belongs to
     */
    public PlanEnumeration getPlanEnumeration() {
        return this.planEnumeration;
    }

    public void setPlanEnumeration(PlanEnumeration planEnumeration) {
        this.planEnumeration = planEnumeration;
    }


    /**
     * Find the {@link InputSlot}s of already picked {@link ExecutionOperator}s that represent the given {@link InputSlot}.
     * <p>Note that we require that this instance either provides all or no {@link ExecutionOperator}s necessary to
     * implement the {@link InputSlot}.</p>
     *
     * @param someInput any {@link InputSlot} of the original {@link WayangPlan}
     * @return the representing {@link InputSlot}s or {@code null} if this instance has no {@link ExecutionOperator}
     * backing the given {@link InputSlot}
     */
    Collection<InputSlot<?>> findExecutionOperatorInputs(final InputSlot<?> someInput) {
        final Operator owner = someInput.getOwner();
        if (owner.isAlternative()) {
            final OperatorAlternative operatorAlternative = (OperatorAlternative) owner;
            final OperatorAlternative.Alternative alternative = this.settledAlternatives.get(operatorAlternative);
            if (alternative == null) return null;
            @SuppressWarnings("unchecked")
            final Collection<InputSlot<?>> innerInputs = (Collection<InputSlot<?>>) (Collection) alternative.followInput(someInput);
            boolean isWithNull = false;
            Collection<InputSlot<?>> result = null;
            for (InputSlot<?> innerInput : innerInputs) {
                final Collection<InputSlot<?>> resolvedInputs = this.findExecutionOperatorInputs(innerInput);
                if (isWithNull && resolvedInputs != null) {
                    throw new IllegalStateException(String.format("Disallowed that %s is required by two different alternatives.", someInput));
                }
                isWithNull |= resolvedInputs == null;
                if (result == null) {
                    result = resolvedInputs;
                } else {
                    assert resolvedInputs != null;
                    result.addAll(resolvedInputs);
                }
            }
            return result;

        } else if (owner.isLoopSubplan()) {
            final LoopSubplan loopSubplan = (LoopSubplan) owner;
            final LoopImplementation loopImplementation = this.getLoopImplementations().get(loopSubplan);
            if (loopImplementation == null) return null;

            // Enter the LoopSubplan.
            final Collection<InputSlot<?>> innerInputs = loopSubplan.followInputUnchecked(someInput);
            if (innerInputs.isEmpty()) return innerInputs;

            // Discern LoopHeadOperator InputSlots and loop body InputSlots.
            final List<LoopImplementation.IterationImplementation> iterationImpls = loopImplementation.getIterationImplementations();
            final Collection<InputSlot<?>> collector = new HashSet<>(innerInputs.size());
            for (InputSlot<?> innerInput : innerInputs) {
                if (innerInput.getOwner() == loopSubplan.getLoopHead()) {
                    final LoopImplementation.IterationImplementation initialIterationImpl = iterationImpls.get(0);
                    collector.addAll(
                            initialIterationImpl.getBodyImplementation().findExecutionOperatorInputs(innerInput)
                    );
                } else {
                    for (LoopImplementation.IterationImplementation iterationImpl : iterationImpls) {
                        collector.addAll(
                                iterationImpl.getBodyImplementation().findExecutionOperatorInputs(innerInput)
                        );
                    }
                }
            }
            return collector;

        } else {
            assert owner.isExecutionOperator();
            Collection<InputSlot<?>> result = new LinkedList<>();
            result.add(someInput);
            return result;
        }

    }

    /**
     * Find the {@link OutputSlot}s of already picked {@link ExecutionOperator}s that represent the given {@link OutputSlot}.
     *
     * @param someOutput any {@link OutputSlot} of the original {@link WayangPlan}
     * @return the representing {@link OutputSlot}s
     */
    Collection<OutputSlot<?>> findExecutionOperatorOutput(OutputSlot<?> someOutput) {
        return this.findExecutionOperatorOutputWithContext(someOutput).stream()
                .map(Tuple::getField0)
                .collect(Collectors.toList());
    }

    /**
     * Find the {@link OutputSlot}s of already picked {@link ExecutionOperator}s that represent the given {@link OutputSlot}.
     *
     * @param someOutput any {@link OutputSlot} of the original {@link WayangPlan}
     * @return the representing {@link OutputSlot}s together with their enclosing {@link PlanImplementation}
     */
    Collection<Tuple<OutputSlot<?>, PlanImplementation>> findExecutionOperatorOutputWithContext(
            OutputSlot<?> someOutput) {
        while (someOutput != null
                && someOutput.getOwner().isAlternative()) {
            final Operator owner = someOutput.getOwner();
            final OperatorAlternative operatorAlternative = (OperatorAlternative) owner;
            final OperatorAlternative.Alternative alternative = this.settledAlternatives.get(operatorAlternative);
            someOutput = alternative == null ? null : alternative.traceOutput(someOutput);
        }

        // If we did not find a terminal OutputSlot.
        if (someOutput == null) {
            return Collections.emptySet();
        }

        // Otherwise, discern LoopSubplans and ExecutionOperators.
        final Operator owner = someOutput.getOwner();
        if (owner.isLoopSubplan()) {
            final LoopSubplan loopSubplan = (LoopSubplan) owner;
            final LoopImplementation loopImplementation = this.getLoopImplementations().get(loopSubplan);
            if (loopImplementation == null) return Collections.emptyList();

            // Enter the LoopSubplan.
            final OutputSlot<?> innerOutput = loopSubplan.traceOutput(someOutput);
            if (innerOutput == null) return Collections.emptyList();
            assert innerOutput.getOwner().isLoopHead();

            // For all the iterations, return the potential OutputSlots.
            final List<LoopImplementation.IterationImplementation> iterationImpls =
                    loopImplementation.getIterationImplementations();
            final Set<Tuple<OutputSlot<?>, PlanImplementation>> collector = new HashSet<>(iterationImpls.size());
            for (LoopImplementation.IterationImplementation iterationImpl : iterationImpls) {
                final Collection<Tuple<OutputSlot<?>, PlanImplementation>> outputsWithContext =
                        iterationImpl.getBodyImplementation().findExecutionOperatorOutputWithContext(innerOutput);
                collector.addAll(outputsWithContext);
            }

            return collector;

        } else {
            assert owner.isExecutionOperator();
            return Collections.singleton(new Tuple<>(someOutput, this));
        }
    }


    /**
     * Creates a new instance that forms the concatenation of this instance with the {@code targetPlans} via the
     * {@code junction}.
     *
     * @param targetPlans              instances to connect to
     * @param junction                 connects this instance with the {@code targetPlans}
     * @param outputPlanImplementation nested instance of this instance that hosts the {@code junction}
     * @param concatenationEnumeration that will host the concatenated instance
     * @return the concatenated instance or {@code null} if the inputs are contradicting each other
     */
    PlanImplementation concatenate(List<PlanImplementation> targetPlans,
                                   Junction junction,
                                   PlanImplementation outputPlanImplementation,
                                   PlanEnumeration concatenationEnumeration) {

        final PlanImplementation concatenation = new PlanImplementation(
                concatenationEnumeration,
                new HashMap<>(this.junctions.size() + 1),
                new HashSet<>(this.settledAlternatives.size(), targetPlans.size() * 4), // ballpark figure
                this.optimizationContext
        );

        concatenation.operators.addAll(this.operators);
        concatenation.junctions.putAll(this.junctions);
        concatenation.settledAlternatives.putAll(this.settledAlternatives);

        // Find the appropriate PlanImplementation for the junction and copy the loop implementations.
        PlanImplementation junctionPlanImplementation;
        if (outputPlanImplementation == this) {
            // Special case: The junction resides inside the top-level PlanImplementation.
            concatenation.loopImplementations.putAll(this.loopImplementations);
            junctionPlanImplementation = concatenation;
        } else {
            // Exhaustively, yet focused, search for the PlanImplementation.
            junctionPlanImplementation = concatenation.copyLoopImplementations(
                    this,
                    outputPlanImplementation,
                    junction.getSourceOutput().getOwner().getLoopStack()
            );
        }
        junctionPlanImplementation.junctions.put(junction.getSourceOutput(), junction);

        for (PlanImplementation targetPlan : targetPlans) {
            // NB: Join semantics at this point weaved in.
            if (concatenation.isSettledAlternativesContradicting(targetPlan)) {
                return null;
            }
            concatenation.operators.addAll(targetPlan.operators);
            concatenation.loopImplementations.putAll(targetPlan.loopImplementations);
            concatenation.junctions.putAll(targetPlan.junctions);
            concatenation.settledAlternatives.putAll(targetPlan.settledAlternatives);
        }

        return concatenation;
    }

    /**
     * Find the a given nested {@link PlanImplementation} in a further {@link PlanImplementation} and copy it to
     * this instance.
     *
     * @param originalPlanImplementation the (top-level) {@link PlanImplementation} to copy from
     * @param targetPlanImplementation   the (nested) {@link PlanImplementation} that should be copied
     * @param loopStack                  of an {@link ExecutionOperator} inside of the {@code targetPlanImplementation}
     * @return the copied {@link PlanImplementation} inside of this instance
     */
    private PlanImplementation copyLoopImplementations(PlanImplementation originalPlanImplementation,
                                                       PlanImplementation targetPlanImplementation,
                                                       LinkedList<LoopSubplan> loopStack) {
        // Descend into the loopStack.
        assert !loopStack.isEmpty();
        final LoopSubplan visitedLoop = loopStack.pop();

        // Copy the LoopImplementations of the originalPlanImplementation.
        this.loopImplementations.putAll(originalPlanImplementation.getLoopImplementations());
        // This one will be altered, so make an instance copy.
        final LoopImplementation loopImplCopy =
                this.loopImplementations.compute(visitedLoop, (key, value) -> new LoopImplementation(value));

        // Find the original counterpart to loopImplCopy.
        final LoopImplementation originalLoopImpl = originalPlanImplementation.loopImplementations.get(visitedLoop);


        // Go over the iterations of the LoopImplementations in parallel to process their PlanImplementations.
        PlanImplementation targetPlanImplementationCopy = null;
        Iterator<LoopImplementation.IterationImplementation>
                originalIterator = originalLoopImpl.getIterationImplementations().iterator(),
                copyIterator = loopImplCopy.getIterationImplementations().iterator();
        while (originalIterator.hasNext()) {
            final LoopImplementation.IterationImplementation nextCopy = copyIterator.next();
            final LoopImplementation.IterationImplementation nextOriginal = originalIterator.next();
            // If we need to descend further, invoke a recursive call.
            if (!loopStack.isEmpty()) {
                targetPlanImplementationCopy = nextCopy.getBodyImplementation().copyLoopImplementations(
                        nextOriginal.getBodyImplementation(),
                        targetPlanImplementation,
                        loopStack);

                // Once, we have found the iteration that contains the targetPlanImplementation, we can stop.
                if (targetPlanImplementationCopy != null) break;
            } else {
                // If we cannot descend futher, we basically need to find the correct iteration only.
                if (nextOriginal.getBodyImplementation() == targetPlanImplementation) {
                    targetPlanImplementationCopy = nextCopy.getBodyImplementation();
                    break;
                }
            }
        }

        // Restore the loopStack.
        loopStack.push(visitedLoop);

        // Return the match.
        return targetPlanImplementationCopy;


    }

    private boolean isSettledAlternativesContradicting(PlanImplementation that) {
        for (Map.Entry<OperatorAlternative, OperatorAlternative.Alternative> entry : this.settledAlternatives.entrySet()) {
            final OperatorAlternative opAlt = entry.getKey();
            final OperatorAlternative.Alternative alternative = entry.getValue();
            final OperatorAlternative.Alternative thatAlternative = that.settledAlternatives.get(opAlt);
            if (thatAlternative != null && alternative != thatAlternative) {
                return true;
            }
        }
        for (Map.Entry<LoopSubplan, LoopImplementation> entry : this.loopImplementations.entrySet()) {
            final LoopSubplan loop = entry.getKey();
            final LoopImplementation thisLoopImplementation = entry.getValue();
            final LoopImplementation thatLoopImplementation = that.loopImplementations.get(loop);
            if (thatLoopImplementation == null) continue;
            if (thisLoopImplementation
                    .getSingleIterationImplementation()
                    .getBodyImplementation()
                    .isSettledAlternativesContradicting(
                            thatLoopImplementation
                                    .getIterationImplementations().get(0)
                                    .getBodyImplementation()
                    )) {
                return true;
            }
        }
        return false;
    }

    /**
     * Escapes the {@link OperatorAlternative} that contains this instance.
     *
     * @param alternative        contains this instance
     * @param newPlanEnumeration will host the new instance
     * @return
     */
    public PlanImplementation escape(OperatorAlternative.Alternative alternative, PlanEnumeration newPlanEnumeration) {
        final PlanImplementation escapedPlanImplementation = new PlanImplementation(
                newPlanEnumeration, this.junctions, this.operators, this.optimizationContext
        );
        escapedPlanImplementation.settledAlternatives.putAll(this.settledAlternatives);
        assert !escapedPlanImplementation.settledAlternatives.containsKey(alternative.getOperatorAlternative());
        escapedPlanImplementation.settledAlternatives.put(alternative.getOperatorAlternative(), alternative);
        escapedPlanImplementation.loopImplementations.putAll(this.getLoopImplementations());
        return escapedPlanImplementation;
    }

    public Canonicalizer<ExecutionOperator> getOperators() {
        return this.operators;
    }

    public Map<LoopSubplan, LoopImplementation> getLoopImplementations() {
        return this.loopImplementations;
    }

    /**
     * Adds a new {@link LoopImplementation} for a given {@link LoopSubplan}.
     *
     * @param loop               the {@link LoopSubplan}
     * @param loopImplementation the {@link LoopImplementation}
     */
    public void addLoopImplementation(LoopSubplan loop, LoopImplementation loopImplementation) {
        this.loopImplementations.put(loop, loopImplementation);
    }

    /**
     * @return those contained {@link ExecutionOperator}s that have a {@link Slot} that is yet to be connected
     * to a further {@link ExecutionOperator} in the further plan enumeration process
     */
    public Collection<ExecutionOperator> getInterfaceOperators() {
        Validate.notNull(this.getPlanEnumeration());
        final Set<OutputSlot> outputSlots = this.getPlanEnumeration().servingOutputSlots.stream()
                .map(Tuple::getField0)
                .distinct()
                .collect(Collectors.toSet());
        final Set<InputSlot<?>> inputSlots = this.getPlanEnumeration().requestedInputSlots;

        return this.operators.stream()
                .filter(operator ->
                        this.allOutermostInputSlots(operator).anyMatch(inputSlots::contains) ||
                                this.allOutermostOutputSlots(operator).anyMatch(outputSlots::contains))
                .collect(Collectors.toList());
    }

    private Stream<OutputSlot> allOutermostOutputSlots(Operator operator) {
        return Arrays.stream(operator.getAllOutputs())
                .flatMap(output -> operator.getOutermostOutputSlots(output).stream());
    }

    private Stream<InputSlot> allOutermostInputSlots(Operator operator) {
        return Arrays.stream(operator.getAllInputs())
                .map(operator::getOutermostInputSlot);
    }

    /**
     * Find the {@link ExecutionOperator} that do not depend on any other {@link ExecutionOperator} as input.
     *
     * @return the start {@link ElementaryOperator}s
     */
    public List<ExecutionOperator> getStartOperators() {
        return this.operators.stream()
                .filter(this::isStartOperator)
                .collect(Collectors.toList());
    }

    /**
     * Detects start {@link ExecutionOperator}s.
     * <p>A start {@link ExecutionOperator} has an {@link InputSlot} that is requested by the {@link #planEnumeration}.</p>
     */
    private boolean isStartOperator(ExecutionOperator executionOperator) {
        ForLoop:
        for (InputSlot<?> inputSlot : executionOperator.getOriginal().getAllInputs()) {
            while (inputSlot != null) {
                if (this.planEnumeration.requestedInputSlots.contains(inputSlot)) {
                    continue ForLoop;
                }
                inputSlot = inputSlot.getOwner().getOuterInputSlot(inputSlot);
            }
            return false;
        }
        return true;
    }

    /**
     * Find for a given {@link OperatorAlternative}, which {@link OperatorAlternative.Alternative} has been picked
     * by this instance
     *
     * @param operatorAlternative the {@link OperatorAlternative} in question
     * @return the {@link OperatorAlternative.Alternative} or {@code null} if none has been chosen in this instance
     */
    public OperatorAlternative.Alternative getChosenAlternative(OperatorAlternative operatorAlternative) {
        return this.settledAlternatives.get(operatorAlternative);
    }

    /**
     * Retrieves the {@link TimeEstimate} for this instance, including platform overhead.
     *
     * @return the {@link TimeEstimate}
     */
    public TimeEstimate getTimeEstimate() {
        return this.getTimeEstimate(true);
    }

    /**
     * Retrieves the {@link TimeEstimate} for this instance, including platform overhead.
     *
     * @param isIncludeOverhead whether to include any incurring global overhead
     * @return the {@link TimeEstimate}
     */
    public TimeEstimate getTimeEstimate(boolean isIncludeOverhead) {
        final TimeEstimate operatorTimeEstimate = this.operators.stream()
                .map(op -> this.optimizationContext.getOperatorContext(op).getTimeEstimate())
                .reduce(TimeEstimate.ZERO, TimeEstimate::plus);
        final TimeEstimate junctionTimeEstimate = this.optimizationContext.getDefaultOptimizationContexts().stream()
                .flatMap(optCtx -> this.junctions.values().stream().map(jct -> jct.getTimeEstimate(optCtx)))
                .reduce(TimeEstimate.ZERO, TimeEstimate::plus);
        final TimeEstimate loopTimeEstimate = this.loopImplementations.values().stream()
                .map(LoopImplementation::getTimeEstimate)
                .reduce(TimeEstimate.ZERO, TimeEstimate::plus);
        TimeEstimate timeEstimate = operatorTimeEstimate.plus(junctionTimeEstimate).plus(loopTimeEstimate);

        if (isIncludeOverhead) {
            final long platformInitializationTime = this.getUtilizedPlatforms().stream()
                    .map(platform -> this.optimizationContext.getConfiguration().getPlatformStartUpTimeProvider().provideFor(platform))
                    .reduce(0L, (a, b) -> a + b);
            timeEstimate = timeEstimate.plus(platformInitializationTime);
        }

        return timeEstimate;
    }

    /**
     * Retrieves the cost estimate for this instance including any overhead.
     *
     * @return the cost estimate
     */
    public ProbabilisticDoubleInterval getCostEstimate() {

        if (this.optimizationContext.getConfiguration().getBooleanProperty("wayang.core.optimizer.enumeration.parallel-tasks")) {
            return this.getParallelCostEstimate(true);
        } else {
            return this.getCostEstimate(true);
        }
    }

    /**
     * Retrieves the cost estimate for this instance.
     *
     * @param isIncludeOverhead whether to include global overhead in the {@link TimeEstimate} (to avoid repeating
     *                          overhead in nested instances)
     * @return the cost estimate
     */
    ProbabilisticDoubleInterval getCostEstimate(boolean isIncludeOverhead) {
        ProbabilisticDoubleInterval costEstimateWithoutOverheadCache, costEstimateCache;
        final ProbabilisticDoubleInterval operatorCosts = this.operators.stream()
                .map(op -> this.optimizationContext.getOperatorContext(op).getCostEstimate())
                .reduce(ProbabilisticDoubleInterval.zero, ProbabilisticDoubleInterval::plus);
        final ProbabilisticDoubleInterval junctionCosts = this.optimizationContext.getDefaultOptimizationContexts().stream()
                .flatMap(optCtx -> this.junctions.values().stream().map(jct -> jct.getCostEstimate(optCtx)))
                .reduce(ProbabilisticDoubleInterval.zero, ProbabilisticDoubleInterval::plus);
        final ProbabilisticDoubleInterval loopCosts = this.loopImplementations.values().stream()
                .map(LoopImplementation::getCostEstimate)
                .reduce(ProbabilisticDoubleInterval.zero, ProbabilisticDoubleInterval::plus);
        costEstimateWithoutOverheadCache = operatorCosts.plus(junctionCosts).plus(loopCosts);
        ProbabilisticDoubleInterval overheadCosts = this.getUtilizedPlatforms().stream()
                .map(platform -> {
                    Configuration configuraiton = this.optimizationContext.getConfiguration();
                    long startUpTime = configuraiton.getPlatformStartUpTimeProvider().provideFor(platform);
                    TimeToCostConverter timeToCostConverter = configuraiton.getTimeToCostConverterProvider().provideFor(platform);
                    return timeToCostConverter.convert(new TimeEstimate(startUpTime, startUpTime, 1d));
                })
                .reduce(ProbabilisticDoubleInterval.zero, ProbabilisticDoubleInterval::plus);
        costEstimateCache = costEstimateWithoutOverheadCache.plus(overheadCosts);
        return isIncludeOverhead ? costEstimateCache : costEstimateWithoutOverheadCache;
    }

    /**
     * Retrieves the cost estimate for this instance including any overhead.
     *
     * @return the cost estimate
     */
    public double getSquashedCostEstimate() {
        // Check if the parallel cost calculation is enabled in the configuration file
        if (this.optimizationContext.getConfiguration().getBooleanProperty("wayang.core.optimizer.enumeration.parallel-tasks")) {
            return this.getSquashedParallelCostEstimate(true);
        } else {
            return this.getSquashedCostEstimate(true);
        }
    }

    /**
     * Retrieves the cost estimate for this instance.
     *
     * @param isIncludeOverhead whether to include global overhead in the {@link TimeEstimate} (to avoid repeating
     *                          overhead in nested instances)
     * @return the squashed cost estimate
     */
    double getSquashedCostEstimate(boolean isIncludeOverhead) {
        assert Double.isNaN(this.squashedCostEstimateCache) == Double.isNaN(this.squashedCostEstimateWithoutOverheadCache);
        if (Double.isNaN(this.squashedCostEstimateCache)) {
            final double operatorCosts = this.operators.stream()
                    .mapToDouble(op -> this.optimizationContext.getOperatorContext(op).getSquashedCostEstimate())
                    .sum();
            final double junctionCosts = this.optimizationContext.getDefaultOptimizationContexts().stream()
                    .flatMapToDouble(optCtx -> this.junctions.values().stream().mapToDouble(jct -> jct.getSquashedCostEstimate(optCtx)))
                    .sum();
            final double loopCosts = this.loopImplementations.values().stream()
                    .mapToDouble(LoopImplementation::getSquashedCostEstimate)
                    .sum();
            this.squashedCostEstimateWithoutOverheadCache = operatorCosts + junctionCosts + loopCosts;
            double overheadCosts = this.getUtilizedPlatforms().stream()
                    .mapToDouble(platform -> {
                        Configuration configuration = this.optimizationContext.getConfiguration();

                        long startUpTime = configuration.getPlatformStartUpTimeProvider().provideFor(platform);

                        TimeToCostConverter timeToCostConverter = configuration.getTimeToCostConverterProvider().provideFor(platform);
                        ProbabilisticDoubleInterval costs = timeToCostConverter.convert(new TimeEstimate(startUpTime, startUpTime, 1d));

                        final ToDoubleFunction<ProbabilisticDoubleInterval> squasher = configuration.getCostSquasherProvider().provide();
                        return squasher.applyAsDouble(costs);
                    })
                    .sum();
            this.squashedCostEstimateCache = this.squashedCostEstimateWithoutOverheadCache + overheadCosts;
        }
        return isIncludeOverhead ? this.squashedCostEstimateCache : this.squashedCostEstimateWithoutOverheadCache;
    }


    /**
     * Retrieves the cost estimate of input {@link Operator} and input {@link Junction} and recurse if there is input Operators
     *
     * @param operator {@link Operator} that will be used to retreive the cost/squashed costs
     * @return list of probabilisticDoubleInterval where First element is the operator cost and second element is the junction cost; and
     * list of double retreived where First element is the operator squashed cost and second element is the junction squashed cost
     * <p>
     * PS: This function will start with the sink operator
     */


    private Tuple<List<ProbabilisticDoubleInterval>, List<Double>> getParallelOperatorJunctionAllCostEstimate(Operator operator) {

        Set<Operator> inputOperators = new HashSet<>();
        Set<Junction> inputJunction = new HashSet<>();

        List<ProbabilisticDoubleInterval> probalisticCost = new ArrayList<>();
        List<Double> squashedCost = new ArrayList<>();

        // check if the operator cost was already calculated and cached
        for (Tuple<Operator, Tuple<List<ProbabilisticDoubleInterval>, List<Double>>> t : calculatedParallelOperatorCostCache) {
            if (t.field0 == operator)
                return t.field1;
        }

        if (this.optimizationContext.getOperatorContext(operator) != null) {
            // Get input junctions
            this.junctions.values()
                    .forEach(j -> {
                        for (int itr = 0; itr < j.getNumTargets(); itr++) {
                            if (j.getTargetOperator(itr) == operator)
                                inputJunction.add(j);
                        }
                    });
            // Get input operators associated with input junctions
            inputJunction
                    .forEach((Junction j) -> {
                        inputOperators.add(j.getSourceOperator());
                    });

            if (inputOperators.size() == 0) {
                // If there is no input operator, only the cost of the current operator is returned
                probalisticCost.add(this.optimizationContext.getOperatorContext(operator).getCostEstimate());
                probalisticCost.add(new ProbabilisticDoubleInterval(0f, 0f, 0f));
                squashedCost.add(this.optimizationContext.getOperatorContext(operator).getSquashedCostEstimate());
                squashedCost.add(.0);
                Tuple<List<ProbabilisticDoubleInterval>, List<Double>> returnedCost = new Tuple(probalisticCost, squashedCost);
                this.calculatedParallelOperatorCostCache.add(new Tuple(operator, returnedCost));
                return returnedCost;
            } else if (inputOperators.size() == 1) {
                // If there is only one input operator the cost of the current operator plus the cost of the input operator is returned

                // Get the operator probalistic cost and put it as a first element in probalisticCost
                probalisticCost.add(this.optimizationContext.getOperatorContext(operator).getCostEstimate()
                        .plus(this.getParallelOperatorJunctionAllCostEstimate(inputOperators.iterator().next()).field0.get(0)));
                // Get the junction probalistic cost and put it as a second element in probalisticCost
                probalisticCost.add(inputJunction.iterator().next().getCostEstimate(this.optimizationContext.getDefaultOptimizationContexts().get(0))
                        .plus(this.getParallelOperatorJunctionAllCostEstimate(inputOperators.iterator().next()).field0.get(1)));
                // Get the operator squashed cost and put it as a first element in squashedCost
                squashedCost.add(this.optimizationContext.getOperatorContext(operator).getSquashedCostEstimate()
                        + this.getParallelOperatorJunctionAllCostEstimate(inputOperators.iterator().next()).field1.get(0));
                // Get the junction squashed cost and put it as a second element in squashedCost
                squashedCost.add(inputJunction.iterator().next().getSquashedCostEstimate(this.optimizationContext.getDefaultOptimizationContexts().get(0))
                        + this.getParallelOperatorJunctionAllCostEstimate(inputOperators.iterator().next()).field1.get(1));

                Tuple<List<ProbabilisticDoubleInterval>, List<Double>> returnedCost = new Tuple(probalisticCost, squashedCost);
                this.calculatedParallelOperatorCostCache.add(new Tuple(operator, returnedCost));
                return returnedCost;
            } else {
                // If multiple input operators, the cost returned is the max of input operators
                ProbabilisticDoubleInterval maxControlProbabilistic = new ProbabilisticDoubleInterval(0f, 0f, 0f);
                ProbabilisticDoubleInterval maxJunctionProbabilistic = new ProbabilisticDoubleInterval(0f, 0f, 0f);

                double maxControlSquash = 0;
                double maxJunctionSquash = 0;

                for (Iterator<Operator> op = inputOperators.iterator(); op.hasNext(); ) {
                    Tuple<List<ProbabilisticDoubleInterval>, List<Double>> val = this.getParallelOperatorJunctionAllCostEstimate(op.next());
                    List<ProbabilisticDoubleInterval> valProbalistic = val.field0;
                    List<Double> valSquash = val.field1;
                    // Take the max of the probalistic cost
                    if (valProbalistic.get(0).getAverageEstimate() + valProbalistic.get(1).getAverageEstimate() >
                            maxControlProbabilistic.getAverageEstimate() + maxJunctionProbabilistic.getAverageEstimate()) {
                        // Get the control probalistic cost
                        maxControlProbabilistic = valProbalistic.get(0);
                        // Get the junction probalistic cost
                        maxJunctionProbabilistic = valProbalistic.get(1);
                    }
                    // Take the cost of the squashed cost
                    if (valSquash.get(0) > maxControlSquash) {
                        maxControlSquash = valSquash.get(0);
                    }
                    if (valSquash.get(1) > maxJunctionSquash) {
                        maxJunctionSquash = valSquash.get(1);
                    }
                }
                // Get the operator probalistic cost and put it as a first element in probalisticCost
                probalisticCost.add(this.optimizationContext.getOperatorContext(operator).getCostEstimate().plus(maxControlProbabilistic));
                // Get the junction probalistic cost and put it as a second element in probalisticCost
                probalisticCost.add(inputJunction.iterator().next().getCostEstimate(this.optimizationContext.getDefaultOptimizationContexts().get(0))
                        .plus(maxJunctionProbabilistic));
                // Get the operator squashed cost and put it as a first element in squashedCost
                squashedCost.add(this.optimizationContext.getOperatorContext(operator).getSquashedCostEstimate()
                        + maxControlSquash);
                // Get the junction squashed cost and put it as a second element in squashedCost
                squashedCost.add(inputJunction.iterator().next().getSquashedCostEstimate(this.optimizationContext.getDefaultOptimizationContexts().get(0))
                        + maxJunctionSquash);

                Tuple<List<ProbabilisticDoubleInterval>, List<Double>> returnedCost = new Tuple(probalisticCost, squashedCost);
                this.calculatedParallelOperatorCostCache.add(new Tuple(operator, returnedCost));
                return returnedCost;
            }
        } else {
            // Handle the case of a control not defined in this.operators (exp: loop operators)
            double controlSquash = 0;
            double junctionSquash = 0;
            ProbabilisticDoubleInterval controlProbabilistic = new ProbabilisticDoubleInterval(0f, 0f, 0f);
            ProbabilisticDoubleInterval junctionProbabilistic = new ProbabilisticDoubleInterval(0f, 0f, 0f);

            probalisticCost.add(controlProbabilistic);
            probalisticCost.add(junctionProbabilistic);
            squashedCost.add(controlSquash);
            squashedCost.add(junctionSquash);

            return new Tuple<>(probalisticCost, squashedCost);
        }
    }

    /**
     * Retrieves the cost estimate for this instance taking into account parallel stage execution.
     *
     * @param isIncludeOverhead whether to include global overhead in the {@link TimeEstimate} (to avoid repeating
     *                          overhead in nested instances)
     * @return the cost estimate taking into account parallel stage execution
     */
    ProbabilisticDoubleInterval getParallelCostEstimate(boolean isIncludeOverhead) {
        ProbabilisticDoubleInterval parallelCostEstimateWithoutOverhead, parallelCostEstimate;

        if (this.parallelCostEstimateCache == null) {
            // It means that the squashed cost is not yet called, might be only one possible execution plan
            this.getSquashedParallelCostEstimate(true);
        }

        final ProbabilisticDoubleInterval loopCosts = this.loopImplementations.values().stream()
                .map(LoopImplementation::getCostEstimate)
                .reduce(ProbabilisticDoubleInterval.zero, ProbabilisticDoubleInterval::plus);
        parallelCostEstimateWithoutOverhead = this.parallelCostEstimateCache.field0.get(0).plus(this.parallelCostEstimateCache.field0.get(1)).plus(loopCosts);
        ProbabilisticDoubleInterval overheadCosts = this.getUtilizedPlatforms().stream()
                .map(platform -> {
                    Configuration configuration = this.optimizationContext.getConfiguration();
                    long startUpTime = configuration.getPlatformStartUpTimeProvider().provideFor(platform);
                    TimeToCostConverter timeToCostConverter = configuration.getTimeToCostConverterProvider().provideFor(platform);
                    return timeToCostConverter.convert(new TimeEstimate(startUpTime, startUpTime, 1d));
                })
                .reduce(ProbabilisticDoubleInterval.zero, ProbabilisticDoubleInterval::plus);
        parallelCostEstimate = parallelCostEstimateWithoutOverhead.plus(overheadCosts);
        return isIncludeOverhead ? parallelCostEstimate : parallelCostEstimateWithoutOverhead;
    }

    /**
     * Retrieves the cost estimate for this instance taking into account parallel stage execution.
     *
     * @param isIncludeOverhead whether to include global overhead in the {@link TimeEstimate} (to avoid repeating
     *                          overhead in nested instances)
     * @return the squashed cost estimate taking into account parallel stage execution
     */
    double getSquashedParallelCostEstimate(boolean isIncludeOverhead) {
        // Collect sink operators by Removing all operators that have an output
        Set<Operator> sinkOperators;
        sinkOperators = this.operators.stream()
                .filter(op -> op.getNumOutputs() == 0)
                .collect(Collectors.toSet());

        // Retrieve operator and junction cost with parallel stage consideration
        double parallelOperatorCosts = 0f;
        double parallelJunctionCosts = 0f;

        // Iterate through all sinks to find the expensive sink
        for (Operator op : sinkOperators) {
            Tuple<List<ProbabilisticDoubleInterval>, List<Double>> tempParallelCostEstimate = this.getParallelOperatorJunctionAllCostEstimate(op);
            List<Double> tempSquashedCost = tempParallelCostEstimate.field1;

            if (tempSquashedCost.get(0) + tempSquashedCost.get(1) > parallelOperatorCosts + parallelJunctionCosts) {
                parallelOperatorCosts = tempSquashedCost.get(0);
                parallelJunctionCosts = tempSquashedCost.get(1);
                this.parallelCostEstimateCache = tempParallelCostEstimate;
            }
        }
        final double loopCosts = this.loopImplementations.values().stream()
                .mapToDouble(LoopImplementation::getSquashedCostEstimate)
                .sum();
        final double parallelSquashedCostEstimateWithoutOverhead = parallelOperatorCosts + parallelJunctionCosts + loopCosts;
        double overheadCosts = this.getUtilizedPlatforms().stream()
                .mapToDouble(platform -> {
                    Configuration configuration = this.optimizationContext.getConfiguration();

                    long startUpTime = configuration.getPlatformStartUpTimeProvider().provideFor(platform);

                    TimeToCostConverter timeToCostConverter = configuration.getTimeToCostConverterProvider().provideFor(platform);
                    ProbabilisticDoubleInterval costs = timeToCostConverter.convert(new TimeEstimate(startUpTime, startUpTime, 1d));

                    final ToDoubleFunction<ProbabilisticDoubleInterval> squasher = configuration.getCostSquasherProvider().provide();
                    return squasher.applyAsDouble(costs);
                })
                .sum();
        final double parallelSquashedCostEstimate = parallelSquashedCostEstimateWithoutOverhead + overheadCosts;
        return isIncludeOverhead ? parallelSquashedCostEstimate : parallelSquashedCostEstimateWithoutOverhead;
    }


    public Junction getJunction(OutputSlot<?> output) {
        return this.junctions.get(output);
    }

    public void putJunction(OutputSlot<?> output, Junction junction) {
        final Junction oldValue = junction == null ?
                this.junctions.remove(output) :
                this.junctions.put(output, junction);
        if (oldValue != null) {
            logger.warn("Replaced {} with {}.", oldValue, junction);
        }
    }

    public OptimizationContext getOptimizationContext() {
        return this.optimizationContext;
    }

    /**
     * Merges the {@link OptimizationContext}s of the {@link Junction}s in this instance into its main
     * {@link OptimizationContext}/
     */
    public void mergeJunctionOptimizationContexts() {
        // Merge the top-level Junctions.
        for (Junction junction : this.junctions.values()) {
            junction.getOptimizationContexts().forEach(OptimizationContext::mergeToBase);
        }

        // Descend into loops.
        this.loopImplementations.values().stream()
                .flatMap(loopImplementation -> loopImplementation.getIterationImplementations().stream())
                .map(LoopImplementation.IterationImplementation::getBodyImplementation)
                .forEach(PlanImplementation::mergeJunctionOptimizationContexts);
    }

    public void logTimeEstimates() {
        if (!this.logger.isDebugEnabled()) return;

        this.logger.debug(">>> Regular operators");
        for (ExecutionOperator operator : this.operators) {
            this.logger.debug("Estimated execution time of {}: {}",
                    operator, this.optimizationContext.getOperatorContext(operator).getTimeEstimate()
            );
        }
        this.logger.debug(">>> Glue operators");
        for (Junction junction : junctions.values()) {
            for (ExecutionTask task : junction.getConversionTasks()) {
                final ExecutionOperator operator = task.getOperator();
                this.logger.debug("Estimated execution time of {}: {}",
                        operator, this.optimizationContext.getOperatorContext(operator).getTimeEstimate()
                );
            }
        }
        this.logger.debug(">>> Loops");
        for (LoopImplementation loopImplementation : this.loopImplementations.values()) {
            for (LoopImplementation.IterationImplementation iterationImplementation : loopImplementation.getIterationImplementations()) {
                iterationImplementation.getBodyImplementation().logTimeEstimates();
            }
        }
    }

    /**
     * Retrieve the {@link Platform}s that are utilized by this instance.
     *
     * @return the {@link Platform}s
     */
    public Set<Platform> getUtilizedPlatforms() {
        if (this.platformCache == null) {
            this.platformCache = this.streamOperators()
                    .map(ExecutionOperator::getPlatform)
                    .collect(Collectors.toSet());
        }
        return this.platformCache;
    }

    /**
     * Stream all the {@link ExecutionOperator}s in this instance.
     *
     * @return a {@link Stream} containing every {@link ExecutionOperator} at least once
     */
    Stream<ExecutionOperator> streamOperators() {
        Stream<ExecutionOperator> operatorStream = Stream.concat(
                this.operators.stream(),
                this.junctions.values().stream().flatMap(j -> j.getConversionTasks().stream()).map(ExecutionTask::getOperator)
        );
        if (!this.loopImplementations.isEmpty()) {
            operatorStream = Stream.concat(
                    operatorStream,
                    this.loopImplementations.values().stream().flatMap(LoopImplementation::streamOperators)
            );
        }
        return operatorStream;
    }

    @Override
    public String toString() {
        return String.format("PlanImplementation[%s, %s, costs=%s]",
                this.getUtilizedPlatforms(), this.getTimeEstimate(), this.getCostEstimate()
        );
    }

    /**
     * Creates a new {@link ConcatenationDescriptor} for this instance.
     *
     * @param output the relevant {@link OutputSlot} or {@code null}
     * @param inputs the relevant {@link InputSlot}s; components can be {@code null}
     * @return the {@link ConcatenationDescriptor}
     */
    ConcatenationDescriptor createConcatenationDescriptor(OutputSlot<?> output, List<InputSlot<?>> inputs) {
        return new ConcatenationDescriptor(output, inputs);
    }

    /**
     * Amends a {@link ConcatenationGroupDescriptor} by {@link PlanImplementation}-specific information.
     */
    class ConcatenationDescriptor {

        final ConcatenationGroupDescriptor groupDescriptor;

        final PlanImplementation execOutputPlanImplementation;

        /**
         * Creates a new instance.
         */
        ConcatenationDescriptor(OutputSlot<?> output, List<InputSlot<?>> inputs) {
            // Find the ExecutionOperator's corresponding OutputSlot along with the nested PlanImplementation.
            OutputSlot<?> execOutput = null;
            PlanImplementation execOutputPlanImplementation = null;
            if (output != null) {
                Collection<Tuple<OutputSlot<?>, PlanImplementation>> execOpOutputsWithContext =
                        PlanImplementation.this.findExecutionOperatorOutputWithContext(output);
                final Tuple<OutputSlot<?>, PlanImplementation> execOpOutputWithCtx =
                        WayangCollections.getSingleOrNull(execOpOutputsWithContext);
                assert execOpOutputsWithContext != null : String.format("No outputs found for %s.", output);
                execOutput = execOpOutputWithCtx.field0;
                execOutputPlanImplementation = execOpOutputWithCtx.field1;
            }

            // Find the ExecutionOperators' corresponding InputSlots.
            List<Set<InputSlot<?>>> execInputs = new ArrayList<>(inputs.size());
            for (InputSlot<?> input : inputs) {
                if (input == null) {
                    execInputs.add(null);
                } else {
                    execInputs.add(WayangCollections.asSet(PlanImplementation.this.findExecutionOperatorInputs(input)));
                }
            }

            this.groupDescriptor = new ConcatenationGroupDescriptor(execOutput, execInputs);
            this.execOutputPlanImplementation = execOutputPlanImplementation;
        }

        PlanImplementation getPlanImplementation() {
            return PlanImplementation.this;
        }

    }

    /**
     * Describes a group of {@link PlanImplementation}s in terms of their implementations for some {@link OutputSlot} and
     * {@link InputSlot}s. These {@link Slot}s are not stored in this class and must be clear from the context.
     */
    static class ConcatenationGroupDescriptor {

        /**
         * A corresponding {@link ExecutionOperator}s {@link OutputSlot} or {@code null}.
         */
        final OutputSlot<?> execOutput;

        /**
         * {@link Set}s of corresponding {@link ExecutionOperator}s' {@link InputSlot}s. Individual components can
         * be {@code null} if the {@link PlanImplementation}s do not implement the corresponding {@link InputSlot}.
         */
        final List<Set<InputSlot<?>>> execInputs;

        ConcatenationGroupDescriptor(OutputSlot<?> execOutput, List<Set<InputSlot<?>>> execInputs) {
            this.execOutput = execOutput;
            this.execInputs = execInputs;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            final ConcatenationGroupDescriptor that = (ConcatenationGroupDescriptor) o;
            return Objects.equals(execOutput, that.execOutput) &&
                    Objects.equals(execInputs, that.execInputs);
        }

        @Override
        public int hashCode() {
            return Objects.hash(execOutput, execInputs);
        }
    }
}