<commit_before>package org.qcri.rheem.core.optimizer.enumeration;

import de.hpi.isg.profiledb.store.model.TimeMeasurement;
import org.qcri.rheem.core.api.Configuration;
import org.qcri.rheem.core.api.exception.RheemException;
import org.qcri.rheem.core.optimizer.OptimizationContext;
import org.qcri.rheem.core.optimizer.OptimizationUtils;
import org.qcri.rheem.core.plan.executionplan.Channel;
import org.qcri.rheem.core.plan.executionplan.ExecutionPlan;
import org.qcri.rheem.core.plan.executionplan.ExecutionTask;
import org.qcri.rheem.core.plan.rheemplan.*;
import org.qcri.rheem.core.util.RheemCollections;
import org.qcri.rheem.core.util.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.ToDoubleFunction;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * The plan partitioner recursively dissects a {@link RheemPlan} into {@link PlanEnumeration}s and then assembles
 * them.
 */
public class PlanEnumerator {

    /**
     * Logger.
     */
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    /**
     * {@link EnumerationActivator}s that are activated and should be followed to create branches.
     */
    private final Queue<EnumerationActivator> activatedEnumerations = new LinkedList<>();

    /**
     * {@link ConcatenationActivator}s that are activated and should be executed.
     */
    private final Queue<ConcatenationActivator> activatedConcatenations = new PriorityQueue<>(
            Comparator.comparingDouble(activator0 -> activator0.priority)
    );

    /**
     * Determines how to calculate the priority of {@link ConcatenationActivator}s.
     */
    private final ToDoubleFunction<ConcatenationActivator> concatenationPriorityFunction;

    /**
     * TODO
     * When this instance enumerates an {@link OperatorAlternative.Alternative}, then this field helps to
     * create correct {@link PlanEnumeration}s by mapping the enumerates {@link Operator}s to the {@link OperatorAlternative}'s
     * slots.
     */
    private final OperatorAlternative.Alternative enumeratedAlternative;

    /**
     * Maintain {@link EnumerationActivator} for {@link Operator}s.
     */
    private final Map<Tuple<Operator, OptimizationContext>, EnumerationActivator> enumerationActivators = new HashMap<>();

    /**
     * Maintain {@link ConcatenationActivator}s for each {@link OutputSlot}.
     */
    private final Map<Tuple<OutputSlot<?>, OptimizationContext>, ConcatenationActivator> concatenationActivators = new HashMap<>();

    /**
     * This instance will put all completed {@link PlanEnumeration}s (which did not cause an activation) here.
     */
    private final Collection<PlanEnumeration> completedEnumerations = new LinkedList<>();

    /**
     * Once this instance has been executed (via {@link #run()}, the result will be stored in this field. Prior to that,
     * it is {@code null}.
     */
    private AtomicReference<PlanEnumeration> resultReference;

    /**
     * {@link OperatorAlternative}s that have been settled already and must be respected during enumeration.
     */
    private final Map<OperatorAlternative, OperatorAlternative.Alternative> presettledAlternatives;

    /**
     * {@link ExecutionTask}s that have already been executed.
     */
    private final Map<ExecutionOperator, ExecutionTask> executedTasks;

    /**
     * {@link Channel}s that are existing and must be reused when re-optimizing.
     */
    private final Map<OutputSlot<?>, Collection<Channel>> openChannels;

    /**
     * {@link OptimizationContext} that holds all relevant task data.
     */
    private final OptimizationContext optimizationContext;

    /**
     * Keeps track of the execution time.
     */
    private TimeMeasurement timeMeasurement;

    /**
     * Tells whether branches should be enumerated first.
     */
    private boolean isEnumeratingBranchesFirst;

    /**
     * Creates a new instance.
     *
     * @param rheemPlan a hyperplan that should be used for enumeration.
     */
    public PlanEnumerator(RheemPlan rheemPlan,
                          OptimizationContext optimizationContext) {
        this(rheemPlan.collectReachableTopLevelSources(),
                optimizationContext,
                null,
                Collections.emptyMap(),
                Collections.emptyMap(),
                Collections.emptyMap());
    }

    /**
     * Creates a new instance, thereby encorporating already executed parts of the {@code rheemPlan}.
     *
     * @param rheemPlan a hyperplan that should be used for enumeration.
     * @param baseplan  an {@link ExecutionPlan} that has been already executed (for re-optimization)
     */
    public PlanEnumerator(RheemPlan rheemPlan,
                          OptimizationContext optimizationContext,
                          ExecutionPlan baseplan,
                          Set<Channel> openChannels) {

        this(rheemPlan.collectReachableTopLevelSources(),
                optimizationContext,
                null,
                new HashMap<>(),
                new HashMap<>(),
                new HashMap<>());

        // Register all the tasks that have been executed already.
        final Set<ExecutionTask> executedTasks = baseplan.collectAllTasks();
        executedTasks.forEach(task -> this.executedTasks.put(task.getOperator(), task));

        // Find out which alternatives have been settled already.
        executedTasks.stream()
                .map(ExecutionTask::getOperator)
                .flatMap(this::streamPickedAlternatives)
                .forEach(alternative -> this.presettledAlternatives.put(alternative.toOperator(), alternative));

        // Index the existing Channels by their user-specified Operator's OutputSlot.
        // Note that we must always take the outermost OutputSlots because only those will be connected if the RheemPlan is sane.
        for (Channel openChannel : openChannels) {
            final OutputSlot<?> outputSlot = OptimizationUtils.findRheemPlanOutputSlotFor(openChannel);
            if (outputSlot != null) {
                for (OutputSlot<?> outerOutput : outputSlot.getOwner().getOutermostOutputSlots(outputSlot)) {
                    final Collection<Channel> channelSet = this.openChannels.computeIfAbsent(outerOutput, k -> new HashSet<>(1));
                    channelSet.add(openChannel);
                }
            } else {
                this.logger.error("Could not find the output slot for the open channel {}.", openChannel);
            }
        }

    }

    /**
     * Basic constructor that will always be called and initializes all fields.
     */
    private PlanEnumerator(Collection<Operator> startOperators,
                           OptimizationContext optimizationContext,
                           OperatorAlternative.Alternative enumeratedAlternative,
                           Map<OperatorAlternative, OperatorAlternative.Alternative> presettledAlternatives,
                           Map<ExecutionOperator, ExecutionTask> executedTasks,
                           Map<OutputSlot<?>, Collection<Channel>> openChannels) {

        this.optimizationContext = optimizationContext;
        this.enumeratedAlternative = enumeratedAlternative;
        this.presettledAlternatives = presettledAlternatives;
        this.executedTasks = executedTasks;
        this.openChannels = openChannels;


        // Set up start Operators.
        for (Operator startOperator : startOperators) {
            this.scheduleForEnumeration(startOperator, optimizationContext);
        }

        // Configure the enumeration.
        final Configuration configuration = this.optimizationContext.getConfiguration();
        this.isEnumeratingBranchesFirst = configuration.getBooleanProperty(
                "rheem.core.optimizer.enumeration.branchesfirst", true
        );

        // Configure the concatenations.
        final String priorityFunctionName = configuration.getStringProperty(
                "rheem.core.optimizer.enumeration.concatenationprio"
        );
        ToDoubleFunction<ConcatenationActivator> concatenationPriorityFunction;
        switch (priorityFunctionName) {
            case "slots":
                concatenationPriorityFunction = ConcatenationActivator::countNumOfOpenSlots;
                break;
            case "plans":
                concatenationPriorityFunction = ConcatenationActivator::estimateNumConcatenatedPlanImplementations;
                break;
            case "random":
                // Randomly generate a priority. However, avoid re-generate priorities, because that would increase
                // of a concatenation activator being processed, the longer it is in the queue (I guess).
                concatenationPriorityFunction = activator -> {
                    if (!Double.isNaN(activator.priority)) return activator.priority;
                    return Math.random();
                };
                break;
            case "none":
                concatenationPriorityFunction = activator -> 0d;
                break;
            default:
                throw new RheemException("Unknown concatenation priority function: " + priorityFunctionName);
        }

        boolean isInvertConcatenationPriorities = configuration.getBooleanProperty(
                "rheem.core.optimizer.enumeration.invertconcatenations", false
        );
        this.concatenationPriorityFunction = isInvertConcatenationPriorities ?
                activator -> -concatenationPriorityFunction.applyAsDouble(activator) :
                concatenationPriorityFunction;


    }

    private void scheduleForEnumeration(Operator operator, OptimizationContext optimizationContext) {
        final EnumerationActivator enumerationActivator = new EnumerationActivator(operator, optimizationContext);
        if (enumerationActivator.canBeActivated()) {
            this.activatedEnumerations.add(enumerationActivator);
        }
        this.enumerationActivators.put(enumerationActivator.getKey(), enumerationActivator);
    }


    /**
     * Produce the {@link PlanEnumeration} for the plan specified during the construction of this instance.
     *
     * @param isRequireResult whether the result is allowed to be {@code null}
     * @return the result {@link PlanEnumeration} or {@code null} if none such exists
     */
    public PlanEnumeration enumerate(boolean isRequireResult) {
        this.run();
        final PlanEnumeration comprehensiveEnumeration = this.resultReference.get();
        if (isRequireResult && comprehensiveEnumeration == null) {
            this.logger.error("No comprehensive PlanEnumeration.");
            this.logger.error("Pending enumerations: {}", this.enumerationActivators.values().stream()
                    .filter(activator -> !activator.wasExecuted())
                    .collect(Collectors.toList())
            );
            this.logger.error("Pending concatenations: {}", this.concatenationActivators.values().stream()
                    .filter(activator -> !activator.wasExecuted())
                    .collect(Collectors.toList())
            );
            throw new RheemException("Could not find a single execution plan.");
        }
        return comprehensiveEnumeration;
    }

    private Stream<OperatorAlternative.Alternative> streamPickedAlternatives(Operator operator) {
        OperatorContainer container = operator.getContainer();
        while (container != null && !(container instanceof OperatorAlternative.Alternative)) {
            container = container.toOperator().getContainer();
        }
        if (container == null) return Stream.empty();
        OperatorAlternative.Alternative alternative = (OperatorAlternative.Alternative) container;
        final OperatorAlternative operatorAlternative = alternative.toOperator();
        return Stream.concat(Stream.of(alternative), this.streamPickedAlternatives(operatorAlternative));
    }

    /**
     * Execute the enumeration. Needs only be run once and will be called by one of the result retrieval methods.
     */
    private synchronized void run() {
        if (this.resultReference == null) {
            while (!this.activatedEnumerations.isEmpty()) {
                // Try to enumerate branches.
                EnumerationActivator enumerationActivator;
                if ((enumerationActivator = this.activatedEnumerations.poll()) != null) {
                    if (this.isTopLevel()) {
                        this.logger.debug("Execute {}.", enumerationActivator);
                    }
                    this.enumerateBranchStartingFrom(enumerationActivator);
                }
            }

            ConcatenationActivator concatenationActivator;
            while ((concatenationActivator = this.activatedConcatenations.poll()) != null) {
                if (this.isTopLevel()) {
                    this.logger.debug("Execute {} (open inputs: {}).",
                            concatenationActivator,
                            concatenationActivator.getBaseEnumeration().getRequestedInputSlots()
                    );
                }
                this.concatenate(concatenationActivator);
            }

            this.constructResultEnumeration();
        }
    }

    /**
     * Enumerate plans from the branch that starts at the given node. The mode of operation is as follows:
     * <ol>
     * <li>Enumerate all {@link Operator}s forming the branch.</li>
     * <li>Create a new {@link PlanEnumeration} for the branch.</li>
     * <li>Join the branch {@link PlanEnumeration} with all existing input {@link PlanEnumeration}s.</li>
     * <li>Activate downstream {@link Operator}s for upcoming branch enumerations.</li>
     * </ol>
     *
     * @param enumerationActivator the activated {@link EnumerationActivator}
     */
    private void enumerateBranchStartingFrom(EnumerationActivator enumerationActivator) {
        assert !enumerationActivator.wasExecuted();
        enumerationActivator.markAsExecuted();

        // Start with the activated operator.
        Operator currentOperator = enumerationActivator.activatableOperator;
        List<Operator> branch = this.collectBranchOperatorsStartingFrom(currentOperator);
        if (branch == null) {
            return;
        }
        if (this.isTopLevel()) {
            this.logger.debug("Enumerating top-level {}.", branch);
        }

        // Go over the branch and create a PlanEnumeration for it.
        final OptimizationContext currentOptimizationCtx = enumerationActivator.getOptimizationContext();
        PlanEnumeration branchEnumeration = this.enumerateBranch(branch, currentOptimizationCtx);
        if (branchEnumeration == null) {
            return;
        }

        this.postProcess(branchEnumeration, currentOptimizationCtx);
    }

    /**
     * Determine the branch (straight of operators) that begins at the given {@link Operator}.
     *
     * @param startOperator starts the branch
     * @return the {@link Operator}s of the branch in their order of appearance or {@code null} if the branch is
     * already known to not yield any enumerations
     */
    private List<Operator> collectBranchOperatorsStartingFrom(Operator startOperator) {
        List<Operator> branch = new LinkedList<>();
        Operator currentOperator = startOperator;
        while (true) {
            boolean isEnumeratable = currentOperator.isExecutionOperator() ||
                    currentOperator.isAlternative() ||
                    currentOperator.isLoopSubplan();
            if (!isEnumeratable) {
                this.logger.trace("Cannot enumerate branch with {}.", currentOperator);
                return null;
            }
            branch.add(currentOperator);

            // Cut branches if requested.
            if (!this.isEnumeratingBranchesFirst) break;

            // Try to advance. This requires certain conditions, though.
            OutputSlot<?> followableOutput;
            if (currentOperator.isLoopHead()) {
                LoopHeadOperator loopHeadOperator = (LoopHeadOperator) currentOperator;
                if (loopHeadOperator.getLoopBodyOutputs().size() != 1) {
                    break;
                }
                followableOutput = RheemCollections.getSingle(loopHeadOperator.getLoopBodyOutputs());

            } else {
                if (currentOperator.getNumOutputs() != 1) {
                    break;
                }
                followableOutput = currentOperator.getOutput(0);
            }

            if (followableOutput.getOccupiedSlots().size() != 1) {
                this.logger.trace("Stopping branch, because operator does not feed exactly one operator.");
                break;
            }
            Operator nextOperator = currentOperator.getOutput(0).getOccupiedSlots().get(0).getOwner();
            if (nextOperator.getNumInputs() != 1) {
                this.logger.trace("Stopping branch, because next operator does not have exactly one input.");
                break;
            }

            if (nextOperator == startOperator) {
                break;
            }

            currentOperator = nextOperator;
        }
        this.logger.trace("Determined branch: {}.", currentOperator);

        return branch;
    }

    /**
     * Create a {@link PlanEnumeration} for the given {@code branch}.
     *
     * @param branch              {@link List} of {@link Operator}s of the branch; ordered downstream
     * @param optimizationContext in which the {@code branch} resides
     * @return a {@link PlanEnumeration} for the given {@code branch}
     */
    private PlanEnumeration enumerateBranch(List<Operator> branch, OptimizationContext optimizationContext) {
        PlanEnumeration branchEnumeration = null;
        Operator lastOperator = null;
        for (Operator operator : branch) {
            PlanEnumeration operatorEnumeration;
            if (operator.isAlternative()) {
                operatorEnumeration = this.enumerateAlternative((OperatorAlternative) operator, optimizationContext);
                if (operatorEnumeration == null || operatorEnumeration.getPlanImplementations().isEmpty()) {
                    this.logger.warn("No implementations enumerated for {}.", operator);
                    return null;
                }
            } else if (operator.isLoopSubplan()) {
                operatorEnumeration = this.enumerateLoop((LoopSubplan) operator, optimizationContext);
            } else {
                assert operator.isExecutionOperator();
                operatorEnumeration = PlanEnumeration.createSingleton((ExecutionOperator) operator, optimizationContext);

                // Check if the operator is filtered.
                OptimizationContext.OperatorContext operatorContext = optimizationContext.getOperatorContext(operator);
                if (operatorContext != null && ((ExecutionOperator) operator).isFiltered(operatorContext)) {
                    this.logger.info("Filtered {} with context {}.", operator, operatorContext);
                    operatorEnumeration.getPlanImplementations().clear();
                }
            }

            if (operatorEnumeration.getPlanImplementations().isEmpty()) {
                if (this.isTopLevel()) {
                    throw new RheemException(String.format("No implementations enumerated for %s.", operator));
                } else {
                    this.logger.warn("No implementations enumerated for {}.", operator);
                }
            }

            if (branchEnumeration == null) {
                branchEnumeration = operatorEnumeration;
            } else {
                final OutputSlot<?> output = lastOperator.getOutput(0);
                branchEnumeration = branchEnumeration.concatenate(
                        output,
                        this.openChannels.get(output),
                        Collections.singletonMap(operator.getInput(0), operatorEnumeration),
                        optimizationContext,
                        this.timeMeasurement);

                if (branchEnumeration.getPlanImplementations().isEmpty()) {
                    if (this.isTopLevel()) {
                        throw new RheemException(String.format("Could not concatenate %s to %s.", lastOperator, operator));<commit_msg>handle re-optimization in the operator filtering<commit_after>package org.qcri.rheem.core.optimizer.enumeration;

import de.hpi.isg.profiledb.store.model.TimeMeasurement;
import org.qcri.rheem.core.api.Configuration;
import org.qcri.rheem.core.api.exception.RheemException;
import org.qcri.rheem.core.optimizer.OptimizationContext;
import org.qcri.rheem.core.optimizer.OptimizationUtils;
import org.qcri.rheem.core.plan.executionplan.Channel;
import org.qcri.rheem.core.plan.executionplan.ExecutionPlan;
import org.qcri.rheem.core.plan.executionplan.ExecutionTask;
import org.qcri.rheem.core.plan.rheemplan.*;
import org.qcri.rheem.core.util.RheemCollections;
import org.qcri.rheem.core.util.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.ToDoubleFunction;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * The plan partitioner recursively dissects a {@link RheemPlan} into {@link PlanEnumeration}s and then assembles
 * them.
 */
public class PlanEnumerator {

    /**
     * Logger.
     */
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    /**
     * {@link EnumerationActivator}s that are activated and should be followed to create branches.
     */
    private final Queue<EnumerationActivator> activatedEnumerations = new LinkedList<>();

    /**
     * {@link ConcatenationActivator}s that are activated and should be executed.
     */
    private final Queue<ConcatenationActivator> activatedConcatenations = new PriorityQueue<>(
            Comparator.comparingDouble(activator0 -> activator0.priority)
    );

    /**
     * Determines how to calculate the priority of {@link ConcatenationActivator}s.
     */
    private final ToDoubleFunction<ConcatenationActivator> concatenationPriorityFunction;

    /**
     * TODO
     * When this instance enumerates an {@link OperatorAlternative.Alternative}, then this field helps to
     * create correct {@link PlanEnumeration}s by mapping the enumerates {@link Operator}s to the {@link OperatorAlternative}'s
     * slots.
     */
    private final OperatorAlternative.Alternative enumeratedAlternative;

    /**
     * Maintain {@link EnumerationActivator} for {@link Operator}s.
     */
    private final Map<Tuple<Operator, OptimizationContext>, EnumerationActivator> enumerationActivators = new HashMap<>();

    /**
     * Maintain {@link ConcatenationActivator}s for each {@link OutputSlot}.
     */
    private final Map<Tuple<OutputSlot<?>, OptimizationContext>, ConcatenationActivator> concatenationActivators = new HashMap<>();

    /**
     * This instance will put all completed {@link PlanEnumeration}s (which did not cause an activation) here.
     */
    private final Collection<PlanEnumeration> completedEnumerations = new LinkedList<>();

    /**
     * Once this instance has been executed (via {@link #run()}, the result will be stored in this field. Prior to that,
     * it is {@code null}.
     */
    private AtomicReference<PlanEnumeration> resultReference;

    /**
     * {@link OperatorAlternative}s that have been settled already and must be respected during enumeration.
     */
    private final Map<OperatorAlternative, OperatorAlternative.Alternative> presettledAlternatives;

    /**
     * {@link ExecutionTask}s that have already been executed.
     */
    private final Map<ExecutionOperator, ExecutionTask> executedTasks;

    /**
     * {@link Channel}s that are existing and must be reused when re-optimizing.
     */
    private final Map<OutputSlot<?>, Collection<Channel>> openChannels;

    /**
     * {@link OptimizationContext} that holds all relevant task data.
     */
    private final OptimizationContext optimizationContext;

    /**
     * Keeps track of the execution time.
     */
    private TimeMeasurement timeMeasurement;

    /**
     * Tells whether branches should be enumerated first.
     */
    private boolean isEnumeratingBranchesFirst;

    /**
     * Creates a new instance.
     *
     * @param rheemPlan a hyperplan that should be used for enumeration.
     */
    public PlanEnumerator(RheemPlan rheemPlan,
                          OptimizationContext optimizationContext) {
        this(rheemPlan.collectReachableTopLevelSources(),
                optimizationContext,
                null,
                Collections.emptyMap(),
                Collections.emptyMap(),
                Collections.emptyMap());
    }

    /**
     * Creates a new instance, thereby encorporating already executed parts of the {@code rheemPlan}.
     *
     * @param rheemPlan a hyperplan that should be used for enumeration.
     * @param baseplan  an {@link ExecutionPlan} that has been already executed (for re-optimization)
     */
    public PlanEnumerator(RheemPlan rheemPlan,
                          OptimizationContext optimizationContext,
                          ExecutionPlan baseplan,
                          Set<Channel> openChannels) {

        this(rheemPlan.collectReachableTopLevelSources(),
                optimizationContext,
                null,
                new HashMap<>(),
                new HashMap<>(),
                new HashMap<>());

        // Register all the tasks that have been executed already.
        final Set<ExecutionTask> executedTasks = baseplan.collectAllTasks();
        executedTasks.forEach(task -> this.executedTasks.put(task.getOperator(), task));

        // Find out which alternatives have been settled already.
        executedTasks.stream()
                .map(ExecutionTask::getOperator)
                .flatMap(this::streamPickedAlternatives)
                .forEach(alternative -> this.presettledAlternatives.put(alternative.toOperator(), alternative));

        // Index the existing Channels by their user-specified Operator's OutputSlot.
        // Note that we must always take the outermost OutputSlots because only those will be connected if the RheemPlan is sane.
        for (Channel openChannel : openChannels) {
            final OutputSlot<?> outputSlot = OptimizationUtils.findRheemPlanOutputSlotFor(openChannel);
            if (outputSlot != null) {
                for (OutputSlot<?> outerOutput : outputSlot.getOwner().getOutermostOutputSlots(outputSlot)) {
                    final Collection<Channel> channelSet = this.openChannels.computeIfAbsent(outerOutput, k -> new HashSet<>(1));
                    channelSet.add(openChannel);
                }
            } else {
                this.logger.error("Could not find the output slot for the open channel {}.", openChannel);
            }
        }

    }

    /**
     * Basic constructor that will always be called and initializes all fields.
     */
    private PlanEnumerator(Collection<Operator> startOperators,
                           OptimizationContext optimizationContext,
                           OperatorAlternative.Alternative enumeratedAlternative,
                           Map<OperatorAlternative, OperatorAlternative.Alternative> presettledAlternatives,
                           Map<ExecutionOperator, ExecutionTask> executedTasks,
                           Map<OutputSlot<?>, Collection<Channel>> openChannels) {

        this.optimizationContext = optimizationContext;
        this.enumeratedAlternative = enumeratedAlternative;
        this.presettledAlternatives = presettledAlternatives;
        this.executedTasks = executedTasks;
        this.openChannels = openChannels;


        // Set up start Operators.
        for (Operator startOperator : startOperators) {
            this.scheduleForEnumeration(startOperator, optimizationContext);
        }

        // Configure the enumeration.
        final Configuration configuration = this.optimizationContext.getConfiguration();
        this.isEnumeratingBranchesFirst = configuration.getBooleanProperty(
                "rheem.core.optimizer.enumeration.branchesfirst", true
        );

        // Configure the concatenations.
        final String priorityFunctionName = configuration.getStringProperty(
                "rheem.core.optimizer.enumeration.concatenationprio"
        );
        ToDoubleFunction<ConcatenationActivator> concatenationPriorityFunction;
        switch (priorityFunctionName) {
            case "slots":
                concatenationPriorityFunction = ConcatenationActivator::countNumOfOpenSlots;
                break;
            case "plans":
                concatenationPriorityFunction = ConcatenationActivator::estimateNumConcatenatedPlanImplementations;
                break;
            case "random":
                // Randomly generate a priority. However, avoid re-generate priorities, because that would increase
                // of a concatenation activator being processed, the longer it is in the queue (I guess).
                concatenationPriorityFunction = activator -> {
                    if (!Double.isNaN(activator.priority)) return activator.priority;
                    return Math.random();
                };
                break;
            case "none":
                concatenationPriorityFunction = activator -> 0d;
                break;
            default:
                throw new RheemException("Unknown concatenation priority function: " + priorityFunctionName);
        }

        boolean isInvertConcatenationPriorities = configuration.getBooleanProperty(
                "rheem.core.optimizer.enumeration.invertconcatenations", false
        );
        this.concatenationPriorityFunction = isInvertConcatenationPriorities ?
                activator -> -concatenationPriorityFunction.applyAsDouble(activator) :
                concatenationPriorityFunction;


    }

    private void scheduleForEnumeration(Operator operator, OptimizationContext optimizationContext) {
        final EnumerationActivator enumerationActivator = new EnumerationActivator(operator, optimizationContext);
        if (enumerationActivator.canBeActivated()) {
            this.activatedEnumerations.add(enumerationActivator);
        }
        this.enumerationActivators.put(enumerationActivator.getKey(), enumerationActivator);
    }


    /**
     * Produce the {@link PlanEnumeration} for the plan specified during the construction of this instance.
     *
     * @param isRequireResult whether the result is allowed to be {@code null}
     * @return the result {@link PlanEnumeration} or {@code null} if none such exists
     */
    public PlanEnumeration enumerate(boolean isRequireResult) {
        this.run();
        final PlanEnumeration comprehensiveEnumeration = this.resultReference.get();
        if (isRequireResult && comprehensiveEnumeration == null) {
            this.logger.error("No comprehensive PlanEnumeration.");
            this.logger.error("Pending enumerations: {}", this.enumerationActivators.values().stream()
                    .filter(activator -> !activator.wasExecuted())
                    .collect(Collectors.toList())
            );
            this.logger.error("Pending concatenations: {}", this.concatenationActivators.values().stream()
                    .filter(activator -> !activator.wasExecuted())
                    .collect(Collectors.toList())
            );
            throw new RheemException("Could not find a single execution plan.");
        }
        return comprehensiveEnumeration;
    }

    private Stream<OperatorAlternative.Alternative> streamPickedAlternatives(Operator operator) {
        OperatorContainer container = operator.getContainer();
        while (container != null && !(container instanceof OperatorAlternative.Alternative)) {
            container = container.toOperator().getContainer();
        }
        if (container == null) return Stream.empty();
        OperatorAlternative.Alternative alternative = (OperatorAlternative.Alternative) container;
        final OperatorAlternative operatorAlternative = alternative.toOperator();
        return Stream.concat(Stream.of(alternative), this.streamPickedAlternatives(operatorAlternative));
    }

    /**
     * Execute the enumeration. Needs only be run once and will be called by one of the result retrieval methods.
     */
    private synchronized void run() {
        if (this.resultReference == null) {
            while (!this.activatedEnumerations.isEmpty()) {
                // Try to enumerate branches.
                EnumerationActivator enumerationActivator;
                if ((enumerationActivator = this.activatedEnumerations.poll()) != null) {
                    if (this.isTopLevel()) {
                        this.logger.debug("Execute {}.", enumerationActivator);
                    }
                    this.enumerateBranchStartingFrom(enumerationActivator);
                }
            }

            ConcatenationActivator concatenationActivator;
            while ((concatenationActivator = this.activatedConcatenations.poll()) != null) {
                if (this.isTopLevel()) {
                    this.logger.debug("Execute {} (open inputs: {}).",
                            concatenationActivator,
                            concatenationActivator.getBaseEnumeration().getRequestedInputSlots()
                    );
                }
                this.concatenate(concatenationActivator);
            }

            this.constructResultEnumeration();
        }
    }

    /**
     * Enumerate plans from the branch that starts at the given node. The mode of operation is as follows:
     * <ol>
     * <li>Enumerate all {@link Operator}s forming the branch.</li>
     * <li>Create a new {@link PlanEnumeration} for the branch.</li>
     * <li>Join the branch {@link PlanEnumeration} with all existing input {@link PlanEnumeration}s.</li>
     * <li>Activate downstream {@link Operator}s for upcoming branch enumerations.</li>
     * </ol>
     *
     * @param enumerationActivator the activated {@link EnumerationActivator}
     */
    private void enumerateBranchStartingFrom(EnumerationActivator enumerationActivator) {
        assert !enumerationActivator.wasExecuted();
        enumerationActivator.markAsExecuted();

        // Start with the activated operator.
        Operator currentOperator = enumerationActivator.activatableOperator;
        List<Operator> branch = this.collectBranchOperatorsStartingFrom(currentOperator);
        if (branch == null) {
            return;
        }
        if (this.isTopLevel()) {
            this.logger.debug("Enumerating top-level {}.", branch);
        }

        // Go over the branch and create a PlanEnumeration for it.
        final OptimizationContext currentOptimizationCtx = enumerationActivator.getOptimizationContext();
        PlanEnumeration branchEnumeration = this.enumerateBranch(branch, currentOptimizationCtx);
        if (branchEnumeration == null) {
            return;
        }

        this.postProcess(branchEnumeration, currentOptimizationCtx);
    }

    /**
     * Determine the branch (straight of operators) that begins at the given {@link Operator}.
     *
     * @param startOperator starts the branch
     * @return the {@link Operator}s of the branch in their order of appearance or {@code null} if the branch is
     * already known to not yield any enumerations
     */
    private List<Operator> collectBranchOperatorsStartingFrom(Operator startOperator) {
        List<Operator> branch = new LinkedList<>();
        Operator currentOperator = startOperator;
        while (true) {
            boolean isEnumeratable = currentOperator.isExecutionOperator() ||
                    currentOperator.isAlternative() ||
                    currentOperator.isLoopSubplan();
            if (!isEnumeratable) {
                this.logger.trace("Cannot enumerate branch with {}.", currentOperator);
                return null;
            }
            branch.add(currentOperator);

            // Cut branches if requested.
            if (!this.isEnumeratingBranchesFirst) break;

            // Try to advance. This requires certain conditions, though.
            OutputSlot<?> followableOutput;
            if (currentOperator.isLoopHead()) {
                LoopHeadOperator loopHeadOperator = (LoopHeadOperator) currentOperator;
                if (loopHeadOperator.getLoopBodyOutputs().size() != 1) {
                    break;
                }
                followableOutput = RheemCollections.getSingle(loopHeadOperator.getLoopBodyOutputs());

            } else {
                if (currentOperator.getNumOutputs() != 1) {
                    break;
                }
                followableOutput = currentOperator.getOutput(0);
            }

            if (followableOutput.getOccupiedSlots().size() != 1) {
                this.logger.trace("Stopping branch, because operator does not feed exactly one operator.");
                break;
            }
            Operator nextOperator = currentOperator.getOutput(0).getOccupiedSlots().get(0).getOwner();
            if (nextOperator.getNumInputs() != 1) {
                this.logger.trace("Stopping branch, because next operator does not have exactly one input.");
                break;
            }

            if (nextOperator == startOperator) {
                break;
            }

            currentOperator = nextOperator;
        }
        this.logger.trace("Determined branch: {}.", currentOperator);

        return branch;
    }

    /**
     * Create a {@link PlanEnumeration} for the given {@code branch}.
     *
     * @param branch              {@link List} of {@link Operator}s of the branch; ordered downstream
     * @param optimizationContext in which the {@code branch} resides
     * @return a {@link PlanEnumeration} for the given {@code branch}
     */
    private PlanEnumeration enumerateBranch(List<Operator> branch, OptimizationContext optimizationContext) {
        PlanEnumeration branchEnumeration = null;
        Operator lastOperator = null;
        for (Operator operator : branch) {
            PlanEnumeration operatorEnumeration;
            if (operator.isAlternative()) {
                operatorEnumeration = this.enumerateAlternative((OperatorAlternative) operator, optimizationContext);
                if (operatorEnumeration == null || operatorEnumeration.getPlanImplementations().isEmpty()) {
                    this.logger.warn("No implementations enumerated for {}.", operator);
                    return null;
                }
            } else if (operator.isLoopSubplan()) {
                operatorEnumeration = this.enumerateLoop((LoopSubplan) operator, optimizationContext);
            } else {
                assert operator.isExecutionOperator();
                operatorEnumeration = PlanEnumeration.createSingleton((ExecutionOperator) operator, optimizationContext);

                // Check if the operator is filtered.
                // However, we must not filter operators that are pre-settled (i.e., that have been executed already).
                boolean isPresettled = false;
                OperatorContainer container = operator.getContainer();
                if (container instanceof OperatorAlternative.Alternative) {
                    OperatorAlternative.Alternative alternative = (OperatorAlternative.Alternative) container;
                    OperatorAlternative operatorAlternative = alternative.getOperatorAlternative();
                    isPresettled = this.presettledAlternatives.get(operatorAlternative) == alternative;
                }
                if (!isPresettled) {
                    OptimizationContext.OperatorContext operatorContext = optimizationContext.getOperatorContext(operator);
                    if (operatorContext != null && ((ExecutionOperator) operator).isFiltered(operatorContext)) {
                        this.logger.info("Filtered {} with context {}.", operator, operatorContext);
                        operatorEnumeration.getPlanImplementations().clear();
                    }
                }
            }

            if (operatorEnumeration.getPlanImplementations().isEmpty()) {
                if (this.isTopLevel()) {
                    throw new RheemException(String.format("No implementations enumerated for %s.", operator));
                } else {
                    this.logger.warn("No implementations enumerated for {}.", operator);
                }
            }

            if (branchEnumeration == null) {
                branchEnumeration = operatorEnumeration;
            } else {
                final OutputSlot<?> output = lastOperator.getOutput(0);
                branchEnumeration = branchEnumeration.concatenate(
                        output,
                        this.openChannels.get(output),
                        Collections.singletonMap(operator.getInput(0), operatorEnumeration),
                        optimizationContext,
                        this.timeMeasurement);

                if (branchEnumeration.getPlanImplementations().isEmpty()) {
                    if (this.isTopLevel()) {
                        throw new RheemException(String.format("Could not concatenate %s to %s.", lastOperator, operator));
{'fraction_non_alphanumeric': 0.0868802422552813, 'fraction_numerical': 0.0007209978610396789, 'mean_word_length': 2.396734693877551, 'pattern_counts': {'":': 8, '<': 111, '<?xml version=': 0, '>': 129, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1698904', 'n_tokens_mistral': 10270, 'n_tokens_neox': 9670, 'n_words': 2461}