/*
 * 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;

import org.apache.wayang.core.api.Configuration;
import org.apache.wayang.core.api.Job;
import org.apache.wayang.core.api.exception.WayangException;
import org.apache.wayang.core.optimizer.cardinality.CardinalityEstimate;
import org.apache.wayang.core.optimizer.channels.ChannelConversionGraph;
import org.apache.wayang.core.optimizer.costs.EstimationContext;
import org.apache.wayang.core.optimizer.costs.LoadProfile;
import org.apache.wayang.core.optimizer.costs.LoadProfileEstimator;
import org.apache.wayang.core.optimizer.costs.LoadProfileEstimators;
import org.apache.wayang.core.optimizer.costs.LoadProfileToTimeConverter;
import org.apache.wayang.core.optimizer.costs.TimeEstimate;
import org.apache.wayang.core.optimizer.costs.TimeToCostConverter;
import org.apache.wayang.core.optimizer.enumeration.PlanEnumerationPruningStrategy;
import org.apache.wayang.core.plan.wayangplan.CompositeOperator;
import org.apache.wayang.core.plan.wayangplan.ExecutionOperator;
import org.apache.wayang.core.plan.wayangplan.InputSlot;
import org.apache.wayang.core.plan.wayangplan.LoopHeadOperator;
import org.apache.wayang.core.plan.wayangplan.LoopSubplan;
import org.apache.wayang.core.plan.wayangplan.Operator;
import org.apache.wayang.core.plan.wayangplan.OperatorContainer;
import org.apache.wayang.core.plan.wayangplan.OutputSlot;
import org.apache.wayang.core.plan.wayangplan.PlanTraversal;
import org.apache.wayang.core.plan.wayangplan.WayangPlan;
import org.apache.wayang.core.plan.wayangplan.Slot;
import org.apache.wayang.core.platform.Platform;
import org.apache.wayang.core.platform.lineage.ExecutionLineageNode;
import org.apache.wayang.core.util.ReflectionUtils;
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.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.ToDoubleFunction;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Manages contextual information required during the optimization of a {@link WayangPlan}.
 * <p>A single {@link Operator} can have multiple contexts in a {@link WayangPlan} - namely if it appears in a loop.
 * We manage these contexts in a hierarchical fashion.</p>
 */
public abstract class OptimizationContext {

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

    /**
     * The {@link Job} whose {@link WayangPlan} is to be optimized.
     */
    protected final Job job;

    /**
     * The instance in that this instance is nested - or {@code null} if it is top-level.
     */
    protected final LoopContext hostLoopContext;

    /**
     * The iteration number of this instance within its {@link #hostLoopContext} (starting from {@code 0}) - or {@code -1}
     * if there is no {@link #hostLoopContext}.
     */
    private int iterationNumber;

    /**
     * Forked {@link OptimizationContext}s can have a base.
     */
    private final OptimizationContext base;

    /**
     * {@link ChannelConversionGraph} used for the optimization.
     */
    private final ChannelConversionGraph channelConversionGraph;

    /**
     * {@link PlanEnumerationPruningStrategy}s to be used during optimization (in the given order).
     */
    private final List<PlanEnumerationPruningStrategy> pruningStrategies;

    /**
     * Create a new, plain instance.
     */
    public OptimizationContext(Job job) {
        this(
                job,
                null,
                null,
                -1,
                new ChannelConversionGraph(job.getConfiguration()),
                initializePruningStrategies(job.getConfiguration())
        );
    }

    /**
     * Creates a new instance. Useful for testing.
     *
     * @param operator the single {@link Operator} of this instance
     */
    public OptimizationContext(Job job, Operator operator) {
        this(
                job,
                null,
                null,
                -1,
                new ChannelConversionGraph(job.getConfiguration()),
                initializePruningStrategies(job.getConfiguration())
        );
        this.addOneTimeOperator(operator);
    }

    /**
     * Base constructor.
     */
    protected OptimizationContext(Job job,
                                  OptimizationContext base,
                                  LoopContext hostLoopContext,
                                  int iterationNumber,
                                  ChannelConversionGraph channelConversionGraph,
                                  List<PlanEnumerationPruningStrategy> pruningStrategies) {
        this.job = job;
        this.base = base;
        this.hostLoopContext = hostLoopContext;
        this.iterationNumber = iterationNumber;
        this.channelConversionGraph = channelConversionGraph;
        this.pruningStrategies = pruningStrategies;
    }

    /**
     * Initializes the {@link PlanEnumerationPruningStrategy}s from the {@link Configuration}.
     *
     * @param configuration defines the {@link PlanEnumerationPruningStrategy}s
     * @return a {@link List} of configured {@link PlanEnumerationPruningStrategy}s
     */
    private static List<PlanEnumerationPruningStrategy> initializePruningStrategies(Configuration configuration) {
        return configuration.getPruningStrategyClassProvider().provideAll().stream()
                .map(strategyClass -> OptimizationUtils.createPruningStrategy(strategyClass, configuration))
                .collect(Collectors.toList());
    }

    /**
     * Add {@link OperatorContext}s for the {@code operator} that is executed once within this instance. Also
     * add its encased {@link Operator}s.
     * Potentially invoke {@link #addOneTimeLoop(OperatorContext)} as well.
     */
    public abstract OperatorContext addOneTimeOperator(Operator operator);

    /**
     * Add {@link OperatorContext}s for all the contained {@link Operator}s of the {@code container}.
     */
    public void addOneTimeOperators(OperatorContainer container) {
        final CompositeOperator compositeOperator = container.toOperator();
        final Stream<Operator> innerOutputOperatorStream = compositeOperator.isSink() ?
                Stream.of(container.getSink()) :
                Arrays.stream(compositeOperator.getAllOutputs())
                        .map(container::traceOutput)
                        .filter(Objects::nonNull)
                        .map(Slot::getOwner);
        PlanTraversal.upstream()
                .withCallback(this::addOneTimeOperator)
                .traverse(innerOutputOperatorStream);
    }

    /**
     * Add {@link OptimizationContext}s for the {@code loop} that is executed once within this instance.
     */
    public abstract void addOneTimeLoop(OperatorContext operatorContext);

    /**
     * Return the {@link OperatorContext} of the {@code operator}.
     *
     * @param operator a one-time {@link Operator} (i.e., not in a nested loop)
     * @return the {@link OperatorContext} for the {@link Operator} or {@code null} if none
     */
    public abstract OperatorContext getOperatorContext(Operator operator);

    /**
     * Retrieve the {@link LoopContext} for the {@code loopSubplan}.
     */
    public abstract LoopContext getNestedLoopContext(LoopSubplan loopSubplan);

    /**
     * @return if this instance describes an iteration within a {@link LoopSubplan}, return the number of that iteration
     * (starting at {@code 0}); otherwise {@code -1}
     */
    public int getIterationNumber() {
        return this.iterationNumber;
    }

    /**
     * @return whether this instance is the first iteration within a {@link LoopContext}; this instance must be embedded
     * in a {@link LoopContext}
     */
    public boolean isInitialIteration() {
        assert this.hostLoopContext != null : "Not within a LoopContext.";
        return this.iterationNumber == 0;
    }


    /**
     * @return whether this instance is the final iteration within a {@link LoopContext}; this instance must be embedded
     * in a {@link LoopContext}
     */
    public boolean isFinalIteration() {
        assert this.hostLoopContext != null;
        return this.iterationNumber == this.hostLoopContext.getIterationContexts().size() - 1;
    }

    /**
     * Retrieve the {@link LoopContext} in which this instance resides.
     *
     * @return the {@link LoopContext} or {@code null} if none
     */
    public LoopContext getLoopContext() {
        return this.hostLoopContext;
    }

    /**
     * @return if this instance describes an iteration within a {@link LoopSubplan}, return the instance in which
     * this instance is nested
     */
    public OptimizationContext getParent() {
        return this.hostLoopContext == null ? null : this.hostLoopContext.getOptimizationContext();
    }

    public OptimizationContext getNextIterationContext() {
        assert this.hostLoopContext != null : String.format("%s is the last iteration.", this);
        assert !this.isFinalIteration();
        return this.hostLoopContext.getIterationContexts().get(this.iterationNumber + 1);
    }

    /**
     * Calls {@link OperatorContext#clearMarks()} for all nested {@link OperatorContext}s.
     */
    public abstract void clearMarks();

    public Configuration getConfiguration() {
        return this.job.getConfiguration();
    }

    /**
     * @return the {@link OperatorContext}s of this instance (exclusive of any base instance)
     */
    public abstract Map<Operator, OperatorContext> getLocalOperatorContexts();

    /**
     * @return whether there is a {@link TimeEstimate} for each {@link ExecutionOperator}
     */
    public abstract boolean isTimeEstimatesComplete();

    public ChannelConversionGraph getChannelConversionGraph() {
        return this.channelConversionGraph;
    }

    public OptimizationContext getBase() {
        return this.base;
    }

    public abstract void mergeToBase();

    public List<PlanEnumerationPruningStrategy> getPruningStrategies() {
        return this.pruningStrategies;
    }

    /**
     * Get the top-level parent containing this instance.
     *
     * @return the top-level parent, which can also be this instance
     */
    public OptimizationContext getRootParent() {
        OptimizationContext optimizationContext = this;
        while (true) {
            final OptimizationContext parent = optimizationContext.getParent();
            if (parent == null) return optimizationContext;
            optimizationContext = parent;
        }
    }

    /**
     * Get the {@link DefaultOptimizationContext}s represented by this instance.
     *
     * @return a {@link Collection} of said {@link DefaultOptimizationContext}s
     */
    public abstract List<DefaultOptimizationContext> getDefaultOptimizationContexts();

    /**
     * Provide the {@link Job} whose optimization is supported by this instance.
     *
     * @return the {@link Job}
     */
    public Job getJob() {
        return this.job;
    }

    /**
     * Stores a value into the {@link Job}-global cache.
     *
     * @param key   identifies the value
     * @param value the value
     * @return the value previously associated with the key or else {@code null}
     */
    public Object putIntoJobCache(String key, Object value) {
        return this.getJob().getCache().put(key, value);
    }

    /**
     * Queries the {@link Job}-global cache.
     *
     * @param key that is associated with the value to be retrieved
     * @return the value associated with the key or else {@code null}
     */
    public Object queryJobCache(String key) {
        return this.getJob().getCache().get(key);
    }

    /**
     * Queries the {@link Job}-global cache.
     *
     * @param key         that is associated with the value to be retrieved
     * @param resultClass the expected {@link Class} of the retrieved value
     * @return the value associated with the key or else {@code null}
     */
    public <T> T queryJobCache(String key, Class<T> resultClass) {
        final Object value = this.queryJobCache(key);
        try {
            return resultClass.cast(value);
        } catch (ClassCastException e) {
            throw new WayangException("Job-cache value cannot be casted as requested.", e);
        }
    }

    /**
     * Represents a single optimization context of an {@link Operator}. This can be thought of as a single, virtual
     * execution of the {@link Operator}.
     */
    public class OperatorContext implements EstimationContext {

        /**
         * The {@link Operator} that is being decorated with this instance.
         */
        private final Operator operator;

        /**
         * {@link CardinalityEstimate}s that align with the {@link #operator}s {@link InputSlot}s and
         * {@link OutputSlot}s.
         */
        private final CardinalityEstimate[] inputCardinalities, outputCardinalities;

        /**
         * Used to mark changed {@link #inputCardinalities} and {@link #outputCardinalities}.
         */
        private final boolean[] inputCardinalityMarkers, outputCardinalityMarkers;

        /**
         * {@link LoadProfile} of the {@link Operator}.
         */
        private LoadProfile loadProfile;

        /**
         * {@link TimeEstimate} for the {@link ExecutionOperator}.
         */
        protected TimeEstimate timeEstimate;

        /**
         * Cost estimate for the {@link ExecutionOperator}.
         */
        private ProbabilisticDoubleInterval costEstimate;

        /**
         * The squashed version of the {@link #costEstimate}.
         */
        private double squashedCostEstimate;

        /**
         * Reflects the number of executions of the {@link #operator}. This, e.g., relevant in {@link LoopContext}s.
         */
        private int numExecutions = 1;

        /**
         * In case this instance corresponds to an execution, this field keeps track of this execution.
         */
        private ExecutionLineageNode lineage;

        /**
         * Creates a new instance.
         */
        protected OperatorContext(Operator operator) {
            this.operator = operator;
            this.inputCardinalities = new CardinalityEstimate[this.operator.getNumInputs()];
            this.inputCardinalityMarkers = new boolean[this.inputCardinalities.length];
            this.outputCardinalities = new CardinalityEstimate[this.operator.getNumOutputs()];
            this.outputCardinalityMarkers = new boolean[this.outputCardinalities.length];
        }

        public Operator getOperator() {
            return this.operator;
        }

        public CardinalityEstimate getOutputCardinality(int outputIndex) {
            return this.outputCardinalities[outputIndex];
        }

        public CardinalityEstimate getInputCardinality(int inputIndex) {
            return this.inputCardinalities[inputIndex];
        }

        public boolean isInputMarked(int inputIndex) {
            return this.inputCardinalityMarkers[inputIndex];
        }

        public boolean isOutputMarked(int outputIndex) {
            return this.outputCardinalityMarkers[outputIndex];
        }

        /**
         * Resets the marks for all {@link InputSlot}s and {@link OutputSlot}s.
         */
        public void clearMarks() {
            Arrays.fill(this.inputCardinalityMarkers, false);
            Arrays.fill(this.outputCardinalityMarkers, false);
        }

        @Override
        public CardinalityEstimate[] getInputCardinalities() {
            return this.inputCardinalities;
        }

        @Override
        public CardinalityEstimate[] getOutputCardinalities() {
            return this.outputCardinalities;
        }

        /**
         * Sets the {@link CardinalityEstimate} for a certain {@link OutputSlot}. If the {@link CardinalityEstimate}
         * changes, the {@link OutputSlot} is marked.
         */
        public void setInputCardinality(int inputIndex, CardinalityEstimate cardinality) {
            this.inputCardinalityMarkers[inputIndex] |= !Objects.equals(this.inputCardinalities[inputIndex], cardinality);
            if (OptimizationContext.this.logger.isDebugEnabled() && this.inputCardinalityMarkers[inputIndex]) {
                OptimizationContext.this.logger.debug(
                        "Changing cardinality of {} from {} to {}.",
                        this.operator.getInput(inputIndex),
                        this.inputCardinalities[inputIndex],
                        cardinality
                );
            }
            this.inputCardinalities[inputIndex] = cardinality;

            // Invalidate dependent estimate caches.
            this.timeEstimate = null;
            this.costEstimate = null;
        }

        /**
         * Sets the {@link CardinalityEstimate} for a certain {@link InputSlot}. If the {@link CardinalityEstimate}
         * changes, the {@link InputSlot} is marked.
         */
        public void setOutputCardinality(int outputIndex, CardinalityEstimate cardinality) {
            this.outputCardinalityMarkers[outputIndex] |= !Objects.equals(this.outputCardinalities[outputIndex], cardinality);
            if (OptimizationContext.this.logger.isDebugEnabled() && this.outputCardinalityMarkers[outputIndex]) {
                OptimizationContext.this.logger.debug(
                        "Changing cardinality of {} from {} to {}.",
                        this.operator.getOutput(outputIndex),
                        this.outputCardinalities[outputIndex],
                        cardinality
                );
            }
            this.outputCardinalities[outputIndex] = cardinality;

            // Invalidate dependent estimate caches.
            this.timeEstimate = null;
            this.costEstimate = null;
        }

        /**
         * Push forward all marked {@link CardinalityEstimate}s within the same {@link OptimizationContext}.
         *
         * @see Operator#propagateOutputCardinality(int, OperatorContext)
         */
        public void pushCardinalitiesForward() {
            OptimizationContext targetContext = this.getOptimizationContext();
            for (int outputIndex = 0; outputIndex < this.outputCardinalities.length; outputIndex++) {
                this.pushCardinalityForward(outputIndex, targetContext);
            }
        }

        /**
         * Pushes the {@link CardinalityEstimate} corresponding to the {@code outputIndex} forward to the
         * {@code targetContext} if it is marked.
         */
        public void pushCardinalityForward(int outputIndex, OptimizationContext targetContext) {
            if (!this.outputCardinalityMarkers[outputIndex]) return;
            this.operator.propagateOutputCardinality(outputIndex, this, targetContext);
        }

        @Override
        public double getDoubleProperty(String propertyKey, double fallback) {
            try {
                return ReflectionUtils.toDouble(ReflectionUtils.getProperty(this.operator, propertyKey));
            } catch (Exception e) {
                logger.error("Could not retrieve property \"{}\" from {}.", propertyKey, this.operator, e);
                return fallback;
            }
        }

        @Override
        public Collection<String> getPropertyKeys() {
            return this.operator.getEstimationContextProperties();
        }

        /**
         * @return the {@link OptimizationContext} in which this instance resides
         */
        public OptimizationContext getOptimizationContext() {
            return OptimizationContext.this;
        }

        /**
         * Update the {@link LoadProfile} and {@link TimeEstimate} of this instance.
         */
        public void updateCostEstimate() {
            this.updateCostEstimate(this.getOptimizationContext().getConfiguration());
        }

        /**
         * Update the {@link LoadProfile} and {@link TimeEstimate} of this instance.
         *
         * @param configuration provides the necessary functions
         */
        private void updateCostEstimate(Configuration configuration) {
            if (!this.operator.isExecutionOperator()) return;

            // Estimate the LoadProfile.
            final LoadProfileEstimator loadProfileEstimator = this.getLoadProfileEstimator();
            try {
                this.loadProfile = LoadProfileEstimators.estimateLoadProfile(this, loadProfileEstimator);
            } catch (Exception e) {
                throw new WayangException(String.format("Load profile estimation for %s failed.", this.operator), e);
            }

            // Calculate the TimeEstimate.
            final ExecutionOperator executionOperator = (ExecutionOperator) this.operator;
            final Platform platform = executionOperator.getPlatform();
            final LoadProfileToTimeConverter timeConverter = configuration.getLoadProfileToTimeConverterProvider().provideFor(platform);
            this.timeEstimate = TimeEstimate.MINIMUM.plus(timeConverter.convert(this.loadProfile));
            if (OptimizationContext.this.logger.isDebugEnabled()) {
                OptimizationContext.this.logger.debug(
                        "Setting time estimate of {} to {}.", this.operator, this.timeEstimate
                );
            }

            // Calculate the cost estimate.
            final TimeToCostConverter timeToCostConverter = configuration.getTimeToCostConverterProvider().provideFor(platform);
            this.costEstimate = timeToCostConverter.convertWithoutFixCosts(this.timeEstimate);

            // Squash the cost estimate.
            final ToDoubleFunction<ProbabilisticDoubleInterval> costSquasher = configuration.getCostSquasherProvider().provide();
            this.squashedCostEstimate = costSquasher.applyAsDouble(this.costEstimate);
        }

        /**
         * Get the {@link LoadProfileEstimator} for the {@link ExecutionOperator} in this instance.
         *
         * @return the {@link LoadProfileEstimator}
         */
        public LoadProfileEstimator getLoadProfileEstimator() {
            if (!(this.operator instanceof ExecutionOperator)) {
                return null;
            }
            return this.getOptimizationContext().getConfiguration()
                    .getOperatorLoadProfileEstimatorProvider()
                    .provideFor((ExecutionOperator) this.operator);
        }

        /**
         * Merges {@code that} instance into this instance. The lineage is not merged, though.
         *
         * @param that the other instance
         * @return this instance
         */
        protected OperatorContext merge(OperatorContext that) {
            assert this.operator == that.operator;
            assert this.inputCardinalities.length == that.inputCardinalities.length;
            assert this.outputCardinalities.length == that.outputCardinalities.length;

            System.arraycopy(that.inputCardinalities, 0, this.inputCardinalities, 0, that.inputCardinalities.length);
            System.arraycopy(that.inputCardinalityMarkers, 0, this.inputCardinalityMarkers, 0, that.inputCardinalityMarkers.length);
            System.arraycopy(that.outputCardinalities, 0, this.outputCardinalities, 0, that.outputCardinalities.length);
            System.arraycopy(that.outputCardinalityMarkers, 0, this.outputCardinalityMarkers, 0, that.outputCardinalityMarkers.length);

            this.loadProfile = that.loadProfile;
            this.timeEstimate = that.timeEstimate;
            this.costEstimate = that.costEstimate;
            this.squashedCostEstimate = that.squashedCostEstimate;
            this.numExecutions = that.numExecutions;

            return this;
        }

        /**
         * Increases the {@link CardinalityEstimate}, {@link TimeEstimate}, and {@link ProbabilisticDoubleInterval cost estimate}
         * of this instance by those of the given instance.
         *
         * @param that the increment
         */
        public void increaseBy(OperatorContext that) {
            assert this.operator.equals(that.operator);
            this.addTo(this.inputCardinalities, that.inputCardinalities);
            this.addTo(this.inputCardinalityMarkers, that.inputCardinalityMarkers);
            this.addTo(this.outputCardinalities, that.outputCardinalities);
            this.addTo(this.outputCardinalityMarkers, that.outputCardinalityMarkers);
            if (that.costEstimate != null) {
                if (this.costEstimate == null) {
                    this.loadProfile = that.loadProfile;
                    this.timeEstimate = that.timeEstimate;
                    this.costEstimate = that.costEstimate;
                    this.squashedCostEstimate = that.squashedCostEstimate;
                } else {
                    this.loadProfile = this.loadProfile.plus(that.loadProfile);
                    this.timeEstimate = this.timeEstimate.plus(that.timeEstimate);
                    this.costEstimate = this.costEstimate.plus(that.costEstimate);
                    this.squashedCostEstimate += that.squashedCostEstimate;
                }
            }
            this.numExecutions += that.numExecutions;
        }

        /**
         * Adds up {@link CardinalityEstimate}s.
         *
         * @param aggregate the accumulator
         * @param delta     the increment
         */
        private void addTo(CardinalityEstimate[] aggregate, CardinalityEstimate[] delta) {
            assert aggregate.length == delta.length;
            for (int i = 0; i < aggregate.length; i++) {
                CardinalityEstimate aggregateCardinality = aggregate[i];
                CardinalityEstimate deltaCardinality = delta[i];
                if (aggregateCardinality == null) {
                    aggregate[i] = deltaCardinality;
                } else if (deltaCardinality != null) {
                    aggregate[i] = aggregateCardinality.plus(deltaCardinality);
                }
            }
        }

        private void addTo(boolean[] aggregate, boolean[] delta) {
            assert aggregate.length == delta.length;
            for (int i = 0; i < aggregate.length; i++) {
                aggregate[i] |= delta[i];
            }
        }

        public void setNumExecutions(int numExecutions) {
            this.numExecutions = numExecutions;
        }

        public int getNumExecutions() {
            return this.numExecutions;
        }

        public LoadProfile getLoadProfile() {
            if (this.loadProfile == null) {
                this.updateCostEstimate();
            }
            return this.loadProfile;
        }

        public TimeEstimate getTimeEstimate() {
            if (this.timeEstimate == null) {
                this.updateCostEstimate();
            }
            return this.timeEstimate;
        }

        /**
         * Get the estimated costs incurred by this instance (without fix costs).
         *
         * @return the cost estimate
         */
        public ProbabilisticDoubleInterval getCostEstimate() {
            if (this.costEstimate == null) {
                this.updateCostEstimate();
            }
            return this.costEstimate;
        }

        /**
         * Get the squashed estimated costs incurred by this instance (without fix costs).
         *
         * @return the squashed cost estimate
         */
        public double getSquashedCostEstimate() {
            if (this.costEstimate == null) {
                this.updateCostEstimate();
            }
            return this.squashedCostEstimate;
        }

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

        /**
         * Resets the estimates of this instance.
         */
        public void resetEstimates() {
            Arrays.fill(this.inputCardinalities, null);
            Arrays.fill(this.inputCardinalityMarkers, false);
            Arrays.fill(this.outputCardinalities, null);
            Arrays.fill(this.outputCardinalityMarkers, false);
            this.loadProfile = null;
            this.timeEstimate = null;
            this.costEstimate = null;
            this.squashedCostEstimate = 0d;
        }
    }

    /**
     * Maintains {@link OptimizationContext}s for the iterations of a {@link LoopSubplan}.
     */
    public class LoopContext {

        private final OperatorContext loopSubplanContext;

        private final List<OptimizationContext> iterationContexts;

        private AggregateOptimizationContext aggregateOptimizationContext;

        protected LoopContext(OperatorContext loopSubplanContext) {
            assert loopSubplanContext.getOptimizationContext() == OptimizationContext.this;
            assert loopSubplanContext.getOperator() instanceof LoopSubplan;

            this.loopSubplanContext = loopSubplanContext;

            LoopSubplan loop = (LoopSubplan) loopSubplanContext.getOperator();
            final int numIterationContexts = loop.getNumExpectedIterations() + 1;
            this.iterationContexts = new ArrayList<>(numIterationContexts);
            for (int iterationNumber = 0; iterationNumber < numIterationContexts; iterationNumber++) {
                this.iterationContexts.add(
                        new DefaultOptimizationContext(loop, this, iterationNumber)
                );
            }
        }

        public OperatorContext getLoopSubplanContext() {
            return this.loopSubplanContext;
        }

        /**
         * Retrieves the iteration {@link OptimizationContext}s.
         * <p>
         * <p>Note that for {@code n} iterations, there are
         * {@code n+1} {@link OptimizationContext}s because the {@link LoopHeadOperator} is triggered {@code n+1} times.
         * The first {@code n} represent the iterations, the final represents the final state of the loop, in which
         * only the {@link LoopHeadOperator} is run the last time.</p>
         *
         * @return the {@link OptimizationContext} for each iteration; order by execution order
         */
        public List<OptimizationContext> getIterationContexts() {
            return this.iterationContexts;
        }

        public OptimizationContext getIterationContext(int iteration) {
            return this.iterationContexts.get(iteration);
        }

        /**
         * @return the {@link OptimizationContext} in that the {@link LoopSubplan} resides
         */
        public OptimizationContext getOptimizationContext() {
            return this.getLoopSubplanContext().getOptimizationContext();
        }

        public OptimizationContext getInitialIterationContext() {
            return this.iterationContexts.get(0);
        }

        public OptimizationContext getFinalIterationContext() {
            return this.iterationContexts.get(this.iterationContexts.size() - 1);
        }

        /**
         * Add a new iteration {@link OptimizationContext} between second-to-last and final iteration.
         *
         * @return the added {@link OptimizationContext}
         */
        public OptimizationContext appendIterationContext() {
            // Shift the final iteration context by one.
            final OptimizationContext finalIterationContext = this.getFinalIterationContext();
            this.iterationContexts.add(finalIterationContext);
            finalIterationContext.iterationNumber++;

            // Copy the second-to-last iteration context.
            OptimizationContext oldSecondToLastIterationContext = this.getIterationContext(this.iterationContexts.size() - 3);
            OptimizationContext newSecondToLastIterationContext = ((DefaultOptimizationContext) oldSecondToLastIterationContext).copy();
            newSecondToLastIterationContext.iterationNumber++;

            // Insert and return the copied iteration context.
            this.iterationContexts.set(iterationContexts.size() - 2, newSecondToLastIterationContext);
            return newSecondToLastIterationContext;
        }

        public LoopSubplan getLoop() {
            return (LoopSubplan) this.loopSubplanContext.getOperator();
        }

        public AggregateOptimizationContext getAggregateContext() {
            if (this.aggregateOptimizationContext == null) {
                this.aggregateOptimizationContext = new AggregateOptimizationContext(this);
            }
            return this.aggregateOptimizationContext;
        }
    }


}
