/*
 *  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.tinkerpop.gremlin.process.traversal.strategy.optimization;

import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.optimization.GraphFilterStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.Compare;
import org.apache.tinkerpop.gremlin.process.traversal.Contains;
import org.apache.tinkerpop.gremlin.process.traversal.P;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.step.GValue;
import org.apache.tinkerpop.gremlin.process.traversal.step.LambdaHolder;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.AndStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.DedupGlobalStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.DropStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.FilterStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.OrStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.TraversalFilterStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.MatchStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.VertexStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.VertexStepContract;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.VertexStepPlaceholder;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.RangeGlobalStepContract;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.HasContainer;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.util.OrP;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.T;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

/**
 * This strategy analyzes filter-steps with child traversals that themselves are pure filters. If the child traversals
 * are pure filters then the wrapping parent filter is not needed and thus, the children can be "inlined." Normalizing
 * pure filters with inlining reduces the number of variations of a filter that a graph provider may need to reason
 * about when writing their own strategies. As a result, this strategy helps increase the likelihood that a provider's
 * filtering optimization will succeed at re-writing the traversal.
 * <p>
 * Has-steps containing traversal-bearing predicates (i.e. {@code has("age", P.gt(__.V(1).values("age")))}) are
 * excluded from inlining because their child traversals must be resolved per-traverser at runtime and cannot be
 * safely merged with adjacent has-steps or folded into edge labels.
 *
 * @author Marko A. Rodriguez (http://markorodriguez.com)
 * @example <pre>
 * __.outE().hasLabel(eq("knows").or(eq("created"))).inV()       // is replaced by __.outE("knows", "created").inV()
 * __.filter(has("name","marko"))                                // is replaced by __.has("name","marko")
 * __.and(has("name"),has("age"))                                // is replaced by __.has("name").has("age")
 * __.and(filter(has("name","marko").has("age")),hasNot("blah")) // is replaced by __.has("name","marko").has("age").hasNot("blah")
 * __.match(as('a').has(key,value),...)                          // is replaced by __.as('a').has(key,value).match(...)
 * </pre>
 */
public final class InlineFilterStrategy extends AbstractTraversalStrategy<TraversalStrategy.OptimizationStrategy> implements TraversalStrategy.OptimizationStrategy {

    private static final InlineFilterStrategy INSTANCE = new InlineFilterStrategy();
    private static final Set<Class<? extends OptimizationStrategy>> POSTS = new HashSet<>(Arrays.asList(
            GraphFilterStrategy.class,
            AdjacentToIncidentStrategy.class,
            PathRetractionStrategy.class));
    private static final Set<Class<? extends OptimizationStrategy>> PRIORS = new HashSet<>(Arrays.asList(
            FilterRankingStrategy.class,
            IdentityRemovalStrategy.class,
            MatchPredicateStrategy.class));

    private InlineFilterStrategy() {
    }

    @Override
    public void apply(final Traversal.Admin<?, ?> traversal) {
        boolean changed = true; // recursively walk child traversals trying to inline them into the current traversal line.
        while (changed) {
            changed = false;
            final Iterator<FilterStep> filterStepIterator = TraversalHelper.getStepsOfAssignableClass(FilterStep.class, traversal).iterator();
            while (!changed && filterStepIterator.hasNext()) {
                final FilterStep<?> step = filterStepIterator.next();
                changed = step instanceof HasStep && processHasStep((HasStep) step, traversal) ||
                        step instanceof TraversalFilterStep && processTraversalFilterStep((TraversalFilterStep) step, traversal) ||
                        step instanceof OrStep && processOrStep((OrStep) step, traversal) ||
                        step instanceof AndStep && processAndStep((AndStep) step, traversal);
            }
            if (!changed && traversal.isRoot()) {
                final Iterator<MatchStep> matchStepIterator = TraversalHelper.getStepsOfClass(MatchStep.class, traversal).iterator();
                while (!changed && matchStepIterator.hasNext()) {
                    if (processMatchStep(matchStepIterator.next(), traversal))
                        changed = true;
                }
            }
        }
    }

    ////////////////////////////
    ///////////////////////////

    private static boolean processHasStep(final HasStep<?> step, final Traversal.Admin<?, ?> traversal) {
        // Skip inlining for has-steps with traversal-bearing predicates. These contain child traversals that
        // must be resolved per-traverser at runtime and cannot be merged into adjacent steps.
        if (step.getHasContainers().stream().anyMatch(HasContainer::hasTraversal)) return false;
        if (step.getPreviousStep() instanceof HasStep) {
            // Also skip if the previous has-step contains traversal-bearing predicates, since merging into
            // it would mix static and dynamic predicates in a single step.
            if (((HasStep<?>) step.getPreviousStep()).getHasContainers().stream().anyMatch(HasContainer::hasTraversal)) return false;
        }
        if (step.getPreviousStep() instanceof HasStep) {
            final HasStep<?> previousStep = (HasStep<?>) step.getPreviousStep();
            final List<HasContainer> hasContainers = new ArrayList<>(step.getHasContainers());
            for (final HasContainer hasContainer : hasContainers) {
                previousStep.addHasContainer(hasContainer);
            }

            TraversalHelper.copyLabels(step, previousStep, false);
            TraversalHelper.removeStep(step, traversal);
            return true;
        } else if (step.getPreviousStep() instanceof VertexStepContract
                && ((VertexStepContract) step.getPreviousStep()).returnsEdge()
                && 0 == ((VertexStepContract) step.getPreviousStep()).getEdgeLabels().length) {
            final VertexStepContract<Edge> previousStep = (VertexStepContract<Edge>) step.getPreviousStep();
            final List<GValue<?>> edgeLabelGValues = new ArrayList<>();
            for (final HasContainer hasContainer : new ArrayList<>(step.getHasContainers())) {
                if (hasContainer.getKey().equals(T.label.getAccessor())) {
                    if (hasContainer.getBiPredicate() == Compare.eq &&
                            hasContainer.getValue() instanceof String &&
                            edgeLabelGValues.isEmpty()) {
                        edgeLabelGValues.add(hasContainer.getPredicate().isParameterized() ?
                                hasContainer.getPredicate().getGValues().iterator().next() :
                                GValue.ofString((String) hasContainer.getValue()));
                        step.removeHasContainer(hasContainer);
                    } else if (hasContainer.getBiPredicate() == Contains.within &&
                            hasContainer.getValue() instanceof Collection &&
                            ((Collection) hasContainer.getValue()).containsAll(Arrays.asList(GValue.resolveToValues(edgeLabelGValues.stream().toArray(GValue[]::new))))) {
                        edgeLabelGValues.addAll(hasContainer.getPredicate().isParameterized() ?
                                hasContainer.getPredicate().getGValues() :
                                Arrays.asList(GValue.ensureGValues(((Collection<String>) hasContainer.getValue()).toArray())));
                        step.removeHasContainer(hasContainer);

                    } else if (hasContainer.getPredicate() instanceof OrP && edgeLabelGValues.isEmpty()) {
                        boolean removeContainer = true;
                        final List<P<?>> orps = ((OrP) hasContainer.getPredicate()).getPredicates();
                        final List<GValue<String>> newEdges = new ArrayList<>();
                        for (int i = 0; i < orps.size(); i++) {
                            if (orps.get(i).getBiPredicate() == Compare.eq && orps.get(i).getValue() instanceof String) {
                                newEdges.add(orps.get(i).isParameterized() ?
                                        (GValue<String>) orps.get(i).getGValues().iterator().next() :
                                        GValue.ofString((String) orps.get(i).getValue()));
                            } else {
                                removeContainer = false;
                                break;
                            }
                        }
                        if (removeContainer) {
                            edgeLabelGValues.addAll(newEdges);
                            step.removeHasContainer(hasContainer);
                        }
                    }
                }
            }
            if (!edgeLabelGValues.isEmpty()) {
                final VertexStepContract<Edge> newVertexStep = GValue.containsVariables(edgeLabelGValues.toArray()) ?
                        new VertexStepPlaceholder<>(traversal, Edge.class, previousStep.getDirection(), edgeLabelGValues.toArray(new GValue[edgeLabelGValues.size()])) :
                        new VertexStep<>(traversal, Edge.class, previousStep.getDirection(), Arrays.stream(GValue.resolveToValues(edgeLabelGValues.toArray(new GValue[edgeLabelGValues.size()]))).toArray(String[]::new));
                TraversalHelper.replaceStep(previousStep, newVertexStep, traversal);
                TraversalHelper.copyLabels(previousStep, newVertexStep, false);
                if (step.getHasContainers().isEmpty()) {
                    TraversalHelper.copyLabels(step, newVertexStep, false);
                    TraversalHelper.removeStep(step, traversal);
                }
                return true;
            }
            return false;
        } else
            return false;
    }

    private static boolean processTraversalFilterStep(final TraversalFilterStep<?> step, final Traversal.Admin<?, ?> traversal) {
        final Traversal.Admin<?, ?> childTraversal = step.getLocalChildren().get(0);
        if (TraversalHelper.hasAllStepsOfClass(childTraversal, FilterStep.class) &&
                !TraversalHelper.hasStepOfClass(childTraversal,
                        DropStep.class,
                        RangeGlobalStepContract.class,
                        DedupGlobalStep.class,
                        LambdaHolder.class)) {
            final Step<?, ?> finalStep = childTraversal.getEndStep();
            TraversalHelper.insertTraversal((Step) step, childTraversal, traversal);
            TraversalHelper.copyLabels(step, finalStep, false);
            TraversalHelper.removeStep(step, traversal);
            return true;
        }
        return false;
    }

    /**
     * Processes an {@link OrStep} within a traversal and attempts to optimize it by replacing it with
     * simpler constructs if possible. The method evaluates the local child traversals of the provided
     * {@code OrStep} and determines if the {@code OrStep} can be simplified into a single step such as
     * a {@link HasStep}.
     */
    private static boolean processOrStep(final OrStep<?> step, final Traversal.Admin<?, ?> traversal) {
        boolean process = false;
        String key = null;
        P predicate = null;
        final List<String> labels = new ArrayList<>();
        for (final Traversal.Admin<?, ?> childTraversal : step.getLocalChildren()) {
            // there is at least one child to the or() so the burden of proof to not process this step is provided
            // by the code below. if the default to process was true then it ends up adding an empty has() to the
            // end of the traversal (which can happen when ConnectiveStrategy isn't executed - subtle)
            process = true;
            
            InlineFilterStrategy.instance().apply(childTraversal);
            for (final Step<?, ?> childStep : childTraversal.getSteps()) {
                if (childStep instanceof HasStep) {
                    // Skip optimization if any HasContainer has a traversal-bearing predicate
                    if (((HasStep<?>) childStep).getHasContainers().stream().anyMatch(HasContainer::hasTraversal)) {
                        process = false;
                        break;
                    }
                    P p = null;
                    for (final HasContainer hasContainer : ((HasStep<?>) childStep).getHasContainers()) {
                        if (null == key)
                            key = hasContainer.getKey();
                        else if (!hasContainer.getKey().equals(key)) {
                            process = false;
                            break;
                        }
                        p = null == p ?
                                hasContainer.getPredicate() :
                                p.and(hasContainer.getPredicate());
                    }
                    if (process) {
                        predicate = null == predicate ? p : predicate.or(p);
                    }
                    labels.addAll(childStep.getLabels());
                } else {
                    process = false;
                    break;
                }
            }
            if (!process)
                break;
        }
        if (process) {
            final HasStep hasStep = new HasStep<>(traversal, new HasContainer(key, predicate));
            TraversalHelper.replaceStep(step, hasStep, traversal);
            TraversalHelper.copyLabels(step, hasStep, false);
            for (final String label : labels) {
                hasStep.addLabel(label);
            }
            return true;
        }
        return false;
    }

    private static boolean processAndStep(final AndStep<?> step, final Traversal.Admin<?, ?> traversal) {
        boolean process = true;
        for (final Traversal.Admin<?, ?> childTraversal : step.getLocalChildren()) {
            if (!TraversalHelper.hasAllStepsOfClass(childTraversal, FilterStep.class) ||
                    TraversalHelper.hasStepOfClass(childTraversal,
                            DropStep.class,
                            RangeGlobalStepContract.class,
                            DedupGlobalStep.class,
                            LambdaHolder.class)) {
                process = false;
                break;
            }
        }
        if (process) {
            final List<Traversal.Admin<?, ?>> childTraversals = (List) step.getLocalChildren();
            Step<?, ?> finalStep = null;
            for (int i = childTraversals.size() - 1; i >= 0; i--) {
                final Traversal.Admin<?, ?> childTraversal = childTraversals.get(i);
                if (null == finalStep)
                    finalStep = childTraversal.getEndStep();
                TraversalHelper.insertTraversal((Step) step, childTraversals.get(i), traversal);

            }
            if (null != finalStep) TraversalHelper.copyLabels(step, finalStep, false);
            TraversalHelper.removeStep(step, traversal);
            return true;
        }
        return false;
    }

    private static boolean processMatchStep(final MatchStep<?, ?> step, final Traversal.Admin<?, ?> traversal) {
        if (step.getPreviousStep() instanceof EmptyStep)
            return false;
        boolean changed = false;
        final String startLabel = MatchStep.Helper.computeStartLabel(step.getGlobalChildren());
        for (final Traversal.Admin<?, ?> matchTraversal : new ArrayList<>(step.getGlobalChildren())) {
            if (TraversalHelper.hasAllStepsOfClass(matchTraversal,
                    HasStep.class,
                    MatchStep.MatchStartStep.class,
                    MatchStep.MatchEndStep.class) &&
                    matchTraversal.getStartStep() instanceof MatchStep.MatchStartStep &&
                    startLabel.equals(((MatchStep.MatchStartStep) matchTraversal.getStartStep()).getSelectKey().orElse(null))) {
                changed = true;
                step.removeGlobalChild(matchTraversal);
                final String endLabel = ((MatchStep.MatchEndStep) matchTraversal.getEndStep()).getMatchKey().orElse(null); // why would this exist? but just in case
                matchTraversal.removeStep(0);                                       // remove MatchStartStep
                matchTraversal.removeStep(matchTraversal.getSteps().size() - 1);    // remove MatchEndStep
                matchTraversal.getEndStep().addLabel(startLabel);
                if (null != endLabel) matchTraversal.getEndStep().addLabel(endLabel);
                TraversalHelper.insertTraversal((Step) step.getPreviousStep(), matchTraversal, traversal);
            }
        }
        if (step.getGlobalChildren().isEmpty())
            TraversalHelper.removeStep(step, traversal);
        return changed;
    }

    @Override
    public Set<Class<? extends OptimizationStrategy>> applyPost() {
        final Set<Class<? extends OptimizationStrategy>> post = OptimizationStrategy.super.applyPost();
        post.addAll(POSTS);
        return post;
    }

    @Override
    public Set<Class<? extends OptimizationStrategy>> applyPrior() {
        return PRIORS;
    }

    public static InlineFilterStrategy instance() {
        return INSTANCE;
    }
}