/*
 * 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 java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.wayang.commons.util.profiledb.instrumentation.StopWatch;
import org.apache.wayang.commons.util.profiledb.model.Experiment;
import org.apache.wayang.commons.util.profiledb.model.measurement.TimeMeasurement;
import org.apache.wayang.core.api.exception.WayangException;
import org.apache.wayang.core.mapping.PlanTransformation;
import org.apache.wayang.core.monitor.DisabledMonitor;
import org.apache.wayang.core.monitor.FileMonitor;
import org.apache.wayang.core.monitor.Monitor;
import org.apache.wayang.core.optimizer.DefaultOptimizationContext;
import org.apache.wayang.core.optimizer.OptimizationContext;
import org.apache.wayang.core.optimizer.ProbabilisticDoubleInterval;
import org.apache.wayang.core.optimizer.cardinality.CardinalityEstimate;
import org.apache.wayang.core.optimizer.cardinality.CardinalityEstimatorManager;
import org.apache.wayang.core.optimizer.costs.TimeEstimate;
import org.apache.wayang.core.optimizer.costs.TimeToCostConverter;
import org.apache.wayang.core.optimizer.enumeration.ExecutionTaskFlow;
import org.apache.wayang.core.optimizer.enumeration.PlanEnumeration;
import org.apache.wayang.core.optimizer.enumeration.PlanEnumerator;
import org.apache.wayang.core.optimizer.enumeration.PlanImplementation;
import org.apache.wayang.core.optimizer.enumeration.StageAssignmentTraversal;
import org.apache.wayang.core.plan.executionplan.Channel;
import org.apache.wayang.core.plan.executionplan.ExecutionPlan;
import org.apache.wayang.core.plan.executionplan.ExecutionStage;
import org.apache.wayang.core.plan.executionplan.ExecutionTask;
import org.apache.wayang.core.plan.wayangplan.ExecutionOperator;
import org.apache.wayang.core.plan.wayangplan.Operator;
import org.apache.wayang.core.plan.wayangplan.OutputSlot;
import org.apache.wayang.core.plan.wayangplan.PlanMetrics;
import org.apache.wayang.core.plan.wayangplan.WayangPlan;
import org.apache.wayang.core.platform.AtomicExecutionGroup;
import org.apache.wayang.core.platform.Breakpoint;
import org.apache.wayang.core.platform.CardinalityBreakpoint;
import org.apache.wayang.core.platform.ConjunctiveBreakpoint;
import org.apache.wayang.core.platform.CrossPlatformExecutor;
import org.apache.wayang.core.platform.ExecutionState;
import org.apache.wayang.core.platform.FixBreakpoint;
import org.apache.wayang.core.platform.NoIterationBreakpoint;
import org.apache.wayang.core.platform.PartialExecution;
import org.apache.wayang.core.platform.Platform;
import org.apache.wayang.core.profiling.CardinalityRepository;
import org.apache.wayang.core.profiling.CostMeasurement;
import org.apache.wayang.core.profiling.ExecutionLog;
import org.apache.wayang.core.profiling.ExecutionPlanMeasurement;
import org.apache.wayang.core.profiling.InstrumentationStrategy;
import org.apache.wayang.core.profiling.PartialExecutionMeasurement;
import org.apache.wayang.core.util.Formats;
import org.apache.wayang.core.util.OneTimeExecutable;
import org.apache.wayang.core.util.ReflectionUtils;
import org.apache.wayang.core.util.WayangCollections;

/**
 * Describes a job that is to be executed using Wayang.
 */
public class Job extends OneTimeExecutable {

    private final Logger logger = LogManager.getLogger(this.getClass());

    /**
     * Guardian to avoid re-execution.
     */
    private final AtomicBoolean hasBeenExecuted = new AtomicBoolean(false);

    /**
     * References the {@link WayangContext} that spawned this instance.
     */
    private final WayangContext wayangContext;

    /**
     * {@link Job}-level {@link Configuration} based on the {@link WayangContext}-level configuration.
     */
    private final Configuration configuration;

    /**
     * The {@link WayangPlan} to be executed by this instance.
     */
    private final WayangPlan wayangPlan;

    /**
     * {@link OptimizationContext} for the {@link #wayangPlan}.
     */
    private DefaultOptimizationContext optimizationContext;

    /**
     * General purpose cache.
     */
    private Map<String, Object> cache = new HashMap<>();

    /**
     * Executes the optimized {@link ExecutionPlan}.
     */
    private CrossPlatformExecutor crossPlatformExecutor;

    /**
     * Manages the {@link CardinalityEstimate}s for the {@link #wayangPlan}.
     */
    private CardinalityEstimatorManager cardinalityEstimatorManager;

    /**
     * Collects metadata w.r.t. the processing of this instance.
     */
    private final Experiment experiment;

    /**
     * {@link StopWatch} to measure some key figures for the {@link #experiment}.
     */
    private final StopWatch stopWatch;

    /**
     * {@link TimeMeasurement}s for the optimization and the execution phases.
     */
    private final TimeMeasurement optimizationRound, executionRound;

    /**
     * Collects the {@link TimeEstimate}s of all (partially) executed {@link PlanImplementation}s.
     */
    private List<TimeEstimate> timeEstimates = new LinkedList<>();

    /**
     * Collects the cost estimates of all (partially) executed {@link PlanImplementation}s.
     */
    private List<ProbabilisticDoubleInterval> costEstimates = new LinkedList<>();

    /**
     * JAR files that are needed to execute the UDFs.
     */
    private final Set<String> udfJarPaths = new HashSet<>();

    private Monitor monitor;

    /**
     * Name for this instance.
     */
    private final String name;

    /**
     * <i>Currently not used.</i>
     */
    private final StageAssignmentTraversal.StageSplittingCriterion stageSplittingCriterion =
            (producerTask, channel, consumerTask) -> false;

    /**
     * The {@link PlanImplementation} that is being executed.
     */
    private PlanImplementation planImplementation;

    /**
     * Controls at which {@link CardinalityEstimate}s the execution should be interrupted.
     */
    private final CardinalityBreakpoint cardinalityBreakpoint;

    private final boolean isProactiveReoptimization;

    /**
     * Creates a new instance.
     *
     * @param name       name for this instance or {@code null} if a default name should be picked
     * @param experiment an {@link Experiment} for that profiling entries will be created
     * @param udfJars    paths to JAR files needed to run the UDFs (see {@link ReflectionUtils#getDeclaringJar(Class)})
     */
    Job(WayangContext wayangContext, String name, Monitor monitor, WayangPlan wayangPlan, Experiment experiment, String... udfJars) {
        this.wayangContext = wayangContext;
        this.name = name == null ? "Wayang app" : name;
        this.configuration = this.wayangContext.getConfiguration().fork(this.name);
        this.wayangPlan = wayangPlan;
        for (String udfJar : udfJars) {
            this.addUdfJar(udfJar);
        }

        // Prepare re-optimization.
        if (this.configuration.getBooleanProperty("wayang.core.optimizer.reoptimize")) {
            this.cardinalityBreakpoint = new CardinalityBreakpoint(this.configuration);
            this.isProactiveReoptimization =
                    this.configuration.getBooleanProperty("wayang.core.optimizer.reoptimize.proactive", false);
        } else {
            this.cardinalityBreakpoint = null;
            this.isProactiveReoptimization = false;
        }

        // Prepare instrumentation.
        this.experiment = experiment;
        this.stopWatch = new StopWatch(experiment);
        this.optimizationRound = this.stopWatch.getOrCreateRound("Optimization");
        this.executionRound = this.stopWatch.getOrCreateRound("Execution");

        // Configure job monitor.
        if (Monitor.isEnabled(this.configuration)) {
            this.monitor = monitor == null ? new FileMonitor() : monitor;
        } else {
            this.monitor = new DisabledMonitor();
        }
    }

    /**
     * Adds a {@code path} to a JAR that is required in one or more UDFs.
     *
     * @see ReflectionUtils#getDeclaringJar(Class)
     */
    public void addUdfJar(String path) {
        this.udfJarPaths.add(path);
    }

    /**
     * Run this instance. Must only be called once.
     *
     * @throws WayangException in case the execution fails for any reason
     */
    @Override
    public void execute() throws WayangException {
        try {
            super.execute();
        } catch (WayangException e) {
            throw e;
        } catch (Throwable t) {
            throw new WayangException("Job execution failed.", t);
        }
    }

    public ExecutionPlan buildInitialExecutionPlan() throws WayangException {
        this.prepareWayangPlan();
        this.estimateKeyFigures();

        // Get initial execution plan.
        ExecutionPlan executionPlan = this.createInitialExecutionPlan();
        return executionPlan;
    }

    // TODO: Move outside of Job class
    public void reportProgress(String opName, Integer progress) {
        HashMap<String, Integer> partialProgress = new HashMap<>();
        partialProgress.put(opName, progress);
        try {
            this.monitor.updateProgress(partialProgress);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    protected void doExecute() {
        // Make sure that each job is only executed once.
        if (this.hasBeenExecuted.getAndSet(true)) {
            throw new WayangException("Job has already been executed.");
        }

        try {

            // Prepare the #wayangPlan for the optimization.
            this.optimizationRound.start();
            this.prepareWayangPlan();

            // Estimate cardinalities and execution times for the #wayangPlan.
            this.estimateKeyFigures();

            // Get an execution plan.
            int executionId = 0;
            ExecutionPlan executionPlan = this.createInitialExecutionPlan();
            this.optimizationRound.stop();
            if (this.experiment != null) {
                this.experiment.addMeasurement(ExecutionPlanMeasurement.capture(
                        executionPlan,
                        String.format("execution-plan-%d", executionId)
                ));
            }

            // TODO: generate run ID. For now we fix this because we can't handle multiple jobs, neither in montoring nor execution.
            String runId = "1";
            try {
                monitor.initialize(this.configuration, runId, executionPlan.toJsonList());
            } catch (Exception e) {
                this.logger.warn("Failed to initialize monitor: {}", e);
            }


            // Take care of the execution.
            while (!this.execute(executionPlan, executionId)) {
                this.optimizationRound.start();
                if (this.postProcess(executionPlan, executionId)) {
                    executionId++;
                    if (this.experiment != null) {
                        this.experiment.addMeasurement(ExecutionPlanMeasurement.capture(
                                executionPlan,
                                String.format("execution-plan-%d", executionId)
                        ));
                    }
                }
                this.optimizationRound.stop();
            }

            this.stopWatch.start("Post-processing");
            if (this.configuration.getBooleanProperty("wayang.core.log.enabled")) {
                this.logExecution();
            }
        } catch (WayangException e) {
            throw e;
        } catch (Throwable t) {
            throw new WayangException("Job execution failed.", t);
        } finally {
            this.stopWatch.stopAll();
            this.stopWatch.start("Post-processing", "Release Resources");
            this.releaseResources();
            this.stopWatch.stop("Post-processing");
            this.logger.info("StopWatch results:\n{}", this.stopWatch.toPrettyString());
        }
    }

    /**
     * Prepares the {@link #wayangPlan}: prunes unused {@link Operator}s, isolates loops, and applies all available
     * {@link PlanTransformation}s.
     */
    private void prepareWayangPlan() {
        this.logger.info("Preparing plan...");

        // Prepare the WayangPlan for the optimization.
        this.optimizationRound.start("Prepare", "Prune&Isolate");
        this.wayangPlan.prepare();
        this.optimizationRound.stop("Prepare", "Prune&Isolate");

        // Apply the mappings to the plan to form a hyperplan.
        this.optimizationRound.start("Prepare", "Transformations");
        final Collection<PlanTransformation> transformations = this.gatherTransformations();
        this.wayangPlan.applyTransformations(transformations);
        this.optimizationRound.stop("Prepare", "Transformations");

        this.optimizationRound.start("Prepare", "Sanity");
        assert this.wayangPlan.isSane();
        this.optimizationRound.stop("Prepare", "Sanity");

        this.optimizationRound.stop("Prepare");
    }

    /**
     * Gather all available {@link PlanTransformation}s from the {@link #configuration}.
     */
    private Collection<PlanTransformation> gatherTransformations() {
        final Set<Platform> platforms = WayangCollections.asSet(this.configuration.getPlatformProvider().provideAll());
        return this.configuration.getMappingProvider().provideAll().stream()
                .flatMap(mapping -> mapping.getTransformations().stream())
                .filter(t -> t.getTargetPlatforms().isEmpty() || platforms.containsAll(t.getTargetPlatforms()))
                .collect(Collectors.toList());
    }


    /**
     * Go over the given {@link WayangPlan} and estimate the cardinalities of data being passed between its
     * {@link Operator}s and the execution profile and time of {@link ExecutionOperator}s.
     */
    private void estimateKeyFigures() {
        this.logger.info("Estimating cardinalities and execution load...");

        this.optimizationRound.start("Cardinality&Load Estimation");
        if (this.cardinalityEstimatorManager == null) {
            this.optimizationRound.start("Cardinality&Load Estimation", "Create OptimizationContext");
            this.optimizationContext = DefaultOptimizationContext.createFrom(this);
            this.optimizationRound.stop("Cardinality&Load Estimation", "Create OptimizationContext");

            this.optimizationRound.start("Cardinality&Load Estimation", "Create CardinalityEstimationManager");
            this.cardinalityEstimatorManager = new CardinalityEstimatorManager(
                    this.wayangPlan, this.optimizationContext, this.configuration);
            this.optimizationRound.stop("Cardinality&Load Estimation", "Create CardinalityEstimationManager");
        }

        this.optimizationRound.start("Cardinality&Load Estimation", "Push Estimation");
        this.cardinalityEstimatorManager.pushCardinalities();
        this.optimizationRound.stop("Cardinality&Load Estimation", "Push Estimation");

        this.optimizationRound.stop("Cardinality&Load Estimation");
    }


    /**
     * Determine a good/the best execution plan from a given {@link WayangPlan}.
     */
    private ExecutionPlan createInitialExecutionPlan() {
        this.logger.info("Enumerating execution plans...");

        this.optimizationRound.start("Create Initial Execution Plan");

        // Enumerate all possible plan.
        final PlanEnumerator planEnumerator = this.createPlanEnumerator();

        final TimeMeasurement enumerateMeasurment = this.optimizationRound.start("Create Initial Execution Plan", "Enumerate");
        planEnumerator.setTimeMeasurement(enumerateMeasurment);
        final PlanEnumeration comprehensiveEnumeration = planEnumerator.enumerate(true);
        planEnumerator.setTimeMeasurement(null);
        this.optimizationRound.stop("Create Initial Execution Plan", "Enumerate");

        final Collection<PlanImplementation> executionPlans = comprehensiveEnumeration.getPlanImplementations();
        this.logger.debug("Enumerated {} plans.", executionPlans.size());
        for (PlanImplementation planImplementation : executionPlans) {
            this.logger.debug("Plan with operators: {}", planImplementation.getOperators());
        }

        // Pick an execution plan.
        // Make sure that an execution plan can be created.
        this.optimizationRound.start("Create Initial Execution Plan", "Pick Best Plan");
        this.pickBestExecutionPlan(executionPlans, null, null, null);
        this.timeEstimates.add(planImplementation.getTimeEstimate());
        this.costEstimates.add(planImplementation.getCostEstimate());
        this.optimizationRound.stop("Create Initial Execution Plan", "Pick Best Plan");

        this.logger.info("Compiling execution plan...");
        this.optimizationRound.start("Create Initial Execution Plan", "Split Stages");
        final ExecutionTaskFlow executionTaskFlow = ExecutionTaskFlow.createFrom(this.planImplementation);
        final ExecutionPlan executionPlan = ExecutionPlan.createFrom(executionTaskFlow, this.stageSplittingCriterion);
        this.optimizationRound.stop("Create Initial Execution Plan", "Split Stages");

        this.planImplementation.mergeJunctionOptimizationContexts();

        this.planImplementation.logTimeEstimates();

        //assert executionPlan.isSane();


        this.optimizationRound.stop("Create Initial Execution Plan");
        return executionPlan;
    }


    private PlanImplementation pickBestExecutionPlan(Collection<PlanImplementation> executionPlans,
                                                     ExecutionPlan existingPlan,
                                                     Set<Channel> openChannels,
                                                     Set<ExecutionStage> executedStages) {

        final PlanImplementation bestPlanImplementation = executionPlans.stream()
                .reduce((p1, p2) -> {
                    final double t1 = p1.getSquashedCostEstimate();
                    final double t2 = p2.getSquashedCostEstimate();
                    return t1 < t2 ? p1 : p2;
                })
                .orElseThrow(() -> new WayangException("Could not find an execution plan."));
        this.logger.info("Picked {} as best plan.", bestPlanImplementation);
        return this.planImplementation = bestPlanImplementation;
    }

    /**
     * Go over the given {@link WayangPlan} and update the cardinalities of data being passed between its
     * {@link Operator}s using the given {@link ExecutionState}.
     *
     * @return whether any cardinalities have been updated
     */
    private boolean reestimateCardinalities(ExecutionState executionState) {
        return this.cardinalityEstimatorManager.pushCardinalityUpdates(executionState, this.planImplementation);
    }

    /**
     * Creates a new {@link PlanEnumerator} for the {@link #wayangPlan} and {@link #configuration}.
     */
    private PlanEnumerator createPlanEnumerator() {
        return this.createPlanEnumerator(null, null);
    }

    /**
     * Creates a new {@link PlanEnumerator} for the {@link #wayangPlan} and {@link #configuration}.
     */
    private PlanEnumerator createPlanEnumerator(ExecutionPlan existingPlan, Set<Channel> openChannels) {
        return existingPlan == null ?
                new PlanEnumerator(this.wayangPlan, this.optimizationContext) :
                new PlanEnumerator(this.wayangPlan, this.optimizationContext, existingPlan, openChannels);
    }

    /**
     * Start executing the given {@link ExecutionPlan} with all bells and whistles, such as instrumentation,
     * logging of the plan, and measuring the execution time.
     *
     * @param executionPlan that should be executed
     * @param executionId   an identifier for the current execution
     * @return whether the execution of the {@link ExecutionPlan} is completed
     */
    private boolean execute(ExecutionPlan executionPlan, int executionId) {
        final TimeMeasurement currentExecutionRound = this.executionRound.start(String.format("Execution %d", executionId));

        // Ensure existence of the #crossPlatformExecutor.
        if (this.crossPlatformExecutor == null) {
            final InstrumentationStrategy instrumentation = this.configuration.getInstrumentationStrategyProvider().provide();
            this.crossPlatformExecutor = new CrossPlatformExecutor(this, instrumentation);
        }

        if (this.configuration.getOptionalBooleanProperty("wayang.core.debug.skipexecution").orElse(false)) {
            return true;
        }
        if (this.configuration.getBooleanProperty("wayang.core.optimizer.reoptimize")) {
            this.setUpBreakpoint(executionPlan, currentExecutionRound);
        }

        // Log the current executionPlan.
        this.logStages(executionPlan);

        // Trigger the execution.
        currentExecutionRound.start("Execute");
        boolean isExecutionComplete = this.crossPlatformExecutor.executeUntilBreakpoint(
                executionPlan, this.optimizationContext
        );
        executionRound.stop();

        // Return.
        return isExecutionComplete;
    }

    /**
     * Sets up a {@link Breakpoint} for an {@link ExecutionPlan}.
     *
     * @param executionPlan for that the {@link Breakpoint} should be set
     * @param round         {@link TimeMeasurement} to be extended for any interesting time measurements
     */
    private void setUpBreakpoint(ExecutionPlan executionPlan, TimeMeasurement round) {

        // Set up appropriate Breakpoints.
        final TimeMeasurement breakpointRound = round.start("Configure Breakpoint");
        FixBreakpoint immediateBreakpoint = new FixBreakpoint();
        final Set<ExecutionStage> completedStages = this.crossPlatformExecutor.getCompletedStages();
        if (completedStages.isEmpty()) {
            executionPlan.getStartingStages().forEach(immediateBreakpoint::breakAfter);
        } else {
            completedStages.stream()
                    .flatMap(stage -> stage.getSuccessors().stream())
                    .filter(stage -> !completedStages.contains(stage))
                    .forEach(immediateBreakpoint::breakAfter);
        }
        this.crossPlatformExecutor.setBreakpoint(new ConjunctiveBreakpoint(
                immediateBreakpoint,
                this.cardinalityBreakpoint,
                new NoIterationBreakpoint() // Avoid re-optimization inside of loops.
        ));
        breakpointRound.stop();
    }

    private void logStages(ExecutionPlan executionPlan) {
        if (this.logger.isInfoEnabled()) {

            StringBuilder sb = new StringBuilder();
            Set<ExecutionStage> seenStages = new HashSet<>();
            Queue<ExecutionStage> stagedStages = new LinkedList<>(executionPlan.getStartingStages());
            ExecutionStage nextStage;
            while ((nextStage = stagedStages.poll()) != null) {
                sb.append(nextStage).append(":\n");
                nextStage.getPlanAsString(sb, "* ");
                nextStage.getSuccessors().stream()
                        .filter(seenStages::add)
                        .forEach(stagedStages::add);
            }

            this.logger.info("Current execution plan:\n{}", executionPlan.toExtensiveString());
        }
    }

    /**
     * Injects the cardinalities obtained from {@link Channel} instrumentation, potentially updates the {@link ExecutionPlan}
     * through re-optimization, and collects measured data.
     *
     * @return whether the {@link ExecutionPlan} has been re-optimized
     */
    private boolean postProcess(ExecutionPlan executionPlan, int executionId) {
        if (this.crossPlatformExecutor.isVetoingPlanChanges()) {
            this.logger.info("The cross-platform executor is currently not allowing re-optimization.");
            return false;
        }

        final TimeMeasurement round = this.optimizationRound.start(String.format("Post-processing %d", executionId));

        round.start("Reestimate Cardinalities&Time");
        boolean isCardinalitiesUpdated = this.reestimateCardinalities(this.crossPlatformExecutor);
        round.stop("Reestimate Cardinalities&Time");

        round.start("Update Execution Plan");
        if (isCardinalitiesUpdated) {
            this.logger.info("Re-optimizing execution plan.");
            this.updateExecutionPlan(executionPlan);
        } else {
            this.logger.info("Skipping re-optimization: no new insights on cardinalities.");
            this.timeEstimates.add(this.timeEstimates.get(this.timeEstimates.size() - 1));
            this.costEstimates.add(this.costEstimates.get(this.costEstimates.size() - 1));

        }
        round.stop("Update Execution Plan");

        round.stop();

        return true;
    }

    /**
     * Enumerate possible execution plans from the given {@link WayangPlan} and determine the (seemingly) best one.
     */
    private void updateExecutionPlan(ExecutionPlan executionPlan) {
        // Defines the plan that we want to use in the end.
        // Find and copy the open Channels.
        final Set<ExecutionStage> completedStages = this.crossPlatformExecutor.getCompletedStages();
        final Set<ExecutionTask> completedTasks = completedStages.stream()
                .flatMap(stage -> stage.getAllTasks().stream())
                .collect(Collectors.toSet());

        // Find Channels that have yet to be consumed by unexecuted ExecutionTasks and scrap unexecuted bits of the plan.
        final Set<Channel> openChannels = executionPlan.retain(completedStages);

        // Enumerate all possible plan.
        final PlanEnumerator planEnumerator = this.createPlanEnumerator(executionPlan, openChannels);
        final PlanEnumeration comprehensiveEnumeration = planEnumerator.enumerate(true);
        final Collection<PlanImplementation> executionPlans = comprehensiveEnumeration.getPlanImplementations();
        this.logger.debug("Enumerated {} plans.", executionPlans.size());
        for (PlanImplementation planImplementation : executionPlans) {
            this.logger.debug("Plan with operators: {}", planImplementation.getOperators());
        }

        // Pick an execution plan.
        // Make sure that an execution plan can be created.
        this.pickBestExecutionPlan(executionPlans, executionPlan, openChannels, completedStages);
        this.timeEstimates.add(this.planImplementation.getTimeEstimate());
        this.costEstimates.add(this.planImplementation.getCostEstimate());

        ExecutionTaskFlow executionTaskFlow = ExecutionTaskFlow.recreateFrom(
                planImplementation, executionPlan, openChannels, completedStages
        );
        final ExecutionPlan executionPlanExpansion = ExecutionPlan.createFrom(executionTaskFlow, this.stageSplittingCriterion);
        executionPlan.expand(executionPlanExpansion);

        this.planImplementation.mergeJunctionOptimizationContexts();

        assert executionPlan.isSane();
    }

    /**
     * Asks this instance to release its critical resources to avoid resource leaks and to enhance durability and
     * consistency of accessed resources.
     */
    private void releaseResources() {
        this.wayangContext.getCardinalityRepository().sleep();
        if (this.crossPlatformExecutor != null) this.crossPlatformExecutor.shutdown();
    }

    private void logExecution() {
        this.stopWatch.start("Post-processing", "Log measurements");

        // For the last time, update the cardinalities and store them.
        this.reestimateCardinalities(this.crossPlatformExecutor);
        final CardinalityRepository cardinalityRepository = this.wayangContext.getCardinalityRepository();
        cardinalityRepository.storeAll(this.crossPlatformExecutor, this.optimizationContext);

        // Execution times.
        final Collection<PartialExecution> partialExecutions = this.crossPlatformExecutor.getPartialExecutions();

        // Add the execution times to the experiment.
        int nextPartialExecutionMeasurementId = 0;
        for (PartialExecution partialExecution : partialExecutions) {
            if (this.logger.isDebugEnabled()) {
                for (AtomicExecutionGroup atomicExecutionGroup : partialExecution.getAtomicExecutionGroups()) {
                    if (!(atomicExecutionGroup.getEstimationContext() instanceof OptimizationContext.OperatorContext)) {
                        continue;
                    }
                    OptimizationContext.OperatorContext operatorContext =
                            (OptimizationContext.OperatorContext) atomicExecutionGroup.getEstimationContext();

                    for (CardinalityEstimate cardinality : operatorContext.getInputCardinalities()) {
                        if (cardinality != null && !cardinality.isExact()) {
                            this.logger.debug(
                                    "Inexact input cardinality estimate {} for {}.",
                                    cardinality, operatorContext.getOperator()
                            );
                        }
                    }
                    for (CardinalityEstimate cardinality : operatorContext.getOutputCardinalities()) {
                        if (cardinality != null && !cardinality.isExact()) {
                            this.logger.debug(
                                    "Inexact output cardinality estimate {} for {}.",
                                    cardinality, operatorContext.getOperator()
                            );
                        }
                    }
                }
            }
            String id = String.format("par-ex-%03d", nextPartialExecutionMeasurementId++);
            final PartialExecutionMeasurement measurement = new PartialExecutionMeasurement(id, partialExecution, this.configuration);
            this.experiment.addMeasurement(measurement);
        }

        // Feed the execution log.
        try (ExecutionLog executionLog = ExecutionLog.open(this.configuration)) {
            executionLog.storeAll(partialExecutions);
        } catch (Exception e) {
            this.logger.error("Storing partial executions failed.", e);
        }
        this.optimizationRound.stop("Post-processing", "Log measurements");

        // Log the execution time.
        long effectiveExecutionMillis = partialExecutions.stream()
                .map(PartialExecution::getMeasuredExecutionTime)
                .reduce(0L, (a, b) -> a + b);
        long measuredExecutionMillis = this.executionRound.getMillis();
        this.logger.info(
                "Accumulated execution time: {} (effective: {}, overhead: {})",
                Formats.formatDuration(measuredExecutionMillis, true),
                Formats.formatDuration(effectiveExecutionMillis, true),
                Formats.formatDuration(measuredExecutionMillis - effectiveExecutionMillis, true)
        );
        int i = 1;
        for (TimeEstimate timeEstimate : this.timeEstimates) {
            this.logger.info("Estimated execution time (plan {}): {}", i, timeEstimate);
            TimeMeasurement lowerEstimate = new TimeMeasurement(String.format("Estimate %d (lower)", i));
            lowerEstimate.setMillis(timeEstimate.getLowerEstimate());
            this.stopWatch.getExperiment().addMeasurement(lowerEstimate);
            TimeMeasurement upperEstimate = new TimeMeasurement(String.format("Estimate %d (upper)", i));
            upperEstimate.setMillis(timeEstimate.getUpperEstimate());
            this.stopWatch.getExperiment().addMeasurement(upperEstimate);
            i++;
        }

        // Log the cost settings.
        final Collection<Platform> consideredPlatforms = this.configuration.getPlatformProvider().provideAll();
        for (Platform consideredPlatform : consideredPlatforms) {
            final TimeToCostConverter timeToCostConverter = this.configuration
                    .getTimeToCostConverterProvider()
                    .provideFor(consideredPlatform);
            this.experiment.getSubject().addConfiguration(
                    String.format("Costs per ms (%s)", consideredPlatform.getName()),
                    timeToCostConverter.getCostsPerMillisecond()
            );
            this.experiment.getSubject().addConfiguration(
                    String.format("Fix costs (%s)", consideredPlatform.getName()),
                    timeToCostConverter.getFixCosts()
            );
        }


        // Log the execution costs.
        double fixCosts = partialExecutions.stream()
                .flatMap(partialExecution -> partialExecution.getInvolvedPlatforms().stream())
                .map(platform -> this.configuration.getTimeToCostConverterProvider().provideFor(platform).getFixCosts())
                .reduce(0d, (a, b) -> a + b);
        double effectiveLowerCosts = fixCosts + partialExecutions.stream()
                .map(PartialExecution::getMeasuredLowerCost)
                .reduce(0d, (a, b) -> a + b);
        double effectiveUpperCosts = fixCosts + partialExecutions.stream()
                .map(PartialExecution::getMeasuredUpperCost)
                .reduce(0d, (a, b) -> a + b);
        this.logger.info("Accumulated costs: {} .. {}",
                String.format("%,.2f", effectiveLowerCosts),
                String.format("%,.2f", effectiveUpperCosts)
        );
        this.experiment.addMeasurement(
                new CostMeasurement("Measured cost", effectiveLowerCosts, effectiveUpperCosts, 1d)
        );
        i = 1;
        for (ProbabilisticDoubleInterval costEstimate : this.costEstimates) {
            this.logger.info("Estimated costs (plan {}): {}", i, costEstimate);
            this.experiment.addMeasurement(new CostMeasurement(
                    String.format("Estimated costs (%d)", i),
                    costEstimate.getLowerEstimate(),
                    costEstimate.getUpperEstimate(),
                    costEstimate.getCorrectnessProbability()
            ));
            i++;
        }

        // Log some plan metrics.
        final PlanMetrics planMetrics = PlanMetrics.createFor(this.wayangPlan, "Plan Metrics");
        this.logger.info("Plan metrics: {} virtual operators, {} execution operators, {} alternatives, {} combinations",
                planMetrics.getNumVirtualOperators(),
                planMetrics.getNumExecutionOperators(),
                planMetrics.getNumAlternatives(),
                planMetrics.getNumCombinations()
        );
        this.experiment.addMeasurement(planMetrics);
    }

    /**
     * Whether a {@link Breakpoint} is requested for the given {@link OutputSlot}.
     *
     * @param output          the {@link OutputSlot}
     * @param operatorContext the {@link OptimizationContext.OperatorContext} for the {@link OutputSlot} owner
     * @return whether a {@link Breakpoint} is requested
     */
    public boolean isRequestBreakpointFor(OutputSlot<?> output, OptimizationContext.OperatorContext operatorContext) {
        return this.isProactiveReoptimization
                && output.getOwner().getInnermostLoop() == null
                && this.cardinalityBreakpoint != null
                && !this.cardinalityBreakpoint.approves(operatorContext.getOutputCardinality(output.getIndex())
        );
    }

    /**
     * Modify the {@link Configuration} to control the {@link Job} execution.
     */
    public Configuration getConfiguration() {
        return this.configuration;
    }

    public Set<String> getUdfJarPaths() {
        return this.udfJarPaths;
    }

    /**
     * Provide the {@link CrossPlatformExecutor} used during the execution of this instance.
     *
     * @return the {@link CrossPlatformExecutor} or {@code null} if there is none allocated
     */
    public CrossPlatformExecutor getCrossPlatformExecutor() {
        return this.crossPlatformExecutor;
    }

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

    /**
     * Retrieves the name of this instance.
     *
     * @return the name
     */
    public String getName() {
        return this.name;
    }

    @Override
    public String toString() {
        return String.format("%s[%s]", this.getClass().getSimpleName(), this.name);
    }

    /**
     * Provide the {@link Experiment} being recorded with the execution of this instance.
     *
     * @return the {@link Experiment}
     */
    public Experiment getExperiment() {
        return this.experiment;
    }

    /**
     * Provide the {@link WayangPlan} executed by this instance.
     *
     * @return the {@link WayangPlan}
     */
    public WayangPlan getWayangPlan() {
        return this.wayangPlan;
    }

    /**
     * Provide the {@link StopWatch} that is used to instrument the execution of this instance.
     *
     * @return the {@link StopWatch}
     */
    public StopWatch getStopWatch() {
        return this.stopWatch;
    }

    /**
     * Provides a general-purpose cache. Can be used to communicate job-global information.
     *
     * @return the cache
     */
    public Map<String, Object> getCache() {
        return this.cache;
    }
}