/*
 * 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.accumulo.entity;

import static org.apache.rya.api.RdfCloudTripleStoreConstants.DELIM_BYTE;
import static org.apache.rya.api.RdfCloudTripleStoreConstants.TYPE_DELIM_BYTE;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.Set;

import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.hadoop.io.Text;
import org.apache.rya.accumulo.AccumuloRdfConfiguration;
import org.apache.rya.accumulo.documentIndex.DocIndexIteratorUtil;
import org.apache.rya.accumulo.documentIndex.DocumentIndexIntersectingIterator;
import org.apache.rya.api.RdfCloudTripleStoreConfiguration;
import org.apache.rya.api.domain.RyaURI;
import org.apache.rya.api.resolver.RyaContext;
import org.apache.rya.api.resolver.RyaToRdfConversions;
import org.apache.rya.api.resolver.RyaTypeResolverException;
import org.apache.rya.indexing.DocIdIndexer;
import org.apache.rya.indexing.accumulo.ConfigUtils;
import org.eclipse.rdf4j.common.iteration.CloseableIteration;
import org.eclipse.rdf4j.query.BindingSet;
import org.eclipse.rdf4j.query.MalformedQueryException;
import org.eclipse.rdf4j.query.QueryEvaluationException;
import org.eclipse.rdf4j.query.algebra.StatementPattern;
import org.eclipse.rdf4j.query.algebra.TupleExpr;
import org.eclipse.rdf4j.query.algebra.evaluation.QueryBindingSet;
import org.eclipse.rdf4j.query.algebra.helpers.StatementPatternCollector;
import org.eclipse.rdf4j.query.parser.ParsedQuery;
import org.eclipse.rdf4j.query.parser.sparql.SPARQLParser;

import com.google.common.base.Preconditions;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Sets;
import com.google.common.primitives.Bytes;

public class AccumuloDocIdIndexer implements DocIdIndexer {


    private BatchScanner bs;
    private final AccumuloRdfConfiguration conf;

    public AccumuloDocIdIndexer(final RdfCloudTripleStoreConfiguration conf) throws AccumuloException, AccumuloSecurityException {
        Preconditions.checkArgument(conf instanceof RdfCloudTripleStoreConfiguration, "conf must be isntance of RdfCloudTripleStoreConfiguration");
        this.conf = (AccumuloRdfConfiguration) conf;
        //Connector conn = ConfigUtils.getConnector(conf);
    }


    public CloseableIteration<BindingSet, QueryEvaluationException> queryDocIndex(final String sparqlQuery,
            final Collection<BindingSet> constraints) throws TableNotFoundException, QueryEvaluationException {

        final SPARQLParser parser = new SPARQLParser();
        ParsedQuery pq1 = null;
        try {
            pq1 = parser.parseQuery(sparqlQuery, null);
        } catch (final MalformedQueryException e) {
            e.printStackTrace();
            throw new QueryEvaluationException("Malformed query. query=" + sparqlQuery, e);
        }

        final TupleExpr te1 = pq1.getTupleExpr();
        final List<StatementPattern> spList1 = StatementPatternCollector.process(te1);

        if(StarQuery.isValidStarQuery(spList1)) {
            final StarQuery sq1 = new StarQuery(spList1);
            return queryDocIndex(sq1, constraints);
        } else {
            throw new IllegalArgumentException("Invalid star query!");
        }

    }


    @Override
    public CloseableIteration<BindingSet, QueryEvaluationException> queryDocIndex(final StarQuery query,
            final Collection<BindingSet> constraints) throws TableNotFoundException, QueryEvaluationException {

        final StarQuery starQ = query;
        final Iterator<BindingSet> bs = constraints.iterator();
        final Iterator<BindingSet> bs2 = constraints.iterator();
        final Set<String> unCommonVarNames;
        final Set<String> commonVarNames;
        if (bs2.hasNext()) {
            final BindingSet currBs = bs2.next();
            commonVarNames = StarQuery.getCommonVars(query, currBs);
            unCommonVarNames = Sets.difference(currBs.getBindingNames(), commonVarNames);
        } else {
            commonVarNames = Sets.newHashSet();
            unCommonVarNames = Sets.newHashSet();
        }

        if( commonVarNames.size() == 1 && !query.commonVarConstant() && commonVarNames.contains(query.getCommonVarName())) {

            final HashMultimap<String, BindingSet> map = HashMultimap.create();
            final String commonVar = starQ.getCommonVarName();
            final Iterator<Entry<Key, Value>> intersections;
            final BatchScanner scan;
            final Set<Range> ranges = Sets.newHashSet();

            while(bs.hasNext()) {

                final BindingSet currentBs = bs.next();

                if(currentBs.getBinding(commonVar) == null) {
                    continue;
                }

                final String row = currentBs.getBinding(commonVar).getValue().stringValue();
                ranges.add(new Range(row));
                map.put(row, currentBs);

            }
            scan = runQuery(starQ, ranges);
            intersections = scan.iterator();


            return new CloseableIteration<BindingSet, QueryEvaluationException>() {


                private QueryBindingSet currentSolutionBs = null;
                private boolean hasNextCalled = false;
                private boolean isEmpty = false;
                private Iterator<BindingSet> inputSet = new ArrayList<BindingSet>().iterator();
                private BindingSet currentBs;
                private Key key;


                @Override
                public boolean hasNext() throws QueryEvaluationException {
                    if (!hasNextCalled && !isEmpty) {
                        while (inputSet.hasNext() || intersections.hasNext()) {
                            if (!inputSet.hasNext()) {
                                key = intersections.next().getKey();
                                inputSet = map.get(key.getRow().toString()).iterator();
                            }
                            currentBs = inputSet.next();
                            currentSolutionBs = deserializeKey(key, starQ, currentBs, unCommonVarNames);

                            if (currentSolutionBs.size() == unCommonVarNames.size() + starQ.getUnCommonVars().size() +1) {
                                hasNextCalled = true;
                                return true;
                            }

                        }

                        isEmpty = true;
                        return false;

                    } else {
                        return !isEmpty;
                    }

                }


                @Override
                public BindingSet next() throws QueryEvaluationException {

                    if (hasNextCalled) {
                        hasNextCalled = false;
                    } else if (isEmpty) {
                        throw new NoSuchElementException();
                    } else {
                        if (this.hasNext()) {
                            hasNextCalled = false;
                        } else {
                            throw new NoSuchElementException();
                        }
                    }

                    return currentSolutionBs;
                }

                @Override
                public void remove() throws QueryEvaluationException {
                    throw new UnsupportedOperationException();
                }

                @Override
                public void close() throws QueryEvaluationException {
                    scan.close();
                }

            };


        } else {

            return new CloseableIteration<BindingSet, QueryEvaluationException>() {

                @Override
                public void remove() throws QueryEvaluationException {
                    throw new UnsupportedOperationException();
                }

                private Iterator<Entry<Key, Value>> intersections = null;
                private QueryBindingSet currentSolutionBs = null;
                private boolean hasNextCalled = false;
                private boolean isEmpty = false;
                private boolean init = false;
                private BindingSet currentBs;
                private StarQuery sq = new StarQuery(starQ);
                private final Set<Range> emptyRangeSet = Sets.newHashSet();
                private BatchScanner scan;

                @Override
                public BindingSet next() throws QueryEvaluationException {
                    if (hasNextCalled) {
                        hasNextCalled = false;
                    } else if (isEmpty) {
                        throw new NoSuchElementException();
                    } else {
                        if (this.hasNext()) {
                            hasNextCalled = false;
                        } else {
                            throw new NoSuchElementException();
                        }
                    }
                    return currentSolutionBs;
                }

                @Override
                public boolean hasNext() throws QueryEvaluationException {

                    if (!init) {
                        if (intersections == null && bs.hasNext()) {
                            currentBs = bs.next();
                            sq = StarQuery.getConstrainedStarQuery(sq, currentBs);
                            scan = runQuery(sq,emptyRangeSet);
                            intersections = scan.iterator();
                            // binding set empty
                        } else if (intersections == null && !bs.hasNext()) {
                            currentBs = new QueryBindingSet();
                            scan = runQuery(starQ,emptyRangeSet);
                            intersections = scan.iterator();
                        }

                        init = true;
                    }

                    if (!hasNextCalled && !isEmpty) {
                        while (intersections.hasNext() || bs.hasNext()) {
                            if (!intersections.hasNext()) {
                                scan.close();
                                currentBs = bs.next();
                                sq = StarQuery.getConstrainedStarQuery(sq, currentBs);
                                scan = runQuery(sq,emptyRangeSet);
                                intersections = scan.iterator();
                            }
                            if (intersections.hasNext()) {
                                currentSolutionBs = deserializeKey(intersections.next().getKey(), sq, currentBs,
                                        unCommonVarNames);
                            } else {
                                continue;
                            }

                            if (sq.commonVarConstant() && currentSolutionBs.size() == unCommonVarNames.size() + sq.getUnCommonVars().size()) {
                                hasNextCalled = true;
                                return true;
                            } else if(currentSolutionBs.size() == unCommonVarNames.size() + sq.getUnCommonVars().size() + 1) {
                                hasNextCalled = true;
                                return true;
                            }
                        }

                        isEmpty = true;
                        return false;

                    } else {
                        return !isEmpty;
                    }
                }

                @Override
                public void close() throws QueryEvaluationException {
                    scan.close();
                }
            };
        }
    }

    private QueryBindingSet deserializeKey(final Key key, final StarQuery sq, final BindingSet currentBs, final Set<String> unCommonVar) {


        final QueryBindingSet currentSolutionBs = new QueryBindingSet();

        final Text row = key.getRow();
        final Text cq = key.getColumnQualifier();


        final String[] cqArray = cq.toString().split(DocIndexIteratorUtil.DOC_ID_INDEX_DELIM);

        boolean commonVarSet = false;

        //if common Var is constant there is no common variable to assign a value to
        if(sq.commonVarConstant()) {
            commonVarSet = true;
        }

        if (!commonVarSet && sq.isCommonVarURI()) {
            final RyaURI rURI = new RyaURI(row.toString());
            currentSolutionBs.addBinding(sq.getCommonVarName(),
                    RyaToRdfConversions.convertValue(rURI));
            commonVarSet = true;
        }

        for (final String s : sq.getUnCommonVars()) {

            final byte[] cqBytes = cqArray[sq.getVarPos().get(s)].getBytes(StandardCharsets.UTF_8);
            final int firstIndex = Bytes.indexOf(cqBytes, DELIM_BYTE);
            final int secondIndex = Bytes.lastIndexOf(cqBytes, DELIM_BYTE);
            final int typeIndex = Bytes.indexOf(cqBytes, TYPE_DELIM_BYTE);
            final String tripleComponent = new String(Arrays.copyOfRange(cqBytes, firstIndex + 1, secondIndex), StandardCharsets.UTF_8);
            final byte[] cqContent = Arrays.copyOfRange(cqBytes, secondIndex + 1, typeIndex);
            final byte[] objType = Arrays.copyOfRange(cqBytes, typeIndex, cqBytes.length);

            if (tripleComponent.equals("object")) {
                final byte[] object = Bytes.concat(cqContent, objType);
                org.eclipse.rdf4j.model.Value v = null;
                try {
                    v = RyaToRdfConversions.convertValue(RyaContext.getInstance().deserialize(
                            object));
                } catch (final RyaTypeResolverException e) {
                    e.printStackTrace();
                }
                currentSolutionBs.addBinding(s, v);

            } else if (tripleComponent.equals("subject")) {
                if (!commonVarSet) {
                    final byte[] object = Bytes.concat(row.getBytes(), objType);
                    org.eclipse.rdf4j.model.Value v = null;
                    try {
                        v = RyaToRdfConversions.convertValue(RyaContext.getInstance().deserialize(
                                object));
                    } catch (final RyaTypeResolverException e) {
                        e.printStackTrace();
                    }
                    currentSolutionBs.addBinding(sq.getCommonVarName(), v);
                    commonVarSet = true;
                }
                final RyaURI rURI = new RyaURI(new String(cqContent, StandardCharsets.UTF_8));
                currentSolutionBs.addBinding(s, RyaToRdfConversions.convertValue(rURI));
            } else {
                throw new IllegalArgumentException("Invalid row.");
            }
        }
        for (final String s : unCommonVar) {
            currentSolutionBs.addBinding(s, currentBs.getValue(s));
        }
        return currentSolutionBs;
    }

    private BatchScanner runQuery(final StarQuery query, Collection<Range> ranges) throws QueryEvaluationException {

        try {
            if (ranges.size() == 0) {
                final String rangeText = query.getCommonVarValue();
                Range r;
                if (rangeText != null) {
                    r = new Range(new Text(query.getCommonVarValue()));
                } else {
                    r = new Range();
                }
                ranges = Collections.singleton(r);
            }

            final Connector accCon = ConfigUtils.getConnector(conf);
            final IteratorSetting is = new IteratorSetting(30, "fii", DocumentIndexIntersectingIterator.class);

            DocumentIndexIntersectingIterator.setColumnFamilies(is, query.getColumnCond());

            if (query.hasContext()) {
                DocumentIndexIntersectingIterator.setContext(is, query.getContextURI());
            }

            final Authorizations auths;
            final String authsStr = conf.get(ConfigUtils.CLOUDBASE_AUTHS);
            if(authsStr == null || authsStr.isEmpty()) {
                auths = new Authorizations();
            } else {
                auths = new Authorizations(authsStr);
            }

            bs = accCon.createBatchScanner(EntityCentricIndex.getTableName(conf), auths, 15);
            bs.addScanIterator(is);
            bs.setRanges(ranges);

            return bs;

        } catch (TableNotFoundException | AccumuloException | AccumuloSecurityException e) {
            throw new QueryEvaluationException(e);
        }
    }

    @Override
    public void close() throws IOException {
        //TODO generate an exception when BS passed in -- scanner closed
//        if (bs != null) {
//            bs.close();
//        }
    }

}