/**
 * 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.rya.indexing;

import java.io.IOException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.commons.lang.Validate;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;
import org.apache.rya.accumulo.AccumuloRdfConfiguration;
import org.apache.rya.indexing.IndexingFunctionRegistry.FUNCTION_TYPE;
import org.apache.rya.indexing.accumulo.ConfigUtils;
import org.apache.rya.indexing.accumulo.freetext.AccumuloFreeTextIndexer;
import org.apache.rya.indexing.accumulo.freetext.FreeTextTupleSet;
import org.apache.rya.indexing.accumulo.geo.GeoParseUtils;
import org.apache.rya.indexing.accumulo.geo.GeoTupleSet;
import org.apache.rya.indexing.accumulo.geo.OptionalConfigUtils;
import org.apache.rya.indexing.accumulo.temporal.AccumuloTemporalIndexer;
import org.apache.rya.mongodb.MongoSecondaryIndex;
import org.apache.rya.mongodb.StatefulMongoDBRdfConfiguration;
import org.eclipse.rdf4j.model.IRI;
import org.eclipse.rdf4j.model.Resource;
import org.eclipse.rdf4j.model.ValueFactory;
import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
import org.eclipse.rdf4j.query.BindingSet;
import org.eclipse.rdf4j.query.Dataset;
import org.eclipse.rdf4j.query.algebra.And;
import org.eclipse.rdf4j.query.algebra.Filter;
import org.eclipse.rdf4j.query.algebra.FunctionCall;
import org.eclipse.rdf4j.query.algebra.Join;
import org.eclipse.rdf4j.query.algebra.LeftJoin;
import org.eclipse.rdf4j.query.algebra.QueryModelNode;
import org.eclipse.rdf4j.query.algebra.StatementPattern;
import org.eclipse.rdf4j.query.algebra.TupleExpr;
import org.eclipse.rdf4j.query.algebra.ValueConstant;
import org.eclipse.rdf4j.query.algebra.Var;
import org.eclipse.rdf4j.query.algebra.evaluation.QueryOptimizer;
import org.eclipse.rdf4j.query.algebra.helpers.AbstractQueryModelVisitor;

import com.google.common.collect.Lists;

public class GeoEnabledFilterFunctionOptimizer implements QueryOptimizer, Configurable {
    private static final Logger LOG = Logger.getLogger(GeoEnabledFilterFunctionOptimizer.class);
    private static final ValueFactory VF = SimpleValueFactory.getInstance();

    private Configuration conf;
    private GeoIndexer geoIndexer;
    private FreeTextIndexer freeTextIndexer;
    private TemporalIndexer temporalIndexer;
    private boolean init = false;

    public GeoEnabledFilterFunctionOptimizer() {
    }

    public GeoEnabledFilterFunctionOptimizer(final AccumuloRdfConfiguration conf) throws AccumuloException, AccumuloSecurityException,
    TableNotFoundException, IOException, TableExistsException, NumberFormatException, UnknownHostException {
        this.conf = conf;
        init();
    }

    //setConf initializes FilterFunctionOptimizer so reflection can be used
    //to create optimizer in RdfCloudTripleStoreConnection
    @Override
    public void setConf(final Configuration conf) {
        this.conf = conf;
        //reset the init.
        init = false;
        init();
    }
    /**
     * Load instances of the selected indexers.  This is tricky because some (geomesa vs geowave) have incompatible dependencies (geotools versions).
     */
    private synchronized void init() {
        if (!init) {
            if (ConfigUtils.getUseMongo(conf)) {
                // create a new MongoGeoIndexer() without having it at compile time.
                StatefulMongoDBRdfConfiguration stateConf = (StatefulMongoDBRdfConfiguration) conf;
                for(final MongoSecondaryIndex indexer : stateConf.getAdditionalIndexers()) {
                    if(indexer instanceof FreeTextIndexer) {
                        freeTextIndexer = (FreeTextIndexer) indexer;
                    } else if(indexer instanceof TemporalIndexer) {
                        temporalIndexer = (TemporalIndexer) indexer;
                    } else if(indexer instanceof GeoIndexer) {
                        geoIndexer = (GeoIndexer) indexer;
                    }
                }
            } else {
                GeoIndexerType geoIndexerType = OptionalConfigUtils.getGeoIndexerType(conf);
                if (geoIndexerType == GeoIndexerType.UNSPECIFIED) {
                    geoIndexer = instantiate(GeoIndexerType.GEO_MESA.getGeoIndexerClassString(), GeoIndexer.class);
                } else {
                    geoIndexer = instantiate(geoIndexerType.getGeoIndexerClassString(), GeoIndexer.class);
                }
                geoIndexer.setConf(conf);
                freeTextIndexer = new AccumuloFreeTextIndexer();
                freeTextIndexer.setConf(conf);
                temporalIndexer = new AccumuloTemporalIndexer();
                temporalIndexer.setConf(conf);
            }
            init = true;
        }
    }


    @Override
    public void optimize(final TupleExpr tupleExpr, final Dataset dataset, final BindingSet bindings) {
        // find variables used in property and resource based searches:
        final SearchVarVisitor searchVars = new SearchVarVisitor();
        tupleExpr.visit(searchVars);
        // rewrites for property searches:
        processPropertySearches(tupleExpr, searchVars.searchProperties);

    }

    /**
     * helper to instantiate a class from a string class name.
     * @param className name of class to instantiate.
     * @param type base interface that the class immplements
     * @return the instance.
     */
    public static <T> T instantiate(final String className, final Class<T> type){
        try{
            return type.cast(Class.forName(className).newInstance());
        } catch(InstantiationException
                | IllegalAccessException
                | ClassNotFoundException e){
            throw new IllegalStateException(e);
        }
    }

    private void processPropertySearches(final TupleExpr tupleExpr, final Collection<Var> searchProperties) {
        final MatchStatementVisitor matchStatements = new MatchStatementVisitor(searchProperties);
        tupleExpr.visit(matchStatements);
        for (final StatementPattern matchStatement: matchStatements.matchStatements) {
            final Var subject = matchStatement.getSubjectVar();
            if (subject.hasValue() && !(subject.getValue() instanceof Resource)) {
                throw new IllegalArgumentException("Query error: Found " + subject.getValue() + ", expected an IRI or BNode");
            }
            Validate.isTrue(subject.hasValue() || subject.getName() != null);
            Validate.isTrue(!matchStatement.getObjectVar().hasValue() && matchStatement.getObjectVar().getName() != null);
            buildQuery(tupleExpr, matchStatement);
        }
    }

    private void buildQuery(final TupleExpr tupleExpr, final StatementPattern matchStatement) {
        //If our IndexerExpr (to be) is the rhs-child of LeftJoin, we can safely make that a Join:
        //  the IndexerExpr will (currently) not return results that can deliver unbound variables.
        //This optimization should probably be generalized into a LeftJoin -> Join optimizer under certain conditions. Until that
        //  has been done, this code path at least takes care of queries generated by OpenSahara SparqTool that filter on OPTIONAL
        //  projections. E.g. summary~'full text search' (summary is optional). See #379
        if (matchStatement.getParentNode() instanceof LeftJoin) {
            final LeftJoin leftJoin = (LeftJoin)matchStatement.getParentNode();
            if (leftJoin.getRightArg() == matchStatement && leftJoin.getCondition() == null) {
                matchStatement.getParentNode().replaceWith(new Join(leftJoin.getLeftArg(), leftJoin.getRightArg()));
            }
        }
        final FilterFunction fVisitor = new FilterFunction(matchStatement.getObjectVar().getName());
        tupleExpr.visit(fVisitor);
        final List<IndexingExpr> results = Lists.newArrayList();
        for(int i = 0; i < fVisitor.func.size(); i++){
            results.add(new IndexingExpr(fVisitor.func.get(i), matchStatement, fVisitor.args.get(i)));
        }
        removeMatchedPattern(tupleExpr, matchStatement, new IndexerExprReplacer(results));
    }

    //find vars contained in filters
    private static class SearchVarVisitor extends AbstractQueryModelVisitor<RuntimeException> {
        private final Collection<Var> searchProperties = new ArrayList<>();

        @Override
        public void meet(final FunctionCall fn) {
            final IRI fun = VF.createIRI(fn.getURI());
            final Var result = IndexingFunctionRegistry.getResultVarFromFunctionCall(fun, fn.getArgs());
            if (result != null && !searchProperties.contains(result)) {
                searchProperties.add(result);
            }
        }
    }

    //find StatementPatterns containing filter variables
    private static class MatchStatementVisitor extends AbstractQueryModelVisitor<RuntimeException> {
        private final Collection<Var> propertyVars;
        private final Collection<Var> usedVars = new ArrayList<>();
        private final List<StatementPattern> matchStatements = new ArrayList<>();

        public MatchStatementVisitor(final Collection<Var> propertyVars) {
            this.propertyVars = propertyVars;
        }

        @Override public void meet(final StatementPattern statement) {
            final Var object = statement.getObjectVar();
            if (propertyVars.contains(object)) {
                if (usedVars.contains(object)) {
                    throw new IllegalArgumentException("Illegal search, variable is used multiple times as object: " + object.getName());
                } else {
                    usedVars.add(object);
                    matchStatements.add(statement);
                }
            }
        }
    }

    private abstract class AbstractEnhanceVisitor extends AbstractQueryModelVisitor<RuntimeException> {
        final String matchVar;
        List<IRI> func = Lists.newArrayList();
        List<Object[]> args = Lists.newArrayList();

        public AbstractEnhanceVisitor(final String matchVar) {
            this.matchVar = matchVar;
        }

        protected void addFilter(final IRI iri, final Object[] values) {
            func.add(iri);
            args.add(values);
        }
    }

    //create indexing expression for each filter matching var in filter StatementPattern
    //replace old filter condition with true condition
    private class FilterFunction extends AbstractEnhanceVisitor {
        public FilterFunction(final String matchVar) {
            super(matchVar);
        }

        @Override
        public void meet(final FunctionCall call) {
            final IRI fnIri = VF.createIRI(call.getURI());
            final Var resultVar = IndexingFunctionRegistry.getResultVarFromFunctionCall(fnIri, call.getArgs());
            if (resultVar != null && resultVar.getName().equals(matchVar)) {
                addFilter(VF.createIRI(call.getURI()), GeoParseUtils.extractArguments(matchVar, call));
                if (call.getParentNode() instanceof Filter || call.getParentNode() instanceof And || call.getParentNode() instanceof LeftJoin) {
                    call.replaceWith(new ValueConstant(VF.createLiteral(true)));
                } else {
                    throw new IllegalArgumentException("Query error: Found " + call + " as part of an expression that is too complex");
                }
            }
        }

        @Override
        public void meet(final Filter filter) {
            //First visit children, then condition (reverse of default):
            filter.getArg().visit(this);
            filter.getCondition().visit(this);
        }
    }

    private void removeMatchedPattern(final TupleExpr tupleExpr, final StatementPattern pattern, final TupleExprReplacer replacer) {
        final List<TupleExpr> indexTuples = replacer.createReplacement(pattern);
        if (indexTuples.size() > 1) {
            final VarExchangeVisitor vev = new VarExchangeVisitor(pattern);
            tupleExpr.visit(vev);
            Join join = new Join(indexTuples.remove(0), indexTuples.remove(0));
            for (final TupleExpr geo : indexTuples) {
                join = new Join(join, geo);
            }
            pattern.replaceWith(join);
        } else if (indexTuples.size() == 1) {
            pattern.replaceWith(indexTuples.get(0));
            pattern.setParentNode(null);
        } else {
            throw new IllegalStateException("Must have at least one replacement for matched StatementPattern.");
        }
    }

    private interface TupleExprReplacer {
        List<TupleExpr> createReplacement(TupleExpr org);
    }

    //replace each filter pertinent StatementPattern with corresponding index expr
    private class IndexerExprReplacer implements TupleExprReplacer {
        private final List<IndexingExpr> indxExpr;
        private final FUNCTION_TYPE type;

        public IndexerExprReplacer(final List<IndexingExpr> indxExpr) {
            this.indxExpr = indxExpr;
            final IRI func = indxExpr.get(0).getFunction();
            type = IndexingFunctionRegistry.getFunctionType(func);
        }

        @Override
        public List<TupleExpr> createReplacement(final TupleExpr org) {
            final List<TupleExpr> indexTuples = Lists.newArrayList();
            switch (type) {
            case GEO:
                for (final IndexingExpr indx : indxExpr) {
                    indexTuples.add(new GeoTupleSet(indx, geoIndexer));
                }
                break;
            case FREETEXT:
                for (final IndexingExpr indx : indxExpr) {
                    indexTuples.add(new FreeTextTupleSet(indx, freeTextIndexer));
                }
                break;
            case TEMPORAL:
                for (final IndexingExpr indx : indxExpr) {
                    indexTuples.add(new TemporalTupleSet(indx, temporalIndexer));
                }
                break;
            default:
                throw new IllegalArgumentException("Incorrect type!");
            }
            return indexTuples;
        }
    }

    private static class VarExchangeVisitor extends AbstractQueryModelVisitor<RuntimeException> {
        private final  StatementPattern exchangeVar;
        public VarExchangeVisitor(final StatementPattern sp) {
            exchangeVar = sp;
        }

        @Override
        public void meet(final Join node) {
            final QueryModelNode lNode = node.getLeftArg();
            if (lNode instanceof StatementPattern) {
                exchangeVar.replaceWith(lNode);
                node.setLeftArg(exchangeVar);
            } else {
                super.meet(node);
            }
        }
    }

    @Override
    public Configuration getConf() {
        return conf;
    }
}